[protocol, et al] Extract gRPC ClientCallMetrics into protocol

geropl · geropl · commit cbfdd789c4bb · 2022-07-14T09:28:39.000Z
diff --git a/components/content-service-api/typescript/BUILD.yaml b/components/content-service-api/typescript/BUILD.yaml
@@ -1,6 +1,8 @@
 packages:
   - name: lib
     type: yarn
+    deps:
+      - components/gitpod-protocol:lib
     srcs:
       - "src/*.ts"
       - "src/*.js"
diff --git a/components/content-service-api/typescript/package.json b/components/content-service-api/typescript/package.json
@@ -11,6 +11,7 @@
     "lib"
   ],
   "dependencies": {
+    "@gitpod/gitpod-protocol": "0.1.5",
     "@grpc/grpc-js": "^1.3.7",
     "google-protobuf": "^3.19.1",
     "inversify": "^5.0.1",
diff --git a/components/content-service-api/typescript/src/client-call-metrics.ts b/components/content-service-api/typescript/src/client-call-metrics.ts
diff --git a/components/content-service-api/typescript/src/sugar.ts b/components/content-service-api/typescript/src/sugar.ts
@@ -6,7 +6,7 @@
 
 import { inject, injectable, interfaces, optional } from "inversify";
 import * as grpc from "@grpc/grpc-js";
-import { createClientCallMetricsInterceptor, IClientCallMetrics } from "./client-call-metrics";
+import { createClientCallMetricsInterceptor, IClientCallMetrics } from "@gitpod/gitpod-protocol/lib/util/grpc";
 import { IDEPluginServiceClient } from "./ideplugin_grpc_pb";
 import { ContentServiceClient } from "./content_grpc_pb";
 import { BlobServiceClient } from "./blobs_grpc_pb";
@@ -16,7 +16,10 @@ import { HeadlessLogServiceClient } from "./headless-log_grpc_pb";
 export const ContentServiceClientConfig = Symbol("ContentServiceClientConfig");
 export const ContentServiceClientCallMetrics = Symbol("ContentServiceClientCallMetrics");
 
-export const contentServiceBinder = (config: (ctx: interfaces.Context) => ContentServiceClientConfig, clientCallMetrics?: IClientCallMetrics): interfaces.ContainerModuleCallBack => {
+export const contentServiceBinder = (
+    config: (ctx: interfaces.Context) => ContentServiceClientConfig,
+    clientCallMetrics?: IClientCallMetrics,
+): interfaces.ContainerModuleCallBack => {
     return (bind, unbind, isBound, rebind) => {
         bind(ContentServiceClientConfig).toDynamicValue(config).inSingletonScope();
         if (clientCallMetrics) {
@@ -50,7 +53,8 @@ export interface ContentServiceClientProvider<T> {
 abstract class CachingClientProvider<T> implements ContentServiceClientProvider<T> {
     @inject(ContentServiceClientConfig) protected readonly clientConfig: ContentServiceClientConfig;
 
-    @inject(ContentServiceClientCallMetrics) @optional()
+    @inject(ContentServiceClientCallMetrics)
+    @optional()
     protected readonly clientCallMetrics: IClientCallMetrics;
 
     protected readonly interceptors: grpc.Interceptor[] = [];
@@ -59,9 +63,7 @@ abstract class CachingClientProvider<T> implements ContentServiceClientProvider<
     // Thus it makes sense to cache them rather than create a new connection for each request.
     protected client: Client<T> | undefined;
 
-    constructor(
-        protected readonly createClient: (config: ContentServiceClientConfig) => Client<T>,
-    ) {
+    constructor(protected readonly createClient: (config: ContentServiceClientConfig) => Client<T>) {
         if (this.clientCallMetrics) {
             this.interceptors.push(createClientCallMetricsInterceptor(this.clientCallMetrics));
         }
@@ -89,8 +91,8 @@ abstract class CachingClientProvider<T> implements ContentServiceClientProvider<
                 options: {
                     ...(config.options || {}),
                     interceptors: [...(config.options?.interceptors || []), ...this.interceptors],
-                }
-            }
+                },
+            };
         }
         return config;
     }
@@ -101,7 +103,7 @@ export class CachingContentServiceClientProvider extends CachingClientProvider<C
     constructor() {
         super((config) => {
             return new ContentServiceClient(config.address, config.credentials, config.options);
-        })
+        });
     }
 }
 
@@ -110,7 +112,7 @@ export class CachingBlobServiceClientProvider extends CachingClientProvider<Blob
     constructor() {
         super((config) => {
             return new BlobServiceClient(config.address, config.credentials, config.options);
-        })
+        });
     }
 }
 
