diff --git a/src/telemetry/client.test.tsx b/src/telemetry/client.test.tsx index e9288fca2..d895a3ef2 100644 --- a/src/telemetry/client.test.tsx +++ b/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 = { diff --git a/src/telemetry/client.tsx b/src/telemetry/client.tsx index 40775efb6..63c1f655a 100644 --- a/src/telemetry/client.tsx +++ b/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( @@ -43,7 +46,7 @@ export class DefaultTelemetryClient implements TelemetryClient { metricAttributes: AttributesOf, ): Promise { 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 { - 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 = 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 { - if (this.resourceAttributes !== undefined) return this.resourceAttributes; + return metricSinks; + }); + private getResourceAttributes: () => Promise = 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(fn: () => Promise): () => Promise { + let cachedPromise: Promise | undefined; + return () => (cachedPromise ??= fn()); } diff --git a/src/telemetry/fileSystemSink.tsx b/src/telemetry/fileSystemSink.tsx new file mode 100644 index 000000000..28d259757 --- /dev/null +++ b/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; + + 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): void { + this.pendingWrite = this.pendingWrite.then(() => + this.appendEntry({ metricName, value, attrs: attributes }), + ); + } + + private async appendEntry(entry: { + metricName: string; + value: number; + attrs: Record; + }): Promise { + await mkdir(dirname(this.filePath), { recursive: true }); + await appendFile(this.filePath, JSON.stringify(entry) + "\n"); + } + + async shutdown(): Promise { + 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; + } +} diff --git a/src/telemetry/loggingSink.tsx b/src/telemetry/loggingSink.tsx deleted file mode 100644 index 4074f03f6..000000000 --- a/src/telemetry/loggingSink.tsx +++ /dev/null @@ -1,31 +0,0 @@ -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, - ): void { - this.logger.child({ metricName, metricValue, metricAttributes }).info("recording telemetry"); - } - - async shutdown(): Promise {} - - getName() { - return "LoggingSink"; - } -}