Skip to content

Commit 3268b99

Browse files
author
Hweinstock
committed
feat(telemetry): setup client with logging sink
1 parent b57ed51 commit 3268b99

8 files changed

Lines changed: 438 additions & 9 deletions

File tree

‎src/index.ts‎

Lines changed: 28 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -14,11 +14,12 @@ import { FsReadWriteJson } from "./io";
1414
import { createFileLogger, LOG_LEVEL } from "./logging";
1515
import { runWithExitCode } from "./runnable";
1616
import { DefaultGlobalConfigAccessor } from "./globalConfig";
17+
import { DefaultTelemetryClient, TelemetryAttributesRecorder } from "./telemetry";
1718

1819
process.exit(
1920
await runWithExitCode(async (argv: string[]) => {
21+
const startTime = Date.now();
2022
// generate a unique identifier corresponding to this process of this CLI. (ex. one command invoke, one TUI session)
21-
// TODO: wire this id into telemetry as well
2223
const cliSessionId = crypto.randomUUID();
2324

2425
const rootLogger = createFileLogger({
@@ -33,15 +34,25 @@ process.exit(
3334
stderr: process.stderr,
3435
};
3536

36-
try {
37-
const globalConfigAccessor = new DefaultGlobalConfigAccessor({
38-
logger: rootLogger.child({ module: "globalConfigAccessor" }),
39-
filePath: join(homedir(), ".agentcore", "config.json"),
40-
json: new FsReadWriteJson({
41-
logger: rootLogger.child({ module: "jsonDataSource" }),
42-
}),
43-
});
37+
const globalConfigAccessor = new DefaultGlobalConfigAccessor({
38+
logger: rootLogger.child({ module: "globalConfigAccessor" }),
39+
filePath: join(homedir(), ".agentcore", "config.json"),
40+
json: new FsReadWriteJson({
41+
logger: rootLogger.child({ module: "jsonDataSource" }),
42+
}),
43+
});
44+
45+
const telemetryClient = new DefaultTelemetryClient({
46+
logger: rootLogger.child({ module: "telemetry" }),
47+
sessionId: cliSessionId,
48+
globalConfigAccessor,
49+
});
50+
51+
const commandRunTelemetryRecorder = new TelemetryAttributesRecorder("cli.command_run", {
52+
exit_reason: "success",
53+
});
4454

55+
try {
4556
rootLogger.info(`running CLI`);
4657

4758
// factories (rather than instances) lets CoreClient build one client per
@@ -69,8 +80,16 @@ process.exit(
6980
rootLogger
7081
.child({ errorName: error.name, errorMessage: error.message, stack: error.stack ?? "" })
7182
.error();
83+
// TODO: add error details to telemetry recorder;
84+
commandRunTelemetryRecorder.set({ exit_reason: "failure" });
7285
throw e;
7386
} finally {
87+
await telemetryClient.emit(
88+
"cli.command_run",
89+
Date.now() - startTime,
90+
commandRunTelemetryRecorder.get(),
91+
);
92+
await telemetryClient.shutdown();
7493
await rootLogger.end();
7594
}
7695
}),

‎src/telemetry/client.test.tsx‎

Lines changed: 151 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,151 @@
1+
import { test, describe, beforeEach, afterEach, expect } from "bun:test";
2+
import { join } from "node:path";
3+
import { mkdtemp, rm } from "node:fs/promises";
4+
import { tmpdir } from "node:os";
5+
import { DefaultTelemetryClient } from "./client";
6+
import { TelemetryAttributesRecorder } from "./recorder";
7+
import { createFileLogger, type Logger } from "../logging";
8+
import { LOG_LEVEL } from "../logging";
9+
import { assertLogsMatch, TestGlobalConfigAccessor } from "../testing";
10+
import type { MetricSink } from "./types";
11+
import { LoggingSink } from "./loggingSink";
12+
13+
describe("DefaultTelemetryClient", () => {
14+
let tempDir: string;
15+
let logger: Logger;
16+
17+
beforeEach(async () => {
18+
tempDir = await mkdtemp(join(tmpdir(), "telemetry-client-test-"));
19+
logger = createFileLogger({
20+
filePath: join(tempDir, "output"),
21+
logLevel: LOG_LEVEL.DEBUG,
22+
});
23+
});
24+
25+
afterEach(async () => {
26+
await rm(tempDir, { recursive: true, force: true });
27+
});
28+
29+
test("emits metrics with resource and command attributes to configured sinks", async () => {
30+
const sink = new LoggingSink({ logger: logger.child({ module: "loggingSink" }) });
31+
const sessionId = "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee";
32+
const globalConfigAccessor = new TestGlobalConfigAccessor();
33+
const client = new DefaultTelemetryClient({
34+
logger,
35+
sessionId,
36+
globalConfigAccessor,
37+
metricSinks: [sink],
38+
});
39+
40+
const recorder = new TelemetryAttributesRecorder("cli.command_run", { exit_reason: "success" });
41+
42+
recorder.set({ exit_reason: "failure" });
43+
44+
await client.emit("cli.command_run", 123, recorder.get());
45+
46+
expect(sink.name).toBe("LoggingSink");
47+
await sink.shutdown();
48+
await client.shutdown();
49+
50+
const { installationId } = await globalConfigAccessor.get();
51+
await assertLogsMatch(tempDir, [
52+
{
53+
filter: (log: any) =>
54+
log.metricName === "cli.command_run" &&
55+
log.metricValue === 123 &&
56+
log.metricAttributes?.["exit_reason"] === "failure" &&
57+
log.metricAttributes?.["service.name"] === "agentcore-cli" &&
58+
log.metricAttributes?.["agentcore-cli.session_id"] === sessionId &&
59+
log.metricAttributes?.["agentcore-cli.installation_id"] === installationId,
60+
expectedCount: 1,
61+
},
62+
]);
63+
});
64+
65+
test("rejects malformed payloads before the sinks sees data", async () => {
66+
const recordedCalls: Array<{ metricName: string; value: number; attributes: any }> = [];
67+
const spySink: MetricSink = {
68+
name: "SpySink",
69+
record: (metricName, value, attributes) => {
70+
recordedCalls.push({ metricName, value, attributes });
71+
},
72+
shutdown: async () => {},
73+
};
74+
75+
const client = new DefaultTelemetryClient({
76+
logger,
77+
sessionId: "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee",
78+
globalConfigAccessor: new TestGlobalConfigAccessor(),
79+
metricSinks: [spySink],
80+
});
81+
82+
// Negative metric value violates z.number().min(0)
83+
await client.emit("cli.command_run", -1, { exit_reason: "success" });
84+
85+
// The sink should never have been called — validation rejects the payload
86+
expect(recordedCalls).toHaveLength(0);
87+
88+
await client.shutdown();
89+
90+
await assertLogsMatch(tempDir, [
91+
{
92+
filter: (log: any) =>
93+
log.msg === "failed to emit telemetry" && log.errorName === "ZodError",
94+
expectedCount: 1,
95+
},
96+
]);
97+
});
98+
99+
test("handles sink errors gracefully without throwing", async () => {
100+
const recordedMetrics: string[] = [];
101+
const goodSink: MetricSink = {
102+
name: "GoodSink",
103+
record: (metricName) => {
104+
recordedMetrics.push(metricName);
105+
},
106+
shutdown: async () => {},
107+
};
108+
109+
const badSink: MetricSink = {
110+
name: "BadSink",
111+
record: () => {
112+
throw new Error("record exploded");
113+
},
114+
shutdown: async () => {
115+
throw new Error("shutdown exploded");
116+
},
117+
};
118+
119+
const client = new DefaultTelemetryClient({
120+
logger,
121+
sessionId: "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee",
122+
globalConfigAccessor: new TestGlobalConfigAccessor(),
123+
metricSinks: [badSink, goodSink],
124+
});
125+
126+
// emit should not throw even though the sink's record() throws
127+
await client.emit("cli.command_run", 100, { exit_reason: "success" });
128+
// shutdown should not throw even though the sink's shutdown() rejects
129+
await client.shutdown();
130+
131+
// GoodSink still receives data despite BadSink throwing
132+
expect(recordedMetrics).toEqual(["cli.command_run"]);
133+
134+
await assertLogsMatch(tempDir, [
135+
{
136+
filter: (log: any) =>
137+
log.msg === "failed to record to sink 'BadSink'" &&
138+
log.errorName === "Error" &&
139+
log.errorMessage === "record exploded",
140+
expectedCount: 1,
141+
},
142+
{
143+
filter: (log: any) =>
144+
log.msg === "failed to shutdown metric sink with name 'BadSink'" &&
145+
log.errorName === "Error" &&
146+
log.errorMessage === "shutdown exploded",
147+
expectedCount: 1,
148+
},
149+
]);
150+
});
151+
});

‎src/telemetry/client.tsx‎

Lines changed: 115 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,115 @@
1+
import type { Logger } from "../logging";
2+
import { LoggingSink } from "./loggingSink";
3+
import { resourceAttributesSchema, type ResourceAttributes } from "./shapes";
4+
import os from "os";
5+
import {
6+
type AttributesOf,
7+
type MetricSink,
8+
type TelemetryClient,
9+
type ValueOf,
10+
METRICS,
11+
type MetricName,
12+
} from "./types";
13+
import type { GlobalConfigAccessor } from "../globalConfig";
14+
15+
export type DefaultTelemetryClientConfig = {
16+
logger: Logger;
17+
globalConfigAccessor: GlobalConfigAccessor;
18+
sessionId: string;
19+
metricSinks?: MetricSink[];
20+
};
21+
22+
/**
23+
* Implements {@link TelemetryClient} by validating and fanning out metrics to a list of {@link MetricSink} implementations.
24+
*/
25+
export class DefaultTelemetryClient implements TelemetryClient {
26+
private logger: Logger;
27+
private readonly sessionId: string;
28+
private globalConfigAccessor: GlobalConfigAccessor;
29+
private resourceAttributes: ResourceAttributes | undefined;
30+
private metricSinks: MetricSink[] | undefined;
31+
32+
constructor(config: DefaultTelemetryClientConfig) {
33+
this.logger = config.logger;
34+
this.sessionId = config.sessionId;
35+
this.globalConfigAccessor = config.globalConfigAccessor;
36+
this.resourceAttributes = undefined;
37+
this.metricSinks = config.metricSinks;
38+
}
39+
40+
async emit<TMetricName extends MetricName>(
41+
metricName: TMetricName,
42+
metricValue: ValueOf<TMetricName>,
43+
metricAttributes: Partial<AttributesOf<TMetricName>>,
44+
): Promise<void> {
45+
try {
46+
const metricSinks = this.getMetricSinks();
47+
const resourceAttributes = await this.getResourceAttributes();
48+
// merge in resource attributes with metric attributes before sending to sink.
49+
const attributes = {
50+
...resourceAttributes,
51+
...METRICS[metricName]["attributeSchema"].parse(metricAttributes),
52+
};
53+
54+
const validatedMetricValue = METRICS[metricName]["valueSchema"].parse(metricValue);
55+
56+
metricSinks.forEach((sink) => {
57+
try {
58+
sink.record(metricName, validatedMetricValue, attributes);
59+
} catch (e) {
60+
const error = e instanceof Error ? e : new Error(String(e));
61+
this.logger
62+
.child({ errorName: error.name, errorMessage: error.message })
63+
.warn(`failed to record to sink '${sink.name}'`);
64+
// do not allow a single sink failure to fail other sinks.
65+
}
66+
});
67+
} catch (e) {
68+
const error = e instanceof Error ? e : new Error(String(e));
69+
this.logger
70+
.child({ errorName: error.name, errorMessage: error.message })
71+
.warn(`failed to emit telemetry`);
72+
// telemetry is best-effort, don't throw.
73+
}
74+
}
75+
76+
async shutdown(): Promise<void> {
77+
const metricSinks = this.getMetricSinks();
78+
79+
const promises = metricSinks.map(async (sink) => {
80+
return sink.shutdown().catch((e) => {
81+
const error = e instanceof Error ? e : new Error(String(e));
82+
this.logger
83+
.child({ errorName: error.name, errorMessage: error.message })
84+
.warn(`failed to shutdown metric sink with name '${sink.name}'`);
85+
});
86+
});
87+
await Promise.all(promises);
88+
}
89+
90+
private getMetricSinks(): MetricSink[] {
91+
if (this.metricSinks !== undefined) return this.metricSinks;
92+
93+
this.metricSinks = [new LoggingSink({ logger: this.logger.child({ module: "loggingSink" }) })];
94+
95+
return this.metricSinks;
96+
}
97+
98+
private async getResourceAttributes(): Promise<ResourceAttributes> {
99+
if (this.resourceAttributes !== undefined) return this.resourceAttributes;
100+
101+
const globalConfig = await this.globalConfigAccessor.get();
102+
this.resourceAttributes = resourceAttributesSchema.parse({
103+
"service.name": "agentcore-cli",
104+
// TODO: wire up real package version.
105+
"service.version": "0.0.0",
106+
"agentcore-cli.installation_id": globalConfig.installationId,
107+
"agentcore-cli.session_id": this.sessionId,
108+
"os.type": os.type(),
109+
"os.version": os.release(),
110+
"host.arch": os.arch(),
111+
"node.version": process.version,
112+
});
113+
return this.resourceAttributes;
114+
}
115+
}

‎src/telemetry/index.tsx‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,3 @@
1+
export { DefaultTelemetryClient } from "./client";
2+
export { TelemetryAttributesRecorder } from "./recorder";
3+
export { type AttributesOf } from "./types";

‎src/telemetry/loggingSink.tsx‎

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,31 @@
1+
import type { Logger } from "../logging";
2+
import type { MetricSink } from "./types";
3+
4+
type LoggingSinkConfig = {
5+
logger: Logger;
6+
};
7+
8+
/**
9+
* An implementation of {@link MetricSink} that logs metrics using the given logger
10+
*/
11+
export class LoggingSink implements MetricSink {
12+
private logger: Logger;
13+
14+
constructor(config: LoggingSinkConfig) {
15+
this.logger = config.logger;
16+
}
17+
18+
record(
19+
metricName: string,
20+
metricValue: number,
21+
metricAttributes: Record<string, string | number>,
22+
): void {
23+
this.logger.child({ metricName, metricValue, metricAttributes }).info("recording telemetry");
24+
}
25+
26+
async shutdown(): Promise<void> {}
27+
28+
get name() {
29+
return "LoggingSink";
30+
}
31+
}

‎src/telemetry/recorder.tsx‎

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,27 @@
1+
import type { MetricName, AttributesOf } from "./types";
2+
3+
/**
4+
* A strongly typed recorder for accumulating metric attributes, bound to a specific metric's schema.
5+
*/
6+
export class TelemetryAttributesRecorder<TMetricName extends MetricName> {
7+
private attributes: Partial<AttributesOf<TMetricName>>;
8+
9+
constructor(
10+
_metricName: TMetricName,
11+
initialAttributes: Partial<AttributesOf<TMetricName>> = {},
12+
) {
13+
this.attributes = initialAttributes;
14+
}
15+
16+
get(): Partial<AttributesOf<TMetricName>> {
17+
return this.attributes;
18+
}
19+
20+
set(data: Partial<AttributesOf<TMetricName>>): Partial<AttributesOf<TMetricName>> {
21+
this.attributes = {
22+
...this.attributes,
23+
...data,
24+
};
25+
return this.attributes;
26+
}
27+
}

0 commit comments

Comments
 (0)