[server] Broadcast prebuild updates from redis

Author: easyCZ

components/gitpod-protocol/src/redis.ts

@@ -4,10 +4,21 @@
  * See License.AGPL.txt in the project root for license information.
  */
 
+import { PrebuiltWorkspaceState } from "./protocol";
+
 export const WorkspaceInstanceUpdatesChannel = "chan:workspace-instances";
+export const PrebuildUpdatesChannel = "chan:prebuilds";
 
 export type RedisWorkspaceInstanceUpdate = {
     ownerID: string;
     instanceID: string;
     workspaceID: string;
 };
+
+export type RedisPrebuildUpdate = {
+    status: PrebuiltWorkspaceState;
+    prebuildID: string;
+    workspaceID: string;
+    instanceID: string;
+    projectID: string;
+};

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

@@ -14,6 +14,9 @@ import { inject, injectable } from "inversify";
 import {
     Disposable,
     DisposableCollection,
+    PrebuildUpdatesChannel,
+    PrebuildWithStatus,
+    RedisPrebuildUpdate,
     RedisWorkspaceInstanceUpdate,
     WorkspaceInstanceUpdatesChannel,
 } from "@gitpod/gitpod-protocol";
@@ -35,11 +38,12 @@ export class RedisSubscriber implements LocalMessageBroker {
     ) {}
 
     protected workspaceInstanceUpdateListeners: Map<string, WorkspaceInstanceUpdateListener[]> = new Map();
+    protected prebuildUpdateListeners: Map<string, PrebuildUpdateListener[]> = new Map();
 
     protected readonly disposables = new DisposableCollection();
 
     async start(): Promise<void> {
-        const channels = [WorkspaceInstanceUpdatesChannel];
+        const channels = [WorkspaceInstanceUpdatesChannel, PrebuildUpdatesChannel];
 
         for (const chan of channels) {
             await this.redis.subscribe(chan);
@@ -65,17 +69,28 @@ 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) {
+                const wsInstanceTypeEnabled = await this.isRedisPubSubByTypeEnabled("workspace-instance");
+                if (!wsInstanceTypeEnabled) {
                     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);
+                return this.onInstanceUpdate(JSON.parse(message) as RedisWorkspaceInstanceUpdate);
+
+            case PrebuildUpdatesChannel:
+                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);
+
             default:
                 throw new Error(`Redis Pub/Sub received message on unknown channel: ${channel}`);
         }
@@ -112,13 +127,48 @@ export class RedisSubscriber implements LocalMessageBroker {
         }
     }
 
+    private async onPrebuildUpdate(update: RedisPrebuildUpdate): Promise<void> {
+        log.debug("[redis] Received prebuild update", { update });
+
+        if (!update.projectID || !update.prebuildID || !update.status) {
+            return;
+        }
+
+        const listeners = this.prebuildUpdateListeners.get(update.projectID) || [];
+        if (listeners.length === 0) {
+            return;
+        }
+
+        const ctx = {};
+        const info = (await this.workspaceDB.findPrebuildInfos([update.prebuildID]))[0];
+        if (!info) {
+            return;
+        }
+
+        const prebuildWithStatus: PrebuildWithStatus = {
+            info: info,
+            status: update.status,
+        };
+
+        for (const l of listeners) {
+            try {
+                l(ctx, prebuildWithStatus);
+            } catch (err) {
+                log.error(
+                    "Failed to broadcast workspace instance update.",
+                    { projectId: update.projectID, instanceId: update.instanceID, workspaceId: update.workspaceID },
+                    err,
+                );
+            }
+        }
+    }
+
     async stop() {
         this.disposables.dispose();
     }
 
     listenForPrebuildUpdates(projectId: string, listener: PrebuildUpdateListener): Disposable {
-        // TODO: not implemented
-        return Disposable.create(() => {});
+        return this.doRegister(projectId, listener, this.prebuildUpdateListeners);
     }
 
     listenForPrebuildUpdatableEvents(listener: HeadlessWorkspaceEventListener): Disposable {

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

@@ -95,7 +95,13 @@ export class PrebuildUpdater {
                 const info = (await this.workspaceDB.trace({ span }).findPrebuildInfos([updatedPrebuild.id]))[0];
                 if (info) {
                     await this.messagebus.notifyOnPrebuildUpdate({ info, status: updatedPrebuild.state });
-                    await this.publisher.publishPrebuildUpdate();
+                    await this.publisher.publishPrebuildUpdate({
+                        instanceID: instanceId,
+                        projectID: prebuild.projectId || "",
+                        prebuildID: updatedPrebuild.id,
+                        status: updatedPrebuild.state,
+                        workspaceID: workspaceId,
+                    });
                 }
             }
         } catch (e) {
@@ -120,7 +126,13 @@ export class PrebuildUpdater {
                 const info = (await this.workspaceDB.trace({ span }).findPrebuildInfos([prebuild.id]))[0];
                 if (info) {
                     await this.messagebus.notifyOnPrebuildUpdate({ info, status: prebuild.state });
-                    await this.publisher.publishPrebuildUpdate();
+                    await this.publisher.publishPrebuildUpdate({
+                        instanceID: instance.id,
+                        projectID: prebuild.projectId || "",
+                        prebuildID: prebuild.id,
+                        status: prebuild.state,
+                        workspaceID: instance.workspaceId,
+                    });
                 }
             }
         }

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

@@ -8,7 +8,12 @@ import { inject, injectable } from "inversify";
 import { log } from "@gitpod/gitpod-protocol/lib/util/logging";
 import { Metrics } from "../metrics";
 import { RedisClient } from "./client";
-import { RedisWorkspaceInstanceUpdate, WorkspaceInstanceUpdatesChannel } from "@gitpod/gitpod-protocol";
+import {
+    PrebuildUpdatesChannel,
+    RedisPrebuildUpdate,
+    RedisWorkspaceInstanceUpdate,
+    WorkspaceInstanceUpdatesChannel,
+} from "@gitpod/gitpod-protocol";
 
 @injectable()
 export class RedisPublisher {
@@ -17,9 +22,20 @@ export class RedisPublisher {
         @inject(Metrics) private readonly metrics: Metrics,
     ) {}
 
-    async publishPrebuildUpdate(): Promise<void> {
+    async publishPrebuildUpdate(update: RedisPrebuildUpdate): Promise<void> {
         log.debug("[redis] Publish prebuild udpate invoked.");
-        this.metrics.reportUpdatePublished("prebuild");
+
+        let err: Error | undefined;
+        try {
+            const serialized = JSON.stringify(update);
+            await this.client.get().publish(PrebuildUpdatesChannel, serialized);
+            log.debug("[redis] Succesfully published prebuild update.", update);
+        } catch (e) {
+            err = e;
+            log.error("[redis] Failed to publish prebuild update.", e, update);
+        } finally {
+            this.metrics.reportUpdatePublished("prebuild", err);
+        }
     }
 
     async publishInstanceUpdate(update: RedisWorkspaceInstanceUpdate): Promise<void> {