Make sync toplevel service

Author: Furisto

components/gitpod-protocol/src/experiments/types.ts

@@ -6,6 +6,8 @@
 
 import { Team } from "../teams-projects-protocol";
 
+export const Client = Symbol("Client");
+
 // Attributes define attributes which can be used to segment audiences.
 // Set the attributes which you want to use to group audiences into.
 export interface Attributes {

components/ws-manager-bridge/src/cluster-service-server.ts

@@ -18,7 +18,6 @@ import {
     AdmissionConstraintHasUserLevel,
     AdmissionConstraintHasMoreResources,
 } from "@gitpod/gitpod-protocol/lib/workspace-cluster";
-import { AdmissionConstraintHasClass } from "@gitpod/gitpod-protocol/src/workspace-cluster";
 import {
     ClusterServiceService,
     ClusterState,
@@ -35,23 +34,19 @@ import {
     UpdateResponse,
     AdmissionConstraint as GRPCAdmissionConstraint,
 } from "@gitpod/ws-manager-bridge-api/lib";
-import {
-    DescribeClusterRequest,
-    DescribeClusterResponse,
-    GetWorkspacesRequest,
-    WorkspaceClass,
-} from "@gitpod/ws-manager/lib";
+import { GetWorkspacesRequest } from "@gitpod/ws-manager/lib";
 import { WorkspaceManagerClientProvider } from "@gitpod/ws-manager/lib/client-provider";
 import {
     WorkspaceManagerClientProviderCompositeSource,
     WorkspaceManagerClientProviderSource,
 } from "@gitpod/ws-manager/lib/client-provider-source";
 import * as grpc from "@grpc/grpc-js";
-import { ServiceError as grpcServiceError } from "@grpc/grpc-js";
 import { inject, injectable } from "inversify";
 import { BridgeController } from "./bridge-controller";
+import { getSupportedWorkspaceClasses } from "./cluster-sync-service";
 import { Configuration } from "./config";
 import { getExperimentsClientForBackend } from "@gitpod/gitpod-protocol/lib/experiments/configcat-server";
+import { GRPCError } from "./rpc";
 
 export interface ClusterServiceServerOptions {
     port: number;
@@ -396,10 +391,6 @@ function mapAdmissionConstraint(c: GRPCAdmissionConstraint | undefined): Admissi
     return;
 }
 
-function mapWorkspaceClass(c: WorkspaceClass): AdmissionConstraintHasClass {
-    return <AdmissionConstraintHasClass>{ type: "has-class", id: c.getId(), displayName: c.getDisplayname() };
-}
-
 function mapPreferabilityToScore(p: Preferability): number | undefined {
     switch (p) {
         case Preferability.PREFER:
@@ -450,66 +441,6 @@ function getClientInfo(call: grpc.ServerUnaryCall<any, any>) {
     return { clientIP, clientName, userAgent };
 }
 
-@injectable()
-export class ClusterSyncService {
-    @inject(Configuration)
-    protected readonly config: Configuration;
-
-    @inject(WorkspaceClusterDB)
-    protected readonly clusterDB: WorkspaceClusterDB;
-
-    @inject(WorkspaceManagerClientProvider)
-    protected readonly clientProvider: WorkspaceManagerClientProvider;
-
-    protected timer: NodeJS.Timer;
-
-    public async start() {
-        const enabled = await getExperimentsClientForBackend().getValueAsync("workspace_classes_backend", false, {});
-        if (enabled) {
-            this.timer = setInterval(() => this.reconcile(), this.config.clusterSyncIntervalSeconds * 1000);
-        }
-    }
-
-    private async reconcile() {
-        log.debug("reconciling workspace classes...");
-        let allClusters = await this.clusterDB.findAll();
-        for (const cluster of allClusters) {
-            try {
-                let supportedClasses = await getSupportedWorkspaceClasses(this.clientProvider, cluster);
-                let existingOtherConstraints = cluster.admissionConstraints?.filter((c) => c.type !== "has-class");
-                cluster.admissionConstraints = existingOtherConstraints?.concat(supportedClasses);
-                await this.clusterDB.save(cluster);
-            } catch (err) {
-                log.error("failed to reconcile workspace classes for cluster", err, { cluster: cluster.name });
-            }
-        }
-        log.debug("done reconciling workspace classes");
-    }
-
-    public async stop() {
-        const enabled = await getExperimentsClientForBackend().getValueAsync("workspace_classes_backend", false, {});
-        if (enabled) {
-            clearInterval(this.timer);
-        }
-    }
-}
-
-async function getSupportedWorkspaceClasses(clientProvider: WorkspaceManagerClientProvider, cluster: WorkspaceCluster) {
-    let constraints = await new Promise<AdmissionConstraintHasClass[]>((resolve, reject) => {
-        const c = clientProvider.createClient(cluster);
-        c.describeCluster(new DescribeClusterRequest(), (err: any, resp: DescribeClusterResponse) => {
-            if (err) {
-                reject(new GRPCError(grpc.status.FAILED_PRECONDITION, `cannot reach ${cluster.url}: ${err.message}`));
-            } else {
-                let classes = resp.getWorkspaceclassesList().map((cl) => mapWorkspaceClass(cl));
-                resolve(classes);
-            }
-        });
-    });
-
-    return constraints;
-}
-
 // "grpc" does not allow additional methods on it's "ServiceServer"s so we have an additional wrapper here
 @injectable()
 export class ClusterServiceServer {
@@ -519,9 +450,6 @@ export class ClusterServiceServer {
     @inject(ClusterService)
     protected readonly service: ClusterService;
 
-    @inject(ClusterSyncService)
-    protected readonly sync: ClusterSyncService;
-
     protected server: grpc.Server | undefined = undefined;
 
     public async start() {
@@ -543,7 +471,6 @@ export class ClusterServiceServer {
             log.info(`gRPC server listening on: ${bindTo}`);
             server.start();
         });
-        this.sync.start();
     }
 
     public async stop() {
@@ -554,30 +481,5 @@ export class ClusterServiceServer {
             });
             this.server = undefined;
         }
-        this.sync.stop();
-    }
-}
-
-class GRPCError extends Error implements Partial<grpcServiceError> {
-    public name = "ServiceError";
-
-    details: string;
-
-    constructor(public readonly status: grpc.status, err: any) {
-        super(GRPCError.errToMessage(err));
-
-        this.details = this.message;
-    }
-
-    static errToMessage(err: any): string | undefined {
-        if (typeof err === "string") {
-            return err;
-        } else if (typeof err === "object") {
-            return err.message;
-        }
-    }
-
-    static isGRPCError(obj: any): obj is GRPCError {
-        return obj !== undefined && typeof obj === "object" && "status" in obj;
     }
 }

components/ws-manager-bridge/src/cluster-sync-service.ts

@@ -0,0 +1,89 @@
+/**
+ * Copyright (c) 2021 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 { log } from "@gitpod/gitpod-protocol/lib/util/logging";
+import { WorkspaceCluster, WorkspaceClusterDB } from "@gitpod/gitpod-protocol/lib/workspace-cluster";
+import { WorkspaceManagerClientProvider } from "@gitpod/ws-manager/lib/client-provider";
+import { inject, injectable } from "inversify";
+import { Configuration } from "./config";
+import { Client } from "@gitpod/gitpod-protocol/lib/experiments/types";
+import { DescribeClusterRequest, DescribeClusterResponse, WorkspaceClass } from "@gitpod/ws-manager/lib";
+import { AdmissionConstraintHasClass } from "@gitpod/gitpod-protocol/src/workspace-cluster";
+import { GRPCError } from "./rpc";
+import * as grpc from "@grpc/grpc-js";
+import { defaultGRPCOptions } from "@gitpod/gitpod-protocol/lib/util/grpc";
+
+@injectable()
+export class ClusterSyncService {
+    @inject(Configuration)
+    protected readonly config: Configuration;
+
+    @inject(WorkspaceClusterDB)
+    protected readonly clusterDB: WorkspaceClusterDB;
+
+    @inject(WorkspaceManagerClientProvider)
+    protected readonly clientProvider: WorkspaceManagerClientProvider;
+
+    @inject(Client)
+    protected readonly featureClient: Client;
+
+    protected timer: NodeJS.Timer;
+
+    public start() {
+        this.timer = setInterval(() => this.reconcile(), this.config.clusterSyncIntervalSeconds * 1000);
+    }
+
+    private async reconcile() {
+        const enabled = await this.featureClient.getValueAsync("workspace_classes_backend", false, {});
+        if (!enabled) {
+            return;
+        }
+
+        log.debug("reconciling workspace classes...");
+        let allClusters = await this.clusterDB.findFiltered({});
+        for (const cluster of allClusters) {
+            try {
+                let supportedClasses = await getSupportedWorkspaceClasses(this.clientProvider, cluster);
+                let existingOtherConstraints = cluster.admissionConstraints?.filter((c) => c.type !== "has-class");
+                cluster.admissionConstraints = existingOtherConstraints?.concat(supportedClasses);
+                await this.clusterDB.save(cluster);
+            } catch (err) {
+                log.error("failed to reconcile workspace classes for cluster", err, { cluster: cluster.name });
+            }
+        }
+        log.debug("done reconciling workspace classes");
+    }
+
+    public stop() {
+        clearInterval(this.timer);
+    }
+}
+
+export async function getSupportedWorkspaceClasses(
+    clientProvider: WorkspaceManagerClientProvider,
+    cluster: WorkspaceCluster,
+) {
+    let constraints = await new Promise<AdmissionConstraintHasClass[]>(async (resolve, reject) => {
+        const grpcOptions: grpc.ClientOptions = {
+            ...defaultGRPCOptions,
+        };
+        const c = await clientProvider.get(cluster.name, grpcOptions);
+        c.client.describeCluster(new DescribeClusterRequest(), (err: any, resp: DescribeClusterResponse) => {
+            if (err) {
+                reject(new GRPCError(grpc.status.FAILED_PRECONDITION, `cannot reach ${cluster.url}: ${err.message}`));
+            } else {
+                let classes = resp.getWorkspaceclassesList().map((cl) => mapWorkspaceClass(cl));
+                resolve(classes);
+            }
+        });
+    });
+
+    return constraints;
+}
+
+function mapWorkspaceClass(c: WorkspaceClass): AdmissionConstraintHasClass {
+    return <AdmissionConstraintHasClass>{ type: "has-class", id: c.getId(), displayName: c.getDisplayname() };
+}

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

@@ -26,7 +26,7 @@ import {
     WorkspaceManagerClientProviderDBSource,
     WorkspaceManagerClientProviderSource,
 } from "@gitpod/ws-manager/lib/client-provider-source";
-import { ClusterService, ClusterServiceServer, ClusterSyncService } from "./cluster-service-server";
+import { ClusterService, ClusterServiceServer } from "./cluster-service-server";
 import { IAnalyticsWriter } from "@gitpod/gitpod-protocol/lib/analytics";
 import { newAnalyticsWriterFromEnv } from "@gitpod/gitpod-protocol/lib/util/analytics";
 import { MetaInstanceController } from "./meta-instance-controller";
@@ -36,6 +36,9 @@ import { PreparingUpdateEmulator, PreparingUpdateEmulatorFactory } from "./prepa
 import { PrebuildStateMapper } from "./prebuild-state-mapper";
 import { PrebuildUpdater, PrebuildUpdaterNoOp } from "./prebuild-updater";
 import { DebugApp } from "@gitpod/gitpod-protocol/lib/util/debug-app";
+import { Client } from "@gitpod/gitpod-protocol/lib/experiments/types";
+import { getExperimentsClientForBackend } from "@gitpod/gitpod-protocol/lib/experiments/configcat-server";
+import { ClusterSyncService } from "./cluster-sync-service";
 
 export const containerModule = new ContainerModule((bind) => {
     bind(MessagebusConfiguration).toSelf().inSingletonScope();
@@ -89,4 +92,6 @@ export const containerModule = new ContainerModule((bind) => {
     bind(PrebuildUpdater).to(PrebuildUpdaterNoOp).inSingletonScope();
 
     bind(DebugApp).toSelf().inSingletonScope();
+
+    bind(Client).toDynamicValue(getExperimentsClientForBackend).inSingletonScope();
 });

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

@@ -15,6 +15,7 @@ import { TracingManager } from "@gitpod/gitpod-protocol/lib/util/tracing";
 import { ClusterServiceServer } from "./cluster-service-server";
 import { BridgeController } from "./bridge-controller";
 import { MetaInstanceController } from "./meta-instance-controller";
+import { ClusterSyncService } from "./cluster-sync-service";
 
 log.enableJSONLogging("ws-manager-bridge", undefined, LogrusLogLevel.getFromEnv());
 
@@ -52,6 +53,10 @@ export const start = async (container: Container) => {
         const metaInstanceController = container.get<MetaInstanceController>(MetaInstanceController);
         metaInstanceController.start();
 
+        log.info("starting sync service");
+        const clusterSyncService = container.get<ClusterSyncService>(ClusterSyncService);
+        clusterSyncService.start();
+
         process.on("SIGTERM", async () => {
             log.info("SIGTERM received, stopping");
             bridgeController.dispose();

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

@@ -0,0 +1,32 @@
+/**
+ * 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 * as grpc from "@grpc/grpc-js";
+import { ServiceError as grpcServiceError } from "@grpc/grpc-js";
+
+export class GRPCError extends Error implements Partial<grpcServiceError> {
+    public name = "ServiceError";
+
+    details: string;
+
+    constructor(public readonly status: grpc.status, err: any) {
+        super(GRPCError.errToMessage(err));
+
+        this.details = this.message;
+    }
+
+    static errToMessage(err: any): string | undefined {
+        if (typeof err === "string") {
+            return err;
+        } else if (typeof err === "object") {
+            return err.message;
+        }
+    }
+
+    static isGRPCError(obj: any): obj is GRPCError {
+        return obj !== undefined && typeof obj === "object" && "status" in obj;
+    }
+}