refactor(tel): move to metric event instead of attributes recorder

Author: Hweinstock

src/index.ts

@@ -16,7 +16,7 @@ import { runWithExitCode } from "./runnable";
 import { DefaultGlobalConfigAccessor } from "./globalConfig";
 import { DefaultTelemetryClient } from "./telemetry";
 import { AgentCoreCLIError } from "./errors";
-import { TelemetryAttributesRecorderKey, ValueContext } from "./router";
+import { CommandRunMetricEventKey, ValueContext } from "./router";
 
 process.exit(
   await runWithExitCode(async (argv: string[]) => {
@@ -50,7 +50,7 @@ process.exit(
       globalConfigAccessor,
     });
 
-    const commandRunTelemetryRecorder = telemetryClient.getAttributesRecorder("cli.command_run", {
+    const commandRunMetricEvent = telemetryClient.startMetricEvent("cli.command_run", {
       exit_reason: "success",
     });
 
@@ -75,29 +75,25 @@ process.exit(
         globalConfigAccessor,
       });
 
-      const context = ValueContext.EmptyContext().withValue<typeof commandRunTelemetryRecorder>(
-        TelemetryAttributesRecorderKey,
-        commandRunTelemetryRecorder,
+      const context = ValueContext.EmptyContext().withValue(
+        CommandRunMetricEventKey,
+        commandRunMetricEvent,
       );
 
       // Handle the request
       await rootHandler.route(argv, context);
     } catch (e) {
       const error = AgentCoreCLIError.fromError(e);
       rootLogger.child({ error: error.json() }).error();
-      commandRunTelemetryRecorder.record({
+      commandRunMetricEvent.setAttributes({
         exit_reason: "failure",
         error_name: error.name,
         error_source: error.source,
       });
       throw error;
     } finally {
       try {
-        await telemetryClient.emit(
-          "cli.command_run",
-          Date.now() - startTime,
-          commandRunTelemetryRecorder,
-        );
+        await commandRunMetricEvent.end(Date.now() - startTime);
       } catch (e) {
         const error = AgentCoreCLIError.fromError(e);
         rootLogger.child({ error: error.json() }).warn("failed to emit telemetry");

src/router/index.tsx

@@ -5,7 +5,7 @@ export {
   PathKey,
   LoggerKey,
   GlobalConfigAccessorKey,
-  TelemetryAttributesRecorderKey,
+  CommandRunMetricEventKey,
   ProjectKey,
   type DefaultHandle,
   type DefaultHandlerProvider,

src/router/router.test.ts

@@ -4,7 +4,7 @@ import z from "zod";
 
 import {
   Router,
-  TelemetryAttributesRecorderKey,
+  CommandRunMetricEventKey,
   ValueContext,
   argument,
   compile,
@@ -677,40 +677,57 @@ test("commands without long-form flag help have no Parameter details section", a
 test.each([
   { scenario: "flag validation succeeds", idFlag: "abc", shouldThrow: false },
   { scenario: "flag validation fails", idFlag: "toolong", shouldThrow: true },
-])(
-  "records command_path on the telemetry recorder when $scenario",
-  async ({ idFlag, shouldThrow }) => {
-    // we use a real command name here so that telemetry schemas accept the path produced below
-    const get = createHandler({
-      name: "config",
-      description: "",
-      flags: [flag("id", "id", z.string().max(3))],
-      handle: async () => {},
-    });
-
-    const telemetryClient = new DefaultTelemetryClient({
-      logger: createSilentLogger(),
-      globalConfigAccessor: new TestGlobalConfigAccessor(),
-      sessionId: "test-session-id",
-    });
-
-    const root = new Router("agentcore");
-    root.handler(get);
-
-    const recorder = telemetryClient.getAttributesRecorder("cli.command_run");
-    const ctx = ValueContext.EmptyContext().withValue(TelemetryAttributesRecorderKey, recorder);
-    const cmd = exitOverrideAll(compile(root, ctx));
-
-    if (shouldThrow) {
-      await expect(
-        cmd.parseAsync(["node", "agentcore", "config", "--id", idFlag]),
-      ).rejects.toThrow();
-      recorder.record({ exit_reason: "failure" });
-    } else {
-      await cmd.parseAsync(["node", "agentcore", "config", "--id", idFlag]);
-      recorder.record({ exit_reason: "success" });
-    }
-
-    expect(recorder.getAttributes()).toMatchObject({ command_path: "/agentcore/config" });
-  },
-);
+])("records command_path on the metric event when $scenario", async ({ idFlag, shouldThrow }) => {
+  // we use a real command name here so that telemetry schemas accept the path produced below
+  const get = createHandler({
+    name: "config",
+    description: "",
+    flags: [flag("id", "id", z.string().max(3))],
+    handle: async () => {},
+  });
+
+  const recordedMetrics: { metricName: string; value: number; attributes: Record<string, any> }[] =
+    [];
+  const inMemorySink = {
+    getName: () => "InMemorySink",
+    send: (metricName: string, value: number, attributes: Record<string, any>) => {
+      recordedMetrics.push({ metricName, value, attributes });
+    },
+    shutdown: async () => {},
+  };
+
+  const telemetryClient = new DefaultTelemetryClient({
+    logger: createSilentLogger(),
+    globalConfigAccessor: new TestGlobalConfigAccessor(),
+    sessionId: "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee",
+    metricSinks: [inMemorySink],
+  });
+
+  const root = new Router("agentcore");
+  root.handler(get);
+
+  const commandRunMetricEvent = telemetryClient.startMetricEvent("cli.command_run");
+  const ctx = ValueContext.EmptyContext().withValue(
+    CommandRunMetricEventKey,
+    commandRunMetricEvent,
+  );
+  const cmd = compile(root, ctx);
+
+  if (shouldThrow) {
+    await expect(cmd.parseAsync(["node", "agentcore", "config", "--id", idFlag])).rejects.toThrow();
+    commandRunMetricEvent.setAttributes({ exit_reason: "failure" });
+  } else {
+    await cmd.parseAsync(["node", "agentcore", "config", "--id", idFlag]);
+    commandRunMetricEvent.setAttributes({ exit_reason: "success" });
+  }
+
+  await commandRunMetricEvent.end(100);
+
+  expect(recordedMetrics).toHaveLength(1);
+  expect(recordedMetrics[0]!.metricName).toBe("cli.command_run");
+  expect(recordedMetrics[0]!.value).toBe(100);
+  expect(recordedMetrics[0]!.attributes).toMatchObject({
+    command_path: "/agentcore/config",
+    exit_reason: shouldThrow ? "failure" : "success",
+  });
+});

src/router/router.tsx

@@ -8,7 +8,7 @@ import { Command } from "commander";
 import type { Logger } from "../logging";
 import type { GlobalConfigAccessor } from "../globalConfig";
 import type { Project } from "../handlers/project/types";
-import { type AttributesOf, type AttributesRecorder } from "../telemetry";
+import { type MetricEvent } from "../telemetry";
 
 // CommandKey exposes the Commander Command for the executing leaf via context.
 export const CommandKey: ContextKey<Command> = contextKey<Command>("commander.command");
@@ -17,9 +17,8 @@ export const PathKey: ContextKey<string> = contextKey<string>("path");
 
 export const LoggerKey = contextKey<Logger>("logger");
 
-export const TelemetryAttributesRecorderKey = contextKey<
-  AttributesRecorder<AttributesOf<"cli.command_run">>
->("telemetryAttributesRecorder");
+export const CommandRunMetricEventKey =
+  contextKey<MetricEvent<"cli.command_run">>("commandRunMetricEvent");
 
 export const GlobalConfigAccessorKey: ContextKey<GlobalConfigAccessor> =
   contextKey<GlobalConfigAccessor>("globalConfigAccessor");
@@ -99,7 +98,7 @@ function globalFlagsOf(node: Handler): GlobalFlag[] {
 
 /** Add the command path to active command run metric **/
 function recordCommandPath(ctx: Context): void {
-  ctx.value(TelemetryAttributesRecorderKey)?.record({ command_path: ctx.value(PathKey) });
+  ctx.value(CommandRunMetricEventKey)?.setAttributes({ command_path: ctx.value(PathKey) });
 }
 
 // compile walks the Handler tree into a Commander Command tree.

src/telemetry/client.test.tsx

@@ -40,15 +40,20 @@ describe("DefaultTelemetryClient", () => {
       metricSinks: [fileSystemSink],
     });
 
-    const recorder = client.getAttributesRecorder("cli.command_run", {
+    const metricEvent = client.startMetricEvent("cli.command_run", {
       exit_reason: "success",
       command_path: "/agentcore",
     });
 
-    await client.emit("cli.command_run", 123, recorder);
+    await metricEvent.end(123);
 
-    recorder.record({ exit_reason: "failure" });
-    await client.emit("cli.command_run", 456, recorder);
+    // Start a second event with failure
+    const metricEvent2 = client.startMetricEvent("cli.command_run", {
+      exit_reason: "failure",
+      command_path: "/agentcore",
+    });
+
+    await metricEvent2.end(456);
     await client.shutdown();
 
     expect(fileSystemSink.getName()).toBe("FileSystemSink");
@@ -129,24 +134,20 @@ describe("DefaultTelemetryClient", () => {
       auditFilePath,
     });
 
-    await enabledClient.emit(
-      "cli.command_run",
-      123,
-      enabledClient.getAttributesRecorder("cli.command_run", {
-        exit_reason: "success",
-        command_path: "/agentcore",
-        is_tui: true,
-      }),
-    );
-    await disabledClient.emit(
-      "cli.command_run",
-      123,
-      enabledClient.getAttributesRecorder("cli.command_run", {
-        exit_reason: "failure",
-        command_path: "/agentcore",
-        is_tui: false,
-      }),
-    );
+    const enabledEvent = enabledClient.startMetricEvent("cli.command_run", {
+      exit_reason: "success",
+      command_path: "/agentcore",
+      is_tui: true,
+    });
+    await enabledEvent.end(123);
+
+    const disabledEvent = disabledClient.startMetricEvent("cli.command_run", {
+      exit_reason: "failure",
+      command_path: "/agentcore",
+      is_tui: false,
+    });
+    await disabledEvent.end(123);
+
     await Promise.all([enabledClient.shutdown(), disabledClient.shutdown()]);
 
     const auditLines = (await readFile(auditFilePath, "utf8")).trimEnd().split("\n");
@@ -170,17 +171,17 @@ describe("DefaultTelemetryClient", () => {
     });
   });
 
-  test("throws when recorder has incomplete attributes", async () => {
+  test("throws when metric event has incomplete attributes", async () => {
     const client = new DefaultTelemetryClient({
       logger,
       sessionId: "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee",
       globalConfigAccessor: new TestGlobalConfigAccessor(),
       metricSinks: [],
     });
 
-    const recorder = client.getAttributesRecorder("cli.command_run");
+    const metricEvent = client.startMetricEvent("cli.command_run");
 
-    expect(() => client.emit("cli.command_run", 100, recorder)).toThrow();
+    await expect(metricEvent.end(100)).rejects.toThrow();
     await client.shutdown();
   });
 
@@ -197,11 +198,11 @@ describe("DefaultTelemetryClient", () => {
       globalConfigAccessor: new TestGlobalConfigAccessor(),
       metricSinks: [sink],
     });
-    const recorder = client.getAttributesRecorder("cli.command_run", {
+    const metricEvent = client.startMetricEvent("cli.command_run", {
       exit_reason: "success",
       command_path: "/agentcore",
     });
-    await client.emit("cli.command_run", 1, recorder);
+    await metricEvent.end(1);
     await client.shutdown();
 
     await assertLogsMatch(tempDir, [
@@ -241,13 +242,13 @@ describe("DefaultTelemetryClient", () => {
       metricSinks: [badSink, goodSink],
     });
 
-    const recorder = client.getAttributesRecorder("cli.command_run", {
+    const metricEvent = client.startMetricEvent("cli.command_run", {
       exit_reason: "success",
       command_path: "/agentcore",
     });
 
-    // emit should not throw even though the sink's record() throws
-    await client.emit("cli.command_run", 100, recorder);
+    // end should not throw even though the sink's send() throws
+    await metricEvent.end(100);
     // shutdown should not throw even though the sink's shutdown() rejects
     await client.shutdown();