feat(telemetry): enable audit mode for telemetry (#1858)

* feat(tel): implement audit sink on the fs

* test(tel): add explicit test for the case where audit is disabled

* refactor(telemetry): remove logging sink in favor of audit sink

* test(telemetry): add a test case for invalid audit path

* refactor(telemetry): simplify the caching of metric sinks and attributes

Author: Hweinstock

src/telemetry/client.test.tsx

@@ -1,14 +1,14 @@
 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 { mkdtemp, readFile, rm } from "node:fs/promises";
+import os, { 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";
+import { FileSystemSink } from "./fileSystemSink";
 
 describe("DefaultTelemetryClient", () => {
   let tempDir: string;
@@ -26,42 +26,120 @@ describe("DefaultTelemetryClient", () => {
     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" }) });
+  test("emits complete metrics to configured JSONL filesystem sinks", async () => {
+    const auditFilePath = join(tempDir, "telemetry", "audit.jsonl");
+    const fileSystemSink = new FileSystemSink({
+      logger: logger.child({ module: "fileSystemSink" }),
+      filePath: auditFilePath,
+    });
     const sessionId = "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee";
     const globalConfigAccessor = new TestGlobalConfigAccessor();
     const client = new DefaultTelemetryClient({
       logger,
       sessionId,
       globalConfigAccessor,
-      metricSinks: [sink],
+      metricSinks: [fileSystemSink],
     });
 
     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();
+    recorder.record({ exit_reason: "failure" });
+    await client.emit("cli.command_run", 456, recorder.getAttributes());
     await client.shutdown();
 
+    expect(fileSystemSink.getName()).toBe("FileSystemSink");
+
     const { installationId } = await globalConfigAccessor.get();
-    await assertLogsMatch(tempDir, [
+
+    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": "0.0.0",
+      "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([
       {
-        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,
+        metricName: "cli.command_run",
+        value: 123,
+        attrs: { ...resourceAttributes, exit_reason: "success" },
+      },
+      {
+        metricName: "cli.command_run",
+        value: 456,
+        attrs: { ...resourceAttributes, exit_reason: "failure" },
       },
     ]);
   });
 
+  test("enables the default audit sink only when global config audit is enabled", async () => {
+    const auditFilePath = join(tempDir, "telemetry", "config-audit.jsonl");
+    const enabledSessionId = "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee";
+    const disabledSessionId = "ffffffff-1111-2222-3333-444444444444";
+
+    const enabledConfigAccessor = new TestGlobalConfigAccessor();
+    const enabledConfig = await enabledConfigAccessor.get();
+    await enabledConfigAccessor.set({
+      ...enabledConfig,
+      telemetry: { ...enabledConfig.telemetry, audit: true },
+    });
+
+    const disabledConfigAccessor = new TestGlobalConfigAccessor();
+    const disabledConfig = await disabledConfigAccessor.get();
+    await disabledConfigAccessor.set({
+      ...disabledConfig,
+      telemetry: { ...disabledConfig.telemetry, audit: false },
+    });
+
+    const enabledClient = new DefaultTelemetryClient({
+      logger,
+      sessionId: enabledSessionId,
+      globalConfigAccessor: enabledConfigAccessor,
+      auditFilePath,
+    });
+    const disabledClient = new DefaultTelemetryClient({
+      logger,
+      sessionId: disabledSessionId,
+      globalConfigAccessor: disabledConfigAccessor,
+      auditFilePath,
+    });
+
+    await enabledClient.emit("cli.command_run", 123, { exit_reason: "success" });
+    await disabledClient.emit("cli.command_run", 456, { exit_reason: "failure" });
+    await Promise.all([enabledClient.shutdown(), disabledClient.shutdown()]);
+
+    const auditLines = (await readFile(auditFilePath, "utf8")).trimEnd().split("\n");
+    expect(auditLines).toHaveLength(1);
+    expect(JSON.parse(auditLines[0]!)).toEqual({
+      metricName: "cli.command_run",
+      value: 123,
+      attrs: {
+        "service.name": "agentcore-cli",
+        "service.version": "0.0.0",
+        "agentcore-cli.installation_id": enabledConfig.installationId,
+        "agentcore-cli.session_id": enabledSessionId,
+        "os.type": os.type(),
+        "os.version": os.release(),
+        "host.arch": os.arch(),
+        "node.version": process.version,
+        exit_reason: "success",
+      },
+    });
+  });
+
   test("throws when recorder has incomplete attributes", async () => {
     const client = new DefaultTelemetryClient({
       logger,
@@ -76,6 +154,33 @@ describe("DefaultTelemetryClient", () => {
     await client.shutdown();
   });
 
+  test("FileSystemSink logs a warning when the file is not writable", async () => {
+    // Point the sink at a directory to trigger EISDIR
+    const sink = new FileSystemSink({
+      logger: logger.child({ module: "fileSystemSink" }),
+      filePath: tempDir,
+    });
+
+    const client = new DefaultTelemetryClient({
+      logger,
+      sessionId: "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee",
+      globalConfigAccessor: new TestGlobalConfigAccessor(),
+      metricSinks: [sink],
+    });
+
+    await client.emit("cli.command_run", 1, { exit_reason: "success" });
+    await client.shutdown();
+
+    await assertLogsMatch(tempDir, [
+      {
+        filter: (log: any) =>
+          log.msg === "failed to append metric data to file" &&
+          log.errorMessage?.includes("EISDIR"),
+        expectedCount: 1,
+      },
+    ]);
+  });
+
   test("handles sink errors gracefully without throwing", async () => {
     const recordedMetrics: string[] = [];
     const goodSink: MetricSink = {

src/telemetry/client.tsx

@@ -1,5 +1,4 @@
 import type { Logger } from "../logging";
-import { LoggingSink } from "./loggingSink";
 import { resourceAttributesSchema, type ResourceAttributes } from "./shapes";
 import os from "os";
 import {
@@ -11,12 +10,15 @@ import {
   type MetricName,
 } from "./types";
 import type { GlobalConfigAccessor } from "../globalConfig";
+import { FileSystemSink } from "./fileSystemSink";
+import path from "path";
 
 export type DefaultTelemetryClientConfig = {
   logger: Logger;
   globalConfigAccessor: GlobalConfigAccessor;
   sessionId: string;
   metricSinks?: MetricSink[];
+  auditFilePath?: string;
 };
 
 /**
@@ -25,16 +27,17 @@ export type DefaultTelemetryClientConfig = {
 export class DefaultTelemetryClient implements TelemetryClient {
   private logger: Logger;
   private readonly sessionId: string;
+  private readonly auditFilePath: string;
   private globalConfigAccessor: GlobalConfigAccessor;
-  private resourceAttributes: ResourceAttributes | undefined;
-  private metricSinks: MetricSink[] | undefined;
-
+  private readonly metricSinksOverride: MetricSink[] | undefined;
   constructor(config: DefaultTelemetryClientConfig) {
     this.logger = config.logger;
     this.sessionId = config.sessionId;
     this.globalConfigAccessor = config.globalConfigAccessor;
-    this.resourceAttributes = undefined;
-    this.metricSinks = config.metricSinks;
+    this.metricSinksOverride = config.metricSinks;
+    this.auditFilePath =
+      config.auditFilePath ??
+      path.join(os.homedir(), ".agentcore", "telemetry", `${this.sessionId}.jsonl`);
   }
 
   async emit<TMetricName extends MetricName>(
@@ -43,7 +46,7 @@ export class DefaultTelemetryClient implements TelemetryClient {
     metricAttributes: AttributesOf<TMetricName>,
   ): Promise<void> {
     try {
-      const metricSinks = this.getMetricSinks();
+      const metricSinks = await this.getMetricSinks();
       const resourceAttributes = await this.getResourceAttributes();
       // merge in resource attributes with metric attributes before sending to sink.
       const attributes = {
@@ -74,7 +77,7 @@ export class DefaultTelemetryClient implements TelemetryClient {
   }
 
   async shutdown(): Promise<void> {
-    const metricSinks = this.getMetricSinks();
+    const metricSinks = await this.getMetricSinks();
 
     const promises = metricSinks.map(async (sink) => {
       return sink.shutdown().catch((e) => {
@@ -87,19 +90,27 @@ export class DefaultTelemetryClient implements TelemetryClient {
     await Promise.all(promises);
   }
 
-  private getMetricSinks(): MetricSink[] {
-    if (this.metricSinks !== undefined) return this.metricSinks;
+  private getMetricSinks: () => Promise<MetricSink[]> = once(async () => {
+    if (this.metricSinksOverride) return this.metricSinksOverride;
 
-    this.metricSinks = [new LoggingSink({ logger: this.logger.child({ module: "loggingSink" }) })];
+    const metricSinks = [];
 
-    return this.metricSinks;
-  }
+    const globalConfig = await this.globalConfigAccessor.get();
+
+    if (globalConfig.telemetry.audit)
+      metricSinks.push(
+        new FileSystemSink({
+          logger: this.logger.child({ module: "fileSystemSink" }),
+          filePath: this.auditFilePath,
+        }),
+      );
 
-  private async getResourceAttributes(): Promise<ResourceAttributes> {
-    if (this.resourceAttributes !== undefined) return this.resourceAttributes;
+    return metricSinks;
+  });
 
+  private getResourceAttributes: () => Promise<ResourceAttributes> = once(async () => {
     const globalConfig = await this.globalConfigAccessor.get();
-    this.resourceAttributes = resourceAttributesSchema.parse({
+    return resourceAttributesSchema.parse({
       "service.name": "agentcore-cli",
       // TODO: wire up real package version.
       "service.version": "0.0.0",
@@ -110,6 +121,11 @@ export class DefaultTelemetryClient implements TelemetryClient {
       "host.arch": os.arch(),
       "node.version": process.version,
     });
-    return this.resourceAttributes;
-  }
+  });
+}
+
+/** wraps an async function such that it only executes once **/
+function once<T>(fn: () => Promise<T>): () => Promise<T> {
+  let cachedPromise: Promise<T> | undefined;
+  return () => (cachedPromise ??= fn());
 }

src/telemetry/fileSystemSink.tsx

@@ -0,0 +1,59 @@
+import type { Logger } from "../logging";
+import type { MetricSink } from "./types";
+import { mkdir, appendFile } from "fs/promises";
+import { dirname } from "path";
+
+export type FileSystemSinkConfig = {
+  logger: Logger;
+  filePath: string;
+};
+
+/** An implementation of {@link MetricSink} that sends all data to the specified file in JSONL format **/
+export class FileSystemSink implements MetricSink {
+  private readonly name: string;
+
+  private readonly filePath: string;
+  private logger: Logger;
+
+  /* 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.pendingWrite = Promise.resolve();
+  }
+
+  send(metricName: string, value: number, attributes: Record<string, string | number>): void {
+    this.pendingWrite = this.pendingWrite.then(() =>
+      this.appendEntry({ metricName, value, attrs: attributes }),
+    );
+  }
+
+  private async appendEntry(entry: {
+    metricName: string;
+    value: number;
+    attrs: Record<string, string | number>;
+  }): Promise<void> {
+    await mkdir(dirname(this.filePath), { recursive: true });
+    await appendFile(this.filePath, JSON.stringify(entry) + "\n");
+  }
+
+  async shutdown(): Promise<void> {
+    try {
+      await this.pendingWrite;
+      this.logger.info(`audit file written to '${this.filePath}'`);
+    } catch (e) {
+      const error = e instanceof Error ? e : new Error(String(e));
+      this.logger
+        .child({ errorName: error.name, errorMessage: error.message })
+        .warn(`failed to append metric data to file`);
+    }
+  }
+
+  getName(): string {
+    return this.name;
+  }
+}