From 7644e0fe48eb0a2dbba50d6db71d4c41a8560d50 Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Tue, 3 Mar 2026 18:48:37 +0530 Subject: [PATCH] Add realtime metrics for connections, messages, and bandwidth in project usage --- app/controllers/api/project.php | 34 ++++ app/init/constants.php | 6 + app/realtime.php | 107 ++++++++++- .../Utopia/Response/Model/UsageProject.php | 39 ++++ tests/e2e/General/UsageTest.php | 172 ++++++++++++++++++ 5 files changed, 355 insertions(+), 3 deletions(-) diff --git a/app/controllers/api/project.php b/app/controllers/api/project.php index d24519e3fb..1fd33a5db0 100644 --- a/app/controllers/api/project.php +++ b/app/controllers/api/project.php @@ -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, diff --git a/app/init/constants.php b/app/init/constants.php index c6a5bc853e..cd77b05746 100644 --- a/app/init/constants.php +++ b/app/init/constants.php @@ -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'; diff --git a/app/realtime.php b/app/realtime.php index 7ec24d03c8..8a066bcbab 100644 --- a/app/realtime.php +++ b/app/realtime.php @@ -1,5 +1,6 @@ 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); diff --git a/src/Appwrite/Utopia/Response/Model/UsageProject.php b/src/Appwrite/Utopia/Response/Model/UsageProject.php index ee644aa845..c219a35f29 100644 --- a/src/Appwrite/Utopia/Response/Model/UsageProject.php +++ b/src/Appwrite/Utopia/Response/Model/UsageProject.php @@ -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.', diff --git a/tests/e2e/General/UsageTest.php b/tests/e2e/General/UsageTest.php index e7b5b1d844..29229ae2bd 100644 --- a/tests/e2e/General/UsageTest.php +++ b/tests/e2e/General/UsageTest.php @@ -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 = '';