[ws-manager-bridge] Make non-governing bridges distribute updates locally

geropl · geropl · commit a85ce6557b1b · 2022-01-10T16:07:31.000Z
diff --git a/components/ws-manager-bridge/ee/src/bridge.ts b/components/ws-manager-bridge/ee/src/bridge.ts
@@ -9,7 +9,7 @@ import { injectable } from "inversify";
 import { TraceContext } from "@gitpod/gitpod-protocol/lib/util/tracing";
 import { WorkspaceStatus, WorkspaceType, WorkspacePhase } from "@gitpod/ws-manager/lib";
 import { WorkspaceInstance } from "@gitpod/gitpod-protocol";
-import { log } from "@gitpod/gitpod-protocol/lib/util/logging";
+import { log, LogContext } from "@gitpod/gitpod-protocol/lib/util/logging";
 
 @injectable()
 export class WorkspaceManagerBridgeEE extends WorkspaceManagerBridge {
@@ -37,17 +37,16 @@ export class WorkspaceManagerBridgeEE extends WorkspaceManagerBridge {
         }
     }
 
-    protected async updatePrebuiltWorkspace(ctx: TraceContext, status: WorkspaceStatus.AsObject) {
+    protected async writePrebuiltUpdateToDB(ctx: TraceContext, userId: string, status: WorkspaceStatus.AsObject) {
         if (status.spec && status.spec.type != WorkspaceType.PREBUILD) {
             return;
         }
 
         const instanceId = status.id!;
         const workspaceId = status.metadata!.metaId!;
-        const userId = status.metadata!.owner!;
         const logCtx = { instanceId, workspaceId, userId };
 
-        const span = TraceContext.startSpan("updatePrebuiltWorkspace", ctx);
+        const span = TraceContext.startSpan("writePrebuiltUpdateToDB", ctx);
         try {
             const prebuild = await this.workspaceDB.trace({span}).findPrebuildByWorkspaceID(status.metadata!.metaId!);
             if (!prebuild) {
@@ -83,12 +82,35 @@ export class WorkspaceManagerBridgeEE extends WorkspaceManagerBridge {
                 }
                 await this.workspaceDB.trace({span}).storePrebuiltWorkspace(prebuild);
             }
+        } catch (e) {
+            TraceContext.setError({span}, e);
+            throw e;
+        } finally {
+            span.finish();
+        }
+    }
 
-            { // notify about prebuild updated
-                const info = (await this.workspaceDB.trace({span}).findPrebuildInfos([prebuild.id]))[0];
-                if (info) {
-                    this.messagebus.notifyOnPrebuildUpdate({ info, status: prebuild.state });
-                }
+    protected async notifyOnPrebuiltUpdate(ctx: TraceContext, userId: string, status: WorkspaceStatus.AsObject) {
+        if (status.spec && status.spec.type != WorkspaceType.PREBUILD) {
+            return;
+        }
+        const instanceId = status.id!;
+        const workspaceId = status.metadata!.metaId!;
+        const logCtx: LogContext = { instanceId, workspaceId, userId };
+
+        const span = TraceContext.startSpan("notifyOnPrebuiltUpdate", ctx);
+        try {
+            const prebuild = await this.workspaceDB.trace({span}).findPrebuildByWorkspaceID(status.metadata!.metaId!);
+            if (!prebuild) {
+                log.warn(logCtx, "headless workspace without prebuild");
+                TraceContext.setError({span}, new Error("headless workspace without prebuild"));
+                return
+            }
+
+            // notify about prebuild updated
+            const info = (await this.workspaceDB.trace({span}).findPrebuildInfos([prebuild.id]))[0];
+            if (info) {
+                this.messagebus.notifyOnPrebuildUpdate({ info, status: prebuild.state });
             }
         } catch (e) {
             TraceContext.setError({span}, e);
diff --git a/components/ws-manager-bridge/src/bridge.ts b/components/ws-manager-bridge/src/bridge.ts
@@ -59,22 +59,32 @@ export class WorkspaceManagerBridge implements Disposable {
     protected cluster: WorkspaceClusterInfo;
 
     public start(cluster: WorkspaceClusterInfo, clientProvider: ClientProvider) {
-        const logPayload = { name: cluster.name, url: cluster.url };
+        const logPayload = { name: cluster.name, url: cluster.url, govern: cluster.govern };
         log.info(`starting bridge to cluster...`, logPayload);
         this.cluster = cluster;
 
-        if (cluster.govern) {
-            log.debug(`starting DB updater: ${cluster.name}`, logPayload);
-            /* no await */ this.startDatabaseUpdater(clientProvider, logPayload)
+        const startStatusUpdateHandler = (writeToDB: boolean) => {
+            log.debug(`starting status update handler: ${cluster.name}`, logPayload);
+            /* no await */ this.startStatusUpdateHandler(clientProvider, writeToDB, logPayload)
                 // this is a mere safe-guard: we do not expect the code inside to fail
-                .catch(err => log.error("cannot start database updater", err));
+                .catch(err => log.error("cannot start status update handler", err));
+        };
 
+        if (cluster.govern) {
+            // notify servers and _update the DB_
+            startStatusUpdateHandler(true);
+
+            // the actual "governing" part
             const controllerInterval = this.config.controllerIntervalSeconds;
             if (controllerInterval <= 0) {
                 throw new Error("controllerInterval <= 0!");
             }
             log.debug(`starting controller: ${cluster.name}`, logPayload);
             this.startController(clientProvider, controllerInterval, this.config.controllerMaxDisconnectSeconds);
+        } else {
+            // _DO NOT_ update the DB (another bridge is responsible for that)
+            // Still, listen to all updates, generate/derive new state and distribute it locally!
+            startStatusUpdateHandler(false);
         }
         log.info(`started bridge to cluster.`, logPayload);
     }
@@ -83,15 +93,15 @@ export class WorkspaceManagerBridge implements Disposable {
         this.dispose();
     }
 
-    protected async startDatabaseUpdater(clientProvider: ClientProvider, logPayload: {}): Promise<void> {
+    protected async startStatusUpdateHandler(clientProvider: ClientProvider, writeToDB: boolean, logPayload: {}): Promise<void> {
         const subscriber = new WsmanSubscriber(clientProvider);
         this.disposables.push(subscriber);
 
         const onReconnect = (ctx: TraceContext, s: WorkspaceStatus[]) => {
-            s.forEach(sx => this.serializeMessagesByInstanceId<WorkspaceStatus>(ctx, sx, m => m.getId(), (ctx, msg) => this.handleStatusUpdate(ctx, msg)))
+            s.forEach(sx => this.serializeMessagesByInstanceId<WorkspaceStatus>(ctx, sx, m => m.getId(), (ctx, msg) => this.handleStatusUpdate(ctx, msg, writeToDB)))
         };
         const onStatusUpdate = (ctx: TraceContext, s: WorkspaceStatus) => {
-            this.serializeMessagesByInstanceId<WorkspaceStatus>(ctx, s, msg => msg.getId(), (ctx, s) => this.handleStatusUpdate(ctx, s))
+            this.serializeMessagesByInstanceId<WorkspaceStatus>(ctx, s, msg => msg.getId(), (ctx, s) => this.handleStatusUpdate(ctx, s, writeToDB))
         };
         await subscriber.subscribe({ onReconnect, onStatusUpdate }, logPayload);
     }
@@ -110,7 +120,7 @@ export class WorkspaceManagerBridge implements Disposable {
         this.queues.set(instanceId, q);
     }
 
-    protected async handleStatusUpdate(ctx: TraceContext, rawStatus: WorkspaceStatus) {
+    protected async handleStatusUpdate(ctx: TraceContext, rawStatus: WorkspaceStatus, writeToDB: boolean) {
         const status = rawStatus.toObject();
         if (!status.spec || !status.metadata || !status.conditions) {
             log.warn("Received invalid status update", status);
@@ -246,26 +256,40 @@ export class WorkspaceManagerBridge implements Disposable {
                     break;
             }
 
-            await this.updatePrebuiltWorkspace({span}, status);
-
             span.setTag("after", JSON.stringify(instance));
-            await this.workspaceDB.trace({span}).storeInstance(instance);
-            await this.messagebus.notifyOnInstanceUpdate({span}, userId, instance);
 
-            // important: call this after the DB update
-            await this.cleanupProbeWorkspace({span}, status);
-
-            if (!!lifecycleHandler) {
-                await lifecycleHandler();
+            // now notify all listeners about updates - and update DB if needed
+            if (writeToDB) {
+                await this.writePrebuiltUpdateToDB({span}, userId, status);
+                await this.writeStatusUpdateToDB({span}, userId, instance, status, lifecycleHandler);
             }
+            await this.notifyOnPrebuiltUpdate({span}, userId, status);
+            await this.notifyOnInstanceUpdate({span}, userId, instance);
+
         } catch (e) {
-            TraceContext.setError({ span }, e);
+            TraceContext.setError({span}, e);
             throw e;
         } finally {
             span.finish();
         }
     }
 
+    protected async writeStatusUpdateToDB(ctx: TraceContext, userId: string, instance: WorkspaceInstance, status: WorkspaceStatus.AsObject, lifecycleHandler: (() => Promise<void>) | undefined): Promise<void> {
+        await this.workspaceDB.trace(ctx).storeInstance(instance);
+
+        // cleanup
+        // important: call this after the DB update
+        await this.cleanupProbeWorkspace(ctx, status);
+
+        if (!!lifecycleHandler) {
+            await lifecycleHandler();
+        }
+    }
+
+    protected async notifyOnInstanceUpdate(ctx: TraceContext, userId: string, instance: WorkspaceInstance) {
+        await this.messagebus.notifyOnInstanceUpdate(ctx, userId, instance);
+    }
+
     protected startController(clientProvider: ClientProvider, controllerIntervalSeconds: number, controllerMaxDisconnectSeconds: number, maxTimeToRunningPhaseSeconds = 60 * 60) {
         let disconnectStarted = Number.MAX_SAFE_INTEGER;
         this.disposables.push(
@@ -321,7 +345,11 @@ export class WorkspaceManagerBridge implements Disposable {
         // probes are an EE feature - we just need the hook here
     }
 
-    protected async updatePrebuiltWorkspace(ctx: TraceContext, status: WorkspaceStatus.AsObject) {
+    protected async writePrebuiltUpdateToDB(ctx: TraceContext, userId: string, status: WorkspaceStatus.AsObject) {
+        // prebuilds are an EE feature - we just need the hook here
+    }
+
+    protected async notifyOnPrebuiltUpdate(ctx: TraceContext, userId: string, status: WorkspaceStatus.AsObject) {
         // prebuilds are an EE feature - we just need the hook here
     }
 
@@ -334,7 +362,7 @@ export class WorkspaceManagerBridge implements Disposable {
 
         try {
             await this.userDB.trace({span}).deleteGitpodTokensNamedLike(ownerUserID, `${instance.id}-%`);
-            await this.analytics.track({
+            this.analytics.track({
                 userId: ownerUserID,
                 event: "workspace-stopped",
                 messageId: `bridge-wsstopped-${instance.id}`,
