From e0fec8f550b098d77192758419bef031cf62ceb1 Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Wed, 22 Apr 2026 15:42:17 +0530 Subject: [PATCH 01/13] updated --- app/realtime.php | 68 +++++++++++++++++++++++ src/Appwrite/SDK/Specification/Format.php | 7 +-- 2 files changed, 71 insertions(+), 4 deletions(-) diff --git a/app/realtime.php b/app/realtime.php index 3461ca83e5..13ebb609a8 100644 --- a/app/realtime.php +++ b/app/realtime.php @@ -38,6 +38,7 @@ use Utopia\DSN\DSN; use Utopia\Logger\Log; use Utopia\Pools\Group; use Utopia\Registry\Registry; +use Utopia\Span\Span; use Utopia\System\System; use Utopia\Telemetry\Adapter\None as NoTelemetry; use Utopia\WebSocket\Adapter; @@ -326,11 +327,32 @@ if (!function_exists('logError')) { } } +if (!function_exists('traceOperationalEvent')) { + function traceOperationalEvent(string $action, string $message, array $context = []): void + { + Span::init($action); + Span::add('realtime.action', $action); + Span::add('realtime.message', $message); + Span::add('realtime.timestamp', DateTime::formatTz(DateTime::now())); + + foreach ($context as $key => $value) { + if (\is_scalar($value) || $value === null) { + Span::add('realtime.' . $key, ($value === null || $value === '') ? 'n/a' : $value); + } + } + + Span::current()?->finish(); + } +} + $server->error(logError(...)); $server->onStart(function () use ($stats, $containerId, &$statsDocument) { sleep(5); // wait for the initial database schema to be ready Console::success('Server started successfully'); + traceOperationalEvent('realtime.server.started', 'Realtime server started', [ + 'container' => $containerId, + ]); /** * Create document for this worker to share stats across Containers. @@ -394,6 +416,9 @@ $server->onStart(function () use ($stats, $containerId, &$statsDocument) { $server->onWorkerStart(function (int $workerId) use ($server, $register, $stats, $realtime) { Console::success('Worker ' . $workerId . ' started successfully'); + traceOperationalEvent('realtime.worker.started', 'Realtime worker started', [ + 'workerId' => $workerId, + ]); $telemetry = getTelemetry($workerId); $realtimeDelayBuckets = [100, 250, 500, 750, 1000, 1500, 2000, 3000, 5000, 7500, 10000, 15000, 30000]; @@ -526,6 +551,9 @@ $server->onWorkerStart(function (int $workerId) use ($server, $register, $stats, if ($pubsub->ping(true)) { $attempts = 0; Console::success('Pub/sub connection established (worker: ' . $workerId . ')'); + traceOperationalEvent('realtime.pubsub.connected', 'Realtime pubsub connected', [ + 'workerId' => $workerId, + ]); } else { Console::error('Pub/sub failed (worker: ' . $workerId . ')'); } @@ -858,6 +886,14 @@ $server->onOpen(function (int $connection, SwooleRequest $request) use ($server, $server->send([$connection], $connectedPayloadJson); $updateStats($project->getId(), $project->getAttribute('teamId'), $connectedPayloadJson); + traceOperationalEvent('realtime.connection.opened', 'Realtime connection established', [ + 'connectionId' => $connection, + 'projectId' => $project->getId(), + 'teamId' => $project->getAttribute('teamId'), + 'userId' => $logUser?->getId() ?: null, + 'channelCount' => \count($names), + 'subscriptionCount' => \count($mapping), + ]); } catch (Throwable $th) { @@ -1053,6 +1089,12 @@ $server->onMessage(function (int $connection, string $message) use ($server, $re ]); $server->send([$connection], $authResponsePayloadJson); + traceOperationalEvent('realtime.authentication.succeeded', 'Realtime authentication succeeded', [ + 'connectionId' => $connection, + 'projectId' => $projectId, + 'userId' => $user['$id'] ?? null, + 'subscriptionDelta' => $subscriptionDelta, + ]); if ($project !== null && !$project->isEmpty()) { $authOutboundBytes = \strlen($authResponsePayloadJson); @@ -1149,6 +1191,12 @@ $server->onMessage(function (int $connection, string $message) use ($server, $re ]); $server->send([$connection], $responsePayload); + traceOperationalEvent('realtime.subscribe.updated', 'Realtime subscriptions updated', [ + 'connectionId' => $connection, + 'projectId' => $projectId, + 'subscriptionCount' => \count($parsedPayloads), + 'subscriptionDelta' => $subscriptionDelta, + ]); if ($project !== null && !$project->isEmpty()) { $subscribeOutboundBytes = \strlen($responsePayload); @@ -1208,6 +1256,13 @@ $server->onMessage(function (int $connection, string $message) use ($server, $re ]); $server->send([$connection], $unsubscribeResponsePayload); + traceOperationalEvent('realtime.unsubscribe.updated', 'Realtime subscriptions removed', [ + 'connectionId' => $connection, + 'projectId' => $projectId, + 'requestedCount' => \count($validatedIds), + 'removedCount' => \count(\array_filter($unsubscribeResults, fn (array $item) => $item['removed'] ?? false)), + 'subscriptionDelta' => $subscriptionDelta, + ]); if ($project !== null && !$project->isEmpty()) { $unsubscribeOutboundBytes = \strlen($unsubscribeResponsePayload); @@ -1255,6 +1310,14 @@ $server->onMessage(function (int $connection, string $message) use ($server, $re }); $server->onClose(function (int $connection) use ($realtime, $stats, $register) { + $projectId = null; + $userId = null; + + if (array_key_exists($connection, $realtime->connections)) { + $projectId = $realtime->connections[$connection]['projectId'] ?? null; + $userId = $realtime->connections[$connection]['userId'] ?? null; + } + try { if (array_key_exists($connection, $realtime->connections)) { $stats->decr($realtime->connections[$connection]['projectId'], 'connectionsTotal'); @@ -1278,6 +1341,11 @@ $server->onClose(function (int $connection) use ($realtime, $stats, $register) { } $realtime->unsubscribe($connection); + traceOperationalEvent('realtime.connection.closed', 'Realtime connection closed', [ + 'connectionId' => $connection, + 'projectId' => $projectId, + 'userId' => $userId, + ]); Console::info('Connection close: ' . $connection); }); diff --git a/src/Appwrite/SDK/Specification/Format.php b/src/Appwrite/SDK/Specification/Format.php index 08f960b2a7..d48a4b8f3f 100644 --- a/src/Appwrite/SDK/Specification/Format.php +++ b/src/Appwrite/SDK/Specification/Format.php @@ -742,6 +742,7 @@ abstract class Format } break; case 'project': + case 'projects': switch ($method) { case 'getUsage': switch ($param) { @@ -749,10 +750,6 @@ abstract class Format return 'ProjectUsageRange'; } break; - } - break; - case 'projects': - switch ($method) { case 'getEmailTemplate': case 'updateEmailTemplate': case 'deleteEmailTemplate': @@ -770,7 +767,9 @@ abstract class Format } break; case 'createSmtpTest': + case 'createSMTPTest': case 'updateSmtp': + case 'updateSMTP': switch ($param) { case 'secure': return 'SMTPSecure'; From ca1cf1982f5a8664a39b8a9752b046c8578aee94 Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Wed, 22 Apr 2026 16:44:20 +0530 Subject: [PATCH 02/13] added wide events inside the structured one logging per message instead of a discrete logs --- app/realtime.php | 105 +++++++++++++++++++---------------------------- 1 file changed, 43 insertions(+), 62 deletions(-) diff --git a/app/realtime.php b/app/realtime.php index 13ebb609a8..835ddf932f 100644 --- a/app/realtime.php +++ b/app/realtime.php @@ -327,32 +327,11 @@ if (!function_exists('logError')) { } } -if (!function_exists('traceOperationalEvent')) { - function traceOperationalEvent(string $action, string $message, array $context = []): void - { - Span::init($action); - Span::add('realtime.action', $action); - Span::add('realtime.message', $message); - Span::add('realtime.timestamp', DateTime::formatTz(DateTime::now())); - - foreach ($context as $key => $value) { - if (\is_scalar($value) || $value === null) { - Span::add('realtime.' . $key, ($value === null || $value === '') ? 'n/a' : $value); - } - } - - Span::current()?->finish(); - } -} - $server->error(logError(...)); $server->onStart(function () use ($stats, $containerId, &$statsDocument) { sleep(5); // wait for the initial database schema to be ready Console::success('Server started successfully'); - traceOperationalEvent('realtime.server.started', 'Realtime server started', [ - 'container' => $containerId, - ]); /** * Create document for this worker to share stats across Containers. @@ -416,9 +395,6 @@ $server->onStart(function () use ($stats, $containerId, &$statsDocument) { $server->onWorkerStart(function (int $workerId) use ($server, $register, $stats, $realtime) { Console::success('Worker ' . $workerId . ' started successfully'); - traceOperationalEvent('realtime.worker.started', 'Realtime worker started', [ - 'workerId' => $workerId, - ]); $telemetry = getTelemetry($workerId); $realtimeDelayBuckets = [100, 250, 500, 750, 1000, 1500, 2000, 3000, 5000, 7500, 10000, 15000, 30000]; @@ -551,9 +527,6 @@ $server->onWorkerStart(function (int $workerId) use ($server, $register, $stats, if ($pubsub->ping(true)) { $attempts = 0; Console::success('Pub/sub connection established (worker: ' . $workerId . ')'); - traceOperationalEvent('realtime.pubsub.connected', 'Realtime pubsub connected', [ - 'workerId' => $workerId, - ]); } else { Console::error('Pub/sub failed (worker: ' . $workerId . ')'); } @@ -886,14 +859,6 @@ $server->onOpen(function (int $connection, SwooleRequest $request) use ($server, $server->send([$connection], $connectedPayloadJson); $updateStats($project->getId(), $project->getAttribute('teamId'), $connectedPayloadJson); - traceOperationalEvent('realtime.connection.opened', 'Realtime connection established', [ - 'connectionId' => $connection, - 'projectId' => $project->getId(), - 'teamId' => $project->getAttribute('teamId'), - 'userId' => $logUser?->getId() ?: null, - 'channelCount' => \count($names), - 'subscriptionCount' => \count($mapping), - ]); } catch (Throwable $th) { @@ -935,10 +900,24 @@ $server->onOpen(function (int $connection, SwooleRequest $request) use ($server, $server->onMessage(function (int $connection, string $message) use ($server, $realtime, $containerId, $register) { $project = null; $authorization = null; + $projectId = $realtime->connections[$connection]['projectId'] ?? null; + $rawSize = \strlen($message); + $messageType = 'invalid'; + $subscriptionDelta = 0; + $subscriptionsRequested = 0; + $subscriptionsRemoved = 0; + $outboundBytes = 0; + $responseCode = 200; + $success = false; + + Span::init('realtime.message'); + Span::add('realtime.connection_id', $connection); + Span::add('realtime.project_id', $projectId ?: 'n/a'); + Span::add('realtime.inbound_bytes', $rawSize); + Span::add('realtime.container_id', $containerId); + try { - $rawSize = \strlen($message); $response = new Response(new SwooleResponse()); - $projectId = $realtime->connections[$connection]['projectId'] ?? null; // Get authorization from connection (stored during onOpen) $authorization = $realtime->connections[$connection]['authorization'] ?? null; @@ -983,6 +962,8 @@ $server->onMessage(function (int $connection, string $message) use ($server, $re } $message = json_decode($message, true); + $messageType = $message['type'] ?? 'invalid'; + Span::add('realtime.message_type', $messageType); if (is_null($message) || (!array_key_exists('type', $message) && !array_key_exists('data', $message))) { throw new Exception(Exception::REALTIME_MESSAGE_FORMAT_INVALID, 'Message format is not valid.'); @@ -1000,6 +981,7 @@ $server->onMessage(function (int $connection, string $message) use ($server, $re ]); $server->send([$connection], $pongPayloadJson); + $outboundBytes += \strlen($pongPayloadJson); if ($project !== null && !$project->isEmpty()) { $pongOutboundBytes = \strlen($pongPayloadJson); @@ -1089,12 +1071,8 @@ $server->onMessage(function (int $connection, string $message) use ($server, $re ]); $server->send([$connection], $authResponsePayloadJson); - traceOperationalEvent('realtime.authentication.succeeded', 'Realtime authentication succeeded', [ - 'connectionId' => $connection, - 'projectId' => $projectId, - 'userId' => $user['$id'] ?? null, - 'subscriptionDelta' => $subscriptionDelta, - ]); + $outboundBytes += \strlen($authResponsePayloadJson); + Span::add('realtime.user_id', $user['$id'] ?? 'n/a'); if ($project !== null && !$project->isEmpty()) { $authOutboundBytes = \strlen($authResponsePayloadJson); @@ -1171,6 +1149,7 @@ $server->onMessage(function (int $connection, string $message) use ($server, $re } $subscriptionsAfter = \count($realtime->getSubscriptionMetadata($connection)); $subscriptionDelta = $subscriptionsAfter - $subscriptionsBefore; + $subscriptionsRequested = \count($parsedPayloads); if ($subscriptionDelta !== 0) { $register->get('telemetry.workerSubscriptionCounter')->add($subscriptionDelta, $register->get('telemetry.workerAttributes')); } @@ -1191,12 +1170,7 @@ $server->onMessage(function (int $connection, string $message) use ($server, $re ]); $server->send([$connection], $responsePayload); - traceOperationalEvent('realtime.subscribe.updated', 'Realtime subscriptions updated', [ - 'connectionId' => $connection, - 'projectId' => $projectId, - 'subscriptionCount' => \count($parsedPayloads), - 'subscriptionDelta' => $subscriptionDelta, - ]); + $outboundBytes += \strlen($responsePayload); if ($project !== null && !$project->isEmpty()) { $subscribeOutboundBytes = \strlen($responsePayload); @@ -1242,6 +1216,8 @@ $server->onMessage(function (int $connection, string $message) use ($server, $re } $subscriptionsAfter = \count($realtime->getSubscriptionMetadata($connection)); $subscriptionDelta = $subscriptionsAfter - $subscriptionsBefore; + $subscriptionsRequested = \count($validatedIds); + $subscriptionsRemoved = \count(\array_filter($unsubscribeResults, fn (array $item) => $item['removed'] ?? false)); if ($subscriptionDelta !== 0) { $register->get('telemetry.workerSubscriptionCounter')->add($subscriptionDelta, $register->get('telemetry.workerAttributes')); } @@ -1256,13 +1232,7 @@ $server->onMessage(function (int $connection, string $message) use ($server, $re ]); $server->send([$connection], $unsubscribeResponsePayload); - traceOperationalEvent('realtime.unsubscribe.updated', 'Realtime subscriptions removed', [ - 'connectionId' => $connection, - 'projectId' => $projectId, - 'requestedCount' => \count($validatedIds), - 'removedCount' => \count(\array_filter($unsubscribeResults, fn (array $item) => $item['removed'] ?? false)), - 'subscriptionDelta' => $subscriptionDelta, - ]); + $outboundBytes += \strlen($unsubscribeResponsePayload); if ($project !== null && !$project->isEmpty()) { $unsubscribeOutboundBytes = \strlen($unsubscribeResponsePayload); @@ -1279,12 +1249,14 @@ $server->onMessage(function (int $connection, string $message) use ($server, $re default: throw new Exception(Exception::REALTIME_MESSAGE_FORMAT_INVALID, 'Message type is not valid.'); } + $success = true; } catch (Throwable $th) { logError($th, 'realtimeMessage', project: $project, authorization: $authorization); $code = $th->getCode(); if (!is_int($code)) { $code = 500; } + $responseCode = $code; $message = $th->getMessage(); @@ -1301,11 +1273,25 @@ $server->onMessage(function (int $connection, string $message) use ($server, $re ] ]; - $server->send([$connection], json_encode($response)); + $responsePayloadJson = json_encode($response); + $server->send([$connection], $responsePayloadJson); + $outboundBytes += \strlen($responsePayloadJson); if ($th->getCode() === 1008) { $server->close($connection, $th->getCode()); } + Span::error($th); + } finally { + Span::add('realtime.success', $success); + Span::add('realtime.response_code', $responseCode); + Span::add('realtime.subscription_delta', $subscriptionDelta); + Span::add('realtime.subscriptions_requested', $subscriptionsRequested); + Span::add('realtime.subscriptions_removed', $subscriptionsRemoved); + Span::add('realtime.outbound_bytes', $outboundBytes); + Span::add('realtime.project_id', $project?->getId() ?: $projectId ?: 'n/a'); + Span::add('realtime.user_id', $realtime->connections[$connection]['userId'] ?? 'n/a'); + Span::add('realtime.message_type', $messageType); + Span::current()?->finish(); } }); @@ -1341,11 +1327,6 @@ $server->onClose(function (int $connection) use ($realtime, $stats, $register) { } $realtime->unsubscribe($connection); - traceOperationalEvent('realtime.connection.closed', 'Realtime connection closed', [ - 'connectionId' => $connection, - 'projectId' => $projectId, - 'userId' => $userId, - ]); Console::info('Connection close: ' . $connection); }); From 57d777f80a21bdd22ae5da1e5fca39eef1ff2113 Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Wed, 22 Apr 2026 16:46:18 +0530 Subject: [PATCH 03/13] revert format --- src/Appwrite/SDK/Specification/Format.php | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/src/Appwrite/SDK/Specification/Format.php b/src/Appwrite/SDK/Specification/Format.php index 243517445e..30df5acf52 100644 --- a/src/Appwrite/SDK/Specification/Format.php +++ b/src/Appwrite/SDK/Specification/Format.php @@ -742,7 +742,6 @@ abstract class Format } break; case 'project': - case 'projects': switch ($method) { case 'getEmailTemplate': case 'updateEmailTemplate': @@ -759,6 +758,10 @@ abstract class Format return 'ProjectUsageRange'; } break; + } + break; + case 'projects': + switch ($method) { case 'getEmailTemplate': case 'updateEmailTemplate': switch ($param) { @@ -775,9 +778,7 @@ abstract class Format } break; case 'createSmtpTest': - case 'createSMTPTest': case 'updateSmtp': - case 'updateSMTP': switch ($param) { case 'secure': return 'SMTPSecure'; From b2ad7237abf61152b4c5b84c5b826d599b90823b Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Wed, 22 Apr 2026 16:52:54 +0530 Subject: [PATCH 04/13] Add detailed telemetry logging for realtime connection events - Introduced span logging for connection open and close events, capturing metrics such as inbound and outbound bytes, subscription counts, and response codes. - Enhanced error handling with logging of exceptions during connection lifecycle. - Updated the structure of the telemetry data to include project and user IDs for better traceability. --- app/realtime.php | 54 ++++++++++++++++++++++++++++++++++++++++++++---- 1 file changed, 50 insertions(+), 4 deletions(-) diff --git a/app/realtime.php b/app/realtime.php index 835ddf932f..8402dda1f0 100644 --- a/app/realtime.php +++ b/app/realtime.php @@ -702,11 +702,24 @@ $server->onOpen(function (int $connection, SwooleRequest $request) use ($server, $project = null; $logUser = null; $authorization = null; + $rawSize = $request->getSize(); + $channelCount = 0; + $subscriptionCount = 0; + $outboundBytes = 0; + $responseCode = 200; + $subscriptionMode = 'message'; + $success = false; + + Span::init('realtime.open'); + Span::add('realtime.connection_id', $connection); + Span::add('realtime.inbound_bytes', $rawSize); + Span::add('realtime.origin', $request->getOrigin() ?: 'n/a'); try { /** @var Document $project */ $project = $connectionContainer->get('project'); $authorization = $connectionContainer->get('authorization'); + Span::add('realtime.project_id', $project->getId() ?: 'n/a'); /* * Project Check @@ -751,8 +764,6 @@ $server->onOpen(function (int $connection, SwooleRequest $request) use ($server, throw new Exception(Exception::REALTIME_TOO_MANY_MESSAGES, 'Too many requests'); } - $rawSize = $request->getSize(); - triggerStats([ METRIC_REALTIME_INBOUND => $rawSize, ], $project->getId()); @@ -770,8 +781,10 @@ $server->onOpen(function (int $connection, SwooleRequest $request) use ($server, } $roles = $user->getRoles($authorization); + Span::add('realtime.user_id', $user->getId() ?: 'n/a'); $channels = Realtime::convertChannels($request->getQuery('channels', []), $user->getId()); + $channelCount = \count($channels); $updateStats = static function (string $projectId, ?string $teamId, string $payloadJson) use ($register, $stats): void { $register->get('telemetry.connectionCounter')->add(1); @@ -809,11 +822,15 @@ $server->onOpen(function (int $connection, SwooleRequest $request) use ($server, $realtime->subscribe($project->getId(), $connection, '', $roles, [], [], $user->getId()); $realtime->connections[$connection]['authorization'] = $authorization; $server->send([$connection], $connectedPayloadJson); + $outboundBytes += \strlen($connectedPayloadJson); $updateStats($project->getId(), $project->getAttribute('teamId'), $connectedPayloadJson); + $subscriptionMode = 'message'; + $success = true; return; } $names = array_keys($channels); + $subscriptionMode = 'url'; try { $subscriptions = Realtime::constructSubscriptions( @@ -840,6 +857,7 @@ $server->onOpen(function (int $connection, SwooleRequest $request) use ($server, $mapping[$index] = $subscriptionId; } + $subscriptionCount = \count($subscriptions); if (!empty($subscriptions)) { $register->get('telemetry.workerSubscriptionCounter')->add(\count($subscriptions), $register->get('telemetry.workerAttributes')); } @@ -858,8 +876,9 @@ $server->onOpen(function (int $connection, SwooleRequest $request) use ($server, ]); $server->send([$connection], $connectedPayloadJson); + $outboundBytes += \strlen($connectedPayloadJson); $updateStats($project->getId(), $project->getAttribute('teamId'), $connectedPayloadJson); - + $success = true; } catch (Throwable $th) { logError($th, 'realtime', project: $project, user: $logUser, authorization: $authorization); @@ -869,6 +888,7 @@ $server->onOpen(function (int $connection, SwooleRequest $request) use ($server, if (!\is_int($code)) { $code = 500; } + $responseCode = $code; $message = $th->getMessage(); @@ -886,7 +906,9 @@ $server->onOpen(function (int $connection, SwooleRequest $request) use ($server, ] ]; - $server->send([$connection], json_encode($response)); + $responsePayloadJson = json_encode($response); + $server->send([$connection], $responsePayloadJson); + $outboundBytes += \strlen($responsePayloadJson); $server->close($connection, $code); if (System::getEnv('_APP_ENV', 'production') === 'development') { @@ -894,6 +916,17 @@ $server->onOpen(function (int $connection, SwooleRequest $request) use ($server, Console::error('[Error] Code: ' . $response['data']['code']); Console::error('[Error] Message: ' . $response['data']['message']); } + Span::error($th); + } finally { + Span::add('realtime.success', $success); + Span::add('realtime.response_code', $responseCode); + Span::add('realtime.subscription_mode', $subscriptionMode); + Span::add('realtime.channel_count', $channelCount); + Span::add('realtime.subscription_count', $subscriptionCount); + Span::add('realtime.outbound_bytes', $outboundBytes); + Span::add('realtime.project_id', $project?->getId() ?: 'n/a'); + Span::add('realtime.user_id', $logUser?->getId() ?: 'n/a'); + Span::current()?->finish(); } }); @@ -1298,6 +1331,11 @@ $server->onMessage(function (int $connection, string $message) use ($server, $re $server->onClose(function (int $connection) use ($realtime, $stats, $register) { $projectId = null; $userId = null; + $subscriptionsBeforeClose = 0; + $success = false; + + Span::init('realtime.close'); + Span::add('realtime.connection_id', $connection); if (array_key_exists($connection, $realtime->connections)) { $projectId = $realtime->connections[$connection]['projectId'] ?? null; @@ -1320,12 +1358,20 @@ $server->onClose(function (int $connection) use ($realtime, $stats, $register) { METRIC_REALTIME_CONNECTIONS => -1, ], $projectId); } + $success = true; } catch (\Throwable $th) { // Log only; do not rethrow. If we let this bubble, Swoole dumps full coroutine // backtraces and unsubscribe() below would never run (connection cleanup would fail). Console::error('Realtime onClose error: ' . $th->getMessage()); + Span::error($th); + } finally { + Span::add('realtime.success', $success); + Span::add('realtime.project_id', $projectId ?: 'n/a'); + Span::add('realtime.user_id', $userId ?: 'n/a'); + Span::add('realtime.subscriptions_before_close', $subscriptionsBeforeClose); } $realtime->unsubscribe($connection); + Span::current()?->finish(); Console::info('Connection close: ' . $connection); }); From 17e3d03b40e1a5792b80104b998c3f7275022e4e Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Wed, 22 Apr 2026 16:55:52 +0530 Subject: [PATCH 05/13] Add telemetry logging for subscribed channels and queries in realtime events - Introduced new arrays to capture subscribed channels and passed queries during connection and message events. - Enhanced span logging to include details about channels and queries for better monitoring and analysis. - Updated telemetry data structure to reflect the new metrics, improving traceability of realtime interactions. --- app/realtime.php | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/app/realtime.php b/app/realtime.php index 8402dda1f0..16aba56250 100644 --- a/app/realtime.php +++ b/app/realtime.php @@ -705,6 +705,8 @@ $server->onOpen(function (int $connection, SwooleRequest $request) use ($server, $rawSize = $request->getSize(); $channelCount = 0; $subscriptionCount = 0; + $urlSubscribedChannels = []; + $urlPassedQueries = []; $outboundBytes = 0; $responseCode = 200; $subscriptionMode = 'message'; @@ -785,6 +787,7 @@ $server->onOpen(function (int $connection, SwooleRequest $request) use ($server, $channels = Realtime::convertChannels($request->getQuery('channels', []), $user->getId()); $channelCount = \count($channels); + $urlSubscribedChannels = \array_values(\array_keys($channels)); $updateStats = static function (string $projectId, ?string $teamId, string $payloadJson) use ($register, $stats): void { $register->get('telemetry.connectionCounter')->add(1); @@ -844,6 +847,10 @@ $server->onOpen(function (int $connection, SwooleRequest $request) use ($server, $mapping = []; foreach ($subscriptions as $index => $subscription) { $subscriptionId = ID::unique(); + $urlPassedQueries[$index] = \array_map( + fn ($query) => $query instanceof Query ? $query->toString() : (string) $query, + $subscription['queries'] ?? [] + ); $realtime->subscribe( $project->getId(), @@ -923,6 +930,8 @@ $server->onOpen(function (int $connection, SwooleRequest $request) use ($server, Span::add('realtime.subscription_mode', $subscriptionMode); Span::add('realtime.channel_count', $channelCount); Span::add('realtime.subscription_count', $subscriptionCount); + Span::add('realtime.channels_subscribed', json_encode($urlSubscribedChannels)); + Span::add('realtime.queries_passed', json_encode($urlPassedQueries)); Span::add('realtime.outbound_bytes', $outboundBytes); Span::add('realtime.project_id', $project?->getId() ?: 'n/a'); Span::add('realtime.user_id', $logUser?->getId() ?: 'n/a'); @@ -939,6 +948,8 @@ $server->onMessage(function (int $connection, string $message) use ($server, $re $subscriptionDelta = 0; $subscriptionsRequested = 0; $subscriptionsRemoved = 0; + $subscribeChannelsPassed = []; + $subscribeQueriesPassed = []; $outboundBytes = 0; $responseCode = 200; $success = false; @@ -1172,6 +1183,9 @@ $server->onMessage(function (int $connection, string $message) use ($server, $re 'channels' => $payload['channels'], 'queries' => $convertedQueries, ]; + + $subscribeChannelsPassed[] = $payload['channels']; + $subscribeQueriesPassed[] = $payload['queries']; } foreach ($parsedPayloads as $parsedPayload) { @@ -1320,6 +1334,9 @@ $server->onMessage(function (int $connection, string $message) use ($server, $re Span::add('realtime.subscription_delta', $subscriptionDelta); Span::add('realtime.subscriptions_requested', $subscriptionsRequested); Span::add('realtime.subscriptions_removed', $subscriptionsRemoved); + Span::add('realtime.subscribe.channels_passed', json_encode($subscribeChannelsPassed)); + Span::add('realtime.subscribe.queries_passed', json_encode($subscribeQueriesPassed)); + Span::add('realtime.subscribe.subscriptions_count', \count($subscribeChannelsPassed)); Span::add('realtime.outbound_bytes', $outboundBytes); Span::add('realtime.project_id', $project?->getId() ?: $projectId ?: 'n/a'); Span::add('realtime.user_id', $realtime->connections[$connection]['userId'] ?? 'n/a'); From 0f81bc2da9fd312e6a9060fb85cdda5d586376f3 Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Wed, 22 Apr 2026 16:57:51 +0530 Subject: [PATCH 06/13] Refactor telemetry logging in realtime events for consistency and clarity - Updated span logging keys to use camelCase for uniformity across connection and message events. - Added checks to ensure project and user IDs are only logged if they are not empty, enhancing data integrity. - Improved error handling and logging structure to maintain consistency in telemetry data. --- app/realtime.php | 92 ++++++++++++++++++++++++++++++------------------ 1 file changed, 57 insertions(+), 35 deletions(-) diff --git a/app/realtime.php b/app/realtime.php index 16aba56250..91d392b77c 100644 --- a/app/realtime.php +++ b/app/realtime.php @@ -713,15 +713,19 @@ $server->onOpen(function (int $connection, SwooleRequest $request) use ($server, $success = false; Span::init('realtime.open'); - Span::add('realtime.connection_id', $connection); - Span::add('realtime.inbound_bytes', $rawSize); - Span::add('realtime.origin', $request->getOrigin() ?: 'n/a'); + Span::add('realtime.connectionId', $connection); + Span::add('realtime.inboundBytes', $rawSize); + if (!empty($request->getOrigin())) { + Span::add('realtime.origin', $request->getOrigin()); + } try { /** @var Document $project */ $project = $connectionContainer->get('project'); $authorization = $connectionContainer->get('authorization'); - Span::add('realtime.project_id', $project->getId() ?: 'n/a'); + if (!empty($project->getId())) { + Span::add('realtime.projectId', $project->getId()); + } /* * Project Check @@ -783,7 +787,7 @@ $server->onOpen(function (int $connection, SwooleRequest $request) use ($server, } $roles = $user->getRoles($authorization); - Span::add('realtime.user_id', $user->getId() ?: 'n/a'); + Span::add('realtime.userId', $user->getId()); $channels = Realtime::convertChannels($request->getQuery('channels', []), $user->getId()); $channelCount = \count($channels); @@ -926,15 +930,19 @@ $server->onOpen(function (int $connection, SwooleRequest $request) use ($server, Span::error($th); } finally { Span::add('realtime.success', $success); - Span::add('realtime.response_code', $responseCode); - Span::add('realtime.subscription_mode', $subscriptionMode); - Span::add('realtime.channel_count', $channelCount); - Span::add('realtime.subscription_count', $subscriptionCount); - Span::add('realtime.channels_subscribed', json_encode($urlSubscribedChannels)); - Span::add('realtime.queries_passed', json_encode($urlPassedQueries)); - Span::add('realtime.outbound_bytes', $outboundBytes); - Span::add('realtime.project_id', $project?->getId() ?: 'n/a'); - Span::add('realtime.user_id', $logUser?->getId() ?: 'n/a'); + Span::add('realtime.responseCode', $responseCode); + Span::add('realtime.subscriptionMode', $subscriptionMode); + Span::add('realtime.channelCount', $channelCount); + Span::add('realtime.subscriptionCount', $subscriptionCount); + Span::add('realtime.channelsSubscribed', json_encode($urlSubscribedChannels)); + Span::add('realtime.queriesPassed', json_encode($urlPassedQueries)); + Span::add('realtime.outboundBytes', $outboundBytes); + if (!empty($project?->getId())) { + Span::add('realtime.projectId', $project->getId()); + } + if (!empty($logUser?->getId())) { + Span::add('realtime.userId', $logUser->getId()); + } Span::current()?->finish(); } }); @@ -955,10 +963,12 @@ $server->onMessage(function (int $connection, string $message) use ($server, $re $success = false; Span::init('realtime.message'); - Span::add('realtime.connection_id', $connection); - Span::add('realtime.project_id', $projectId ?: 'n/a'); - Span::add('realtime.inbound_bytes', $rawSize); - Span::add('realtime.container_id', $containerId); + Span::add('realtime.connectionId', $connection); + if (!empty($projectId)) { + Span::add('realtime.projectId', $projectId); + } + Span::add('realtime.inboundBytes', $rawSize); + Span::add('realtime.containerId', $containerId); try { $response = new Response(new SwooleResponse()); @@ -1007,7 +1017,7 @@ $server->onMessage(function (int $connection, string $message) use ($server, $re $message = json_decode($message, true); $messageType = $message['type'] ?? 'invalid'; - Span::add('realtime.message_type', $messageType); + Span::add('realtime.messageType', $messageType); if (is_null($message) || (!array_key_exists('type', $message) && !array_key_exists('data', $message))) { throw new Exception(Exception::REALTIME_MESSAGE_FORMAT_INVALID, 'Message format is not valid.'); @@ -1116,7 +1126,9 @@ $server->onMessage(function (int $connection, string $message) use ($server, $re $server->send([$connection], $authResponsePayloadJson); $outboundBytes += \strlen($authResponsePayloadJson); - Span::add('realtime.user_id', $user['$id'] ?? 'n/a'); + if (!empty($user['$id'] ?? null)) { + Span::add('realtime.userId', $user['$id']); + } if ($project !== null && !$project->isEmpty()) { $authOutboundBytes = \strlen($authResponsePayloadJson); @@ -1330,17 +1342,23 @@ $server->onMessage(function (int $connection, string $message) use ($server, $re Span::error($th); } finally { Span::add('realtime.success', $success); - Span::add('realtime.response_code', $responseCode); - Span::add('realtime.subscription_delta', $subscriptionDelta); - Span::add('realtime.subscriptions_requested', $subscriptionsRequested); - Span::add('realtime.subscriptions_removed', $subscriptionsRemoved); - Span::add('realtime.subscribe.channels_passed', json_encode($subscribeChannelsPassed)); - Span::add('realtime.subscribe.queries_passed', json_encode($subscribeQueriesPassed)); - Span::add('realtime.subscribe.subscriptions_count', \count($subscribeChannelsPassed)); - Span::add('realtime.outbound_bytes', $outboundBytes); - Span::add('realtime.project_id', $project?->getId() ?: $projectId ?: 'n/a'); - Span::add('realtime.user_id', $realtime->connections[$connection]['userId'] ?? 'n/a'); - Span::add('realtime.message_type', $messageType); + Span::add('realtime.responseCode', $responseCode); + Span::add('realtime.subscriptionDelta', $subscriptionDelta); + Span::add('realtime.subscriptionsRequested', $subscriptionsRequested); + Span::add('realtime.subscriptionsRemoved', $subscriptionsRemoved); + Span::add('realtime.subscribe.channelsPassed', json_encode($subscribeChannelsPassed)); + Span::add('realtime.subscribe.queriesPassed', json_encode($subscribeQueriesPassed)); + Span::add('realtime.subscribe.subscriptionsCount', \count($subscribeChannelsPassed)); + Span::add('realtime.outboundBytes', $outboundBytes); + if (!empty($project?->getId())) { + Span::add('realtime.projectId', $project->getId()); + } elseif (!empty($projectId)) { + Span::add('realtime.projectId', $projectId); + } + if (!empty($realtime->connections[$connection]['userId'] ?? null)) { + Span::add('realtime.userId', $realtime->connections[$connection]['userId']); + } + Span::add('realtime.messageType', $messageType); Span::current()?->finish(); } }); @@ -1352,7 +1370,7 @@ $server->onClose(function (int $connection) use ($realtime, $stats, $register) { $success = false; Span::init('realtime.close'); - Span::add('realtime.connection_id', $connection); + Span::add('realtime.connectionId', $connection); if (array_key_exists($connection, $realtime->connections)) { $projectId = $realtime->connections[$connection]['projectId'] ?? null; @@ -1383,9 +1401,13 @@ $server->onClose(function (int $connection) use ($realtime, $stats, $register) { Span::error($th); } finally { Span::add('realtime.success', $success); - Span::add('realtime.project_id', $projectId ?: 'n/a'); - Span::add('realtime.user_id', $userId ?: 'n/a'); - Span::add('realtime.subscriptions_before_close', $subscriptionsBeforeClose); + if (!empty($projectId)) { + Span::add('realtime.projectId', $projectId); + } + if (!empty($userId)) { + Span::add('realtime.userId', $userId); + } + Span::add('realtime.subscriptionsBeforeClose', $subscriptionsBeforeClose); } $realtime->unsubscribe($connection); Span::current()?->finish(); From 59e0383264950063a4d65fd085a29b0505f144f8 Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Wed, 22 Apr 2026 17:01:08 +0530 Subject: [PATCH 07/13] updated --- app/realtime.php | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/app/realtime.php b/app/realtime.php index 91d392b77c..e161f08545 100644 --- a/app/realtime.php +++ b/app/realtime.php @@ -1017,12 +1017,17 @@ $server->onMessage(function (int $connection, string $message) use ($server, $re $message = json_decode($message, true); $messageType = $message['type'] ?? 'invalid'; - Span::add('realtime.messageType', $messageType); if (is_null($message) || (!array_key_exists('type', $message) && !array_key_exists('data', $message))) { throw new Exception(Exception::REALTIME_MESSAGE_FORMAT_INVALID, 'Message format is not valid.'); } + if (!\is_scalar($messageType) && $messageType !== null) { + throw new Exception(Exception::REALTIME_MESSAGE_FORMAT_INVALID, 'Message type is not valid.'); + } + + Span::add('realtime.messageType', $messageType); + // Ping does not require project context; other messages do (e.g. after unsubscribe during auth) if (empty($projectId) && ($message['type'] ?? '') !== 'ping') { throw new Exception(Exception::REALTIME_POLICY_VIOLATION, 'Missing project context. Reconnect to the project first.'); From fd9fe5d9ce79c4bc04428409aef5f37d14cf25f8 Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Wed, 22 Apr 2026 17:07:53 +0530 Subject: [PATCH 08/13] corrected the position --- app/realtime.php | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/app/realtime.php b/app/realtime.php index e161f08545..fcca7c9c42 100644 --- a/app/realtime.php +++ b/app/realtime.php @@ -1195,9 +1195,12 @@ $server->onMessage(function (int $connection, string $message) use ($server, $re throw new Exception(Exception::REALTIME_MESSAGE_FORMAT_INVALID, 'Invalid query: ' . $e->getMessage()); } + $convertedChannels = \array_keys(Realtime::convertChannels($payload['channels'], $userId)); + $parsedPayloads[] = [ 'subscriptionId' => $subscriptionId, 'channels' => $payload['channels'], + 'convertedChannels' => $convertedChannels, 'queries' => $convertedQueries, ]; @@ -1207,7 +1210,7 @@ $server->onMessage(function (int $connection, string $message) use ($server, $re foreach ($parsedPayloads as $parsedPayload) { $subscriptionId = $parsedPayload['subscriptionId']; - $channels = \array_keys(Realtime::convertChannels($parsedPayload['channels'], $userId)); + $channels = $parsedPayload['convertedChannels']; $queries = $parsedPayload['queries']; $realtime->subscribe($projectId, $connection, $subscriptionId, $roles, $channels, $queries); } @@ -1226,7 +1229,7 @@ $server->onMessage(function (int $connection, string $message) use ($server, $re 'subscriptions' => \array_map(function (array $parsedPayload) { return [ 'subscriptionId' => $parsedPayload['subscriptionId'], - 'channels' => $parsedPayload['channels'], + 'channels' => $parsedPayload['convertedChannels'], 'queries' => \array_map(fn ($q) => $q->toString(), $parsedPayload['queries']), ]; }, $parsedPayloads), @@ -1413,9 +1416,9 @@ $server->onClose(function (int $connection) use ($realtime, $stats, $register) { Span::add('realtime.userId', $userId); } Span::add('realtime.subscriptionsBeforeClose', $subscriptionsBeforeClose); + Span::current()?->finish(); } $realtime->unsubscribe($connection); - Span::current()?->finish(); Console::info('Connection close: ' . $connection); }); From 46e778ea90269414eed652f6d1b7a98f25b13a31 Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Wed, 22 Apr 2026 17:13:35 +0530 Subject: [PATCH 09/13] updated --- app/realtime.php | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/app/realtime.php b/app/realtime.php index fcca7c9c42..521a80a82f 100644 --- a/app/realtime.php +++ b/app/realtime.php @@ -1016,12 +1016,13 @@ $server->onMessage(function (int $connection, string $message) use ($server, $re } $message = json_decode($message, true); - $messageType = $message['type'] ?? 'invalid'; if (is_null($message) || (!array_key_exists('type', $message) && !array_key_exists('data', $message))) { throw new Exception(Exception::REALTIME_MESSAGE_FORMAT_INVALID, 'Message format is not valid.'); } + $messageType = $message['type'] ?? 'invalid'; + if (!\is_scalar($messageType) && $messageType !== null) { throw new Exception(Exception::REALTIME_MESSAGE_FORMAT_INVALID, 'Message type is not valid.'); } @@ -1408,6 +1409,13 @@ $server->onClose(function (int $connection) use ($realtime, $stats, $register) { Console::error('Realtime onClose error: ' . $th->getMessage()); Span::error($th); } finally { + try { + $realtime->unsubscribe($connection); + } catch (\Throwable $th) { + Console::error('Realtime onClose unsubscribe error: ' . $th->getMessage()); + Span::error($th); + } + Span::add('realtime.success', $success); if (!empty($projectId)) { Span::add('realtime.projectId', $projectId); @@ -1418,7 +1426,6 @@ $server->onClose(function (int $connection) use ($realtime, $stats, $register) { Span::add('realtime.subscriptionsBeforeClose', $subscriptionsBeforeClose); Span::current()?->finish(); } - $realtime->unsubscribe($connection); Console::info('Connection close: ' . $connection); }); From f0f1e1c412bad80f33dbe3591370a400feb5fa7b Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Wed, 22 Apr 2026 17:44:18 +0530 Subject: [PATCH 10/13] updated --- app/realtime.php | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/app/realtime.php b/app/realtime.php index 521a80a82f..eeaa56d30a 100644 --- a/app/realtime.php +++ b/app/realtime.php @@ -1023,7 +1023,7 @@ $server->onMessage(function (int $connection, string $message) use ($server, $re $messageType = $message['type'] ?? 'invalid'; - if (!\is_scalar($messageType) && $messageType !== null) { + if (!\is_scalar($messageType)) { throw new Exception(Exception::REALTIME_MESSAGE_FORMAT_INVALID, 'Message type is not valid.'); } @@ -1285,7 +1285,7 @@ $server->onMessage(function (int $connection, string $message) use ($server, $re $subscriptionsAfter = \count($realtime->getSubscriptionMetadata($connection)); $subscriptionDelta = $subscriptionsAfter - $subscriptionsBefore; $subscriptionsRequested = \count($validatedIds); - $subscriptionsRemoved = \count(\array_filter($unsubscribeResults, fn (array $item) => $item['removed'] ?? false)); + $subscriptionsRemoved = \count(\array_filter($unsubscribeResults, fn (array $item) => $item['removed'])); if ($subscriptionDelta !== 0) { $register->get('telemetry.workerSubscriptionCounter')->add($subscriptionDelta, $register->get('telemetry.workerAttributes')); } From 6d1def7716c1265c3df3fa3630a8a86fa1a1cac9 Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Wed, 22 Apr 2026 17:45:58 +0530 Subject: [PATCH 11/13] removed redundant span attributes --- app/realtime.php | 6 ------ 1 file changed, 6 deletions(-) diff --git a/app/realtime.php b/app/realtime.php index eeaa56d30a..572256eec6 100644 --- a/app/realtime.php +++ b/app/realtime.php @@ -723,9 +723,6 @@ $server->onOpen(function (int $connection, SwooleRequest $request) use ($server, /** @var Document $project */ $project = $connectionContainer->get('project'); $authorization = $connectionContainer->get('authorization'); - if (!empty($project->getId())) { - Span::add('realtime.projectId', $project->getId()); - } /* * Project Check @@ -787,7 +784,6 @@ $server->onOpen(function (int $connection, SwooleRequest $request) use ($server, } $roles = $user->getRoles($authorization); - Span::add('realtime.userId', $user->getId()); $channels = Realtime::convertChannels($request->getQuery('channels', []), $user->getId()); $channelCount = \count($channels); @@ -1027,8 +1023,6 @@ $server->onMessage(function (int $connection, string $message) use ($server, $re throw new Exception(Exception::REALTIME_MESSAGE_FORMAT_INVALID, 'Message type is not valid.'); } - Span::add('realtime.messageType', $messageType); - // Ping does not require project context; other messages do (e.g. after unsubscribe during auth) if (empty($projectId) && ($message['type'] ?? '') !== 'ping') { throw new Exception(Exception::REALTIME_POLICY_VIOLATION, 'Missing project context. Reconnect to the project first.'); From b006858d0c34e3df26623324723a9c113e5436bd Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Wed, 22 Apr 2026 17:52:45 +0530 Subject: [PATCH 12/13] dedupe --- app/realtime.php | 16 ++-------------- 1 file changed, 2 insertions(+), 14 deletions(-) diff --git a/app/realtime.php b/app/realtime.php index 572256eec6..d1177ac07b 100644 --- a/app/realtime.php +++ b/app/realtime.php @@ -960,9 +960,6 @@ $server->onMessage(function (int $connection, string $message) use ($server, $re Span::init('realtime.message'); Span::add('realtime.connectionId', $connection); - if (!empty($projectId)) { - Span::add('realtime.projectId', $projectId); - } Span::add('realtime.inboundBytes', $rawSize); Span::add('realtime.containerId', $containerId); @@ -1126,9 +1123,6 @@ $server->onMessage(function (int $connection, string $message) use ($server, $re $server->send([$connection], $authResponsePayloadJson); $outboundBytes += \strlen($authResponsePayloadJson); - if (!empty($user['$id'] ?? null)) { - Span::add('realtime.userId', $user['$id']); - } if ($project !== null && !$project->isEmpty()) { $authOutboundBytes = \strlen($authResponsePayloadJson); @@ -1353,14 +1347,8 @@ $server->onMessage(function (int $connection, string $message) use ($server, $re Span::add('realtime.subscribe.queriesPassed', json_encode($subscribeQueriesPassed)); Span::add('realtime.subscribe.subscriptionsCount', \count($subscribeChannelsPassed)); Span::add('realtime.outboundBytes', $outboundBytes); - if (!empty($project?->getId())) { - Span::add('realtime.projectId', $project->getId()); - } elseif (!empty($projectId)) { - Span::add('realtime.projectId', $projectId); - } - if (!empty($realtime->connections[$connection]['userId'] ?? null)) { - Span::add('realtime.userId', $realtime->connections[$connection]['userId']); - } + Span::add('realtime.projectId', $project?->getId() ?? $projectId); + Span::add('realtime.userId', $realtime->connections[$connection]['userId'] ?? null); Span::add('realtime.messageType', $messageType); Span::current()?->finish(); } From c2e5bbe0f738ac4a16a1a9924656f29afd1fa19f Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Wed, 22 Apr 2026 18:11:32 +0530 Subject: [PATCH 13/13] updated --- app/realtime.php | 18 +----------------- 1 file changed, 1 insertion(+), 17 deletions(-) diff --git a/app/realtime.php b/app/realtime.php index d1177ac07b..71aa251069 100644 --- a/app/realtime.php +++ b/app/realtime.php @@ -705,8 +705,6 @@ $server->onOpen(function (int $connection, SwooleRequest $request) use ($server, $rawSize = $request->getSize(); $channelCount = 0; $subscriptionCount = 0; - $urlSubscribedChannels = []; - $urlPassedQueries = []; $outboundBytes = 0; $responseCode = 200; $subscriptionMode = 'message'; @@ -787,7 +785,6 @@ $server->onOpen(function (int $connection, SwooleRequest $request) use ($server, $channels = Realtime::convertChannels($request->getQuery('channels', []), $user->getId()); $channelCount = \count($channels); - $urlSubscribedChannels = \array_values(\array_keys($channels)); $updateStats = static function (string $projectId, ?string $teamId, string $payloadJson) use ($register, $stats): void { $register->get('telemetry.connectionCounter')->add(1); @@ -847,10 +844,6 @@ $server->onOpen(function (int $connection, SwooleRequest $request) use ($server, $mapping = []; foreach ($subscriptions as $index => $subscription) { $subscriptionId = ID::unique(); - $urlPassedQueries[$index] = \array_map( - fn ($query) => $query instanceof Query ? $query->toString() : (string) $query, - $subscription['queries'] ?? [] - ); $realtime->subscribe( $project->getId(), @@ -930,8 +923,6 @@ $server->onOpen(function (int $connection, SwooleRequest $request) use ($server, Span::add('realtime.subscriptionMode', $subscriptionMode); Span::add('realtime.channelCount', $channelCount); Span::add('realtime.subscriptionCount', $subscriptionCount); - Span::add('realtime.channelsSubscribed', json_encode($urlSubscribedChannels)); - Span::add('realtime.queriesPassed', json_encode($urlPassedQueries)); Span::add('realtime.outboundBytes', $outboundBytes); if (!empty($project?->getId())) { Span::add('realtime.projectId', $project->getId()); @@ -952,8 +943,6 @@ $server->onMessage(function (int $connection, string $message) use ($server, $re $subscriptionDelta = 0; $subscriptionsRequested = 0; $subscriptionsRemoved = 0; - $subscribeChannelsPassed = []; - $subscribeQueriesPassed = []; $outboundBytes = 0; $responseCode = 200; $success = false; @@ -1192,9 +1181,6 @@ $server->onMessage(function (int $connection, string $message) use ($server, $re 'convertedChannels' => $convertedChannels, 'queries' => $convertedQueries, ]; - - $subscribeChannelsPassed[] = $payload['channels']; - $subscribeQueriesPassed[] = $payload['queries']; } foreach ($parsedPayloads as $parsedPayload) { @@ -1343,9 +1329,7 @@ $server->onMessage(function (int $connection, string $message) use ($server, $re Span::add('realtime.subscriptionDelta', $subscriptionDelta); Span::add('realtime.subscriptionsRequested', $subscriptionsRequested); Span::add('realtime.subscriptionsRemoved', $subscriptionsRemoved); - Span::add('realtime.subscribe.channelsPassed', json_encode($subscribeChannelsPassed)); - Span::add('realtime.subscribe.queriesPassed', json_encode($subscribeQueriesPassed)); - Span::add('realtime.subscribe.subscriptionsCount', \count($subscribeChannelsPassed)); + Span::add('realtime.subscribe.subscriptionsCount', $subscriptionsRequested); Span::add('realtime.outboundBytes', $outboundBytes); Span::add('realtime.projectId', $project?->getId() ?? $projectId); Span::add('realtime.userId', $realtime->connections[$connection]['userId'] ?? null);