[server] Broadcast headless updates to subscribers

Author: easyCZ

components/gitpod-protocol/src/redis.ts

@@ -4,10 +4,12 @@
  * See License.AGPL.txt in the project root for license information.
  */
 
+import { HeadlessWorkspaceEventType } from "./headless-workspace-log";
 import { PrebuiltWorkspaceState } from "./protocol";
 
 export const WorkspaceInstanceUpdatesChannel = "chan:workspace-instances";
 export const PrebuildUpdatesChannel = "chan:prebuilds";
+export const HeadlessUpdatesChannel = "chan:headless";
 
 export type RedisWorkspaceInstanceUpdate = {
     ownerID: string;
@@ -21,3 +23,8 @@ export type RedisPrebuildUpdate = {
     workspaceID: string;
     projectID: string;
 };
+
+export type RedisHeadlessUpdate = {
+    workspaceID: string;
+    type: HeadlessWorkspaceEventType;
+};

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

@@ -7,15 +7,19 @@
 import {
     HeadlessWorkspaceEventListener,
     LocalMessageBroker,
+    LocalRabbitMQBackedMessageBroker,
     PrebuildUpdateListener,
     WorkspaceInstanceUpdateListener,
 } from "./local-message-broker";
 import { inject, injectable } from "inversify";
 import {
     Disposable,
     DisposableCollection,
+    HeadlessUpdatesChannel,
+    HeadlessWorkspaceEvent,
     PrebuildUpdatesChannel,
     PrebuildWithStatus,
+    RedisHeadlessUpdate,
     RedisPrebuildUpdate,
     RedisWorkspaceInstanceUpdate,
     WorkspaceInstanceUpdatesChannel,
@@ -39,11 +43,16 @@ export class RedisSubscriber implements LocalMessageBroker {
 
     protected workspaceInstanceUpdateListeners: Map<string, WorkspaceInstanceUpdateListener[]> = new Map();
     protected prebuildUpdateListeners: Map<string, PrebuildUpdateListener[]> = new Map();
+    protected headlessWorkspaceEventListeners: Map<string, HeadlessWorkspaceEventListener[]> = new Map();
 
     protected readonly disposables = new DisposableCollection();
 
     async start(): Promise<void> {
+<<<<<<< HEAD
         const channels = [WorkspaceInstanceUpdatesChannel, PrebuildUpdatesChannel];
+=======
+        const channels = [WorkspaceInstanceUpdatesChannel, HeadlessUpdatesChannel];
+>>>>>>> 65ff502c2 ([server] Broadcast headless updates to subscribers)
 
         for (const chan of channels) {
             await this.redis.subscribe(chan);
@@ -81,16 +90,27 @@ export class RedisSubscriber implements LocalMessageBroker {
                 return this.onInstanceUpdate(JSON.parse(message) as RedisWorkspaceInstanceUpdate);
 
             case PrebuildUpdatesChannel:
-                const headlessUpdateEnabled = await this.isRedisPubSubByTypeEnabled("prebuild-updatable");
-                if (!headlessUpdateEnabled) {
-                    log.debug("[redis] Redis headless update is disabled through feature flag", {
+                const prebuildTypeEnabled = await this.isRedisPubSubByTypeEnabled("prebuild");
+                if (!prebuildTypeEnabled) {
+                    log.debug("[redis] Redis prebuild update is disabled through feature flag", {
                         channel,
                         message,
                     });
                     return;
                 }
                 return this.onPrebuildUpdate(JSON.parse(message) as RedisPrebuildUpdate);
 
+            case HeadlessUpdatesChannel:
+                const headlessTypeEnabled = await this.isRedisPubSubByTypeEnabled("prebuild-updatable");
+                if (!headlessTypeEnabled) {
+                    log.debug("[redis] Redis headless update is disabled through feature flag", {
+                        channel,
+                        message,
+                    });
+                    return;
+                }
+                return this.onHeadlessUpdate(JSON.parse(message) as RedisHeadlessUpdate);
+
             default:
                 throw new Error(`Redis Pub/Sub received message on unknown channel: ${channel}`);
         }
@@ -163,6 +183,29 @@ export class RedisSubscriber implements LocalMessageBroker {
         }
     }
 
+    private async onHeadlessUpdate(update: RedisHeadlessUpdate): Promise<void> {
+        log.debug("[redis] Received prebuild update", { update });
+
+        if (!update.type || !update.workspaceID) {
+            return;
+        }
+
+        const listeners =
+            this.headlessWorkspaceEventListeners.get(LocalRabbitMQBackedMessageBroker.UNDEFINED_KEY) || [];
+        if (listeners.length === 0) {
+            return;
+        }
+
+        const ctx = {};
+        for (const l of listeners) {
+            try {
+                l(ctx, update);
+            } catch (err) {
+                log.error("Failed to broadcast headless update.", update, err);
+            }
+        }
+    }
+
     async stop() {
         this.disposables.dispose();
     }
@@ -172,8 +215,12 @@ export class RedisSubscriber implements LocalMessageBroker {
     }
 
     listenForPrebuildUpdatableEvents(listener: HeadlessWorkspaceEventListener): Disposable {
-        // TODO: not implemented
-        return Disposable.create(() => {});
+        // we're being cheap here in re-using a map where it just needs to be a plain array.
+        return this.doRegister(
+            LocalRabbitMQBackedMessageBroker.UNDEFINED_KEY,
+            listener,
+            this.headlessWorkspaceEventListeners,
+        );
     }
 
     listenForWorkspaceInstanceUpdates(userId: string, listener: WorkspaceInstanceUpdateListener): Disposable {

components/ws-manager-bridge/src/prebuild-updater.ts

@@ -7,7 +7,7 @@
 import { inject, injectable } from "inversify";
 import { TraceContext } from "@gitpod/gitpod-protocol/lib/util/tracing";
 import { WorkspaceStatus, WorkspaceType } from "@gitpod/ws-manager/lib";
-import { WorkspaceInstance } from "@gitpod/gitpod-protocol";
+import { HeadlessWorkspaceEventType, WorkspaceInstance } from "@gitpod/gitpod-protocol";
 import { log, LogContext } from "@gitpod/gitpod-protocol/lib/util/logging";
 import { PrebuildStateMapper } from "./prebuild-state-mapper";
 import { DBWithTracing, TracedWorkspaceDB } from "@gitpod/gitpod-db/lib/traced-db";
@@ -89,7 +89,12 @@ export class PrebuildUpdater {
                     type: update.type,
                     workspaceID: workspaceId,
                 });
-                await this.publisher.publishHeadlessUpdate();
+                if (!HeadlessWorkspaceEventType.isRunning(update.type)) {
+                    await this.publisher.publishHeadlessUpdate({
+                        type: update.type,
+                        workspaceID: workspaceId,
+                    });
+                }
 
                 // prebuild info
                 const info = (await this.workspaceDB.trace({ span }).findPrebuildInfos([updatedPrebuild.id]))[0];

components/ws-manager-bridge/src/redis/publisher.ts

@@ -9,7 +9,9 @@ import { log } from "@gitpod/gitpod-protocol/lib/util/logging";
 import { Metrics } from "../metrics";
 import { RedisClient } from "./client";
 import {
+    HeadlessUpdatesChannel,
     PrebuildUpdatesChannel,
+    RedisHeadlessUpdate,
     RedisPrebuildUpdate,
     RedisWorkspaceInstanceUpdate,
     WorkspaceInstanceUpdatesChannel,
@@ -52,8 +54,19 @@ export class RedisPublisher {
         }
     }
 
-    async publishHeadlessUpdate(): Promise<void> {
+    async publishHeadlessUpdate(update: RedisHeadlessUpdate): Promise<void> {
         log.debug("[redis] Publish headless udpate invoked.");
-        this.metrics.reportUpdatePublished("headless");
+
+        let err: Error | undefined;
+        try {
+            const serialized = JSON.stringify(update);
+            await this.client.get().publish(HeadlessUpdatesChannel, serialized);
+            log.debug("[redis] Succesfully published headless update.", update);
+        } catch (e) {
+            err = e;
+            log.error("[redis] Failed to publish headless update.", e, update);
+        } finally {
+            this.metrics.reportUpdatePublished("headless", err);
+        }
     }
 }