diff --git a/src/core/observability.test.ts b/src/core/observability.test.ts index cccbe7622..30777dc93 100644 --- a/src/core/observability.test.ts +++ b/src/core/observability.test.ts @@ -1,7 +1,6 @@ import { describe, expect, test } from "bun:test"; import { GetQueryResultsCommand, - ResourceNotFoundException, StartQueryCommand, type CloudWatchLogsClient, } from "@aws-sdk/client-cloudwatch-logs"; @@ -266,9 +265,12 @@ function insightsLogs(results: { field: string; value: string }[][]) { return { logs, queries }; } -describe("ObservabilityClient.listRuntimeTraces", () => { - const INPUT = { - runtimeId: "my_agent-AbC123XyZ9", +const TRACE_SOURCE = { + logGroupName: "/aws/bedrock-agentcore/runtimes/my_agent-AbC123XyZ9-DEFAULT", +}; + +describe("ObservabilityClient.listTraces", () => { + const QUERY = { startTimeMs: 1_700_000_000_123, endTimeMs: 1_700_003_600_456, limit: 5, @@ -277,12 +279,10 @@ describe("ObservabilityClient.listRuntimeTraces", () => { test("aggregates traces with a stats-by-traceId query over the runtime log group", async () => { const { logs, queries } = insightsLogs([]); - await clientWith(logs).listRuntimeTraces(INPUT, OPTIONS); + await clientWith(logs).listTraces(TRACE_SOURCE, QUERY, OPTIONS); expect(queries).toHaveLength(1); - expect(queries[0]!.logGroupNames).toEqual([ - "/aws/bedrock-agentcore/runtimes/my_agent-AbC123XyZ9-DEFAULT", - ]); + expect(queries[0]!.logGroupNames).toEqual([TRACE_SOURCE.logGroupName]); // Epoch ms narrows to whole seconds. expect(queries[0]!.startTime).toBe(1_700_000_000); expect(queries[0]!.endTime).toBe(1_700_003_600); @@ -311,7 +311,7 @@ describe("ObservabilityClient.listRuntimeTraces", () => { ], ]); - const traces = await clientWith(logs).listRuntimeTraces(INPUT, OPTIONS); + const traces = await clientWith(logs).listTraces(TRACE_SOURCE, QUERY, OPTIONS); expect(traces).toEqual([ { @@ -324,21 +324,10 @@ describe("ObservabilityClient.listRuntimeTraces", () => { { traceId: "def456", timestamp: "1700000002000", sessionId: undefined, spanCount: undefined }, ]); }); - - test("translates a missing log group into invoked-yet guidance", async () => { - const logs = fakeLogs(async () => { - throw new ResourceNotFoundException({ message: "no such group", $metadata: {} }); - }); - - await expect(clientWith(logs).listRuntimeTraces(INPUT, OPTIONS)).rejects.toThrow( - "Has the runtime been invoked yet?", - ); - }); }); -describe("ObservabilityClient.getRuntimeTrace", () => { - const INPUT = { - runtimeId: "my_agent-AbC123XyZ9", +describe("ObservabilityClient.getTrace", () => { + const QUERY = { traceId: "68b2fabc0000000000abcdef", startTimeMs: 1_700_000_000_000, endTimeMs: 1_700_003_600_000, @@ -350,7 +339,7 @@ describe("ObservabilityClient.getRuntimeTrace", () => { }); await expect( - clientWith(logs).getRuntimeTrace({ ...INPUT, traceId: "not'a$trace" }, OPTIONS), + clientWith(logs).getTrace(TRACE_SOURCE, { ...QUERY, traceId: "not'a$trace" }, OPTIONS), ).rejects.toThrow("Invalid trace ID format. Expected a hex string (e.g., abc123def456)."); }); @@ -367,7 +356,7 @@ describe("ObservabilityClient.getRuntimeTrace", () => { ], ]); - const records = await clientWith(logs).getRuntimeTrace(INPUT, OPTIONS); + const records = await clientWith(logs).getTrace(TRACE_SOURCE, QUERY, OPTIONS); expect(queries[0]!.queryString).toBe( "fields @timestamp, @message\n" + @@ -388,8 +377,20 @@ describe("ObservabilityClient.getRuntimeTrace", () => { test("fails when the trace has no records", async () => { const { logs } = insightsLogs([]); - await expect(clientWith(logs).getRuntimeTrace(INPUT, OPTIONS)).rejects.toThrow( + await expect(clientWith(logs).getTrace(TRACE_SOURCE, QUERY, OPTIONS)).rejects.toThrow( "No trace data found for trace ID: 68b2fabc0000000000abcdef", ); }); + + test("returns every record when the trace reaches the 10,000-record query limit", async () => { + const { logs } = insightsLogs( + Array.from({ length: 10_000 }, (_, index) => [ + { field: "@message", value: `record-${index}` }, + ]), + ); + + const records = await clientWith(logs).getTrace(TRACE_SOURCE, QUERY, OPTIONS); + + expect(records).toHaveLength(10_000); + }); }); diff --git a/src/core/observability.ts b/src/core/observability.ts index 747e6b57d..97a38ec6c 100644 --- a/src/core/observability.ts +++ b/src/core/observability.ts @@ -1,6 +1,5 @@ import { GetQueryResultsCommand, - ResourceNotFoundException, StartQueryCommand, type CloudWatchLogsClient, type ResultField, @@ -17,22 +16,14 @@ import { } from "../errors"; import type { ReadWriteJson } from "../io"; import type { Project } from "../handlers/project/types"; -import type { - CoreObservabilityClient, - DeployedRuntime, - GetRuntimeTraceInput, - ListRuntimeTracesInput, - TraceRecord, - TraceSummary, -} from "../handlers/runtime/types"; +import type { CoreObservabilityClient, DeployedRuntime } from "../handlers/runtime/types"; import { AwsDeploymentTargetsSchema } from "../projectSchemas/aws-targets"; import { CloudWatchClient, ObservabilityClient as GenericObservabilityClient, } from "./observability/index"; import { isStackNotFound } from "./project/backends/cdk/environment"; -import type { AwsClients, CoreOptions } from "./types"; -import { toClientConfig } from "./utils"; +import type { AwsClients } from "./types"; // Shared CloudWatch observability helpers. AgentCore Runtimes write their logs // and OTel telemetry to per-runtime CloudWatch log groups; both the eval flows @@ -243,21 +234,18 @@ export interface ObservabilityClientDeps { } /** - * Runtime-specific observability APIs retained for the existing trace and - * project-resolution commands. Generic log reads are inherited from the new - * shared observability client. + * Project-resolution APIs retained for the existing project command path. + * Shared observability operations are inherited from the generic client. */ export class ObservabilityClient extends GenericObservabilityClient implements CoreObservabilityClient { - private readonly clients: AwsClients; private readonly readJson: ReadWriteJson; private readonly describeStackOutputs: DescribeStackOutputs; constructor(clients: AwsClients, deps: ObservabilityClientDeps) { super(new CloudWatchClient(clients)); - this.clients = clients; this.readJson = deps.readJson; this.describeStackOutputs = deps.describeStackOutputs ?? describeStackOutputsWithSdk; } @@ -320,124 +308,4 @@ export class ObservabilityClient targetName: target.name, }; } - - /** - * Lists the runtime's recent traces by aggregating its telemetry records with - * a Logs Insights `stats … by traceId` query (mirrors the old CLI's - * list-traces operation), newest first. - */ - async listRuntimeTraces( - input: ListRuntimeTracesInput, - options: CoreOptions, - ): Promise { - const logGroupName = runtimeLogGroup(input.runtimeId, DEFAULT_ENDPOINT_QUALIFIER); - // Infrastructure records carry an empty traceId; excluding them before the - // aggregation keeps them from occupying one of the `limit` buckets (the old - // CLI filtered afterwards, silently returning one trace fewer). - const queryString = - `filter ispresent(traceId) and traceId != ""\n` + - `| stats earliest(@timestamp) as firstSeen, latest(@timestamp) as lastSeen, ` + - `count(*) as spanCount, earliest(attributes.session.id) as sessionId by traceId\n` + - `| sort lastSeen desc\n` + - `| limit ${Math.floor(input.limit)}`; - - const rows = await this.runTraceQuery(input, logGroupName, queryString, options); - - const traces: TraceSummary[] = []; - for (const row of rows) { - const fields = fieldMap(row); - if (!fields.traceId) continue; - traces.push({ - traceId: fields.traceId, - timestamp: fields.lastSeen ?? fields.firstSeen ?? "unknown", - sessionId: fields.sessionId, - spanCount: fields.spanCount, - }); - } - return traces; - } - - /** - * Downloads every log record belonging to one trace, oldest first. The - * `@message` body is JSON-parsed when possible; other Insights fields pass - * through as returned. - */ - async getRuntimeTrace(input: GetRuntimeTraceInput, options: CoreOptions): Promise { - if (!TRACE_ID_PATTERN.test(input.traceId)) { - throw new InputValidationError( - "Invalid trace ID format. Expected a hex string (e.g., abc123def456).", - { meta: { traceId: input.traceId } }, - ); - } - - const logGroupName = runtimeLogGroup(input.runtimeId, DEFAULT_ENDPOINT_QUALIFIER); - const queryString = - `fields @timestamp, @message\n` + - `| filter traceId = '${sanitizeQueryValue(input.traceId)}'\n` + - `| sort @timestamp asc\n` + - `| limit 10000`; - - const rows = await this.runTraceQuery(input, logGroupName, queryString, options); - if (rows.length === 0) { - throw new ResourceNotFoundError(`No trace data found for trace ID: ${input.traceId}`, { - meta: { traceId: input.traceId }, - }); - } - - return rows.map((row) => { - const record: TraceRecord = fieldMap(row); - const message = record["@message"]; - if (typeof message === "string") { - try { - record["@message"] = JSON.parse(message); - } catch { - // Keep the original string when the body is not valid JSON. - } - } - return record; - }); - } - - private async runTraceQuery( - input: { runtimeId: string; startTimeMs: number; endTimeMs: number }, - logGroupName: string, - queryString: string, - options: CoreOptions, - ): Promise { - const logs = this.clients.logs(toClientConfig(options)); - const startSec = Math.floor(input.startTimeMs / 1000); - const endSec = Math.floor(input.endTimeMs / 1000); - try { - return await runInsightsQuery(logs, [logGroupName], queryString, startSec, endSec); - } catch (error) { - if (error instanceof ResourceNotFoundException) { - throw missingLogGroupError(input.runtimeId, logGroupName, error); - } - throw error; - } - } -} - -// Trace ids are hex strings, optionally dash-separated (mirrors the old CLI). -const TRACE_ID_PATTERN = /^[a-fA-F0-9-]+$/; - -// fieldMap flattens one Insights result row into a name -> value record. -function fieldMap(row: ResultField[]): Record { - const fields: Record = {}; - for (const field of row) { - if (field.field && field.value !== undefined) fields[field.field] = field.value; - } - return fields; -} - -function missingLogGroupError( - runtimeId: string, - logGroupName: string, - cause?: unknown, -): ResourceNotFoundError { - return new ResourceNotFoundError( - `No logs found for runtime '${runtimeId}': log group ${logGroupName} does not exist. ` + - `Has the runtime been invoked yet?`, - { cause, meta: { runtimeId, logGroupName } }, - ); } diff --git a/src/core/observability/client.ts b/src/core/observability/client.ts index c648121fa..59267c3be 100644 --- a/src/core/observability/client.ts +++ b/src/core/observability/client.ts @@ -2,12 +2,22 @@ import type { CoreOptions } from "../types"; import { CloudWatchClient } from "./cloudWatchClient"; import type { CloudWatchLogEvent, + GetTraceQuery, InsightsQuery, InsightsQueryRow, + ListTracesQuery, LogSearchQuery, LogSource, LogTailQuery, + TraceRecord, + TraceSummary, } from "./types"; +import { + getTraceInsightsQuery, + listTracesInsightsQuery, + normalizeTraceRecords, + normalizeTraceSummaries, +} from "./traces"; /** Shared observability API over explicit CloudWatch log-group targets. */ export class ObservabilityClient { @@ -39,4 +49,24 @@ export class ObservabilityClient { ): Promise { return this.cloudWatch.queryLogs(source, query, options, signal); } + + async listTraces( + source: LogSource, + query: ListTracesQuery, + options: CoreOptions, + signal?: AbortSignal, + ): Promise { + const rows = await this.queryLogs(source, listTracesInsightsQuery(query), options, signal); + return normalizeTraceSummaries(rows); + } + + async getTrace( + source: LogSource, + query: GetTraceQuery, + options: CoreOptions, + signal?: AbortSignal, + ): Promise { + const rows = await this.queryLogs(source, getTraceInsightsQuery(query), options, signal); + return normalizeTraceRecords(rows, query.traceId); + } } diff --git a/src/core/observability/index.ts b/src/core/observability/index.ts index bcf839bb9..d8a755d02 100644 --- a/src/core/observability/index.ts +++ b/src/core/observability/index.ts @@ -6,11 +6,16 @@ export { sanitizeQueryValue, type InsightsRowLimit, } from "./insights"; +export { TRACE_RECORD_LIMIT } from "./traces"; export type { CloudWatchLogEvent, + GetTraceQuery, InsightsQuery, InsightsQueryRow, + ListTracesQuery, LogSearchQuery, LogSource, LogTailQuery, + TraceRecord, + TraceSummary, } from "./types"; diff --git a/src/core/observability/traces.ts b/src/core/observability/traces.ts new file mode 100644 index 000000000..ba7f880ba --- /dev/null +++ b/src/core/observability/traces.ts @@ -0,0 +1,89 @@ +import { InputValidationError, ResourceNotFoundError } from "../../errors"; +import { sanitizeQueryValue } from "./insights"; +import type { + GetTraceQuery, + InsightsQuery, + InsightsQueryRow, + ListTracesQuery, + TraceRecord, + TraceSummary, +} from "./types"; + +export const TRACE_RECORD_LIMIT = 10_000; + +// Trace ids are hex strings, optionally dash-separated (mirrors the old CLI). +const TRACE_ID_PATTERN = /^[a-fA-F0-9-]+$/; + +export function listTracesInsightsQuery(query: ListTracesQuery): InsightsQuery { + const limit = Math.floor(query.limit); + if (!Number.isFinite(limit) || limit <= 0) { + throw new InputValidationError("Trace limit must be a positive integer", { + meta: { limit: query.limit }, + }); + } + + return { + startTimeMs: query.startTimeMs, + endTimeMs: query.endTimeMs, + queryString: + `filter ispresent(traceId) and traceId != ""\n` + + `| stats earliest(@timestamp) as firstSeen, latest(@timestamp) as lastSeen, ` + + `count(*) as spanCount, earliest(attributes.session.id) as sessionId by traceId\n` + + `| sort lastSeen desc\n` + + `| limit ${limit}`, + }; +} + +export function normalizeTraceSummaries(rows: InsightsQueryRow[]): TraceSummary[] { + const traces: TraceSummary[] = []; + for (const row of rows) { + if (!row.traceId) continue; + traces.push({ + traceId: row.traceId, + timestamp: row.lastSeen ?? row.firstSeen ?? "unknown", + sessionId: row.sessionId, + spanCount: row.spanCount, + }); + } + return traces; +} + +export function getTraceInsightsQuery(query: GetTraceQuery): InsightsQuery { + if (!TRACE_ID_PATTERN.test(query.traceId)) { + throw new InputValidationError( + "Invalid trace ID format. Expected a hex string (e.g., abc123def456).", + { meta: { traceId: query.traceId } }, + ); + } + + return { + startTimeMs: query.startTimeMs, + endTimeMs: query.endTimeMs, + queryString: + `fields @timestamp, @message\n` + + `| filter traceId = '${sanitizeQueryValue(query.traceId)}'\n` + + `| sort @timestamp asc\n` + + `| limit ${TRACE_RECORD_LIMIT}`, + }; +} + +export function normalizeTraceRecords(rows: InsightsQueryRow[], traceId: string): TraceRecord[] { + if (rows.length === 0) { + throw new ResourceNotFoundError(`No trace data found for trace ID: ${traceId}`, { + meta: { traceId }, + }); + } + + return rows.map((row) => { + const record: TraceRecord = { ...row }; + const message = record["@message"]; + if (typeof message === "string") { + try { + record["@message"] = JSON.parse(message); + } catch { + // Keep the original string when the body is not valid JSON. + } + } + return record; + }); +} diff --git a/src/core/observability/types.ts b/src/core/observability/types.ts index 9e2d3e6ed..ffeb0c8c1 100644 --- a/src/core/observability/types.ts +++ b/src/core/observability/types.ts @@ -32,3 +32,26 @@ export type InsightsQuery = { }; export type InsightsQueryRow = Record; + +export type ListTracesQuery = { + startTimeMs: number; + endTimeMs: number; + limit: number; +}; + +export type GetTraceQuery = { + traceId: string; + startTimeMs: number; + endTimeMs: number; +}; + +/** One trace aggregated from telemetry records, newest first. */ +export type TraceSummary = { + traceId: string; + timestamp: string; + sessionId?: string; + spanCount?: string; +}; + +/** One telemetry record belonging to a trace */ +export type TraceRecord = Record; diff --git a/src/handlers/observability/traces.ts b/src/handlers/observability/traces.ts new file mode 100644 index 000000000..9779b3db1 --- /dev/null +++ b/src/handlers/observability/traces.ts @@ -0,0 +1,216 @@ +import { mkdir } from "node:fs/promises"; +import { dirname } from "node:path"; +import z from "zod"; +import type { + GetTraceQuery, + ListTracesQuery, + TraceRecord, + TraceSummary, +} from "../../core/observability/index"; +import { TRACE_RECORD_LIMIT } from "../../core/observability/index"; +import { FileWriteError } from "../../errors"; +import { atomicWrite, type AppIO } from "../../io"; +import { + argument, + createHandler, + flag, + Router, + type Context, + type Flag, + type Handler, +} from "../../router"; +import { withUserCancellation } from "../../runnable"; +import { JsonRendererKey } from "../../tui"; +import { JsonKey } from "../keys"; +import { resolveTimeWindow } from "./time"; +import type { ResourceFlagValues } from "./types"; + +const DEFAULT_TRACES_WINDOW_MS = 12 * 3_600_000; +const TRACE_ID_WIDTH = 34; +const TIMESTAMP_WIDTH = 22; + +const traceWindowFlags = [ + flag( + "since", + 'window start: "5m", "1h", "2d", ISO 8601, epoch ms, or "now" (default 12h ago)', + z.string().min(1).optional(), + ), + flag( + "until", + 'window end: "5m", "1h", "2d", ISO 8601, epoch ms, or "now" (default now)', + z.string().min(1).optional(), + ), +] as const; + +const outputSchema = z.string().min(1, "requires a nonempty path").optional(); + +const listTraceFlags = [ + flag("limit", "maximum number of traces to display", z.number().int().positive().default(20)), + ...traceWindowFlags, +] as const; + +const getTraceFlags = [ + flag("output", "the output file path", outputSchema), + ...traceWindowFlags, +] as const; + +type ListTraceFlagValues = ResourceFlagValues; +type GetTraceFlagValues = ResourceFlagValues; + +export function formatTraceTimestamp(timestamp: string): string { + const epochMs = Number(timestamp); + if (isNaN(epochMs)) return timestamp; + return new Date(epochMs) + .toISOString() + .replace("T", " ") + .replace(/\.\d+Z$/, "Z"); +} + +export function formatTraceTable(traces: TraceSummary[]): string { + const lines = [ + `${"TRACE ID".padEnd(TRACE_ID_WIDTH)}${"TIMESTAMP".padEnd(TIMESTAMP_WIDTH)}SESSION ID`, + ]; + for (const trace of traces) { + lines.push( + trace.traceId.padEnd(TRACE_ID_WIDTH) + + formatTraceTimestamp(trace.timestamp).padEnd(TIMESTAMP_WIDTH) + + (trace.sessionId ?? "-"), + ); + } + return lines.join("\n") + "\n"; +} + +export function createListTracesHandler[]>( + io: AppIO, + config: { + description: string; + flags: F; + read( + ctx: Context, + values: ResourceFlagValues & ListTraceFlagValues, + query: ListTracesQuery, + signal: AbortSignal, + ): Promise; + }, +): Handler { + const listFlags = [...config.flags, ...listTraceFlags] as const; + + return createHandler({ + name: "list", + description: config.description, + flags: listFlags, + handle: async (ctx, values) => { + const parsed = values as unknown as ResourceFlagValues & ListTraceFlagValues; + const query: ListTracesQuery = { + ...resolveTimeWindow({ + since: parsed.since, + until: parsed.until, + defaultWindowMs: DEFAULT_TRACES_WINDOW_MS, + }), + limit: parsed.limit, + }; + const traces = await withUserCancellation((signal) => + config.read(ctx, parsed, query, signal), + ); + + if (ctx.require(JsonKey)) { + ctx.require(JsonRendererKey).renderJson({ traces }); + return; + } + if (traces.length === 0) { + io.stderr.write( + "No traces found in the specified time range. Traces take 2-3 minutes " + + "to appear after an invocation.\n", + ); + return; + } + io.stdout.write(formatTraceTable(traces)); + }, + }); +} + +export function createGetTraceHandler[]>( + io: AppIO, + config: { + description: string; + outputDescription: string; + flags: F; + read( + ctx: Context, + values: ResourceFlagValues & GetTraceFlagValues, + query: GetTraceQuery, + signal: AbortSignal, + ): Promise; + resolveOutputPath( + ctx: Context, + values: ResourceFlagValues & GetTraceFlagValues, + request: { traceId: string; output?: string }, + ): string | Promise; + }, +): Handler { + const getFlags = [ + ...config.flags, + flag("output", config.outputDescription, outputSchema), + ...traceWindowFlags, + ] as const; + + return createHandler({ + name: "get", + description: config.description, + arguments: [argument("trace-id", "the trace ID to download", z.string().min(1))], + flags: getFlags, + handle: async (ctx, values, args) => { + const parsed = values as unknown as ResourceFlagValues & GetTraceFlagValues; + const traceId = args["trace-id"]; + const query: GetTraceQuery = { + ...resolveTimeWindow({ + since: parsed.since, + until: parsed.until, + defaultWindowMs: DEFAULT_TRACES_WINDOW_MS, + }), + traceId, + }; + const records = await withUserCancellation((signal) => + config.read(ctx, parsed, query, signal), + ); + const filePath = await config.resolveOutputPath(ctx, parsed, { + traceId, + output: parsed.output, + }); + + try { + await mkdir(dirname(filePath), { recursive: true }); + await atomicWrite(filePath, JSON.stringify(records, null, 2)); + } catch (error) { + throw new FileWriteError( + `Could not write the trace file at ${filePath}: ` + + `${error instanceof Error ? error.message : String(error)}`, + { cause: error, meta: { filePath } }, + ); + } + + if (records.length >= TRACE_RECORD_LIMIT) { + io.stderr.write( + `Warning: The trace query returned ${TRACE_RECORD_LIMIT.toLocaleString("en-US")} ` + + "records, the maximum; the saved file may be incomplete. Narrow the time window " + + "with --since and --until if needed.\n", + ); + } + + if (ctx.require(JsonKey)) { + ctx.require(JsonRendererKey).renderJson({ filePath, recordCount: records.length }); + return; + } + io.stderr.write(`Saved ${records.length} records for trace ${traceId}\n`); + io.stdout.write(`${filePath}\n`); + }, + }); +} + +export function createTracesHandler(config: { + description: string; + list: Handler; + get: Handler; +}): Router { + return new Router("traces", config.description).handler(config.list).handler(config.get); +} diff --git a/src/handlers/runtime/traces/get/index.tsx b/src/handlers/runtime/traces/get/index.tsx index ab8da87bc..39b789e29 100644 --- a/src/handlers/runtime/traces/get/index.tsx +++ b/src/handlers/runtime/traces/get/index.tsx @@ -1,81 +1,24 @@ -import { mkdir, writeFile } from "node:fs/promises"; -import { dirname } from "node:path"; -import z from "zod"; -import { parseTimeString } from "../../../../core/observability"; -import { FileWriteError } from "../../../../errors"; +import { DEFAULT_ENDPOINT_QUALIFIER, runtimeLogGroup } from "../../../../core/observability"; import type { AppIO } from "../../../../io"; -import { argument, createHandler, flag } from "../../../../router"; -import { JsonRendererKey } from "../../../../tui"; -import { JsonKey } from "../../../keys"; +import { flag } from "../../../../router"; +import { createGetTraceHandler } from "../../../observability/traces"; import type { Core } from "../../../types"; +import { coreOptsFromCtx } from "../../../utils"; import { runtimeIdSchema } from "../../invoke/request"; -import { resolveRuntimeTarget } from "../../resolveRuntimeTarget"; -import { DEFAULT_TRACES_WINDOW_MS } from "../index"; -import { resolveTraceOutputPath } from "./outputPath"; +import { resolveTraceOutputPath } from "../outputPath"; + +const runtimeFlags = [flag("id", "the ID of the Runtime", runtimeIdSchema)] as const; export const createGetRuntimeTraceHandler = (core: Core, io: AppIO) => - createHandler({ - name: "get", + createGetTraceHandler(io, { description: "download a trace's log records to a JSON file", - arguments: [argument("trace-id", "the trace ID to download", z.string().min(1))], - flags: [ - flag( - "id", - "the ID of the Runtime (defaults to the project's deployed runtime)", - runtimeIdSchema.optional(), - ), - flag( - "output", - "the output file path (default: agentcore/.cli/traces/-.json in a project)", - z.string().min(1, "requires a nonempty path").optional(), - ), - flag( - "since", - 'window start: "5m", "1h", "2d", ISO 8601, epoch ms, or "now" (default 12h ago)', - z.string().min(1).optional(), - ), - flag( - "until", - 'window end: "5m", "1h", "2d", ISO 8601, epoch ms, or "now" (default now)', - z.string().min(1).optional(), - ), - ], - handle: async (ctx, flags, args) => { - const traceId = args["trace-id"]; - const startTimeMs = - flags.since !== undefined - ? parseTimeString(flags.since) - : Date.now() - DEFAULT_TRACES_WINDOW_MS; - const endTimeMs = flags.until !== undefined ? parseTimeString(flags.until) : Date.now(); - - const target = await resolveRuntimeTarget(core, ctx, flags.id); - const records = await core.observability.getRuntimeTrace( - { runtimeId: target.runtimeId, traceId, startTimeMs, endTimeMs }, - target.options, - ); - - const filePath = resolveTraceOutputPath({ - output: flags.output, - project: target.project, - runtimeId: target.runtimeId, - traceId, - }); - try { - await mkdir(dirname(filePath), { recursive: true }); - await writeFile(filePath, JSON.stringify(records, null, 2)); - } catch (error) { - throw new FileWriteError( - `Could not write the trace file at ${filePath}: ` + - `${error instanceof Error ? error.message : String(error)}`, - { cause: error, meta: { filePath } }, - ); - } - - if (ctx.require(JsonKey)) { - ctx.require(JsonRendererKey).renderJson({ filePath, recordCount: records.length }); - return; - } - io.stderr.write(`Saved ${records.length} records for trace ${traceId}\n`); - io.stdout.write(`${filePath}\n`); + outputDescription: "the output file path (default: .json in the current directory)", + flags: runtimeFlags, + read: (ctx, flags, query, signal) => { + const source = { + logGroupName: runtimeLogGroup(flags.id, DEFAULT_ENDPOINT_QUALIFIER), + }; + return core.observability.getTrace(source, query, coreOptsFromCtx(ctx), signal); }, + resolveOutputPath: (_ctx, _flags, request) => resolveTraceOutputPath(request), }); diff --git a/src/handlers/runtime/traces/get/outputPath.ts b/src/handlers/runtime/traces/get/outputPath.ts deleted file mode 100644 index 2eee13731..000000000 --- a/src/handlers/runtime/traces/get/outputPath.ts +++ /dev/null @@ -1,29 +0,0 @@ -import { join, resolve } from "node:path"; -import type { Project } from "../../../project/types"; - -/** - * Resolves where `runtime traces get` writes its JSON file: an explicit - * --output wins; inside a project the file lands under the project's - * `agentcore/.cli/traces/` (keyed by runtime and trace so downloads never - * collide); outside a project it lands in the working directory. - */ -export function resolveTraceOutputPath(config: { - output?: string; - project?: Project; - runtimeId: string; - traceId: string; - cwd?: string; -}): string { - const cwd = config.cwd ?? process.cwd(); - if (config.output) return resolve(cwd, config.output); - if (config.project) { - return join( - config.project.rootPath, - "agentcore", - ".cli", - "traces", - `${config.runtimeId}-${config.traceId}.json`, - ); - } - return resolve(cwd, `${config.traceId}.json`); -} diff --git a/src/handlers/runtime/traces/index.tsx b/src/handlers/runtime/traces/index.tsx index a8f11cccb..a936e541c 100644 --- a/src/handlers/runtime/traces/index.tsx +++ b/src/handlers/runtime/traces/index.tsx @@ -1,15 +1,12 @@ -import { Router } from "../../../router"; import type { AppIO } from "../../../io"; +import { createTracesHandler } from "../../observability/traces"; import type { Core } from "../../types"; import { createGetRuntimeTraceHandler } from "./get"; import { createListRuntimeTracesHandler } from "./list"; -// The default window traces commands look back over when --since is omitted, -// matching the old CLI's 12h Insights lookback. -export const DEFAULT_TRACES_WINDOW_MS = 12 * 3_600_000; - -export function createRuntimeTracesHandler(core: Core, io: AppIO): Router { - return new Router("traces", "inspect a Runtime's traces") - .handler(createListRuntimeTracesHandler(core, io)) - .handler(createGetRuntimeTraceHandler(core, io)); -} +export const createRuntimeTracesHandler = (core: Core, io: AppIO) => + createTracesHandler({ + description: "inspect a Runtime's traces", + list: createListRuntimeTracesHandler(core, io), + get: createGetRuntimeTraceHandler(core, io), + }); diff --git a/src/handlers/runtime/traces/list/index.tsx b/src/handlers/runtime/traces/list/index.tsx index 36246cea1..9a5901d8e 100644 --- a/src/handlers/runtime/traces/list/index.tsx +++ b/src/handlers/runtime/traces/list/index.tsx @@ -1,92 +1,21 @@ -import z from "zod"; -import { parseTimeString } from "../../../../core/observability"; +import { DEFAULT_ENDPOINT_QUALIFIER, runtimeLogGroup } from "../../../../core/observability"; import type { AppIO } from "../../../../io"; -import { createHandler, flag } from "../../../../router"; -import { JsonRendererKey } from "../../../../tui"; -import { JsonKey } from "../../../keys"; +import { flag } from "../../../../router"; +import { createListTracesHandler } from "../../../observability/traces"; import type { Core } from "../../../types"; +import { coreOptsFromCtx } from "../../../utils"; import { runtimeIdSchema } from "../../invoke/request"; -import { resolveRuntimeTarget } from "../../resolveRuntimeTarget"; -import type { TraceSummary } from "../../types"; -import { DEFAULT_TRACES_WINDOW_MS } from "../index"; -const TRACE_ID_WIDTH = 34; -const TIMESTAMP_WIDTH = 22; - -/** - * Renders a Logs Insights timestamp for the table. Aggregations return epoch - * milliseconds as a string; anything non-numeric passes through untouched. - */ -export function formatTraceTimestamp(timestamp: string): string { - const epochMs = Number(timestamp); - if (isNaN(epochMs)) return timestamp; - return new Date(epochMs) - .toISOString() - .replace("T", " ") - .replace(/\.\d+Z$/, "Z"); -} - -export function formatTraceTable(traces: TraceSummary[]): string { - const lines = [ - `${"TRACE ID".padEnd(TRACE_ID_WIDTH)}${"TIMESTAMP".padEnd(TIMESTAMP_WIDTH)}SESSION ID`, - ]; - for (const trace of traces) { - lines.push( - trace.traceId.padEnd(TRACE_ID_WIDTH) + - formatTraceTimestamp(trace.timestamp).padEnd(TIMESTAMP_WIDTH) + - (trace.sessionId ?? "-"), - ); - } - return lines.join("\n") + "\n"; -} +const runtimeFlags = [flag("id", "the ID of the Runtime", runtimeIdSchema)] as const; export const createListRuntimeTracesHandler = (core: Core, io: AppIO) => - createHandler({ - name: "list", + createListTracesHandler(io, { description: "list a Runtime's recent traces", - flags: [ - flag( - "id", - "the ID of the Runtime (defaults to the project's deployed runtime)", - runtimeIdSchema.optional(), - ), - flag("limit", "maximum number of traces to display", z.number().int().positive().default(20)), - flag( - "since", - 'window start: "5m", "1h", "2d", ISO 8601, epoch ms, or "now" (default 12h ago)', - z.string().min(1).optional(), - ), - flag( - "until", - 'window end: "5m", "1h", "2d", ISO 8601, epoch ms, or "now" (default now)', - z.string().min(1).optional(), - ), - ], - handle: async (ctx, flags) => { - const startTimeMs = - flags.since !== undefined - ? parseTimeString(flags.since) - : Date.now() - DEFAULT_TRACES_WINDOW_MS; - const endTimeMs = flags.until !== undefined ? parseTimeString(flags.until) : Date.now(); - - const target = await resolveRuntimeTarget(core, ctx, flags.id); - const traces = await core.observability.listRuntimeTraces( - { runtimeId: target.runtimeId, startTimeMs, endTimeMs, limit: flags.limit }, - target.options, - ); - - if (ctx.require(JsonKey)) { - ctx.require(JsonRendererKey).renderJson({ traces }); - return; - } - - if (traces.length === 0) { - io.stderr.write( - "No traces found in the specified time range. Traces take 2-3 minutes " + - "to appear after an invocation.\n", - ); - return; - } - io.stdout.write(formatTraceTable(traces)); + flags: runtimeFlags, + read: (ctx, flags, query, signal) => { + const source = { + logGroupName: runtimeLogGroup(flags.id, DEFAULT_ENDPOINT_QUALIFIER), + }; + return core.observability.listTraces(source, query, coreOptsFromCtx(ctx), signal); }, }); diff --git a/src/handlers/runtime/traces/outputPath.ts b/src/handlers/runtime/traces/outputPath.ts new file mode 100644 index 000000000..efcdb04d8 --- /dev/null +++ b/src/handlers/runtime/traces/outputPath.ts @@ -0,0 +1,11 @@ +import { resolve } from "node:path"; + +export function resolveTraceOutputPath(config: { + output?: string; + traceId: string; + cwd?: string; +}): string { + const cwd = config.cwd ?? process.cwd(); + if (config.output) return resolve(cwd, config.output); + return resolve(cwd, `${config.traceId}.json`); +} diff --git a/src/handlers/runtime/traces/traces.test.tsx b/src/handlers/runtime/traces/traces.test.tsx index 487d7c8f7..118d64ab4 100644 --- a/src/handlers/runtime/traces/traces.test.tsx +++ b/src/handlers/runtime/traces/traces.test.tsx @@ -6,10 +6,9 @@ import { join, resolve } from "node:path"; import { createSilentLogger, TestCoreClient, testIO } from "../../../testing"; import { TestGlobalConfigAccessor } from "../../../testing/globalConfig"; import { createRootHandler } from "../../index"; -import type { GetRuntimeTraceInput, ListRuntimeTracesInput } from "../types"; -import { formatTraceTable, formatTraceTimestamp } from "./list"; -import { resolveTraceOutputPath } from "./get/outputPath"; -import type { Project } from "../../project/types"; +import type { GetTraceQuery, ListTracesQuery } from "../../../core/observability/index"; +import { formatTraceTable, formatTraceTimestamp } from "../../observability/traces"; +import { resolveTraceOutputPath } from "./outputPath"; const REGION = "us-west-2"; const SINCE_MS = 1_709_391_000_000; @@ -60,13 +59,16 @@ describe("runtime traces list", () => { expect(core.observability.calls).toHaveLength(1); const call = core.observability.calls[0]!; - expect(call.method).toBe("listRuntimeTraces"); - expect(call.args[0] as ListRuntimeTracesInput).toEqual({ - runtimeId: "my_agent-AbC123XyZ9", + expect(call.method).toBe("listTraces"); + expect(call.args[0]).toEqual({ + logGroupName: "/aws/bedrock-agentcore/runtimes/my_agent-AbC123XyZ9-DEFAULT", + }); + expect(call.args[1] as ListTracesQuery).toEqual({ startTimeMs: SINCE_MS, endTimeMs: UNTIL_MS, limit: 5, }); + expect(call.args[2]).toMatchObject({ region: REGION }); const [header, first, second] = io.stdout().split("\n"); expect(header).toMatch(/^TRACE ID\s+TIMESTAMP\s+SESSION ID$/); @@ -80,7 +82,7 @@ describe("runtime traces list", () => { await route(["runtime", "traces", "list", "--id", "rt-1", "--since", `${SINCE_MS}`]); - expect((core.observability.calls[0]!.args[0] as ListRuntimeTracesInput).limit).toBe(20); + expect((core.observability.calls[0]!.args[1] as ListTracesQuery).limit).toBe(20); }); test("--json renders a single JSON document", async () => { @@ -127,9 +129,11 @@ describe("runtime traces get", () => { ]); const call = core.observability.calls[0]!; - expect(call.method).toBe("getRuntimeTrace"); - expect(call.args[0] as GetRuntimeTraceInput).toMatchObject({ - runtimeId: "my_agent-AbC123XyZ9", + expect(call.method).toBe("getTrace"); + expect(call.args[0]).toEqual({ + logGroupName: "/aws/bedrock-agentcore/runtimes/my_agent-AbC123XyZ9-DEFAULT", + }); + expect(call.args[1] as GetTraceQuery).toMatchObject({ traceId: "abc123def456", startTimeMs: SINCE_MS, }); @@ -161,6 +165,21 @@ describe("runtime traces get", () => { expect(JSON.parse(io.stdout())).toEqual({ filePath: output, recordCount: 2 }); }); + test("saves and warns when the trace reaches the 10,000-record query limit", async () => { + const { core, io, route } = testTracesCommand(); + core.observability.traceRecords = Array.from({ length: 10_000 }, (_, index) => ({ + "@message": `record-${index}`, + })); + const output = join(mkdtempSync(join(tmpdir(), "trace-out-")), "trace.json"); + + await route(["runtime", "traces", "get", "abc123", "--id", "rt-1", "--output", output]); + + expect(JSON.parse(await readFile(output, "utf8"))).toHaveLength(10_000); + expect(io.stderr()).toContain( + "The trace query returned 10,000 records, the maximum; the saved file may be incomplete", + ); + }); + test("surfaces core errors (e.g. no trace data) unchanged", async () => { const { core, route } = testTracesCommand(); core.observability.error = new Error("No trace data found for trace ID: abc123"); @@ -172,32 +191,18 @@ describe("runtime traces get", () => { }); describe("resolveTraceOutputPath", () => { - const project = { name: "Proj", rootPath: "/work/proj", spec: {} } as unknown as Project; - - // Expected paths are built with the same node:path primitives the resolver - // uses: what these tests pin down is which branch wins (--output > project - // > cwd), not the platform's separator (Windows resolves to drive-letter - // backslash paths). test("an explicit --output wins, resolved against the cwd", () => { expect( resolveTraceOutputPath({ output: "out/trace.json", - project, - runtimeId: "rt-1", traceId: "abc", cwd: "/work/elsewhere", }), ).toBe(resolve("/work/elsewhere", "out/trace.json")); }); - test("inside a project the file lands under agentcore/.cli/traces", () => { - expect( - resolveTraceOutputPath({ project, runtimeId: "my_agent-AbC", traceId: "abc123", cwd: "/x" }), - ).toBe(join("/work/proj", "agentcore", ".cli", "traces", "my_agent-AbC-abc123.json")); - }); - - test("outside a project the file lands in the working directory", () => { - expect(resolveTraceOutputPath({ runtimeId: "rt-1", traceId: "abc123", cwd: "/tmp/x" })).toBe( + test("the default file lands in the working directory", () => { + expect(resolveTraceOutputPath({ traceId: "abc123", cwd: "/tmp/x" })).toBe( resolve("/tmp/x", "abc123.json"), ); }); diff --git a/src/handlers/runtime/types.tsx b/src/handlers/runtime/types.tsx index 35df3e928..d47d31cc5 100644 --- a/src/handlers/runtime/types.tsx +++ b/src/handlers/runtime/types.tsx @@ -7,11 +7,15 @@ import type { } from "@aws-sdk/client-bedrock-agentcore-control"; import type { CloudWatchLogEvent, + GetTraceQuery, InsightsQuery, InsightsQueryRow, + ListTracesQuery, LogSearchQuery, LogSource, LogTailQuery, + TraceRecord, + TraceSummary, } from "../../core/observability/types"; import type { CoreOptions } from "../../core/types"; import type { Project } from "../project/types"; @@ -99,41 +103,6 @@ export type DeployedRuntime = { targetName: string; }; -/** One trace aggregated from a runtime's telemetry, newest first. */ -export type TraceSummary = { - traceId: string; - /** Last-seen time as reported by Logs Insights (epoch ms rendered as a string). */ - timestamp: string; - sessionId?: string; - spanCount?: string; -}; - -/** - * One raw log record belonging to a trace. `@message` is the parsed JSON body - * when it parses, otherwise the original string; other Insights fields (e.g. - * `@timestamp`, `@ptr`) pass through as returned. - */ -export type TraceRecord = Record; - -export type ListRuntimeTracesInput = { - runtimeId: string; - /** Window start, epoch milliseconds. */ - startTimeMs: number; - /** Window end, epoch milliseconds. */ - endTimeMs: number; - /** Maximum number of traces to return. */ - limit: number; -}; - -export type GetRuntimeTraceInput = { - runtimeId: string; - traceId: string; - /** Window start, epoch milliseconds. */ - startTimeMs: number; - /** Window end, epoch milliseconds. */ - endTimeMs: number; -}; - export interface CoreObservabilityClient { resolveDeployedRuntime(project: Project, targetName: string): Promise; searchLogs( @@ -154,8 +123,16 @@ export interface CoreObservabilityClient { options: CoreOptions, signal?: AbortSignal, ): Promise; - /** Lists recent traces in the runtime's log group, newest first. */ - listRuntimeTraces(input: ListRuntimeTracesInput, options: CoreOptions): Promise; - /** Downloads every log record of one trace, oldest first. */ - getRuntimeTrace(input: GetRuntimeTraceInput, options: CoreOptions): Promise; + listTraces( + source: LogSource, + query: ListTracesQuery, + options: CoreOptions, + signal?: AbortSignal, + ): Promise; + getTrace( + source: LogSource, + query: GetTraceQuery, + options: CoreOptions, + signal?: AbortSignal, + ): Promise; } diff --git a/src/testing/TestCoreClient.tsx b/src/testing/TestCoreClient.tsx index 43f81ffa8..78638a2ea 100644 --- a/src/testing/TestCoreClient.tsx +++ b/src/testing/TestCoreClient.tsx @@ -135,22 +135,22 @@ import type { import type { CoreMemoryClient } from "../handlers/memory/types"; import type { CloudWatchLogEvent, + GetTraceQuery, InsightsQuery, InsightsQueryRow, + ListTracesQuery, LogSearchQuery, LogSource, LogTailQuery, + TraceRecord, + TraceSummary, } from "../core/observability/types"; import type { CoreObservabilityClient, CoreRuntimeClient, DeployedRuntime, - GetRuntimeTraceInput, - ListRuntimeTracesInput, RuntimeInvokeRequest, RuntimeInvokeResponse, - TraceRecord, - TraceSummary, } from "../handlers/runtime/types"; import type { BatchEvaluationDetail, @@ -2387,17 +2387,24 @@ export class TestObservabilityClient implements CoreObservabilityClient { traceSummaries: TraceSummary[] = []; traceRecords: TraceRecord[] = []; - async listRuntimeTraces( - input: ListRuntimeTracesInput, + async listTraces( + source: LogSource, + query: ListTracesQuery, options: CoreOptions, + signal?: AbortSignal, ): Promise { - this.calls.push({ method: "listRuntimeTraces", args: [input, options] }); + this.calls.push({ method: "listTraces", args: [source, query, options, signal] }); if (this.error) throw this.error; return this.traceSummaries; } - async getRuntimeTrace(input: GetRuntimeTraceInput, options: CoreOptions): Promise { - this.calls.push({ method: "getRuntimeTrace", args: [input, options] }); + async getTrace( + source: LogSource, + query: GetTraceQuery, + options: CoreOptions, + signal?: AbortSignal, + ): Promise { + this.calls.push({ method: "getTrace", args: [source, query, options, signal] }); if (this.error) throw this.error; return this.traceRecords; }