feat(telemetry): setup client with logging sink

src/index.ts

@@ -14,12 +14,13 @@ import { FsReadWriteJson } from "./io";
 import { createFileLogger, LOG_LEVEL } from "./logging";
 import { runWithExitCode } from "./runnable";
 import { DefaultGlobalConfigAccessor } from "./globalConfig";
+import { DefaultTelemetryClient, TelemetryAttributesRecorder } from "./telemetry";
 import { AgentCoreCLIError } from "./errors";
 
 process.exit(
   await runWithExitCode(async (argv: string[]) => {
+    const startTime = Date.now();
     // generate a unique identifier corresponding to this process of this CLI. (ex. one command invoke, one TUI session)
-    // TODO: wire this id into telemetry as well
     const cliSessionId = crypto.randomUUID();
 
     const rootLogger = createFileLogger({
@@ -34,15 +35,25 @@ process.exit(
       stderr: process.stderr,
     };
 
-    try {
-      const globalConfigAccessor = new DefaultGlobalConfigAccessor({
-        logger: rootLogger.child({ module: "globalConfigAccessor" }),
-        filePath: join(homedir(), ".agentcore", "config.json"),
-        json: new FsReadWriteJson({
-          logger: rootLogger.child({ module: "jsonDataSource" }),
-        }),
-      });
+    const globalConfigAccessor = new DefaultGlobalConfigAccessor({
+      logger: rootLogger.child({ module: "globalConfigAccessor" }),
+      filePath: join(homedir(), ".agentcore", "config.json"),
+      json: new FsReadWriteJson({
+        logger: rootLogger.child({ module: "jsonDataSource" }),
+      }),
+    });
+
+    const telemetryClient = new DefaultTelemetryClient({
+      logger: rootLogger.child({ module: "telemetry" }),
+      sessionId: cliSessionId,
+      globalConfigAccessor,
+    });
 
+    const commandRunTelemetryRecorder = new TelemetryAttributesRecorder("cli.command_run", {
+      exit_reason: "success",
+    });
+
+    try {
       rootLogger.info(`running CLI`);
 
       // factories (rather than instances) lets CoreClient build one client per
@@ -68,8 +79,20 @@ process.exit(
     } catch (e) {
       const error = AgentCoreCLIError.fromError(e);
       rootLogger.child({ error: error.json() }).error();
+      // TODO: add error details to telemetry recorder;
+      commandRunTelemetryRecorder.record({ exit_reason: "failure" });
+
       throw error;
     } finally {
+      try {
+        const attributes = commandRunTelemetryRecorder.getAttributes();
+        await telemetryClient.emit("cli.command_run", Date.now() - startTime, attributes);
+      } catch (e) {
+        const error = AgentCoreCLIError.fromError(e);
+        rootLogger.child({ error: error.json() }).warn("failed to emit telemetry");
+        // telemetry is best-effort
+      }
+      await telemetryClient.shutdown();
       await rootLogger.end();
     }
   }),

src/telemetry/client.test.tsx

@@ -0,0 +1,131 @@
+import { test, describe, beforeEach, afterEach, expect } from "bun:test";
+import { join } from "node:path";
+import { mkdtemp, rm } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import { DefaultTelemetryClient } from "./client";
+import { TelemetryAttributesRecorder } from "./recorder";
+import { createFileLogger, type Logger } from "../logging";
+import { LOG_LEVEL } from "../logging";
+import { assertLogsMatch, TestGlobalConfigAccessor } from "../testing";
+import type { MetricSink } from "./types";
+import { LoggingSink } from "./loggingSink";
+
+describe("DefaultTelemetryClient", () => {
+  let tempDir: string;
+  let logger: Logger;
+
+  beforeEach(async () => {
+    tempDir = await mkdtemp(join(tmpdir(), "telemetry-client-test-"));
+    logger = createFileLogger({
+      filePath: join(tempDir, "output"),
+      logLevel: LOG_LEVEL.DEBUG,
+    });
+  });
+
+  afterEach(async () => {
+    await rm(tempDir, { recursive: true, force: true });
+  });
+
+  test("emits metrics with resource and command attributes to configured sinks", async () => {
+    const sink = new LoggingSink({ logger: logger.child({ module: "loggingSink" }) });
+    const sessionId = "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee";
+    const globalConfigAccessor = new TestGlobalConfigAccessor();
+    const client = new DefaultTelemetryClient({
+      logger,
+      sessionId,
+      globalConfigAccessor,
+      metricSinks: [sink],
+    });
+
+    const recorder = new TelemetryAttributesRecorder("cli.command_run", { exit_reason: "success" });
+
+    recorder.record({ exit_reason: "failure" });
+
+    await client.emit("cli.command_run", 123, recorder.getAttributes());
+
+    expect(sink.getName()).toBe("LoggingSink");
+    await sink.shutdown();
+    await client.shutdown();
+
+    const { installationId } = await globalConfigAccessor.get();
+    await assertLogsMatch(tempDir, [
+      {
+        filter: (log: any) =>
+          log.metricName === "cli.command_run" &&
+          log.metricValue === 123 &&
+          log.metricAttributes?.["exit_reason"] === "failure" &&
+          log.metricAttributes?.["service.name"] === "agentcore-cli" &&
+          log.metricAttributes?.["agentcore-cli.session_id"] === sessionId &&
+          log.metricAttributes?.["agentcore-cli.installation_id"] === installationId,
+        expectedCount: 1,
+      },
+    ]);
+  });
+
+  test("throws when recorder has incomplete attributes", async () => {
+    const client = new DefaultTelemetryClient({
+      logger,
+      sessionId: "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee",
+      globalConfigAccessor: new TestGlobalConfigAccessor(),
+      metricSinks: [],
+    });
+
+    const recorder = new TelemetryAttributesRecorder("cli.command_run");
+
+    expect(() => client.emit("cli.command_run", 100, recorder.getAttributes())).toThrow();
+    await client.shutdown();
+  });
+
+  test("handles sink errors gracefully without throwing", async () => {
+    const recordedMetrics: string[] = [];
+    const goodSink: MetricSink = {
+      getName: () => "GoodSink",
+      send: (metricName) => {
+        recordedMetrics.push(metricName);
+      },
+      shutdown: async () => {},
+    };
+
+    const badSink: MetricSink = {
+      getName: () => "BadSink",
+      send: () => {
+        throw new Error("record exploded");
+      },
+      shutdown: async () => {
+        throw new Error("shutdown exploded");
+      },
+    };
+
+    const client = new DefaultTelemetryClient({
+      logger,
+      sessionId: "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee",
+      globalConfigAccessor: new TestGlobalConfigAccessor(),
+      metricSinks: [badSink, goodSink],
+    });
+
+    // emit should not throw even though the sink's record() throws
+    await client.emit("cli.command_run", 100, { exit_reason: "success" });
+    // shutdown should not throw even though the sink's shutdown() rejects
+    await client.shutdown();
+
+    // GoodSink still receives data despite BadSink throwing
+    expect(recordedMetrics).toEqual(["cli.command_run"]);
+
+    await assertLogsMatch(tempDir, [
+      {
+        filter: (log: any) =>
+          log.msg === "failed to record to sink 'BadSink'" &&
+          log.errorName === "Error" &&
+          log.errorMessage === "record exploded",
+        expectedCount: 1,
+      },
+      {
+        filter: (log: any) =>
+          log.msg === "failed to shutdown metric sink with name 'BadSink'" &&
+          log.errorName === "Error" &&
+          log.errorMessage === "shutdown exploded",
+        expectedCount: 1,
+      },
+    ]);
+  });
+});

src/telemetry/client.tsx

@@ -0,0 +1,115 @@
+import type { Logger } from "../logging";
+import { LoggingSink } from "./loggingSink";
+import { resourceAttributesSchema, type ResourceAttributes } from "./shapes";
+import os from "os";
+import {
+  type AttributesOf,
+  type MetricSink,
+  type TelemetryClient,
+  type ValueOf,
+  METRICS,
+  type MetricName,
+} from "./types";
+import type { GlobalConfigAccessor } from "../globalConfig";
+
+export type DefaultTelemetryClientConfig = {
+  logger: Logger;
+  globalConfigAccessor: GlobalConfigAccessor;
+  sessionId: string;
+  metricSinks?: MetricSink[];
+};
+
+/**
+ * Implements {@link TelemetryClient} by validating and fanning out metrics to a list of {@link MetricSink} implementations.
+ */
+export class DefaultTelemetryClient implements TelemetryClient {
+  private logger: Logger;
+  private readonly sessionId: string;
+  private globalConfigAccessor: GlobalConfigAccessor;
+  private resourceAttributes: ResourceAttributes | undefined;
+  private metricSinks: MetricSink[] | undefined;
+
+  constructor(config: DefaultTelemetryClientConfig) {
+    this.logger = config.logger;
+    this.sessionId = config.sessionId;
+    this.globalConfigAccessor = config.globalConfigAccessor;
+    this.resourceAttributes = undefined;
+    this.metricSinks = config.metricSinks;
+  }
+
+  async emit<TMetricName extends MetricName>(
+    metricName: TMetricName,
+    metricValue: ValueOf<TMetricName>,
+    metricAttributes: AttributesOf<TMetricName>,
+  ): Promise<void> {
+    try {
+      const metricSinks = this.getMetricSinks();
+      const resourceAttributes = await this.getResourceAttributes();
+      // merge in resource attributes with metric attributes before sending to sink.
+      const attributes = {
+        ...resourceAttributes,
+        ...metricAttributes,
+      };
+
+      const validatedMetricValue = METRICS[metricName]["valueSchema"].parse(metricValue);
+
+      metricSinks.forEach((sink) => {
+        try {
+          sink.send(metricName, validatedMetricValue, attributes);
+        } catch (e) {
+          const error = e instanceof Error ? e : new Error(String(e));
+          this.logger
+            .child({ errorName: error.name, errorMessage: error.message })
+            .warn(`failed to record to sink '${sink.getName()}'`);
+          // do not allow a single sink failure to fail other sinks.
+        }
+      });
+    } catch (e) {
+      const error = e instanceof Error ? e : new Error(String(e));
+      this.logger
+        .child({ errorName: error.name, errorMessage: error.message })
+        .warn(`failed to emit telemetry`);
+      // telemetry is best-effort, don't throw.
+    }
+  }
+
+  async shutdown(): Promise<void> {
+    const metricSinks = this.getMetricSinks();
+
+    const promises = metricSinks.map(async (sink) => {
+      return sink.shutdown().catch((e) => {
+        const error = e instanceof Error ? e : new Error(String(e));
+        this.logger
+          .child({ errorName: error.name, errorMessage: error.message })
+          .warn(`failed to shutdown metric sink with name '${sink.getName()}'`);
+      });
+    });
+    await Promise.all(promises);
+  }
+
+  private getMetricSinks(): MetricSink[] {
+    if (this.metricSinks !== undefined) return this.metricSinks;
+
+    this.metricSinks = [new LoggingSink({ logger: this.logger.child({ module: "loggingSink" }) })];
+
+    return this.metricSinks;
+  }
+
+  private async getResourceAttributes(): Promise<ResourceAttributes> {
+    if (this.resourceAttributes !== undefined) return this.resourceAttributes;
+
+    const globalConfig = await this.globalConfigAccessor.get();
+    this.resourceAttributes = resourceAttributesSchema.parse({
+      "service.name": "agentcore-cli",
+      // TODO: wire up real package version.
+      "service.version": "0.0.0",
+      "agentcore-cli.installation_id": globalConfig.installationId,
+      "agentcore-cli.session_id": this.sessionId,
+      "os.type": os.type(),
+      "os.version": os.release(),
+      "host.arch": os.arch(),
+      "node.version": process.version,
+    });
+    return this.resourceAttributes;
+  }
+}

src/telemetry/index.tsx

@@ -0,0 +1,3 @@
+export { DefaultTelemetryClient } from "./client";
+export { TelemetryAttributesRecorder } from "./recorder";
+export { type AttributesOf } from "./types";

src/telemetry/loggingSink.tsx

@@ -0,0 +1,31 @@
+import type { Logger } from "../logging";
+import type { MetricSink } from "./types";
+
+type LoggingSinkConfig = {
+  logger: Logger;
+};
+
+/**
+ * An implementation of {@link MetricSink} that logs metrics using the given logger
+ */
+export class LoggingSink implements MetricSink {
+  private logger: Logger;
+
+  constructor(config: LoggingSinkConfig) {
+    this.logger = config.logger;
+  }
+
+  send(
+    metricName: string,
+    metricValue: number,
+    metricAttributes: Record<string, string | number>,
+  ): void {
+    this.logger.child({ metricName, metricValue, metricAttributes }).info("recording telemetry");
+  }
+
+  async shutdown(): Promise<void> {}
+
+  getName() {
+    return "LoggingSink";
+  }
+}

src/telemetry/recorder.tsx

@@ -0,0 +1,35 @@
+import { type MetricName, type AttributesOf, METRICS } from "./types";
+
+/**
+ * A strongly typed recorder for accumulating metric attributes, bound to a specific metric's schema.
+ */
+export class TelemetryAttributesRecorder<TMetricName extends MetricName> {
+  private attributes: Partial<AttributesOf<TMetricName>>;
+
+  constructor(
+    private readonly metricName: TMetricName,
+    initialAttributes: Partial<AttributesOf<TMetricName>> = {},
+  ) {
+    this.attributes = initialAttributes;
+  }
+
+  /**
+   * Retrieves the underlying attributes and validates them against the metric schema.
+   * Throws if metric shape is invalid.
+   */
+  getAttributes(): AttributesOf<TMetricName> {
+    const attributes = METRICS[this.metricName]["attributeSchema"].parse(this.attributes);
+    return attributes as AttributesOf<TMetricName>;
+  }
+
+  /**
+   * Add attributes that overwrite existing values if already set.
+   */
+  record(data: Partial<AttributesOf<TMetricName>>): Partial<AttributesOf<TMetricName>> {
+    this.attributes = {
+      ...this.attributes,
+      ...data,
+    };
+    return this.attributes;
+  }
+}