Add realtime metrics for connections, messages, and bandwidth in project usage

2026-05-26 13:51:13 +00:00 · 2026-03-03 18:48:37 +05:30
parent 598c71fb11
commit 7644e0fe48
5 changed files with 355 additions and 3 deletions
@@ -72,6 +72,10 @@ Http::get('/v1/project/usage')
                METRIC_DATABASES_OPERATIONS_READS,
                METRIC_DATABASES_OPERATIONS_WRITES,
                METRIC_FILES_IMAGES_TRANSFORMED,
+                METRIC_REALTIME_CONNECTIONS,
+                METRIC_REALTIME_CONNECTIONS_MESSAGES_SENT,
+                METRIC_REALTIME_INBOUND,
+                METRIC_REALTIME_OUTBOUND,
            ],
            'period' => [
                METRIC_NETWORK_REQUESTS,
@@ -85,6 +89,10 @@ Http::get('/v1/project/usage')
                METRIC_DATABASES_OPERATIONS_READS,
                METRIC_DATABASES_OPERATIONS_WRITES,
                METRIC_FILES_IMAGES_TRANSFORMED,
+                METRIC_REALTIME_CONNECTIONS,
+                METRIC_REALTIME_CONNECTIONS_MESSAGES_SENT,
+                METRIC_REALTIME_INBOUND,
+                METRIC_REALTIME_OUTBOUND,
            ]
        ];

@@ -347,6 +355,26 @@ Http::get('/v1/project/usage')
            ];
        }

