[server] Use a text feature flag to toggle which message updates come from redis and messagebus (#18209)

Author: easyCZ

components/server/src/messaging/local-message-broker.ts

@@ -15,6 +15,7 @@ import { log } from "@gitpod/gitpod-protocol/lib/util/logging";
 import { TraceContext } from "@gitpod/gitpod-protocol/lib/util/tracing";
 import { inject, injectable } from "inversify";
 import { MessageBusIntegration } from "../workspace/messagebus-integration";
+import { getExperimentsClientForBackend } from "@gitpod/gitpod-protocol/lib/experiments/configcat-server";
 
 export interface PrebuildUpdateListener {
     (ctx: TraceContext, evt: PrebuildWithStatus): void;
@@ -69,7 +70,12 @@ export class LocalRabbitMQBackedMessageBroker implements LocalMessageBroker {
         this.disposables.push(
             this.messageBusIntegration.listenForPrebuildUpdates(
                 undefined,
-                (ctx: TraceContext, update: PrebuildWithStatus) => {
+                async (ctx: TraceContext, update: PrebuildWithStatus) => {
+                    const enabled = await this.isRedisPubSubByTypeEnabled("prebuild");
+                    if (enabled) {
+                        log.debug("[messagebus] Prebuild listener is disabled through feature flag");
+                        return;
+                    }
                     TraceContext.setOWI(ctx, { workspaceId: update.info.buildWorkspaceId });
 
                     const listeners = this.prebuildUpdateListeners.get(update.info.projectId) || [];
@@ -91,7 +97,12 @@ export class LocalRabbitMQBackedMessageBroker implements LocalMessageBroker {
         );
         this.disposables.push(
             this.messageBusIntegration.listenForPrebuildUpdatableQueue(
-                (ctx: TraceContext, evt: HeadlessWorkspaceEvent) => {
+                async (ctx: TraceContext, evt: HeadlessWorkspaceEvent) => {
+                    const enabled = await this.isRedisPubSubByTypeEnabled("prebuild-updatable");
+                    if (enabled) {
+                        log.debug("[messagebus] Prebuild updatable listener is disabled through feature flag");
+                        return;
+                    }
                     TraceContext.setOWI(ctx, { workspaceId: evt.workspaceID });
 
                     const listeners =
@@ -110,7 +121,13 @@ export class LocalRabbitMQBackedMessageBroker implements LocalMessageBroker {
         this.disposables.push(
             this.messageBusIntegration.listenForWorkspaceInstanceUpdates(
                 undefined,
-                (ctx: TraceContext, instance: WorkspaceInstance, userId: string | undefined) => {
+                async (ctx: TraceContext, instance: WorkspaceInstance, userId: string | undefined) => {
+                    const enabled = await this.isRedisPubSubByTypeEnabled("workspace-instance");
+                    if (enabled) {
+                        log.debug("[messagebus] Workspace instance listner is disabled through feature flag");
+                        return;
+                    }
+
                     TraceContext.setOWI(ctx, { userId, instanceId: instance.id });
 
                     if (!userId) {
@@ -170,4 +187,15 @@ export class LocalRabbitMQBackedMessageBroker implements LocalMessageBroker {
             }
         });
     }
+
+    private async isRedisPubSubByTypeEnabled(
+        type: "workspace-instance" | "prebuild" | "prebuild-updatable",
+    ): Promise<boolean> {
+        const enabledTypes = await getExperimentsClientForBackend().getValueAsync(
+            "enableRedisPubSubByUpdateType",
+            "none",
+            {},
+        );
+        return enabledTypes.indexOf(type) >= 0;
+    }
 }

components/server/src/messaging/redis-subscriber.ts

@@ -19,7 +19,6 @@ import {
 } from "@gitpod/gitpod-protocol";
 import { log } from "@gitpod/gitpod-protocol/lib/util/logging";
 import { getExperimentsClientForBackend } from "@gitpod/gitpod-protocol/lib/experiments/configcat-server";
-import { Attributes } from "@gitpod/gitpod-protocol/lib/experiments/types";
 import { reportRedisUpdateCompleted, reportRedisUpdateReceived } from "../prometheus-metrics";
 import { Redis } from "ioredis";
 
@@ -42,12 +41,6 @@ export class RedisSubscriber implements LocalMessageBroker {
         this.redis.on("message", async (channel: string, message: string) => {
             reportRedisUpdateReceived(channel);
 
-            const featureEnabled = await this.isRedisPubSubEnabled({});
-            if (!featureEnabled) {
-                log.debug("[redis] Redis listener is disabled through feature flag", { channel, message });
-                return;
-            }
-
             let err: Error | undefined;
             try {
                 await this.onMessage(channel, message);
@@ -64,6 +57,15 @@ export class RedisSubscriber implements LocalMessageBroker {
     private async onMessage(channel: string, message: string): Promise<void> {
         switch (channel) {
             case WorkspaceInstanceUpdatesChannel:
+                const enabled = await this.isRedisPubSubByTypeEnabled("workspace-instance");
+                if (!enabled) {
+                    log.debug("[redis] Redis workspace instance update is disabled through feature flag", {
+                        channel,
+                        message,
+                    });
+                    return;
+                }
+
                 const parsed = JSON.parse(message) as RedisWorkspaceInstanceUpdate;
                 return this.onInstanceUpdate(parsed);
             default:
@@ -112,8 +114,14 @@ export class RedisSubscriber implements LocalMessageBroker {
         });
     }
 
-    private async isRedisPubSubEnabled(attributes: Attributes): Promise<boolean> {
-        const enabled = await getExperimentsClientForBackend().getValueAsync("enableRedisPubSub", false, attributes);
-        return enabled;
+    private async isRedisPubSubByTypeEnabled(
+        type: "workspace-instance" | "prebuild" | "prebuild-updatable",
+    ): Promise<boolean> {
+        const enabledTypes = await getExperimentsClientForBackend().getValueAsync(
+            "enableRedisPubSubByUpdateType",
+            "none",
+            {},
+        );
+        return enabledTypes.indexOf(type) >= 0;
     }
 }