feat(tel): implement otel sink

package.json

@@ -52,10 +52,15 @@
     "@aws-sdk/client-bedrock-agentcore": "^3.1092.0",
     "@aws-sdk/client-bedrock-agentcore-control": "^3.1079.0",
     "@aws-sdk/client-iam": "^3.1080.0",
+    "@opentelemetry/api": "^1.9.1",
+    "@opentelemetry/exporter-metrics-otlp-http": "^0.221.0",
+    "@opentelemetry/resources": "^2.10.0",
+    "@opentelemetry/sdk-metrics": "^2.10.0",
     "@smithy/core": "3.29.3",
     "@tanstack/react-query": "^5.101.2",
     "cli-truncate": "^6.1.1",
     "commander": "^15.0.0",
+    "handlebars": "^4.7.9",
     "ink": "^7.1.0",
     "ink-scroll-view": "^0.3.7",
     "lodash": "^4.18.1",
@@ -65,7 +70,6 @@
     "string-width": "^8.2.2",
     "winston": "^3.19.0",
     "winston-daily-rotate-file": "^5.0.0",
-    "handlebars": "^4.7.9",
     "zod": "^4.4.3"
   }
 }

src/telemetry/client.test.tsx

@@ -5,9 +5,10 @@ import os, { tmpdir } from "node:os";
 import { DefaultTelemetryClient } from "./client";
 import { createFileLogger, type Logger } from "../logging";
 import { LOG_LEVEL } from "../logging";
-import { assertLogsMatch, TestGlobalConfigAccessor } from "../testing";
+import { assertLogsMatch, createSilentLogger, TestGlobalConfigAccessor } from "../testing";
 import type { MetricSink } from "./types";
 import { FileSystemSink } from "./fileSystemSink";