+        // Realtime bandwidth = realtime.inbound + realtime.outbound (per bucket)
+        $realtimeProjectBandwidth = [];
+        foreach ($usage[METRIC_REALTIME_INBOUND] as $item) {
+            $realtimeProjectBandwidth[$item['date']] ??= 0;
+            $realtimeProjectBandwidth[$item['date']] += $item['value'];
+        }
+
+        foreach ($usage[METRIC_REALTIME_OUTBOUND] as $item) {
+            $realtimeProjectBandwidth[$item['date']] ??= 0;
+            $realtimeProjectBandwidth[$item['date']] += $item['value'];
+        }
+
+        $realtimeBandwidth = [];
+        foreach ($realtimeProjectBandwidth as $date => $value) {
+            $realtimeBandwidth[] = [
+                'date' => $date,
+                'value' => $value
+            ];
+        }
+
        $response->dynamic(new Document([
            'requests' => ($usage[METRIC_NETWORK_REQUESTS]),
            'network' => $network,
@@ -367,10 +395,16 @@ Http::get('/v1/project/usage')
            'deploymentsStorageTotal' => $total[METRIC_DEPLOYMENTS_STORAGE],
            'databasesReadsTotal' => $total[METRIC_DATABASES_OPERATIONS_READS],
            'databasesWritesTotal' => $total[METRIC_DATABASES_OPERATIONS_WRITES],
+            'realtimeConnectionsTotal' => $total[METRIC_REALTIME_CONNECTIONS],
+            'realtimeMessagesTotal' => $total[METRIC_REALTIME_CONNECTIONS_MESSAGES_SENT],
+            'realtimeBandwidthTotal' => ($total[METRIC_REALTIME_INBOUND] ?? 0) + ($total[METRIC_REALTIME_OUTBOUND] ?? 0),
            'executionsBreakdown' => $executionsBreakdown,
            'bucketsBreakdown' => $bucketsBreakdown,
            'databasesReads' => $usage[METRIC_DATABASES_OPERATIONS_READS],
            'databasesWrites' => $usage[METRIC_DATABASES_OPERATIONS_WRITES],
+            'realtimeConnections' => $usage[METRIC_REALTIME_CONNECTIONS],
+            'realtimeMessages' => $usage[METRIC_REALTIME_CONNECTIONS_MESSAGES_SENT],
+            'realtimeBandwidth' => $realtimeBandwidth,
            'databasesStorageBreakdown' => $databasesStorageBreakdown,
            'executionsMbSecondsBreakdown' => $executionsMbSecondsBreakdown,
            'buildsMbSecondsBreakdown' => $buildsMbSecondsBreakdown,
@@ -361,6 +361,12 @@ const METRIC_AVATARS_SCREENSHOTS_GENERATED = 'avatars.screenshotsGenerated';
 const METRIC_FUNCTIONS_RUNTIME = 'functions.runtimes.{runtime}';
 const METRIC_SITES_FRAMEWORK = 'sites.frameworks.{framework}';

+// Realtime metrics
+const METRIC_REALTIME_CONNECTIONS = 'realtime.connections';
+const METRIC_REALTIME_CONNECTIONS_MESSAGES_SENT = 'realtime.messages.sent';
+const METRIC_REALTIME_INBOUND = 'realtime.inbound';
+const METRIC_REALTIME_OUTBOUND = 'realtime.outbound';
+
 // Resource types
 const RESOURCE_TYPE_PROJECTS = 'projects';
 const RESOURCE_TYPE_FUNCTIONS = 'functions';
@@ -1,5 +1,6 @@
 <?php

+use Appwrite\Event\StatsUsage;
 use Appwrite\Extend\Exception;
 use Appwrite\Extend\Exception as AppwriteException;
 use Appwrite\Messaging\Adapter\Realtime;
@@ -37,6 +38,7 @@ use Utopia\DSN\DSN;
 use Utopia\Http\Http;
 use Utopia\Logger\Log;
 use Utopia\Pools\Group;
+use Utopia\Queue\Broker\Pool as BrokerPool;
 use Utopia\Registry\Registry;
 use Utopia\System\System;
 use Utopia\Telemetry\Adapter\None as NoTelemetry;
@@ -224,6 +226,31 @@ if (!function_exists('getTelemetry')) {
    }
 }

+if (!function_exists('queueForStatsUsage')) {
+    function getQueueForStatsUsageForProject(Document $project): StatsUsage
+    {
+        $ctx = Coroutine::getContext();
+
+        if (!isset($ctx['queueForStatsUsage'])) {
+            $ctx['queueForStatsUsage'] = [];
+        }
+
+        if (isset($ctx['queueForStatsUsage'][$project->getSequence()])) {
+            return $ctx['queueForStatsUsage'][$project->getSequence()];
+        }
+
+        global $register;
+
+        /** @var Group $pools */
+        $pools = $register->get('pools');
+
+        $queue = new StatsUsage(new BrokerPool(publisher: $pools->get('publisher')));
+        $queue->setProject($project);
+
+        return $ctx['queueForStatsUsage'][$project->getSequence()] = $queue;
+    }
+}
+
 $realtime = getRealtime();

 /**
@@ -545,20 +572,59 @@ $server->onWorkerStart(function (int $workerId) use ($server, $register, $stats,
                }

                $total = 0;
+                $outboundBytes = 0;
+
                foreach ($groups as $group) {
                    $data = $event['data'];
                    $data['subscriptions'] = $group['subscriptions'];

-                    $server->send($group['ids'], json_encode([
+                    $payloadJson = json_encode([
                        'type' => 'event',
                        'data' => $data
-                    ]));
-                    $total += count($group['ids']);
+                    ]);
+
+                    $server->send($group['ids'], $payloadJson);
+
+                    $count = count($group['ids']);
+                    $total += $count;
+                    $outboundBytes += strlen($payloadJson) * $count;
                }

                if ($total > 0) {
                    $register->get('telemetry.messageSentCounter')->add($total);
                    $stats->incr($event['project'], 'messages', $total);
+
+                    $projectId = $event['project'] ?? null;
+
+                    if (!empty($projectId)) {
+                        try {
+                            $consoleDB = getConsoleDB();
+                            /** @var Document $project */
+                            $project = $consoleDB->getAuthorization()->skip(
+                                fn () => $consoleDB->getDocument('projects', $projectId)
+                            );
+
+                            if (!$project->isEmpty()) {
+                                $queueForStatsUsage = getQueueForStatsUsageForProject($project);
+
+                                $queueForStatsUsage->addMetric(
+                                    METRIC_REALTIME_CONNECTIONS_MESSAGES_SENT,
+                                    $total
+                                );
+
+                                if ($outboundBytes > 0) {
+                                    $queueForStatsUsage->addMetric(
+                                        METRIC_REALTIME_OUTBOUND,
+                                        $outboundBytes
+                                    );
+                                }
+
+                                $queueForStatsUsage->trigger();
+                            }
+                        } catch (Throwable $th) {
+                            logError($th, 'realtimeUsageOutbound', tags: ['projectId' => $projectId]);
+                        }
+                    }
                }
            });
        } catch (Throwable $th) {
@@ -707,6 +773,16 @@ $server->onOpen(function (int $connection, SwooleRequest $request) use ($server,
        ]);
        $stats->incr($project->getId(), 'connections');
        $stats->incr($project->getId(), 'connectionsTotal');
+
+        try {
+            $queueForStatsUsage = getQueueForStatsUsageForProject($project);
+            $queueForStatsUsage
+                ->addMetric(METRIC_REALTIME_CONNECTIONS, 1)
+                ->trigger();
+        } catch (\Throwable $th) {
+            logError($th, 'realtimeUsageConnections', project: $project);
+        }
+
    } catch (Throwable $th) {
        logError($th, 'realtime', project: $project, user: $logUser, authorization: $authorization);

@@ -748,6 +824,7 @@ $server->onMessage(function (int $connection, string $message) use ($server, $re
    $authorization = null;

    try {
+        $rawSize = \strlen($message);
        $response = new Response(new SwooleResponse());
        $projectId = $realtime->connections[$connection]['projectId'] ?? null;

@@ -786,6 +863,18 @@ $server->onMessage(function (int $connection, string $message) use ($server, $re
            throw new Exception(Exception::REALTIME_TOO_MANY_MESSAGES, 'Too many messages.');
        }

+        // Record realtime inbound bytes for this project
+        if ($project !== null && !$project->isEmpty()) {
+            try {
+                $queueForStatsUsage = getQueueForStatsUsageForProject($project);
+                $queueForStatsUsage
+                    ->addMetric(METRIC_REALTIME_INBOUND, $rawSize)
+                    ->trigger();
+            } catch (Throwable $th) {
+                logError($th, 'realtimeUsageInbound', project: $project);
+            }
+        }
+
        $message = json_decode($message, true);

        if (is_null($message) || (!array_key_exists('type', $message) && !array_key_exists('data', $message))) {
@@ -905,6 +994,18 @@ $server->onClose(function (int $connection) use ($realtime, $stats, $register) {
    if (array_key_exists($connection, $realtime->connections)) {
        $stats->decr($realtime->connections[$connection]['projectId'], 'connectionsTotal');
        $register->get('telemetry.connectionCounter')->add(-1);
+
+        $projectId = $realtime->connections[$connection]['projectId'];
+
+        $consoleDB = getConsoleDB();
+        $project = $consoleDB->getAuthorization()->skip(
+            fn () => $consoleDB->getDocument('projects', $projectId)
+        );
+
+        if (!$project->isEmpty()) {
+            $queue = getQueueForStatsUsageForProject($project);
+            $queue->addMetric(METRIC_REALTIME_CONNECTIONS, -1)->trigger();
+        }
    }
    $realtime->unsubscribe($connection);

@@ -100,6 +100,24 @@ class UsageProject extends Model
                'default' => 0,
                'example' => 0,
            ])
+            ->addRule('realtimeConnectionsTotal', [
+                'type' => self::TYPE_INTEGER,
+                'description' => 'Current aggregated number of open Realtime connections.',
+                'default' => 0,
+                'example' => 0,
+            ])
+            ->addRule('realtimeMessagesTotal', [
+                'type' => self::TYPE_INTEGER,
+                'description' => 'Total number of Realtime messages sent to clients.',
+                'default' => 0,
+                'example' => 0,
+            ])
+            ->addRule('realtimeBandwidthTotal', [
+                'type' => self::TYPE_INTEGER,
+                'description' => 'Total consumed Realtime bandwidth (in bytes).',
+                'default' => 0,
+                'example' => 0,
+            ])
            ->addRule('requests', [
                'type' => Response::MODEL_METRIC,
                'description' => 'Aggregated  number of requests per period.',
@@ -114,6 +132,27 @@ class UsageProject extends Model
                'example' => [],
                'array' => true
            ])
+            ->addRule('realtimeConnections', [
+                'type' => Response::MODEL_METRIC,
+                'description' => 'Aggregated number of open Realtime connections per period.',
+                'default' => [],
+                'example' => [],
+                'array' => true
+            ])
+            ->addRule('realtimeMessages', [
+                'type' => Response::MODEL_METRIC,
+                'description' => 'Aggregated number of Realtime messages sent to clients per period.',
+                'default' => [],
+                'example' => [],
+                'array' => true
+            ])
+            ->addRule('realtimeBandwidth', [
+                'type' => Response::MODEL_METRIC,
+                'description' => 'Aggregated consumed Realtime bandwidth (in bytes) per period.',
+                'default' => [],
+                'example' => [],
+                'array' => true
+            ])
            ->addRule('users', [
                'type' => Response::MODEL_METRIC,
                'description' => 'Aggregated number of users per period.',
@@ -11,6 +11,7 @@ use Tests\E2E\Scopes\ProjectCustom;
 use Tests\E2E\Scopes\Scope;
 use Tests\E2E\Scopes\SideServer;
 use Tests\E2E\Services\Functions\FunctionsBase;
+use Tests\E2E\Services\Realtime\RealtimeBase;
 use Tests\E2E\Services\Sites\SitesBase;
 use Utopia\Database\Helpers\ID;
 use Utopia\Database\Helpers\Permission;
@@ -23,6 +24,7 @@ class UsageTest extends Scope
    use ProjectCustom;
    use SideServer;
    use FunctionsBase;
+    use RealtimeBase;
    use SitesBase {
        FunctionsBase::createDeployment insteadof SitesBase;
        FunctionsBase::setupDeployment insteadof SitesBase;
@@ -1408,6 +1410,176 @@ class UsageTest extends Scope
        });
    }

+    public function testRealtimeUsageMetrics(): void
+    {
+        $user = $this->getUser();
+        $session = $user['session'] ?? '';
+        $projectId = $this->getProject()['$id'];
+
+        // Baseline realtime usage before opening a new connection
+        $baseline = $this->client->call(
+            Client::METHOD_GET,
+            '/project/usage',
+            $this->getConsoleHeaders(),
+            [
+                'period' => '1h',
+                'startDate' => self::getToday(),
+                'endDate' => self::getTomorrow(),
+            ]
+        );
+
+        $connectionsBefore = $baseline['body']['realtimeConnectionsTotal'] ?? 0;
+        $messagesBefore = $baseline['body']['realtimeMessagesTotal'] ?? 0;
+
+        $connectionCount = 3;
+        $clients = [];
+
+        for ($i = 0; $i < $connectionCount; $i++) {
+            $client = $this->getWebsocket(['documents'], [
+                'origin' => 'http://localhost',
+                'cookie' => 'a_session_' . $projectId . '=' . $session,
+            ], null);
+
+            $connected = json_decode($client->receive(), true);
+            $this->assertEquals('connected', $connected['type']);
+
+            $clients[] = $client;
+        }
+
+        try {
+            $database = $this->client->call(Client::METHOD_POST, '/databases', [
+                'content-type' => 'application/json',
+                'x-appwrite-project' => $projectId,
+                'x-appwrite-key' => $this->getProject()['apiKey'],
+            ], [
+                'databaseId' => ID::unique(),
+                'name' => 'Realtime Usage DB',
+            ]);
+
+            $databaseId = $database['body']['$id'];
+
+            $collection = $this->client->call(Client::METHOD_POST, '/databases/' . $databaseId . '/collections', [
+                'content-type' => 'application/json',
+                'x-appwrite-project' => $projectId,
+                'x-appwrite-key' => $this->getProject()['apiKey'],
+            ], [
+                'collectionId' => ID::unique(),
+                'name' => 'Realtime Usage Collection',
+                'permissions' => [
+                    Permission::create(Role::user($user['$id'])),
+                ],
+                'documentSecurity' => true,
+            ]);
+
+            $collectionId = $collection['body']['$id'];
+
+            $attribute = $this->client->call(Client::METHOD_POST, '/databases/' . $databaseId . '/collections/' . $collectionId . '/attributes/string', [
+                'content-type' => 'application/json',
+                'x-appwrite-project' => $projectId,
+                'x-appwrite-key' => $this->getProject()['apiKey'],
+            ], [
+                'key' => 'name',
+                'size' => 256,
+                'required' => true,
+            ]);
+
+            $this->assertEquals(202, $attribute['headers']['status-code']);
+
+            $this->assertEventually(function () use ($databaseId, $collectionId, $projectId) {
+                $response = $this->client->call(Client::METHOD_GET, '/databases/' . $databaseId . '/collections/' . $collectionId . '/attributes/name', [
+                    'content-type' => 'application/json',
+                    'x-appwrite-project' => $projectId,
+                    'x-appwrite-key' => $this->getProject()['apiKey'],
+                ]);
+                $this->assertEquals('available', $response['body']['status']);
+            }, 30000, 250);
+
+            $document = $this->client->call(Client::METHOD_POST, '/databases/' . $databaseId . '/collections/' . $collectionId . '/documents', array_merge([
+                'content-type' => 'application/json',
+                'x-appwrite-project' => $projectId,
+            ], $this->getHeaders()), [
+                'documentId' => ID::unique(),
+                'data' => [
+                    'name' => 'Realtime Usage Doc',
+                ],
+                'permissions' => [
+                    Permission::read(Role::any()),
+                    Permission::update(Role::any()),
+                    Permission::delete(Role::any()),
+                ],
+            ]);
+
+            $this->assertEquals(201, $document['headers']['status-code']);
+
+            $event = json_decode($clients[0]->receive(), true);
+            $this->assertEquals('event', $event['type']);
+
+            // After creating a document we expect all connections to receive an event
+            $this->assertEventually(function () use ($connectionsBefore, $messagesBefore, $connectionCount) {
+                $response = $this->client->call(
+                    Client::METHOD_GET,
+                    '/project/usage',
+                    $this->getConsoleHeaders(),
+                    [
+                        'period' => '1h',
+                        'startDate' => self::getToday(),
+                        'endDate' => self::getTomorrow(),
+                    ]
+                );
+
+                $this->assertEquals(200, $response['headers']['status-code']);
+
+                $this->assertArrayHasKey('realtimeConnectionsTotal', $response['body']);
+                $this->assertArrayHasKey('realtimeMessagesTotal', $response['body']);
+                $this->assertArrayHasKey('realtimeBandwidthTotal', $response['body']);
+                $this->assertArrayHasKey('realtimeConnections', $response['body']);
+                $this->assertArrayHasKey('realtimeMessages', $response['body']);
+                $this->assertArrayHasKey('realtimeBandwidth', $response['body']);
+
+                // We expect exactly $connectionCount additional open connections and $connectionCount additional message deliveries
+                $this->assertEquals($connectionsBefore + $connectionCount, $response['body']['realtimeConnectionsTotal']);
+                $this->assertEquals($messagesBefore + $connectionCount, $response['body']['realtimeMessagesTotal']);
+                $this->assertGreaterThan(0, $response['body']['realtimeBandwidthTotal']);
+
+                $this->validateDates($response['body']['realtimeConnections']);
+                $this->validateDates($response['body']['realtimeMessages']);
+                $this->validateDates($response['body']['realtimeBandwidth']);
+            }, 60000, 2000);
+
+            // Now close a single connection and ensure the counters reflect it
+            $clients[0]->close();
+
+            $this->assertEventually(function () use ($connectionsBefore, $messagesBefore, $connectionCount) {
+                $response = $this->client->call(
+                    Client::METHOD_GET,
+                    '/project/usage',
+                    $this->getConsoleHeaders(),
+                    [
+                        'period' => '1h',
+                        'startDate' => self::getToday(),
+                        'endDate' => self::getTomorrow(),
+                    ]
+                );
+
+                $this->assertEquals(200, $response['headers']['status-code']);
+
+                $this->assertArrayHasKey('realtimeConnectionsTotal', $response['body']);
+                $this->assertArrayHasKey('realtimeMessagesTotal', $response['body']);
+                $this->assertArrayHasKey('realtimeBandwidthTotal', $response['body']);
+
+                // One of the connections is closed, so we expect one less open connection.
+                // Messages and bandwidth are cumulative and should not decrease.
+                $this->assertEquals($connectionsBefore + $connectionCount - 1, $response['body']['realtimeConnectionsTotal']);
+                $this->assertEquals($messagesBefore + $connectionCount, $response['body']['realtimeMessagesTotal']);
+                $this->assertGreaterThan(0, $response['body']['realtimeBandwidthTotal']);
+            }, 60000, 2000);
+        } finally {
+            foreach ($clients as $client) {
+                $client->close();
+            }
+        }
+    }
+
    public function tearDown(): void
    {
        $this->projectId = '';