@@ -119,7 +121,7 @@ export class CachingWorkspaceServiceClientProvider extends CachingClientProvider
     constructor() {
         super((config) => {
             return new WorkspaceServiceClient(config.address, config.credentials, config.options);
-        })
+        });
     }
 }
 
@@ -128,7 +130,7 @@ export class CachingIDEPluginClientProvider extends CachingClientProvider<IDEPlu
     constructor() {
         super((config) => {
             return new IDEPluginServiceClient(config.address, config.credentials, config.options);
-        })
+        });
     }
 }
 
@@ -137,11 +139,15 @@ export class CachingHeadlessLogServiceClientProvider extends CachingClientProvid
     constructor() {
         super((config) => {
             return new HeadlessLogServiceClient(config.address, config.credentials, config.options);
-        })
+        });
     }
 }
 
 function isConnectionAlive(client: grpc.Client) {
     const cs = client.getChannel().getConnectivityState(false);
-    return cs == grpc.connectivityState.CONNECTING || cs == grpc.connectivityState.IDLE || cs == grpc.connectivityState.READY;
+    return (
+        cs == grpc.connectivityState.CONNECTING ||
+        cs == grpc.connectivityState.IDLE ||
+        cs == grpc.connectivityState.READY
+    );
 }
diff --git a/components/gitpod-protocol/package.json b/components/gitpod-protocol/package.json
@@ -14,6 +14,7 @@
         "@types/chai-subset": "^1.3.3",
         "@types/cookie": "^0.4.1",
         "@types/express": "^4.17.13",
+        "@grpc/grpc-js": "^1.3.7",
         "@types/jaeger-client": "^3.18.3",
         "@types/js-yaml": "^3.10.1",
         "@types/mocha": "^5.2.7",
diff --git a/components/gitpod-protocol/src/messaging/client-call-metrics.ts b/components/gitpod-protocol/src/messaging/client-call-metrics.ts
@@ -6,26 +6,7 @@
 
 import { injectable } from "inversify";
 import * as prometheusClient from "prom-client";
