[ws-manager-bridge] Emulate InstanceUpdates for 'Preparing' phase to decouple messagebus

Author: geropl

chart/templates/ws-manager-bridge-configmap.yaml

@@ -31,6 +31,7 @@ data:
           "stoppingPhaseSeconds": 3600,
           "unknownPhaseSeconds": 600
         },
+        "emulatePreparingIntervalSeconds": "10",
         "staticBridges": {{ index (include "ws-manager-list" (dict "root" . "gp" $.Values "comp" .Values.components.server) | fromYaml) "manager" | default list | toJson }}
     }
 {{- end -}}

components/gitpod-db/src/typeorm/entity/db-workspace-instance.ts

@@ -13,6 +13,7 @@ import { Transformer } from "../transformer";
 
 @Entity()
 @Index("ind_find_wsi_ws_in_period", ['workspaceId', 'startedTime', 'stoppedTime'])   // findInstancesWithWorkspaceInPeriod
+@Index("ind_phasePersisted_region", ['phasePersisted', 'region'])   // findInstancesByPhaseAndRegion
 // on DB but not Typeorm: @Index("ind_lastModified", ["_lastModified"])   // DBSync
 export class DBWorkspaceInstance implements WorkspaceInstance {
     @PrimaryColumn(TypeORM.UUID_COLUMN_TYPE)

components/gitpod-db/src/typeorm/migration/1645019483643-InstancesByPhaseAndRegion.ts

@@ -0,0 +1,24 @@
+/**
+ * Copyright (c) 2022 Gitpod GmbH. All rights reserved.
+ * Licensed under the GNU Affero General Public License (AGPL).
+ * See License-AGPL.txt in the project root for license information.
+ */
+
+import {MigrationInterface, QueryRunner} from "typeorm";
+import { indexExists } from "./helper/helper";
+
+const TABLE_NAME = "d_b_workspace_instance";
+const INDEX_NAME = "ind_phasePersisted_region";
+
+export class InstancesByPhaseAndRegion1645019483643 implements MigrationInterface {
+
+    public async up(queryRunner: QueryRunner): Promise<void> {
+        if(!(await indexExists(queryRunner, TABLE_NAME, INDEX_NAME))) {
+            await queryRunner.query(`CREATE INDEX ${INDEX_NAME} ON ${TABLE_NAME} (phasePersisted, region)`);
+        }
+    }
+
+    public async down(queryRunner: QueryRunner): Promise<void> {
+    }
+
+}

components/gitpod-db/src/typeorm/workspace-db-impl.ts

@@ -852,6 +852,15 @@ export abstract class AbstractTypeORMWorkspaceDBImpl implements WorkspaceDB {
         return <WorkspaceAndInstance>(res);
     }
 
+    async findInstancesByPhaseAndRegion(phase: string, region: string): Promise<WorkspaceInstance[]> {
+        const repo = await this.getWorkspaceInstanceRepo();
+        // uses index: ind_phasePersisted_region
+        const qb = repo.createQueryBuilder("wsi")
+            .where("wsi.phasePersisted = :phase", { phase })
+            .andWhere("wsi.region = :region", { region });
+        return qb.getMany();
+    }
+
     async findPrebuiltWorkspacesByProject(projectId: string, branch?: string, limit?: number): Promise<PrebuiltWorkspace[]> {
         const repo = await this.getPrebuiltWorkspaceRepo();
 

components/gitpod-db/src/workspace-db.ts

@@ -83,6 +83,7 @@ export interface WorkspaceDB {
     findAllWorkspaces(offset: number, limit: number, orderBy: keyof Workspace, orderDir: "ASC" | "DESC", ownerId?: string, searchTerm?: string, minCreationTime?: Date, maxCreationDateTime?: Date, type?: WorkspaceType): Promise<{ total: number, rows: Workspace[] }>;
     findAllWorkspaceAndInstances(offset: number, limit: number, orderBy: keyof WorkspaceAndInstance, orderDir: "ASC" | "DESC", query?: AdminGetWorkspacesQuery, searchTerm?: string): Promise<{ total: number, rows: WorkspaceAndInstance[] }>;
     findWorkspaceAndInstance(id: string): Promise<WorkspaceAndInstance | undefined>;
+    findInstancesByPhaseAndRegion(phase: string, region: string): Promise<WorkspaceInstance[]>;
 
     findAllWorkspaceInstances(offset: number, limit: number, orderBy: keyof WorkspaceInstance, orderDir: "ASC" | "DESC", ownerId?: string, minCreationTime?: Date, maxCreationTime?: Date, onlyRunning?: boolean, type?: WorkspaceType): Promise<{ total: number, rows: WorkspaceInstance[] }>;
 

components/installer/pkg/components/ws-manager-bridge/configmap.go

@@ -30,7 +30,8 @@ func configmap(ctx *common.RenderContext) ([]runtime.Object, error) {
 			StoppingPhaseSeconds:             3600,
 			UnknownPhaseSeconds:              600,
 		},
-		StaticBridges: WSManagerList(),
+		EmulatePreparingIntervalSeconds: 10,
+		StaticBridges:                   WSManagerList(),
 	}
 
 	fc, err := common.ToJSONString(wsmbcfg)

components/installer/pkg/components/ws-manager-bridge/types.go

@@ -13,6 +13,7 @@ type Configuration struct {
 	ControllerIntervalSeconds           int32              `json:"controllerIntervalSeconds"`
 	ControllerMaxDisconnectSeconds      int32              `json:"controllerMaxDisconnectSeconds"`
 	MaxTimeToRunningPhaseSeconds        int32              `json:"maxTimeToRunningPhaseSeconds"`
+	EmulatePreparingIntervalSeconds     int32              `json:"emulatePreparingIntervalSeconds"`
 	Timeouts                            Timeouts           `json:"timeouts"`
 }
 

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

@@ -6,7 +6,7 @@
 
 import { inject, injectable } from "inversify";
 import { MessageBusIntegration } from "./messagebus-integration";
-import { Disposable, WorkspaceInstance, Queue, WorkspaceInstancePort, PortVisibility, RunningWorkspaceInfo } from "@gitpod/gitpod-protocol";
+import { Disposable, WorkspaceInstance, Queue, WorkspaceInstancePort, PortVisibility, RunningWorkspaceInfo, DisposableCollection } from "@gitpod/gitpod-protocol";
 import { WorkspaceStatus, WorkspacePhase, GetWorkspacesRequest, WorkspaceConditionBool, PortVisibility as WsManPortVisibility, WorkspaceType, PromisifiedWorkspaceManagerClient } from "@gitpod/ws-manager/lib";
 import { WorkspaceDB } from "@gitpod/gitpod-db/lib/workspace-db";
 import { UserDB } from "@gitpod/gitpod-db/lib/user-db";
@@ -20,6 +20,7 @@ import { Timestamp } from "google-protobuf/google/protobuf/timestamp_pb";
 import { Configuration } from "./config";
 import { WorkspaceCluster } from "@gitpod/gitpod-protocol/lib/workspace-cluster";
 import { repeat } from "@gitpod/gitpod-protocol/lib/util/repeat";
+import { PreparingUpdateEmulator } from "./preparing-update-emulator";
 
 export const WorkspaceManagerBridgeFactory = Symbol("WorkspaceManagerBridgeFactory");
 
@@ -53,7 +54,7 @@ export class WorkspaceManagerBridge implements Disposable {
     @inject(IAnalyticsWriter)
     protected readonly analytics: IAnalyticsWriter;
 
-    protected readonly disposables: Disposable[] = [];
+    protected readonly disposables = new DisposableCollection();
     protected readonly queues = new Map<string, Queue>();
 
     protected cluster: WorkspaceClusterInfo;
@@ -85,6 +86,11 @@ export class WorkspaceManagerBridge implements Disposable {
             // _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);
+
+            // emulate WorkspaceInstance updates for all Workspaces in the "preparing" phase in this cluster
+            const updateEmulator = new PreparingUpdateEmulator();
+            this.disposables.push(updateEmulator);
+            updateEmulator.start(cluster.name);
         }
         log.info(`started bridge to cluster.`, logPayload);
     }
@@ -366,7 +372,7 @@ export class WorkspaceManagerBridge implements Disposable {
     }
 
     public dispose() {
-        this.disposables.forEach(d => d.dispose());
+        this.disposables.dispose();
     }
 
 }

components/ws-manager-bridge/src/config.ts

@@ -36,4 +36,7 @@ export interface Configuration {
         stoppingPhaseSeconds: number;
         unknownPhaseSeconds: number;
     }
+
+    // emulatePreparingIntervalSeconds configures how often we check for Workspaces in phase "preparing" for clusters we do not govern
+    emulatePreparingIntervalSeconds: number;
 }

