Introduce gitpod_server_topic_read_total counter metrics, which counts the amount of times that server reads messagebus topic for each workspace.

Signed-off-by: ArthurSens <[email protected]>

Author: ArthurSens

components/gitpod-messagebus/src/messagebus.ts

@@ -442,11 +442,11 @@ export abstract class AbstractTopicListener<T> implements MessagebusListener {
     constructor(protected readonly exchangeName: string, protected readonly listener: TopicListener<T>) { }
 
     async establish(channel: Channel): Promise<void> {
-        const topic = await this.topic();
+        const topic = this.topic();
         return this.doEstablish(channel, topic);
     }
 
-    abstract topic(): Promise<string>;
+    abstract topic(): string;
 
     protected async doEstablish(channel: Channel, topic: string): Promise<void> {
         if (channel === undefined) {

components/server/src/consensus/rabbitmq-consensus-leader-messenger.ts

@@ -98,7 +98,7 @@ export class RabbitMQConsensusLeaderMessenger extends AbstractMessageBusIntegrat
 
 class EventListener extends AbstractTopicListener<RaftMessage> {
 
-    async topic(): Promise<string> {
+    topic(): string {
         return '';
     }
 

components/server/src/prometheus-metrics.ts

@@ -88,3 +88,16 @@ const httpRequestDuration = new prometheusClient.Histogram({
 export function observeHttpRequestDuration(method: string, route: string, statusCode: number, durationInSeconds: number) {
     httpRequestDuration.observe({ method, route, statusCode }, durationInSeconds);
 }
+
+const messagebusTopicReads = new prometheusClient.Counter({
+   name: 'gitpod_server_topic_reads_total',
+   help: 'The amount of reads from messagebus topics.',
+   labelNames: ['topic'],
+   registers: [prometheusClient.register]
+});
+
+export function increaseMessagebusTopicReads(topic: string) {
+    messagebusTopicReads.inc({
+        topic,
+    })
+}

components/server/src/workspace/messagebus-integration.ts

@@ -13,14 +13,15 @@ import { Channel, Message } from "amqplib";
 import { TraceContext } from "@gitpod/gitpod-protocol/lib/util/tracing";
 import * as opentracing from "opentracing";
 import { CancellationTokenSource } from "vscode-ws-jsonrpc";
+import { increaseMessagebusTopicReads } from '../prometheus-metrics';
 
 export class WorkspaceInstanceUpdateListener extends AbstractTopicListener<WorkspaceInstance> {
 
     constructor(protected readonly messageBusHelper: MessageBusHelper, listener: TopicListener<WorkspaceInstance>, protected readonly userId?: string) {
         super(messageBusHelper.workspaceExchange, listener);
     }
 
-    async topic() {
+    topic() {
         return this.messageBusHelper.getWsTopicForListening(this.userId, undefined, "updates");
     }
 }
@@ -31,7 +32,7 @@ export class HeadlessWorkspaceLogListener extends AbstractTopicListener<Headless
         super(messageBusHelper.workspaceExchange, listener);
     }
 
-    async topic() {
+    topic() {
         return this.messageBusHelper.getWsTopicForListening(undefined, this.workspaceID, "headless-log");
     }
 
@@ -117,6 +118,7 @@ export class MessageBusIntegration extends AbstractMessageBusIntegration {
         const listener = new HeadlessWorkspaceLogListener(this.messageBusHelper, callback, workspaceID);
         const cancellationTokenSource = new CancellationTokenSource()
         this.listen(listener, cancellationTokenSource.token);
+        increaseMessagebusTopicReads(listener.topic())
         return Disposable.create(() => cancellationTokenSource.cancel())
     }
 
@@ -131,6 +133,7 @@ export class MessageBusIntegration extends AbstractMessageBusIntegration {
         const listener = new WorkspaceInstanceUpdateListener(this.messageBusHelper, callback, userId);
         const cancellationTokenSource = new CancellationTokenSource()
         this.listen(listener, cancellationTokenSource.token);
+        increaseMessagebusTopicReads(listener.topic())
         return Disposable.create(() => cancellationTokenSource.cancel())
     }
 

components/ws-manager-bridge/src/messagebus-integration.ts

@@ -104,7 +104,7 @@ class WorkspaceStatusUpdateListener extends AbstractTopicListener<WorkspaceStatu
         super(MessageBusIntegration.WORKSPACE_EXCHANGE, listener);
     }
 
-    async topic() {
+    topic() {
         return this._topic;
     }
 }
@@ -115,7 +115,7 @@ class WorkspaceLogListener extends AbstractTopicListener<WorkspaceLogMessage.AsO
         super(MessageBusIntegration.WORKSPACE_EXCHANGE, listener);
     }
 
-    async topic() {
+    topic() {
         return this._topic;
     }
 }