-
-type GrpcMethodType = "unary" | "client_stream" | "server_stream" | "bidi_stream";
-export interface IGrpcCallMetricsLabels {
-    service: string;
-    method: string;
-    type: GrpcMethodType;
-}
-
-export interface IGrpcCallMetricsLabelsWithCode extends IGrpcCallMetricsLabels {
-    code: string;
-}
-
-export const IClientCallMetrics = Symbol("IClientCallMetrics");
-
-export interface IClientCallMetrics {
-    started(labels: IGrpcCallMetricsLabels): void;
-    sent(labels: IGrpcCallMetricsLabels): void;
-    received(labels: IGrpcCallMetricsLabels): void;
-    handled(labels: IGrpcCallMetricsLabelsWithCode): void;
-}
+import { IClientCallMetrics, IGrpcCallMetricsLabels, IGrpcCallMetricsLabelsWithCode } from "../util/grpc";
 
 @injectable()
 export class PrometheusClientCallMetrics implements IClientCallMetrics {
diff --git a/components/gitpod-protocol/src/util/grpc.ts b/components/gitpod-protocol/src/util/grpc.ts
@@ -4,6 +4,9 @@
  * See License-AGPL.txt in the project root for license information.
  */
 
+import * as grpc from "@grpc/grpc-js";
+import { Status } from "@grpc/grpc-js/build/src/constants";
+
 export const defaultGRPCOptions = {
     "grpc.keepalive_timeout_ms": 10000,
     "grpc.keepalive_time_ms": 60000,
@@ -13,3 +16,89 @@ export const defaultGRPCOptions = {
     "grpc.max_reconnect_backoff_ms": 5000,
     "grpc.max_receive_message_length": 1024 * 1024 * 16,
 };
+
+export type GrpcMethodType = "unary" | "client_stream" | "server_stream" | "bidi_stream";
+
+export interface IGrpcCallMetricsLabels {
+    service: string;
+    method: string;
+    type: GrpcMethodType;
+}
+
+export interface IGrpcCallMetricsLabelsWithCode extends IGrpcCallMetricsLabels {
+    code: string;
+}
+
+export const IClientCallMetrics = Symbol("IClientCallMetrics");
+
+export interface IClientCallMetrics {
+    started(labels: IGrpcCallMetricsLabels): void;
+    sent(labels: IGrpcCallMetricsLabels): void;
+    received(labels: IGrpcCallMetricsLabels): void;
+    handled(labels: IGrpcCallMetricsLabelsWithCode): void;
+}
+
+export function getGrpcMethodType(requestStream: boolean, responseStream: boolean): GrpcMethodType {
+    if (requestStream) {
+        if (responseStream) {
+            return "bidi_stream";
+        } else {
+            return "client_stream";
+        }
+    } else {
+        if (responseStream) {
+            return "server_stream";
+        } else {
+            return "unary";
+        }
+    }
+}
+
+export function createClientCallMetricsInterceptor(metrics: IClientCallMetrics): grpc.Interceptor {
+    return (options, nextCall): grpc.InterceptingCall => {
+        const methodDef = options.method_definition;
+        const method = methodDef.path.substring(methodDef.path.lastIndexOf("/") + 1);
+        const service = methodDef.path.substring(1, methodDef.path.length - method.length - 1);
+        const labels = {
+            service,
+            method,
+            type: getGrpcMethodType(options.method_definition.requestStream, options.method_definition.responseStream),
+        };
+        const requester = new grpc.RequesterBuilder()
+            .withStart((metadata, listener, next) => {
+                const newListener = new grpc.ListenerBuilder()
+                    .withOnReceiveStatus((status, next) => {
+                        try {
+                            metrics.handled({
+                                ...labels,
+                                code: Status[status.code],
+                            });
+                        } finally {
+                            next(status);
+                        }
+                    })
+                    .withOnReceiveMessage((message, next) => {
+                        try {
+                            metrics.received(labels);
+                        } finally {
+                            next(message);
+                        }
+                    })
+                    .build();
+                try {
+                    metrics.started(labels);
+                } finally {
+                    next(metadata, newListener);
+                }
+            })
+            .withSendMessage((message, next) => {
+                try {
+                    metrics.sent(labels);
+                } finally {
+                    next(message);
+                }
+            })
+            .build();
+        return new grpc.InterceptingCall(nextCall(options), requester);
+    };
+}
diff --git a/components/image-builder-api/typescript/src/sugar.ts b/components/image-builder-api/typescript/src/sugar.ts
@@ -8,10 +8,7 @@ import { ImageBuilderClient } from "./imgbuilder_grpc_pb";
 import { TraceContext } from "@gitpod/gitpod-protocol/lib/util/tracing";
 import { Deferred } from "@gitpod/gitpod-protocol/lib/util/deferred";
 import { log } from "@gitpod/gitpod-protocol/lib/util/logging";
-import {
-    createClientCallMetricsInterceptor,
-    IClientCallMetrics,
-} from "@gitpod/content-service/lib/client-call-metrics";
+import { createClientCallMetricsInterceptor, IClientCallMetrics } from "@gitpod/gitpod-protocol/lib/util/grpc";
 import * as opentracing from "opentracing";
 import { Metadata } from "@grpc/grpc-js";
 import {
diff --git a/components/server/src/container-module.ts b/components/server/src/container-module.ts
@@ -91,7 +91,7 @@ import { Config, ConfigFile } from "./config";
 import { defaultGRPCOptions } from "@gitpod/gitpod-protocol/lib/util/grpc";
 import { IDEConfigService } from "./ide-config";
 import { PrometheusClientCallMetrics } from "@gitpod/gitpod-protocol/lib/messaging/client-call-metrics";
-import { IClientCallMetrics } from "@gitpod/content-service/lib/client-call-metrics";
+import { IClientCallMetrics } from "@gitpod/gitpod-protocol/lib/util/grpc";
 import { DebugApp } from "@gitpod/gitpod-protocol/lib/util/debug-app";
 import { LocalMessageBroker, LocalRabbitMQBackedMessageBroker } from "./messaging/local-message-broker";
 import { contentServiceBinder } from "@gitpod/content-service/lib/sugar";
diff --git a/components/server/src/workspace/workspace-cluster-imagebuilder-client-provider.ts b/components/server/src/workspace/workspace-cluster-imagebuilder-client-provider.ts
@@ -5,8 +5,7 @@
  */
 
 import { Workspace, WorkspaceInstance } from "@gitpod/gitpod-protocol";
-import { IClientCallMetrics } from "@gitpod/gitpod-protocol/lib/messaging/client-call-metrics";
-import { defaultGRPCOptions } from "@gitpod/gitpod-protocol/lib/util/grpc";
+import { defaultGRPCOptions, IClientCallMetrics } from "@gitpod/gitpod-protocol/lib/util/grpc";
 import {
     ImageBuilderClient,
     ImageBuilderClientCallMetrics,
diff --git a/components/usage-api/typescript/src/usage/v1/sugar.ts b/components/usage-api/typescript/src/usage/v1/sugar.ts
diff --git a/components/ws-manager-api/typescript/src/client-provider.ts b/components/ws-manager-api/typescript/src/client-provider.ts
diff --git a/components/ws-manager-bridge/src/container-module.ts b/components/ws-manager-bridge/src/container-module.ts
