From bb3a0d4222061f629c97933d2ab6135be356d04d Mon Sep 17 00:00:00 2001
From: Diego Sampaio <chinello@gmail.com>
Date: Mon, 11 Sep 2023 17:41:24 -0300
Subject: [PATCH] chore: add subscription metrics to ddp-streamer (#30289)

---
 ee/apps/ddp-streamer/src/DDPStreamer.ts     | 11 +++++++++++
 ee/apps/ddp-streamer/src/Server.ts          | 11 +++++++++++
 packages/core-services/src/types/IBroker.ts |  1 +
 3 files changed, 23 insertions(+)

diff --git a/ee/apps/ddp-streamer/src/DDPStreamer.ts b/ee/apps/ddp-streamer/src/DDPStreamer.ts
index 868ad8fec202..bccb35d2b326 100644
--- a/ee/apps/ddp-streamer/src/DDPStreamer.ts
+++ b/ee/apps/ddp-streamer/src/DDPStreamer.ts
@@ -72,6 +72,15 @@ export class DDPStreamer extends ServiceClass {
 			return;
 		}
 
+		metrics.register({
+			name: 'rocketchat_subscription',
+			type: 'histogram',
+			labelNames: ['subscription'],
+			description: 'Client subscriptions to Rocket.Chat',
+			unit: 'millisecond',
+			quantiles: true,
+		});
+
 		metrics.register({
 			name: 'users_connected',
 			type: 'gauge',
@@ -86,6 +95,8 @@ export class DDPStreamer extends ServiceClass {
 			description: 'Users logged by streamer',
 		});
 
+		server.setMetrics(metrics);
+
 		server.on(DDP_EVENTS.CONNECTED, () => {
 			metrics.increment('users_connected', { nodeID }, 1);
 		});
diff --git a/ee/apps/ddp-streamer/src/Server.ts b/ee/apps/ddp-streamer/src/Server.ts
index 01c7c63511ff..af083621230d 100644
--- a/ee/apps/ddp-streamer/src/Server.ts
+++ b/ee/apps/ddp-streamer/src/Server.ts
@@ -1,5 +1,6 @@
 import { EventEmitter } from 'events';
 
+import type { IServiceMetrics } from '@rocket.chat/core-services';
 import { MeteorService, isMeteorError, MeteorError } from '@rocket.chat/core-services';
 import { Logger } from '@rocket.chat/logger';
 import ejson from 'ejson';
@@ -38,6 +39,8 @@ export class Server extends EventEmitter {
 
 	private _methods = new Map<string, MethodFn>();
 
+	private metrics?: IServiceMetrics;
+
 	public readonly id = uuidv1();
 
 	serialize = ejson.stringify;
@@ -52,6 +55,10 @@ export class Server extends EventEmitter {
 		return ejson.parse(payload);
 	};
 
+	setMetrics(metrics: IServiceMetrics): void {
+		this.metrics = metrics;
+	}
+
 	async call(client: Client, packet: IPacket): Promise<void> {
 		// if client is not connected we don't need to do anything
 		if (client.ws.readyState !== WebSocket.OPEN) {
@@ -103,9 +110,13 @@ export class Server extends EventEmitter {
 				throw new MeteorError(404, `Subscription '${packet.name}' not found`);
 			}
 
+			const end = this.metrics?.timer('rocketchat_subscription', { subscription: packet.name });
+
 			const publication = new Publication(client, packet, this);
 			const [eventName, options] = packet.params;
 			await fn.call(publication, eventName, options);
+
+			end?.();
 		} catch (err: unknown) {
 			return this.nosub(client, packet, handleInternalException(err, 'Subscription error'));
 		}
diff --git a/packages/core-services/src/types/IBroker.ts b/packages/core-services/src/types/IBroker.ts
index 8647d04a56dc..4bd48afef0ff 100644
--- a/packages/core-services/src/types/IBroker.ts
+++ b/packages/core-services/src/types/IBroker.ts
@@ -27,6 +27,7 @@ export type BaseMetricOptions = {
 	labelNames?: Array<string>;
 	unit?: string;
 	aggregator?: string;
+	[key: string]: unknown;
 };
 
 export interface IServiceMetrics {