[server] SpiceDB: add client call metrics

Author: geropl

components/server/src/authorization/spicedb.ts

@@ -7,8 +7,11 @@
 import { v1 } from "@authzed/authzed-node";
 import { log } from "@gitpod/gitpod-protocol/lib/util/logging";
 import * as grpc from "@grpc/grpc-js_1_9_0";
+import { Status } from "@grpc/grpc-js_1_9_0/build/src/constants";
+import { inject, optional } from "inversify";
 
 export const SpiceDBClientProvider = Symbol("SpiceDBClientProvider");
+export const SpiceDBClientCallMetrics = Symbol("SpiceDBClientCallMetrics");
 
 export interface SpiceDBClientConfig {
     address: string;
@@ -57,7 +60,17 @@ function spicedbClientFromConfig(config: SpiceDBClientConfig): Client {
 export class CachingSpiceDBClientProvider implements SpiceDBClientProvider {
     private client: Client | undefined;
 
-    constructor(private readonly clientConfig: SpiceDBClientConfig) {}
+    @inject(SpiceDBClientCallMetrics)
+    @optional()
+    private readonly clientCallMetrics: IClientCallMetrics;
+
+    private readonly interceptors: grpc.Interceptor[] = [];
+
+    constructor(private readonly _clientConfig: SpiceDBClientConfig) {
+        if (this.clientCallMetrics) {
+            this.interceptors.push(createClientCallMetricsInterceptor(this.clientCallMetrics));
+        }
+    }
 
     getClient(): SpiceDBClient {
         let client = this.client;
@@ -76,4 +89,107 @@ export class CachingSpiceDBClientProvider implements SpiceDBClientProvider {
 
         return client.promises;
     }
+
+    protected get clientConfig() {
+        const config = this._clientConfig;
+        if (this.interceptors) {
+            return {
+                ...config,
+                options: {
+                    interceptors: [...this.interceptors],
+                },
+            };
+        }
+        return config;
+    }
+}
+
+// TODO(gpl): remove all of this and merge again with "util/grpc.ts" once we are beyond grpc-js 1.9.0
+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;
+    startHandleTimer(
+        labels: IGrpcCallMetricsLabels,
+    ): (labels?: Partial<Record<string, string | number>> | undefined) => number;
+}
+
+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);
+    };
 }