diff --git a/src/index.ts b/src/index.ts index b36ac6789..6fd6b3e73 100644 --- a/src/index.ts +++ b/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(); } }), diff --git a/src/telemetry/client.test.tsx b/src/telemetry/client.test.tsx new file mode 100644 index 000000000..e9288fca2 --- /dev/null +++ b/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, + }, + ]); + }); +}); diff --git a/src/telemetry/client.tsx b/src/telemetry/client.tsx new file mode 100644 index 000000000..40775efb6 --- /dev/null +++ b/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( + metricName: TMetricName, + metricValue: ValueOf, + metricAttributes: AttributesOf, + ): Promise { + 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 { + 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 { + 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; + } +} diff --git a/src/telemetry/index.tsx b/src/telemetry/index.tsx new file mode 100644 index 000000000..103df657c --- /dev/null +++ b/src/telemetry/index.tsx @@ -0,0 +1,3 @@ +export { DefaultTelemetryClient } from "./client"; +export { TelemetryAttributesRecorder } from "./recorder"; +export { type AttributesOf } from "./types"; diff --git a/src/telemetry/loggingSink.tsx b/src/telemetry/loggingSink.tsx new file mode 100644 index 000000000..4074f03f6 --- /dev/null +++ b/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, + ): void { + this.logger.child({ metricName, metricValue, metricAttributes }).info("recording telemetry"); + } + + async shutdown(): Promise {} + + getName() { + return "LoggingSink"; + } +} diff --git a/src/telemetry/recorder.tsx b/src/telemetry/recorder.tsx new file mode 100644 index 000000000..ba83df103 --- /dev/null +++ b/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 { + private attributes: Partial>; + + constructor( + private readonly metricName: TMetricName, + initialAttributes: Partial> = {}, + ) { + this.attributes = initialAttributes; + } + + /** + * Retrieves the underlying attributes and validates them against the metric schema. + * Throws if metric shape is invalid. + */ + getAttributes(): AttributesOf { + const attributes = METRICS[this.metricName]["attributeSchema"].parse(this.attributes); + return attributes as AttributesOf; + } + + /** + * Add attributes that overwrite existing values if already set. + */ + record(data: Partial>): Partial> { + this.attributes = { + ...this.attributes, + ...data, + }; + return this.attributes; + } +} diff --git a/src/telemetry/shapes.tsx b/src/telemetry/shapes.tsx new file mode 100644 index 000000000..7141aba99 --- /dev/null +++ b/src/telemetry/shapes.tsx @@ -0,0 +1,31 @@ +import z from "zod"; + +const UUID_PATTERN = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i; +const SEMVER_PATTERN = /^\d+\.\d+\.\d+$/; +const NODE_VERSION_PATTERN = /^v\d+\.\d+\.\d+$/; +const MAX_ATTR_LENGTH = 64; + +/** + * Resource attributes that are attached to every metric datapoint. + * Set once per session, not metric. + */ +export const resourceAttributesSchema = z.object({ + "service.name": z.literal("agentcore-cli"), + "service.version": z.string().regex(SEMVER_PATTERN), + "agentcore-cli.installation_id": z.string().regex(UUID_PATTERN), + "agentcore-cli.session_id": z.string().regex(UUID_PATTERN), + "os.type": z.string().min(1).max(MAX_ATTR_LENGTH), + "os.version": z.string().min(1).max(MAX_ATTR_LENGTH), + "host.arch": z.string().min(1).max(MAX_ATTR_LENGTH), + "node.version": z.string().regex(NODE_VERSION_PATTERN), +}); + +/** + * Type derived from {@link resourceAttributesSchema} + */ +export type ResourceAttributes = z.infer; + +// TODO: expand this definition. +export const commandRunSchema = z.object({ + exit_reason: z.enum(["success", "failure"]), +}); diff --git a/src/telemetry/types.tsx b/src/telemetry/types.tsx new file mode 100644 index 000000000..f29e01f3d --- /dev/null +++ b/src/telemetry/types.tsx @@ -0,0 +1,52 @@ +import z from "zod"; +import { commandRunSchema } from "./shapes"; + +/** + * The primary interface for telemetry that orchestrates the emitting of metrics + */ +export interface TelemetryClient { + emit( + metricName: TMetricName, + metricValue: ValueOf, + attributes: AttributesOf, + ): Promise; + shutdown(): Promise; +} + +/** + * A destination to send metric data. + */ +export interface MetricSink { + /** Send data to the given metric sink **/ + send(metricName: string, value: number, attributes: Record): void; + /** Flush and close the given metric sink **/ + shutdown(): Promise; + getName(): string; +} + +/** + * Static definition of all metrics the CLI emits. + */ +export const METRICS = { + "cli.command_run": { + attributeSchema: commandRunSchema, + // value describes duration (ms) of the command + valueSchema: z.number().min(0), + }, +} satisfies Record; + +export type MetricName = keyof typeof METRICS; + +/** + * Describes the value type for the given {@link Metric} + */ +export type ValueOf = z.input< + (typeof METRICS)[TMetricName]["valueSchema"] +>; + +/** + * Describes the attributes type for the given {@link Metric} + */ +export type AttributesOf = z.input< + (typeof METRICS)[TMetricName]["attributeSchema"] +>;