tracing: provider shutdown/flush, optional sampler, error status (add-ons for #61)

.github/workflows/ambar-core.yaml

@@ -26,7 +26,10 @@ jobs:
           version: 10
 
       - name: Install
-        run: pnpm install
+        # pnpm 10 ignores env-var auth in the committed .npmrc; write the token to ~/.npmrc instead.
+        run: |
+          echo "//npm.pkg.github.com/:_authToken=${GITHUB_TOKEN}" >> ~/.npmrc
+          pnpm install
         env:
           GITHUB_TOKEN: ${{ secrets.READ_ACCESS_TO_REPOS }}
 

.github/workflows/ambar-task-explorer.yaml

@@ -30,7 +30,10 @@ jobs:
           version: 10
 
       - name: Install
-        run: pnpm install
+        # pnpm 10 ignores env-var auth in the committed .npmrc; write the token to ~/.npmrc instead.
+        run: |
+          echo "//npm.pkg.github.com/:_authToken=${GITHUB_TOKEN}" >> ~/.npmrc
+          pnpm install
         env:
           GITHUB_TOKEN: ${{ secrets.READ_ACCESS_TO_REPOS }}
 
@@ -74,7 +77,10 @@ jobs:
           version: 10
 
       - name: Install
-        run: pnpm install
+        # pnpm 10 ignores env-var auth in the committed .npmrc; write the token to ~/.npmrc instead.
+        run: |
+          echo "//npm.pkg.github.com/:_authToken=${GITHUB_TOKEN}" >> ~/.npmrc
+          pnpm install
         env:
           GITHUB_TOKEN: ${{ secrets.READ_ACCESS_TO_REPOS }}
 

.github/workflows/ambar-tasks.yaml

@@ -40,7 +40,10 @@ jobs:
           version: 10
 
       - name: Install
-        run: pnpm install
+        # pnpm 10 ignores env-var auth in the committed .npmrc; write the token to ~/.npmrc instead.
+        run: |
+          echo "//npm.pkg.github.com/:_authToken=${GITHUB_TOKEN}" >> ~/.npmrc
+          pnpm install
         env:
           GITHUB_TOKEN: ${{ secrets.READ_ACCESS_TO_REPOS }}
 

core/src/tracing/opentelemetry.ts

@@ -20,14 +20,15 @@ import { Future } from "../future";
 import {
   context,
   trace as otel,
+  SpanStatusCode,
   type Context,
   type Span,
   type Tracer as OtelTracer,
   type HrTime,
   type AttributeValue,
 } from "@opentelemetry/api";
 import { NodeTracerProvider } from "@opentelemetry/sdk-trace-node";
-import type { SpanProcessor, ReadableSpan } from "@opentelemetry/sdk-trace-base";
+import type { SpanProcessor, ReadableSpan, Sampler } from "@opentelemetry/sdk-trace-base";
 import { AsyncLocalStorageContextManager } from "@opentelemetry/context-async-hooks";
 import { resourceFromAttributes } from "@opentelemetry/resources";
 import { ATTR_SERVICE_NAME } from "@opentelemetry/semantic-conventions";
@@ -154,19 +155,32 @@ interface OpenTelemetryTracerOptions {
   // Where finished spans go. Use `new OtlpJsonStdoutProcessor()` for the stdout
   // OTLP/JSON output, or a BatchSpanProcessor + OTLP exporter for a backend.
   spanProcessor: SpanProcessor;
+  // Head sampling decision. When omitted the SDK default applies (honouring the
+  // OTEL_TRACES_SAMPLER env), so sampling can be left to the env or the collector.
+  sampler?: Sampler;
+}
+
+// Mark a span as failed: record the error and set its status to ERROR. Accepts
+// any thrown/rejected value (Futures reject with an arbitrary `E`), coercing
+// non-Error values to a string.
+function recordError(span: Span, err: unknown): void {
+  span.recordException(err instanceof Error ? err : String(err));
+  span.setStatus({ code: SpanStatusCode.ERROR });
 }
 
 class OpenTelemetryTracer implements Tracer {
   private readonly tracer: OtelTracer;
+  private readonly provider: NodeTracerProvider;
 
   constructor(options: OpenTelemetryTracerOptions) {
-    const provider = new NodeTracerProvider({
+    this.provider = new NodeTracerProvider({
       resource: resourceFromAttributes({ [ATTR_SERVICE_NAME]: options.serviceName }),
       spanProcessors: [options.spanProcessor],
+      ...(options.sampler ? { sampler: options.sampler } : {}),
     });
     // The AsyncLocalStorage context manager is what lets `context.active()`
     // return the right parent span inside `await`s and Fluture continuations.
-    provider.register({ contextManager: new AsyncLocalStorageContextManager().enable() });
+    this.provider.register({ contextManager: new AsyncLocalStorageContextManager().enable() });
     this.tracer = otel.getTracer(options.serviceName);
   }
 
@@ -183,6 +197,9 @@ class OpenTelemetryTracer implements Tracer {
     const { span, ctx } = this.begin(name, attributes);
     try {
       return context.with(ctx, f);
+    } catch (err) {
+      recordError(span, err);
+      throw err;
     } finally {
       span.end();
     }
@@ -197,6 +214,9 @@ class OpenTelemetryTracer implements Tracer {
     const { span, ctx } = this.begin(name, attributes);
     try {
       return await context.with(ctx, f);
+    } catch (err) {
+      recordError(span, err);
+      throw err;
     } finally {
       span.end();
     }
@@ -215,6 +235,7 @@ class OpenTelemetryTracer implements Tracer {
       return context.with(ctx, () =>
         f.fork(
           err => {
+            recordError(span, err);
             span.end();
             context.with(parentCtx, () => reject(err));
           },
@@ -232,4 +253,11 @@ class OpenTelemetryTracer implements Tracer {
   event(name: string, attributes?: Attributes): void {
     otel.getActiveSpan()?.addEvent(name, attributes);
   }
+
+  // Flush any pending spans, then shut down the provider and its span processors.
+  // Call on graceful shutdown (SIGTERM/SIGINT) so the final batch is not dropped.
+  async shutdown(): Promise<void> {
+    await this.provider.forceFlush();
+    await this.provider.shutdown();
+  }
 }