+import { DEFAULT_GLOBAL_CONFIG } from "../globalConfig";
 import { PACKAGE_VERSION } from "../constants";
 
 describe("DefaultTelemetryClient", () => {
@@ -28,9 +29,20 @@ describe("DefaultTelemetryClient", () => {
 
   test("emits complete metrics to configured JSONL filesystem sinks", async () => {
     const auditFilePath = join(tempDir, "telemetry", "audit.jsonl");
+    const sinkResourceAttributes = {
+      "service.name": "agentcore-cli" as const,
+      "service.version": "0.0.0",
+      "agentcore-cli.installation_id": "00000000-0000-0000-0000-000000000000",
+      "agentcore-cli.session_id": "00000000-0000-0000-0000-000000000000",
+      "os.type": os.type(),
+      "os.version": os.release(),
+      "host.arch": os.arch(),
+      "node.version": process.version,
+    };
     const fileSystemSink = new FileSystemSink({
       logger: logger.child({ module: "fileSystemSink" }),
       filePath: auditFilePath,
+      resourceAttributes: sinkResourceAttributes,
     });
     const sessionId = "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee";
     const globalConfigAccessor = new TestGlobalConfigAccessor();
@@ -191,6 +203,16 @@ describe("DefaultTelemetryClient", () => {
     const sink = new FileSystemSink({
       logger: logger.child({ module: "fileSystemSink" }),
       filePath: tempDir,
+      resourceAttributes: {
+        "service.name": "agentcore-cli",
+        "service.version": "0.0.0",
+        "agentcore-cli.installation_id": "00000000-0000-0000-0000-000000000000",
+        "agentcore-cli.session_id": "00000000-0000-0000-0000-000000000000",
+        "os.type": os.type(),
+        "os.version": os.release(),
+        "host.arch": os.arch(),
+        "node.version": process.version,
+      },
     });
 
     const client = new DefaultTelemetryClient({
@@ -274,3 +296,109 @@ describe("DefaultTelemetryClient", () => {
     ]);
   });
 });
+
+describe("OtelHistogramSink", () => {
+  let testCollector: ReturnType<typeof Bun.serve>;
+  let receivedBodies: any[];
+
+  const logger = createSilentLogger();
+
+  beforeEach(async () => {
+    receivedBodies = [];
+    testCollector = Bun.serve({
+      port: 0,
+      async fetch(req) {
+        const body = await req.json();
+        receivedBodies.push(body);
+        return new Response("", { status: 200 });
+      },
+    });
+  });
+
+  afterEach(async () => {
+    testCollector.stop(true);
+  });
+
+  test.each([
+    { enabled: true, expectRequests: true },
+    { enabled: false, expectRequests: false },
+  ])(
+    "telemetry.enabled=$enabled → collector receives requests=$expectRequests",
+    async ({ enabled, expectRequests }) => {
+      const sessionId = "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee";
+      const exitReason = "success";
+      const commandPath = "/agentcore";
+      const metricName = "cli.command_run";
+      const scopeName = "agentcore-cli";
+      const serviceName = "agentcore-cli";
+      const globalConfigAccessor = new TestGlobalConfigAccessor({
+        initialConfigData: {
+          ...DEFAULT_GLOBAL_CONFIG,
+          telemetry: {
+            enabled,
+            audit: false,
+            endpoint: `http://localhost:${testCollector.port}`,
+          },
+        },
+      });
+
+      const client = new DefaultTelemetryClient({
+        logger,
+        sessionId,
+        globalConfigAccessor,
+      });
+
+      const event = client.createMetricEvent(metricName, {
+        exit_reason: exitReason,
+        command_path: commandPath,
+      });
+      await event.emit(100);
+      await client.shutdown();
+
+      if (expectRequests) {
+        expect(receivedBodies.length).toBeGreaterThan(0);
+
+        const body = receivedBodies[0];
+        expect(body).toMatchObject({
+          resourceMetrics: [
+            {
+              resource: {
+                attributes: expect.arrayContaining([
+                  { key: "service.name", value: { stringValue: serviceName } },
+                  {
+                    key: "agentcore-cli.session_id",
+                    value: { stringValue: sessionId },
+                  },
+                  { key: "os.type", value: { stringValue: os.type() } },
+                  { key: "host.arch", value: { stringValue: os.arch() } },
+                ]),
+              },
+              scopeMetrics: [
+                {
+                  scope: { name: scopeName },
+                  metrics: [
+                    {
+                      name: metricName,
+                      histogram: {
+                        dataPoints: [
+                          {
+                            attributes: expect.arrayContaining([
+                              { key: "exit_reason", value: { stringValue: exitReason } },
+                              { key: "command_path", value: { stringValue: commandPath } },
+                            ]),
+                          },
+                        ],
+                      },
+                    },
+                  ],
+                },
+              ],
+            },
+          ],
+        });
+      } else {
+        expect(receivedBodies).toHaveLength(0);
+      }
+    },
+  );
+});

src/telemetry/client.tsx

@@ -13,6 +13,7 @@ import {
 import type { GlobalConfigAccessor } from "../globalConfig";
 import { FileSystemSink } from "./fileSystemSink";
 import path from "path";
+import { OtelHistogramSink } from "./otelSink";
 import { PACKAGE_VERSION } from "../constants";
 
 export type DefaultTelemetryClientConfig = {
@@ -72,6 +73,7 @@ export class DefaultTelemetryClient implements TelemetryClient {
 
   private getMetricSinks: () => Promise<MetricSink[]> = once(async () => {
     if (this.metricSinksOverride) return this.metricSinksOverride;
+    const resourceAttributes = await this.getResourceAttributes();
 
     const metricSinks = [];
 
@@ -82,6 +84,16 @@ export class DefaultTelemetryClient implements TelemetryClient {
         new FileSystemSink({
           logger: this.logger.child({ module: "fileSystemSink" }),
           filePath: this.auditFilePath,
+          resourceAttributes,
+        }),
+      );
+
+    if (globalConfig.telemetry.enabled)
+      metricSinks.push(
+        new OtelHistogramSink({
+          logger: this.logger.child({ module: "otelCollectorSink" }),
+          collectorEndpoint: globalConfig.telemetry.endpoint,
+          resourceAttributes,
         }),
       );
 

src/telemetry/fileSystemSink.tsx

@@ -1,11 +1,13 @@
 import type { Logger } from "../logging";
+import type { ResourceAttributes } from "./shapes";
 import type { MetricSink } from "./types";
 import { mkdir, appendFile } from "fs/promises";
 import { dirname } from "path";
 
 export type FileSystemSinkConfig = {
   logger: Logger;
   filePath: string;
+  resourceAttributes: ResourceAttributes;
 };
 
 /** An implementation of {@link MetricSink} that sends all data to the specified file in JSONL format **/
@@ -15,13 +17,16 @@ export class FileSystemSink implements MetricSink {
   private readonly filePath: string;
   private logger: Logger;
 
+  private readonly resourceAttributes: ResourceAttributes;
+
   /* a chain of promises describing the pending writes to the audit file */
   private pendingWrite: Promise<void>;
 
   constructor(config: FileSystemSinkConfig) {
     this.filePath = config.filePath;
     this.logger = config.logger.child({ fsSinkFilePath: this.filePath });
     this.name = new.target.name;
+    this.resourceAttributes = config.resourceAttributes;
 
     this.pendingWrite = Promise.resolve();
   }
@@ -32,7 +37,7 @@ export class FileSystemSink implements MetricSink {
     attributes: Record<string, string | number | boolean>,
   ): void {
     this.pendingWrite = this.pendingWrite.then(() =>
-      this.appendEntry({ metricName, value, attrs: attributes }),
+      this.appendEntry({ metricName, value, attrs: { ...this.resourceAttributes, ...attributes } }),
     );
   }
 

src/telemetry/otelSink.tsx

@@ -0,0 +1,97 @@
+import { MeterProvider, PeriodicExportingMetricReader } from "@opentelemetry/sdk-metrics";
+import { resourceFromAttributes } from "@opentelemetry/resources";
+import type { Logger } from "../logging";
+import { type MetricSink } from "./types";
+import { OTLPMetricExporter } from "@opentelemetry/exporter-metrics-otlp-http";
+import { type Histogram, type Meter } from "@opentelemetry/api";
+import type { ResourceAttributes } from "./shapes";
+
+export type OtelCollectorSinkConfig = {
+  collectorEndpoint: string;
+  logger: Logger;
+  /** The resource attributes to attach to all metrics.**/
+  resourceAttributes: ResourceAttributes;
+  /** The time period between export flushes **/
+  exportIntervalMs?: number;
+  /** Describes the maximum time to wait when flushing metrics to the sink.**/
+  flushTimeoutMs?: number;
+  /** Describes the maximum time to wait when shutting down the sink.**/
+  shutdownTimeoutMs?: number;
+  /**
+   * Describes the scope to attach to the given metrics. Defaults to `agentcore-cli`
+   * See https://opentelemetry.io/docs/concepts/instrumentation-scope/ for more information.
+   **/
+  instrumentationScope?: string;
+};
+
+/**
+ * An implementation of {@link MetricSink} that sends data to a collector using an otel histogram.
+ */
+export class OtelHistogramSink implements MetricSink {
+  private readonly name: string;
+  private readonly endpoint: string;
+  private meterProvider: MeterProvider;
+  private histograms: Map<string, Histogram>;
+  private logger: Logger;
+
+  private readonly flushTimeoutMs: number;
+  private readonly shutdownTimeoutMs: number;
+  private readonly scopedMeter: Meter;
+
+  constructor(config: OtelCollectorSinkConfig) {
+    this.endpoint = config.collectorEndpoint;
+    this.logger = config.logger.child({ telemetryEndpoint: config.collectorEndpoint });
+    this.name = new.target.name;
+
+    this.flushTimeoutMs = config.flushTimeoutMs ?? 500;
+    this.shutdownTimeoutMs = config.shutdownTimeoutMs ?? 500;
+
+    this.meterProvider = new MeterProvider({
+      resource: resourceFromAttributes(config.resourceAttributes),
+      readers: [
+        new PeriodicExportingMetricReader({
+          exporter: new OTLPMetricExporter({
+            url: this.endpoint.endsWith("/v1/metrics")
+              ? this.endpoint
+              : `${this.endpoint}/v1/metrics`,
+          }),
+          exportIntervalMillis: config.exportIntervalMs ?? 5_000,
+        }),
+      ],
+    });
+    this.scopedMeter = this.meterProvider.getMeter(config.instrumentationScope ?? "agentcore-cli");
+    this.histograms = new Map<string, Histogram>();
+  }
+
+  private getHistogram(metricName: string): Histogram {
+    if (!this.histograms.has(metricName)) {
+      this.histograms.set(metricName, this.scopedMeter.createHistogram(metricName));
+    }
+    return this.histograms.get(metricName)!;
+  }
+
+  send(metricName: string, value: number, attributes: Record<string, string | number>): void {
+    this.logger
+      .child({ metricName, metricValue: value, metricAttributes: attributes })
+      .info(`sending telemetry metric to collector`);
+
+    this.getHistogram(metricName).record(value, attributes);
+  }
+
+  async shutdown(): Promise<void> {
+    try {
+      await this.meterProvider.forceFlush({ timeoutMillis: this.flushTimeoutMs });
+    } catch (e) {
+      const error = e instanceof Error ? e : new Error(String(e));
+      this.logger
+        .child({ errorName: error.name, errorMessage: error.message })
+        .warn(`failed to flush metrics to ${this.getName()}`);
+      // don't let flush failures prevent shutdown
+    }
+    await this.meterProvider.shutdown({ timeoutMillis: this.shutdownTimeoutMs });
+  }
+
+  getName(): string {
+    return this.name;
+  }
+}