Skip to content

Commit 01eb60b

Browse files
HweinstockHweinstock
authored
feat(tel): implement otel sink (#1888)
* feat(tel): implement otel sink * docs(tel): add docstring comments to otel sink * test(tel): add a test of a local server for the collector sink * chore: remove extra space from telemetry test file * docs: add note in comment about default value * fix(tel): avoid letting flush failures stop shutdown * feat(tel): swap to cli from error method * fix(tel): remove duplicate resource attributes * feat(tel): add installationId header and temporarily preference * test(tel): adjust test to use passed down resource attributes
1 parent 64d601f commit 01eb60b

6 files changed

Lines changed: 286 additions & 28 deletions

File tree

‎bun.lock‎

Lines changed: 26 additions & 0 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎package.json‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -53,6 +53,10 @@
5353
"@aws-sdk/client-bedrock-agentcore-control": "^3.1102.0",
5454
"@aws-sdk/client-cloudwatch-logs": "^3.1092.0",
5555
"@aws-sdk/client-iam": "^3.1080.0",
56+
"@opentelemetry/api": "^1.9.1",
57+
"@opentelemetry/exporter-metrics-otlp-http": "^0.221.0",
58+
"@opentelemetry/resources": "^2.10.0",
59+
"@opentelemetry/sdk-metrics": "^2.10.0",
5660
"@smithy/core": "3.29.3",
5761
"@tanstack/react-query": "^5.101.2",
5862
"cli-truncate": "^6.1.1",

‎src/telemetry/client.test.tsx‎

Lines changed: 131 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -5,9 +5,10 @@ import os, { tmpdir } from "node:os";
55
import { DefaultTelemetryClient } from "./client";
66
import { createFileLogger, type Logger } from "../logging";
77
import { LOG_LEVEL } from "../logging";
8-
import { assertLogsMatch, TestGlobalConfigAccessor } from "../testing";
8+
import { assertLogsMatch, createSilentLogger, TestGlobalConfigAccessor } from "../testing";
99
import type { MetricSink } from "./types";
1010
import { FileSystemSink } from "./fileSystemSink";
11+
import { DEFAULT_GLOBAL_CONFIG } from "../globalConfig";
1112
import { PACKAGE_VERSION } from "../constants";
1213

1314
describe("DefaultTelemetryClient", () => {
@@ -28,9 +29,20 @@ describe("DefaultTelemetryClient", () => {
2829

2930
test("emits complete metrics to configured JSONL filesystem sinks", async () => {
3031
const auditFilePath = join(tempDir, "telemetry", "audit.jsonl");
32+
const sinkResourceAttributes = {
33+
"service.name": "agentcore-cli" as const,
34+
"service.version": "0.0.0",
35+
"agentcore-cli.installation_id": "00000000-0000-0000-0000-000000000000",
36+
"agentcore-cli.session_id": "00000000-0000-0000-0000-000000000000",
37+
"os.type": os.type(),
38+
"os.version": os.release(),
39+
"host.arch": os.arch(),
40+
"node.version": process.version,
41+
};
3142
const fileSystemSink = new FileSystemSink({
3243
logger: logger.child({ module: "fileSystemSink" }),
3344
filePath: auditFilePath,
45+
resourceAttributes: sinkResourceAttributes,
3446
});
3547
const sessionId = "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee";
3648
const globalConfigAccessor = new TestGlobalConfigAccessor();
@@ -59,32 +71,19 @@ describe("DefaultTelemetryClient", () => {
5971

6072
expect(fileSystemSink.getName()).toBe("FileSystemSink");
6173

62-
const { installationId } = await globalConfigAccessor.get();
63-
6474
const auditContents = await readFile(auditFilePath, "utf8");
6575

6676
const entries = auditContents
6777
.trimEnd()
6878
.split("\n")
6979
.map((line) => JSON.parse(line));
7080

71-
const resourceAttributes = {
72-
"service.name": "agentcore-cli",
73-
"service.version": PACKAGE_VERSION,
74-
"agentcore-cli.installation_id": installationId,
75-
"agentcore-cli.session_id": sessionId,
76-
"os.type": os.type(),
77-
"os.version": os.release(),
78-
"host.arch": os.arch(),
79-
"node.version": process.version,
80-
};
81-
8281
expect(entries).toEqual([
8382
{
8483
metricName: "cli.command_run",
8584
value: 123,
8685
attrs: {
87-
...resourceAttributes,
86+
...sinkResourceAttributes,
8887
exit_reason: "success",
8988
command_path: "/agentcore",
9089
is_tui: false,
@@ -94,7 +93,7 @@ describe("DefaultTelemetryClient", () => {
9493
metricName: "cli.command_run",
9594
value: 456,
9695
attrs: {
97-
...resourceAttributes,
96+
...sinkResourceAttributes,
9897
exit_reason: "failure",
9998
command_path: "/agentcore",
10099
is_tui: false,
@@ -191,6 +190,16 @@ describe("DefaultTelemetryClient", () => {
191190
const sink = new FileSystemSink({
192191
logger: logger.child({ module: "fileSystemSink" }),
193192
filePath: tempDir,
193+
resourceAttributes: {
194+
"service.name": "agentcore-cli",
195+
"service.version": "0.0.0",
196+
"agentcore-cli.installation_id": "00000000-0000-0000-0000-000000000000",
197+
"agentcore-cli.session_id": "00000000-0000-0000-0000-000000000000",
198+
"os.type": os.type(),
199+
"os.version": os.release(),
200+
"host.arch": os.arch(),
201+
"node.version": process.version,
202+
},
194203
});
195204

196205
const client = new DefaultTelemetryClient({
@@ -274,3 +283,109 @@ describe("DefaultTelemetryClient", () => {
274283
]);
275284
});
276285
});
286+
287+
describe("OtelHistogramSink", () => {
288+
let testCollector: ReturnType<typeof Bun.serve>;
289+
let receivedBodies: any[];
290+
291+
const logger = createSilentLogger();
292+
293+
beforeEach(async () => {
294+
receivedBodies = [];
295+
testCollector = Bun.serve({
296+
port: 0,
297+
async fetch(req) {
298+
const body = await req.json();
299+
receivedBodies.push(body);
300+
return new Response("", { status: 200 });
301+
},
302+
});
303+
});
304+
305+
afterEach(async () => {
306+
testCollector.stop(true);
307+
});
308+
309+
test.each([
310+
{ enabled: true, expectRequests: true },
311+
{ enabled: false, expectRequests: false },
312+
])(
313+
"telemetry.enabled=$enabled → collector receives requests=$expectRequests",
314+
async ({ enabled, expectRequests }) => {
315+
const sessionId = "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee";
316+
const exitReason = "success";
317+
const commandPath = "/agentcore";
318+
const metricName = "cli.command_run";
319+
const scopeName = "agentcore-cli";
320+
const serviceName = "agentcore-cli";
321+
const globalConfigAccessor = new TestGlobalConfigAccessor({
322+
initialConfigData: {
323+
...DEFAULT_GLOBAL_CONFIG,
324+
telemetry: {
325+
enabled,
326+
audit: false,
327+
endpoint: `http://localhost:${testCollector.port}`,
328+
},
329+
},
330+
});
331+
332+
const client = new DefaultTelemetryClient({
333+
logger,
334+
sessionId,
335+
globalConfigAccessor,
336+
});
337+
338+
const event = client.createMetricEvent(metricName, {
339+
exit_reason: exitReason,
340+
command_path: commandPath,
341+
});
342+
await event.emit(100);
343+
await client.shutdown();
344+
345+
if (expectRequests) {
346+
expect(receivedBodies.length).toBeGreaterThan(0);
347+
348+
const body = receivedBodies[0];
349+
expect(body).toMatchObject({
350+
resourceMetrics: [
351+
{
352+
resource: {
353+
attributes: expect.arrayContaining([
354+
{ key: "service.name", value: { stringValue: serviceName } },
355+
{
356+
key: "agentcore-cli.session_id",
357+
value: { stringValue: sessionId },
358+
},
359+
{ key: "os.type", value: { stringValue: os.type() } },
360+
{ key: "host.arch", value: { stringValue: os.arch() } },
361+
]),
362+
},
363+
scopeMetrics: [
364+
{
365+
scope: { name: scopeName },
366+
metrics: [
367+
{
368+
name: metricName,
369+
histogram: {
370+
dataPoints: [
371+
{
372+
attributes: expect.arrayContaining([
373+
{ key: "exit_reason", value: { stringValue: exitReason } },
374+
{ key: "command_path", value: { stringValue: commandPath } },
375+
]),
376+
},
377+
],
378+
},
379+
},
380+
],
381+
},
382+
],
383+
},
384+
],
385+
});
386+
} else {
387+
expect(receivedBodies).toHaveLength(0);
388+
}
389+
},
390+
);
391+
});

‎src/telemetry/client.tsx‎

Lines changed: 13 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,7 @@ import {
1313
import type { GlobalConfigAccessor } from "../globalConfig";
1414
import { FileSystemSink } from "./fileSystemSink";
1515
import path from "path";
16+
import { OtelHistogramSink } from "./otelSink";
1617
import { PACKAGE_VERSION } from "../constants";
1718

1819
export type DefaultTelemetryClientConfig = {
@@ -52,7 +53,6 @@ export class DefaultTelemetryClient implements TelemetryClient {
5253
initialAttributes,
5354
logger: this.logger,
5455
getSinks: () => this.getMetricSinks(),
55-
getResourceAttributes: () => this.getResourceAttributes(),
5656
});
5757
}
5858

@@ -72,6 +72,7 @@ export class DefaultTelemetryClient implements TelemetryClient {
7272

7373
private getMetricSinks: () => Promise<MetricSink[]> = once(async () => {
7474
if (this.metricSinksOverride) return this.metricSinksOverride;
75+
const resourceAttributes = await this.getResourceAttributes();
7576

7677
const metricSinks = [];
7778

@@ -82,6 +83,16 @@ export class DefaultTelemetryClient implements TelemetryClient {
8283
new FileSystemSink({
8384
logger: this.logger.child({ module: "fileSystemSink" }),
8485
filePath: this.auditFilePath,
86+
resourceAttributes,
87+
}),
88+
);
89+
90+
if (globalConfig.telemetry.enabled)
91+
metricSinks.push(
92+
new OtelHistogramSink({
93+
logger: this.logger.child({ module: "otelCollectorSink" }),
94+
collectorEndpoint: globalConfig.telemetry.endpoint,
95+
resourceAttributes,
8596
}),
8697
);
8798

@@ -114,7 +125,6 @@ type InMemoryMetricEventConfig<TMetricName extends MetricName> = {
114125
initialAttributes?: Partial<AttributesOf<TMetricName>>;
115126
logger: Logger;
116127
getSinks: () => Promise<MetricSink[]>;
117-
getResourceAttributes: () => Promise<ResourceAttributes>;
118128
};
119129

120130
/** An in-memory implementation of {@link MetricEvent} that accumulates attributes and emits on end() **/
@@ -123,14 +133,12 @@ class InMemoryMetricEvent<TMetricName extends MetricName> implements MetricEvent
123133
private readonly metricName: TMetricName;
124134
private readonly logger: Logger;
125135
private readonly getSinks: () => Promise<MetricSink[]>;
126-
private readonly getResourceAttributes: () => Promise<ResourceAttributes>;
127136

128137
constructor(config: InMemoryMetricEventConfig<TMetricName>) {
129138
this.metricName = config.metricName;
130139
this.data = config.initialAttributes ?? {};
131140
this.logger = config.logger;
132141
this.getSinks = config.getSinks;
133-
this.getResourceAttributes = config.getResourceAttributes;
134142
}
135143

136144
setAttributes(newData: Partial<AttributesOf<TMetricName>>): void {
@@ -143,18 +151,12 @@ class InMemoryMetricEvent<TMetricName extends MetricName> implements MetricEvent
143151
async emit(value: ValueOf<TMetricName>): Promise<void> {
144152
const metricAttributes = METRICS[this.metricName]["attributeSchema"].parse(this.data);
145153
const validatedValue = METRICS[this.metricName]["valueSchema"].parse(value);
146-
const resourceAttributes = await this.getResourceAttributes();
147-
148-
const attributes = {
149-
...resourceAttributes,
150-
...metricAttributes,
151-
};
152154

153155
const sinks = await this.getSinks();
154156

155157
sinks.forEach((sink) => {
156158
try {
157-
sink.send(this.metricName, validatedValue, attributes);
159+
sink.send(this.metricName, validatedValue, metricAttributes);
158160
} catch (e) {
159161
const error = e instanceof Error ? e : new Error(String(e));
160162
this.logger

‎src/telemetry/fileSystemSink.tsx‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,11 +1,13 @@
11
import type { Logger } from "../logging";
2+
import type { ResourceAttributes } from "./shapes";
23
import type { MetricSink } from "./types";
34
import { mkdir, appendFile } from "fs/promises";
45
import { dirname } from "path";
56

67
export type FileSystemSinkConfig = {
78
logger: Logger;
89
filePath: string;
10+
resourceAttributes: ResourceAttributes;
911
};
1012

1113
/** An implementation of {@link MetricSink} that sends all data to the specified file in JSONL format **/
@@ -15,13 +17,16 @@ export class FileSystemSink implements MetricSink {
1517
private readonly filePath: string;
1618
private logger: Logger;
1719

20+
private readonly resourceAttributes: ResourceAttributes;
21+
1822
/* a chain of promises describing the pending writes to the audit file */
1923
private pendingWrite: Promise<void>;
2024

2125
constructor(config: FileSystemSinkConfig) {
2226
this.filePath = config.filePath;
2327
this.logger = config.logger.child({ fsSinkFilePath: this.filePath });
2428
this.name = new.target.name;
29+
this.resourceAttributes = config.resourceAttributes;
2530

2631
this.pendingWrite = Promise.resolve();
2732
}
@@ -32,7 +37,7 @@ export class FileSystemSink implements MetricSink {
3237
attributes: Record<string, string | number | boolean>,
3338
): void {
3439
this.pendingWrite = this.pendingWrite.then(() =>
35-
this.appendEntry({ metricName, value, attrs: attributes }),
40+
this.appendEntry({ metricName, value, attrs: { ...this.resourceAttributes, ...attributes } }),
3641
);
3742
}
3843

0 commit comments

Comments
 (0)