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();
@@ -59,32 +71,19 @@ describe("DefaultTelemetryClient", () => {
 
     expect(fileSystemSink.getName()).toBe("FileSystemSink");
 
-    const { installationId } = await globalConfigAccessor.get();
-
     const auditContents = await readFile(auditFilePath, "utf8");
 
     const entries = auditContents
       .trimEnd()
       .split("\n")
       .map((line) => JSON.parse(line));
 
-    const resourceAttributes = {
-      "service.name": "agentcore-cli",
-      "service.version": PACKAGE_VERSION,
-      "agentcore-cli.installation_id": installationId,
-      "agentcore-cli.session_id": sessionId,
-      "os.type": os.type(),
-      "os.version": os.release(),
-      "host.arch": os.arch(),
-      "node.version": process.version,
-    };
-
     expect(entries).toEqual([
       {
         metricName: "cli.command_run",
         value: 123,
         attrs: {
-          ...resourceAttributes,
+          ...sinkResourceAttributes,
           exit_reason: "success",
           command_path: "/agentcore",
           is_tui: false,
@@ -94,7 +93,7 @@ describe("DefaultTelemetryClient", () => {
         metricName: "cli.command_run",
         value: 456,
         attrs: {
-          ...resourceAttributes,
+          ...sinkResourceAttributes,
           exit_reason: "failure",
           command_path: "/agentcore",
           is_tui: false,
@@ -191,6 +190,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 +283,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 = {
@@ -52,7 +53,6 @@ export class DefaultTelemetryClient implements TelemetryClient {
       initialAttributes,
       logger: this.logger,
       getSinks: () => this.getMetricSinks(),
-      getResourceAttributes: () => this.getResourceAttributes(),
     });
   }
 
@@ -72,6 +72,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 +83,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,
         }),
       );
 
@@ -114,7 +125,6 @@ type InMemoryMetricEventConfig<TMetricName extends MetricName> = {
   initialAttributes?: Partial<AttributesOf<TMetricName>>;
   logger: Logger;
   getSinks: () => Promise<MetricSink[]>;
-  getResourceAttributes: () => Promise<ResourceAttributes>;
 };
 
 /** An in-memory implementation of {@link MetricEvent} that accumulates attributes and emits on end() **/
@@ -123,14 +133,12 @@ class InMemoryMetricEvent<TMetricName extends MetricName> implements MetricEvent
   private readonly metricName: TMetricName;
   private readonly logger: Logger;
   private readonly getSinks: () => Promise<MetricSink[]>;
-  private readonly getResourceAttributes: () => Promise<ResourceAttributes>;
 
   constructor(config: InMemoryMetricEventConfig<TMetricName>) {
     this.metricName = config.metricName;
     this.data = config.initialAttributes ?? {};
     this.logger = config.logger;
     this.getSinks = config.getSinks;
-    this.getResourceAttributes = config.getResourceAttributes;
   }
 
   setAttributes(newData: Partial<AttributesOf<TMetricName>>): void {
@@ -143,18 +151,12 @@ class InMemoryMetricEvent<TMetricName extends MetricName> implements MetricEvent
   async emit(value: ValueOf<TMetricName>): Promise<void> {
     const metricAttributes = METRICS[this.metricName]["attributeSchema"].parse(this.data);
     const validatedValue = METRICS[this.metricName]["valueSchema"].parse(value);
-    const resourceAttributes = await this.getResourceAttributes();
-
-    const attributes = {
-      ...resourceAttributes,
-      ...metricAttributes,
-    };
 
     const sinks = await this.getSinks();
 
     sinks.forEach((sink) => {
       try {
-        sink.send(this.metricName, validatedValue, attributes);
+        sink.send(this.metricName, validatedValue, metricAttributes);
       } catch (e) {
         const error = e instanceof Error ? e : new Error(String(e));
         this.logger

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 } }),
     );
   }