[DEV-792] Support tracing in multi stream append (#430)

Author: w1am

packages/opentelemetry/src/attributes.ts

@@ -12,6 +12,7 @@ export const KurrentAttributes = {
   SERVER_PORT: `${server}.port`,
 
   STREAM_APPEND: `${streams}.append`,
+  STREAM_MULTI_APPEND: `${streams}.multi-append`,
   STREAM_SUBSCRIBE: `${streams}.subscribe`,
 
   KURRENT_DB_STREAM: `${kurrentdb}.stream`,

packages/opentelemetry/src/instrumentation.ts

@@ -8,6 +8,7 @@ import {
   Span,
   SpanKind,
   SpanStatusCode,
+  TimeInput,
   trace,
   TraceFlags,
   Tracer,
@@ -24,6 +25,7 @@ import type {
   EventData,
   EventType,
   JSONEventType,
+  MultiAppendResult,
   ResolvedEvent,
   SubscribeToAllOptions,
   SubscribeToPersistentSubscriptionToAllOptions,
@@ -38,6 +40,7 @@ import type { Subscription } from "@kurrent/kurrentdb-client/src/streams/utils/S
 import { INSTRUMENTATION_NAME, INSTRUMENTATION_VERSION } from "./version";
 import type {
   AppendToStreamParams,
+  MultiStreamAppendParams,
   PersistentSubscribeParameters,
   SubscribeParameters,
 } from "./types";
@@ -67,6 +70,11 @@ export class Instrumentation extends InstrumentationBase {
         "appendToStream",
         this._patchAppendToStream()
       );
+      this.wrap(
+        moduleExports.KurrentDBClient.prototype,
+        "multiStreamAppend",
+        this._patchMultiStreamAppend()
+      );
       this.wrap(
         moduleExports.KurrentDBClient.prototype,
         "subscribeToStream",
@@ -106,6 +114,10 @@ export class Instrumentation extends InstrumentationBase {
       this._diag.debug("un-patching");
 
       this._unwrap(moduleExports.KurrentDBClient.prototype, "appendToStream");
+      this._unwrap(
+        moduleExports.KurrentDBClient.prototype,
+        "multiStreamAppend"
+      );
       this._unwrap(
         moduleExports.KurrentDBClient.prototype,
         "subscribeToStream"
@@ -196,6 +208,118 @@ export class Instrumentation extends InstrumentationBase {
     };
   }
 
+  private _patchMultiStreamAppend(): (
+    original: Function,
+    operation: keyof kurrentdb.KurrentDBClient
+  ) => (...args: MultiStreamAppendParams) => Promise<MultiAppendResult> {
+    const instrumentation = this;
+    const tracer = instrumentation.tracer;
+
+    return function multiStreamAppend(
+      original: Function,
+      operation: keyof kurrentdb.KurrentDBClient
+    ) {
+      return async function (
+        this: kurrentdb.KurrentDBClient,
+        ...args: MultiStreamAppendParams
+      ): Promise<MultiAppendResult> {
+        const [requests] = [...args];
+
+        const uri = await this.resolveUri();
+        const { hostname, port } = Instrumentation.getServerAddress(uri);
+
+        const requestStartTime: TimeInput = Date.now();
+
+        const span = tracer.startSpan(KurrentAttributes.STREAM_MULTI_APPEND, {
+          kind: SpanKind.CLIENT,
+          startTime: requestStartTime,
+          attributes: {
+            [KurrentAttributes.SERVER_ADDRESS]: hostname,
+            [KurrentAttributes.SERVER_PORT]: port,
+            [KurrentAttributes.DATABASE_SYSTEM]: INSTRUMENTATION_NAME,
+            [KurrentAttributes.DATABASE_OPERATION]: operation,
+          },
+        });
+
+        requests.forEach((request) => {
+          const traceId = span.spanContext().traceId;
+          const spanId = span.spanContext().spanId;
+
+          request.events.forEach((event) => {
+            const metadata = (event.metadata = event.metadata || {});
+            if (isJSONEventData(event) && typeof metadata === "object") {
+              event.metadata = {
+                ...metadata,
+                [TRACE_ID]: traceId,
+                [SPAN_ID]: spanId,
+              };
+            }
+          });
+        });
+
+        try {
+          const result = await original.apply(this, [requests]);
+
+          const requestEndTime: TimeInput = Date.now();
+
+          if (!result.success) {
+            const failures: kurrentdb.AppendStreamFailure[] = result.output;
+
+            span.setStatus({
+              code: SpanStatusCode.ERROR,
+            });
+
+            failures.forEach((failure) => {
+              switch (failure.details.type) {
+                case "wrong_expected_revision":
+                  span.addEvent("exception", {
+                    "exception.type": "wrong_expected_revision",
+                    "exception.revision":
+                      failure.details.revision.toLocaleString(),
+                  });
+                  break;
+
+                case "access_denied":
+                  span.addEvent("exception", {
+                    "exception.type": failure.details.type,
+                    "exception.message": failure.details.reason,
+                  });
+                  break;
+
+                case "stream_deleted":
+                  span.addEvent("exception", {
+                    "exception.type": failure.details.type,
+                  });
+                  break;
+
+                case "transaction_max_size_exceeded":
+                  span.addEvent("exception", {
+                    "exception.type": failure.details.type,
+                    "exception.max_size":
+                      failure.details.maxSize.toLocaleString(),
+                  });
+                  break;
+
+                case "unknown":
+                  span.addEvent("exception", {
+                    "exception.type": "unknown",
+                  });
+                  break;
+              }
+            });
+          }
+
+          span.end(requestEndTime);
+
+          return result;
+        } catch (error) {
+          Instrumentation.handleError(error, span);
+          throw error;
+        }
+      };
+    };
+  }
+
   static applySubscriptionInstrumentation<KnownEventType>(
     spanName: string,
     subscription:

packages/opentelemetry/src/types.ts

@@ -23,3 +23,7 @@ export type PersistentSubscribeParameters =
 export type AppendToStreamParams = Parameters<
   kdb.KurrentDBClient["appendToStream"]
 >;
+
+export type MultiStreamAppendParams = Parameters<
+  kdb.KurrentDBClient["multiStreamAppend"]
+>;

packages/test/package.json

@@ -37,6 +37,7 @@
     "@opentelemetry/api": "^1.9.0",
     "@opentelemetry/exporter-trace-otlp-grpc": "^0.51.1",
     "@opentelemetry/instrumentation": "^0.56.0",
+    "@opentelemetry/resources": "^2.0.1",
     "@opentelemetry/sdk-trace-node": "^1.30.0",
     "@opentelemetry/semantic-conventions": "^1.28.0",
     "@types/debug": "^4.1.12",

packages/test/src/opentelemetry/instrumentation.test.ts

@@ -1,3 +1,5 @@
+/** @jest-environment ./src/utils/enableVersionCheck.ts */
+
 import { createTestNode, Defer, delay, jsonTestEvents } from "@test-utils";
 import {
   NodeTracerProvider,