components/ws-manager-bridge/src/container-module.ts

@@ -25,6 +25,7 @@ import { newAnalyticsWriterFromEnv } from '@gitpod/gitpod-protocol/lib/util/anal
 import { MetaInstanceController } from './meta-instance-controller';
 import { IClientCallMetrics } from '@gitpod/content-service/lib/client-call-metrics';
 import { PrometheusClientCallMetrics } from "@gitpod/gitpod-protocol/lib/messaging/client-call-metrics";
+import { PreparingUpdateEmulator } from './preparing-update-emulator';
 
 export const containerModule = new ContainerModule(bind => {
 
@@ -68,4 +69,6 @@ export const containerModule = new ContainerModule(bind => {
     }).inSingletonScope();
 
     bind(IAnalyticsWriter).toDynamicValue(newAnalyticsWriterFromEnv).inSingletonScope();
+
+    bind(PreparingUpdateEmulator).toSelf().inSingletonScope();
 });

components/ws-manager-bridge/src/preparing-update-emulator.ts

@@ -0,0 +1,89 @@
+/**
+ * Copyright (c) 2020 Gitpod GmbH. All rights reserved.
+ * Licensed under the GNU Affero General Public License (AGPL).
+ * See License-AGPL.txt in the project root for license information.
+ */
+import { WorkspaceDB } from "@gitpod/gitpod-db/lib/workspace-db";
+import { Disposable, DisposableCollection, WorkspaceInstance } from "@gitpod/gitpod-protocol";
+import { log } from "@gitpod/gitpod-protocol/lib/util/logging";
+import { repeat } from "@gitpod/gitpod-protocol/lib/util/repeat";
+import { TraceContext } from "@gitpod/gitpod-protocol/lib/util/tracing";
+import { inject, injectable } from "inversify";
+import { Configuration } from "./config";
+import { MessageBusIntegration } from "./messagebus-integration";
+import { GarbageCollectedCache } from "@gitpod/gitpod-protocol/lib/util/garbage-collected-cache";
+import * as crypto from 'crypto';
+
+interface CacheEntry {
+    instance: WorkspaceInstance,
+    userId: string,
+    hash: string,
+}
+
+/**
+ * The purpose of this class is to emulate WorkspaceInstance updates for workspaces instances that are not governed by this bridge.
+ * It does so by polling the DB for the specific region, and if anything changed, push that update into the local messagebus.
+ * This is a work-around to enable decoupling cross-cluster messagebus instances from each other.
+ */
+@injectable()
+export class PreparingUpdateEmulator implements Disposable {
+
+    @inject(Configuration) protected readonly config: Configuration;
+    @inject(WorkspaceDB) protected readonly workspaceDb: WorkspaceDB;
+    @inject(MessageBusIntegration) protected readonly messagebus: MessageBusIntegration;
+
+
+    protected readonly cachedResponses = new GarbageCollectedCache<CacheEntry>(600, 150);
+    protected readonly disposables = new DisposableCollection();
+
+    start(region: string) {
+        this.disposables.push(
+            repeat(async () => {
+                const span = TraceContext.startSpan("preparingUpdateEmulatorRun");
+                const ctx = {span};
+                try {
+                    const instances = await this.workspaceDb.findInstancesByPhaseAndRegion("preparing", region);
+                    span.setTag("preparingUpdateEmulatorRun.nrOfInstances", instances.length);
+                    for (const instance of instances) {
+                        const hash = hasher(instance);
+                        const entry = this.cachedResponses.get(instance.id);
+                        if (entry && entry.hash === hash) {
+                            continue;
+                        }
+
+                        let userId = entry?.userId;
+                        if (!userId) {
+                            const ws = await this.workspaceDb.findById(instance.workspaceId);
+                            if (!ws) {
+                                log.debug({ instanceId: instance.id, workspaceId: instance.workspaceId }, "no workspace found for workspace instance");
+                                continue;
+                            }
+                            userId = ws.ownerId;
+                        }
+
+                        await this.messagebus.notifyOnInstanceUpdate(ctx, userId, instance);
+                        this.cachedResponses.set(instance.id, {
+                            instance,
+                            hash,
+                            userId,
+                        });
+                    }
+                } catch (err) {
+                    TraceContext.setError(ctx, err);
+                } finally {
+                    span.finish();
+                }
+            }, this.config.emulatePreparingIntervalSeconds * 1000)
+        );
+    }
+
+    dispose() {
+        this.disposables.dispose();
+    }
+}
+
+function hasher(o: {}): string {
+    return crypto.createHash('md5')
+        .update(JSON.stringify(o))
+        .digest('hex');
+}