[server] Publish events in workspace starter - WEB-622 (gitpod-io#18234)

* [server] Publish events during workspace starter

* add copy comment

* retest

* fix

* fix

* fix

* fix

* fix

* retest

* retest

* try?

components/gitpod-protocol/src/redis.ts

@@ -19,6 +19,5 @@ export type RedisPrebuildUpdate = {
     status: PrebuiltWorkspaceState;
     prebuildID: string;
     workspaceID: string;
-    instanceID: string;
     projectID: string;
 };

components/server/BUILD.yaml

@@ -8,6 +8,7 @@ packages:
       - .eslintrc
       - package.json
       - mocha.opts
+      - ../../install/installer/pkg/components/spicedb/data/schema.yaml
     deps:
       - components/content-service-api/typescript:lib
       - components/gitpod-db:lib

components/server/src/container-module.ts

@@ -131,6 +131,7 @@ import { SpiceDBAuthorizer } from "./authorization/spicedb-authorizer";
 import { OrganizationService } from "./orgs/organization-service";
 import { RedisSubscriber } from "./messaging/redis-subscriber";
 import { Redis } from "ioredis";
+import { RedisPublisher } from "./redis/publisher";
 
 export const productionContainerModule = new ContainerModule(
     (bind, unbind, isBound, rebind, unbindAsync, onActivation, onDeactivation) => {
@@ -364,5 +365,6 @@ export const productionContainerModule = new ContainerModule(
 
         bind(RedisMutex).toSelf().inSingletonScope();
         bind(RedisSubscriber).toSelf().inSingletonScope();
+        bind(RedisPublisher).toSelf().inSingletonScope();
     },
 );

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

@@ -156,7 +156,7 @@ export class RedisSubscriber implements LocalMessageBroker {
             } catch (err) {
                 log.error(
                     "Failed to broadcast workspace instance update.",
-                    { projectId: update.projectID, instanceId: update.instanceID, workspaceId: update.workspaceID },
+                    { projectId: update.projectID, workspaceId: update.workspaceID },
                     err,
                 );
             }

components/server/src/prebuilds/prebuilt-status-maintainer.ts

@@ -20,6 +20,7 @@ import {
 import { TraceContext } from "@gitpod/gitpod-protocol/lib/util/tracing";
 import { LocalMessageBroker } from "../messaging/local-message-broker";
 import { repeat } from "@gitpod/gitpod-protocol/lib/util/repeat";
+import { RedisSubscriber } from "../messaging/redis-subscriber";
 
 export interface CheckRunInfo {
     owner: string;
@@ -41,20 +42,25 @@ export type AuthenticatedGithubProvider = (
 
 @injectable()
 export class PrebuildStatusMaintainer implements Disposable {
-    @inject(TracedWorkspaceDB) protected readonly workspaceDB: DBWithTracing<WorkspaceDB>;
-    @inject(LocalMessageBroker) protected readonly localMessageBroker: LocalMessageBroker;
+    constructor(
+        @inject(TracedWorkspaceDB) private readonly workspaceDB: DBWithTracing<WorkspaceDB>,
+        @inject(LocalMessageBroker) private readonly localMessageBroker: LocalMessageBroker,
+        @inject(RedisSubscriber) private readonly subscriber: RedisSubscriber,
+    ) {}
+
     protected githubApiProvider: AuthenticatedGithubProvider;
     protected readonly disposables = new DisposableCollection();
 
     start(githubApiProvider: AuthenticatedGithubProvider): void {
         // set github before registering the msgbus listener - otherwise an incoming message and the github set might race
         this.githubApiProvider = githubApiProvider;
 
-        this.disposables.push(
+        this.disposables.pushAll([
             this.localMessageBroker.listenForPrebuildUpdatableEvents((ctx, msg) =>
                 this.handlePrebuildFinished(ctx, msg),
             ),
-        );
+            this.subscriber.listenForPrebuildUpdatableEvents((ctx, msg) => this.handlePrebuildFinished(ctx, msg)),
+        ]);
         this.disposables.push(repeat(this.periodicUpdatableCheck.bind(this), MAX_UPDATABLE_AGE / 2));
         log.debug("prebuild updatable status maintainer started");
     }

components/server/src/prometheus-metrics.ts

@@ -34,6 +34,7 @@ export function registerServerMetrics(registry: prometheusClient.Registry) {
     registry.registerMetric(redisUpdatesReceived);
     registry.registerMetric(redisUpdatesCompletedTotal);
     registry.registerMetric(updateSubscribersRegistered);
+    registry.registerMetric(updatesPublishedTotal);
 }
 
 const loginCounter = new prometheusClient.Counter({
@@ -337,3 +338,13 @@ export const updateSubscribersRegistered = new prometheusClient.Gauge({
     help: "Gauge of subscribers registered",
     labelNames: ["type"],
 });
+
+export const updatesPublishedTotal = new prometheusClient.Counter({
+    name: "gitpod_server_updates_published_total",
+    help: "Counter of events published to Redis by type and error",
+    labelNames: ["type", "error"],
+});
+
+export function reportUpdatePublished(type: "workspace-instance" | "prebuild" | "headless", err?: Error): void {
+    updatesPublishedTotal.labels(type, err ? "true" : "false").inc();
+}

components/server/src/redis/publisher.ts

@@ -0,0 +1,59 @@
+/**
+ * Copyright (c) 2023 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 { inject, injectable } from "inversify";
+import { log } from "@gitpod/gitpod-protocol/lib/util/logging";
+import {
+    PrebuildUpdatesChannel,
+    RedisPrebuildUpdate,
+    RedisWorkspaceInstanceUpdate,
+    WorkspaceInstanceUpdatesChannel,
+} from "@gitpod/gitpod-protocol";
+import { Redis } from "ioredis";
+import { reportUpdatePublished } from "../prometheus-metrics";
+
+@injectable()
+// RedisPublisher is a copy from ws-manager-bridge/src/redis/publisher.go until we find a better
+// way to share the publisher across packages.
+// WEB-621
+export class RedisPublisher {
+    constructor(@inject(Redis) private readonly client: Redis) {}
+
+    async publishPrebuildUpdate(update: RedisPrebuildUpdate): Promise<void> {
+        log.debug("[redis] Publish prebuild udpate invoked.");
+
+        let err: Error | undefined;
+        try {
+            const serialized = JSON.stringify(update);
+            await this.client.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 {
+            reportUpdatePublished("prebuild", err);
+        }
+    }
+
+    async publishInstanceUpdate(update: RedisWorkspaceInstanceUpdate): Promise<void> {
+        let err: Error | undefined;
+        try {
+            const serialized = JSON.stringify(update);
+            await this.client.publish(WorkspaceInstanceUpdatesChannel, serialized);
+            log.debug("[redis] Succesfully published instance update.", update);
+        } catch (e) {
+            err = e;
+            log.error("[redis] Failed to publish instance update.", e, update);
+        } finally {
+            reportUpdatePublished("workspace-instance", err);
+        }
+    }
+
+    async publishHeadlessUpdate(): Promise<void> {
+        log.debug("[redis] Publish headless udpate invoked.");
+        reportUpdatePublished("headless");
+    }
+}

components/server/src/workspace/workspace-starter.ts

@@ -119,6 +119,7 @@ import { ResolvedEnvVars } from "./env-var-service";
 import { ImageSourceProvider } from "./image-source-provider";
 import { MessageBusIntegration } from "./messagebus-integration";
 import { WorkspaceClassesConfig } from "./workspace-classes";
+import { RedisPublisher } from "../redis/publisher";
 
 export interface StartWorkspaceOptions extends GitpodServer.StartWorkspaceOptions {
     rethrow?: boolean;
@@ -209,6 +210,7 @@ export class WorkspaceStarter {
         @inject(BlockedRepositoryDB) private readonly blockedRepositoryDB: BlockedRepositoryDB,
         @inject(EntitlementService) private readonly entitlementService: EntitlementService,
         @inject(RedisMutex) private readonly redisMutex: RedisMutex,
+        @inject(RedisPublisher) private readonly publisher: RedisPublisher,
     ) {}
     public async startWorkspace(
         ctx: TraceContext,
@@ -452,6 +454,11 @@ export class WorkspaceStarter {
                 });
                 await this.userDB.trace({ span }).deleteGitpodTokensNamedLike(workspace.ownerId, `${instance.id}-%`);
                 await this.messagebus.notifyOnInstanceUpdate(workspace.ownerId, updated);
+                await this.publisher.publishInstanceUpdate({
+                    instanceID: updated.id,
+                    ownerID: workspace.ownerId,
+                    workspaceID: workspace.id,
+                });
             }
             return;
         }
@@ -715,6 +722,11 @@ export class WorkspaceStarter {
                 await this.workspaceDb.trace(ctx).storeInstance(instance);
                 try {
                     await this.messagebus.notifyOnInstanceUpdate(workspace.ownerId, instance);
+                    await this.publisher.publishInstanceUpdate({
+                        instanceID: instance.id,
+                        ownerID: workspace.ownerId,
+                        workspaceID: workspace.id,
+                    });
                 } catch (err) {
                     // if sending the notification fails that's no reason to stop the workspace creation.
                     // If the dashboard misses this event it will catch up at the next one.
@@ -768,6 +780,12 @@ export class WorkspaceStarter {
                 const info = (await this.workspaceDb.trace({ span }).findPrebuildInfos([prebuild.id]))[0];
                 if (info) {
                     await this.messagebus.notifyOnPrebuildUpdate({ info, status: "queued" });
+                    await this.publisher.publishPrebuildUpdate({
+                        prebuildID: prebuild.id,
+                        projectID: info.projectId,
+                        status: "queued",
+                        workspaceID: workspaceId,
+                    });
                 }
             }
         } catch (e) {
@@ -796,6 +814,11 @@ export class WorkspaceStarter {
             instance.status.message = `Workspace cannot be started: ${err}`;
             await this.workspaceDb.trace({ span }).storeInstance(instance);
             await this.messagebus.notifyOnInstanceUpdate(workspace.ownerId, instance);
+            await this.publisher.publishInstanceUpdate({
+                instanceID: instance.id,
+                ownerID: workspace.ownerId,
+                workspaceID: workspace.id,
+            });
 
             // If we just attempted to start a workspace for a prebuild - and that failed, we have to fail the prebuild itself.
             await this.failPrebuildWorkspace({ span }, err, workspace);
@@ -825,6 +848,7 @@ export class WorkspaceStarter {
                         type: HeadlessWorkspaceEventType.Failed,
                         workspaceID: workspace.id, // required in prebuild-queue-maintainer.ts
                     });
+                    await this.publisher.publishHeadlessUpdate();
                 }
             }
         } catch (err) {
@@ -1209,6 +1233,11 @@ export class WorkspaceStarter {
                 .trace({ span })
                 .updateInstancePartial(instance.id, { workspaceImage, status });
             await this.messagebus.notifyOnInstanceUpdate(workspace.ownerId, instance);
+            await this.publisher.publishInstanceUpdate({
+                instanceID: instance.id,
+                ownerID: workspace.ownerId,
+                workspaceID: workspace.id,
+            });
 
             let buildResult: BuildResponse;
             try {
@@ -1283,6 +1312,11 @@ export class WorkspaceStarter {
 
             // Push updated workspace instance over messagebus
             await this.messagebus.notifyOnInstanceUpdate(workspace.ownerId, instance);
+            await this.publisher.publishInstanceUpdate({
+                workspaceID: workspace.ownerId,
+                instanceID: instance.id,
+                ownerID: workspace.ownerId,
+            });
 
             TraceContext.setError({ span }, err);
             const looksLikeUserError = (msg: string): boolean => {

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

@@ -96,7 +96,6 @@ export class PrebuildUpdater {
                 if (info) {
                     await this.messagebus.notifyOnPrebuildUpdate({ info, status: updatedPrebuild.state });
                     await this.publisher.publishPrebuildUpdate({
-                        instanceID: instanceId,
                         projectID: prebuild.projectId || "",
                         prebuildID: updatedPrebuild.id,
                         status: updatedPrebuild.state,
@@ -127,7 +126,6 @@ export class PrebuildUpdater {
                 if (info) {
                     await this.messagebus.notifyOnPrebuildUpdate({ info, status: prebuild.state });
                     await this.publisher.publishPrebuildUpdate({
-                        instanceID: instance.id,
                         projectID: prebuild.projectId || "",
                         prebuildID: prebuild.id,
                         status: prebuild.state,