diff --git a/src/core/observability.test.ts b/src/core/observability.test.ts index 3758af14d..cccbe7622 100644 --- a/src/core/observability.test.ts +++ b/src/core/observability.test.ts @@ -1,13 +1,9 @@ import { describe, expect, test } from "bun:test"; import { - DescribeLogGroupsCommand, - FilterLogEventsCommand, GetQueryResultsCommand, ResourceNotFoundException, - StartLiveTailCommand, StartQueryCommand, type CloudWatchLogsClient, - type StartLiveTailResponseStream, } from "@aws-sdk/client-cloudwatch-logs"; import { mkdtempSync, mkdirSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; @@ -24,7 +20,6 @@ import type { Project } from "../handlers/project/types"; import type { AwsClients } from "./types"; import { ObservabilityClient, - parseTimeString, runInsightsQuery, runtimeLogGroup, sanitizeQueryValue, @@ -46,46 +41,6 @@ describe("sanitizeQueryValue", () => { }); }); -describe("parseTimeString", () => { - const NOW = 1_700_000_000_000; - const now = () => NOW; - - test('parses "now" as the current time', () => { - expect(parseTimeString("now", now)).toBe(NOW); - }); - - test("parses relative durations for every unit as that long ago", () => { - expect(parseTimeString("30s", now)).toBe(NOW - 30_000); - expect(parseTimeString("5m", now)).toBe(NOW - 5 * 60_000); - expect(parseTimeString("1h", now)).toBe(NOW - 3_600_000); - expect(parseTimeString("2d", now)).toBe(NOW - 2 * 86_400_000); - }); - - test("parses epoch milliseconds (13+ digits) literally", () => { - expect(parseTimeString("1709391000000", now)).toBe(1_709_391_000_000); - }); - - test("parses ISO 8601 timestamps", () => { - expect(parseTimeString("2026-03-02T14:30:00Z", now)).toBe(Date.parse("2026-03-02T14:30:00Z")); - }); - - test("trims surrounding whitespace", () => { - expect(parseTimeString(" 15m ", now)).toBe(NOW - 15 * 60_000); - }); - - test("rejects empty input with a typed error", () => { - expect(() => parseTimeString(" ", now)).toThrow(InputValidationError); - expect(() => parseTimeString("", now)).toThrow("Time string cannot be empty"); - }); - - test("rejects garbage with a typed error naming the accepted forms", () => { - expect(() => parseTimeString("yesterday-ish", now)).toThrow(InputValidationError); - expect(() => parseTimeString("5x", now)).toThrow( - 'Invalid time string: "5x". Use relative durations (5m, 1h, 2d), ISO 8601, epoch ms, or "now".', - ); - }); -}); - type Send = (command: unknown) => Promise; function fakeLogs(send: Send): CloudWatchLogsClient { @@ -291,253 +246,6 @@ describe("ObservabilityClient.resolveDeployedRuntime", () => { }); }); -describe("ObservabilityClient.searchRuntimeLogs", () => { - const SEARCH = { - runtimeId: "my_agent-AbC123XyZ9", - startTimeMs: 1_000, - endTimeMs: 2_000, - }; - - async function collect(events: AsyncGenerator<{ timestamp: number; message: string }>) { - const out: { timestamp: number; message: string }[] = []; - for await (const event of events) out.push(event); - return out; - } - - test("paginates FilterLogEvents to completion, oldest to newest", async () => { - const inputs: unknown[] = []; - const logs = fakeLogs(async (command) => { - expect(command).toBeInstanceOf(FilterLogEventsCommand); - const input = (command as FilterLogEventsCommand).input; - inputs.push(input); - if (input.nextToken === "page-2") { - return { events: [{ timestamp: 3, message: "three" }] }; - } - return { - events: [ - { timestamp: 1, message: "one" }, - { timestamp: 2, message: "two" }, - ], - nextToken: "page-2", - }; - }); - - const events = await collect(clientWith(logs).searchRuntimeLogs(SEARCH, OPTIONS)); - - expect(events).toEqual([ - { timestamp: 1, message: "one" }, - { timestamp: 2, message: "two" }, - { timestamp: 3, message: "three" }, - ]); - expect(inputs[0]).toEqual({ - logGroupName: "/aws/bedrock-agentcore/runtimes/my_agent-AbC123XyZ9-DEFAULT", - startTime: 1_000, - endTime: 2_000, - }); - expect(inputs[1]).toMatchObject({ nextToken: "page-2" }); - }); - - test("caps yielded events at limit and requests no more than needed", async () => { - const limits: (number | undefined)[] = []; - const logs = fakeLogs(async (command) => { - const input = (command as FilterLogEventsCommand).input; - limits.push(input.limit); - if (input.nextToken === "page-2") { - return { - events: [ - { timestamp: 3, message: "three" }, - { timestamp: 4, message: "four" }, - ], - }; - } - return { - events: [ - { timestamp: 1, message: "one" }, - { timestamp: 2, message: "two" }, - ], - nextToken: "page-2", - }; - }); - - const events = await collect( - clientWith(logs).searchRuntimeLogs({ ...SEARCH, limit: 3 }, OPTIONS), - ); - - expect(events.map((event) => event.message)).toEqual(["one", "two", "three"]); - expect(limits).toEqual([3, 1]); - }); - - test("passes the filter pattern through to FilterLogEvents", async () => { - const logs = fakeLogs(async (command) => { - expect((command as FilterLogEventsCommand).input.filterPattern).toBe("ERROR database"); - return { events: [] }; - }); - - await collect( - clientWith(logs).searchRuntimeLogs({ ...SEARCH, filterPattern: "ERROR database" }, OPTIONS), - ); - }); - - test("translates a missing log group into invoked-yet guidance", async () => { - const logs = fakeLogs(async () => { - throw new ResourceNotFoundException({ - message: "The specified log group does not exist.", - $metadata: {}, - }); - }); - - await expect(collect(clientWith(logs).searchRuntimeLogs(SEARCH, OPTIONS))).rejects.toThrow( - "No logs found for runtime 'my_agent-AbC123XyZ9': log group " + - "/aws/bedrock-agentcore/runtimes/my_agent-AbC123XyZ9-DEFAULT does not exist. " + - "Has the runtime been invoked yet?", - ); - }); -}); - -describe("ObservabilityClient.streamRuntimeLogs", () => { - const STREAM = { runtimeId: "my_agent-AbC123XyZ9" }; - const LOG_GROUP = "/aws/bedrock-agentcore/runtimes/my_agent-AbC123XyZ9-DEFAULT"; - const GROUP_ARN = `arn:aws:logs:us-east-1:111122223333:log-group:${LOG_GROUP}`; - - type LiveTailEvent = Partial; - - function liveTailLogs( - sessions: (LiveTailEvent[] | Error)[], - groups: { logGroupName?: string; logGroupArn?: string; arn?: string }[] = [ - { logGroupName: LOG_GROUP, logGroupArn: GROUP_ARN }, - ], - ) { - const starts: unknown[] = []; - const logs = fakeLogs(async (command) => { - if (command instanceof DescribeLogGroupsCommand) { - expect(command.input.logGroupNamePrefix).toBe(LOG_GROUP); - return { logGroups: groups }; - } - expect(command).toBeInstanceOf(StartLiveTailCommand); - starts.push((command as StartLiveTailCommand).input); - const session = sessions[starts.length - 1] ?? []; - return { - responseStream: (async function* () { - if (session instanceof Error) throw session; - yield* session as StartLiveTailResponseStream[]; - })(), - }; - }); - return { logs, starts }; - } - - function update(...messages: string[]): LiveTailEvent { - return { - sessionUpdate: { - sessionResults: messages.map((message, i) => ({ timestamp: 1_000 + i, message })), - }, - }; - } - - async function collect(client: ObservabilityClient, signal: AbortSignal) { - const out: string[] = []; - for await (const event of client.streamRuntimeLogs(STREAM, OPTIONS, signal)) { - out.push(event.message); - } - return out; - } - - test("yields live-tail session updates and stops when the stream ends normally", async () => { - const { logs, starts } = liveTailLogs([[update("one", "two"), update("three")]]); - - const messages = await collect(clientWith(logs), new AbortController().signal); - - expect(messages).toEqual(["one", "two", "three"]); - expect(starts).toEqual([{ logGroupIdentifiers: [GROUP_ARN] }]); - }); - - test("reconnects when the session reports a timeout event", async () => { - const { logs, starts } = liveTailLogs([ - [update("one"), { SessionTimeoutException: { name: "SessionTimeoutException" } } as never], - [update("two")], - ]); - - const messages = await collect(clientWith(logs), new AbortController().signal); - - expect(messages).toEqual(["one", "two"]); - expect(starts).toHaveLength(2); - }); - - test("reconnects when the stream throws a session timeout", async () => { - const timeout = Object.assign(new Error("session timed out"), { - name: "SessionTimeoutException", - }); - const { logs, starts } = liveTailLogs([timeout, [update("after-reconnect")]]); - - const messages = await collect(clientWith(logs), new AbortController().signal); - - expect(messages).toEqual(["after-reconnect"]); - expect(starts).toHaveLength(2); - }); - - test("propagates non-timeout stream errors", async () => { - const { logs } = liveTailLogs([new Error("stream exploded")]); - - await expect(collect(clientWith(logs), new AbortController().signal)).rejects.toThrow( - "stream exploded", - ); - }); - - test("returns cleanly when aborted mid-session", async () => { - const controller = new AbortController(); - const { logs, starts } = liveTailLogs([ - [update("one"), { SessionTimeoutException: { name: "SessionTimeoutException" } } as never], - ]); - - const messages: string[] = []; - for await (const event of clientWith(logs).streamRuntimeLogs( - STREAM, - OPTIONS, - controller.signal, - )) { - messages.push(event.message); - controller.abort(); - } - - // The timeout after the abort must not trigger a reconnect. - expect(messages).toEqual(["one"]); - expect(starts).toHaveLength(1); - }); - - test("passes the filter pattern to the live tail", async () => { - const { logs, starts } = liveTailLogs([[]]); - - for await (const _ of clientWith(logs).streamRuntimeLogs( - { ...STREAM, filterPattern: "ERROR" }, - OPTIONS, - new AbortController().signal, - )) { - // drain - } - - expect(starts).toEqual([{ logGroupIdentifiers: [GROUP_ARN], logEventFilterPattern: "ERROR" }]); - }); - - test("strips the legacy ARN's trailing :* when the modern field is absent", async () => { - const { logs, starts } = liveTailLogs( - [[]], - [{ logGroupName: LOG_GROUP, arn: `${GROUP_ARN}:*` }], - ); - - await collect(clientWith(logs), new AbortController().signal); - - expect(starts).toEqual([{ logGroupIdentifiers: [GROUP_ARN] }]); - }); - - test("fails with invoked-yet guidance when the log group does not exist", async () => { - const { logs } = liveTailLogs([[]], []); - - await expect(collect(clientWith(logs), new AbortController().signal)).rejects.toThrow( - "Has the runtime been invoked yet?", - ); - }); -}); - // insightsLogs fakes the StartQuery/GetQueryResults protocol: every query // completes immediately with `results`, and each StartQuery input is recorded. function insightsLogs(results: { field: string; value: string }[][]) { diff --git a/src/core/observability.ts b/src/core/observability.ts index 81736d6e1..747e6b57d 100644 --- a/src/core/observability.ts +++ b/src/core/observability.ts @@ -1,9 +1,6 @@ import { - DescribeLogGroupsCommand, - FilterLogEventsCommand, GetQueryResultsCommand, ResourceNotFoundException, - StartLiveTailCommand, StartQueryCommand, type CloudWatchLogsClient, type ResultField, @@ -25,13 +22,14 @@ import type { DeployedRuntime, GetRuntimeTraceInput, ListRuntimeTracesInput, - RuntimeLogEvent, - SearchRuntimeLogsInput, - StreamRuntimeLogsInput, TraceRecord, TraceSummary, } 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"; @@ -245,18 +243,20 @@ export interface ObservabilityClientDeps { } /** - * ObservabilityClient reads the CloudWatch-backed telemetry of deployed - * AgentCore Runtimes: live-tail and search over the per-runtime log group, and - * resolution of a project's deployed runtime id from its CloudFormation stack - * outputs (runtime ids are not persisted locally, so the stack is the source - * of truth). + * Runtime-specific observability APIs retained for the existing trace and + * project-resolution commands. Generic log reads are inherited from the new + * shared observability client. */ -export class ObservabilityClient implements CoreObservabilityClient { +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; @@ -321,131 +321,6 @@ export class ObservabilityClient implements CoreObservabilityClient { }; } - /** - * Live-tails a runtime's log group via StartLiveTail, yielding events as they - * arrive. A live-tail session is server-capped (~3h); when it times out a new - * session is started transparently, so the stream runs until `signal` aborts - * (in which case the generator simply returns). - */ - async *streamRuntimeLogs( - input: StreamRuntimeLogsInput, - options: CoreOptions, - signal: AbortSignal, - ): AsyncGenerator { - const logGroupName = runtimeLogGroup(input.runtimeId, DEFAULT_ENDPOINT_QUALIFIER); - const logs = this.clients.logs(toClientConfig(options)); - - // StartLiveTail addresses log groups by ARN. DescribeLogGroups resolves it - // without hand-assembling one (partition/account), and doubles as the - // existence check so a never-invoked runtime fails with guidance instead of - // an opaque service error. - const described = await logs.send( - new DescribeLogGroupsCommand({ logGroupNamePrefix: logGroupName }), - { abortSignal: signal }, - ); - const group = (described.logGroups ?? []).find( - (candidate) => candidate.logGroupName === logGroupName, - ); - // The legacy `arn` field carries a trailing `:*` that StartLiveTail rejects. - const logGroupArn = group?.logGroupArn ?? group?.arn?.replace(/:\*$/, ""); - if (!logGroupArn) { - throw missingLogGroupError(input.runtimeId, logGroupName); - } - - while (!signal.aborted) { - let response; - try { - response = await logs.send( - new StartLiveTailCommand({ - logGroupIdentifiers: [logGroupArn], - ...(input.filterPattern ? { logEventFilterPattern: input.filterPattern } : {}), - }), - { abortSignal: signal }, - ); - } catch (error) { - if (signal.aborted) return; - throw error; - } - if (!response.responseStream) return; - - let sessionTimedOut = false; - try { - for await (const event of response.responseStream) { - if (signal.aborted) return; - if (event.sessionUpdate) { - for (const logEvent of event.sessionUpdate.sessionResults ?? []) { - yield { - timestamp: logEvent.timestamp ?? Date.now(), - message: logEvent.message ?? "", - }; - } - } - if (event.SessionTimeoutException) { - sessionTimedOut = true; - break; - } - } - } catch (error) { - if (signal.aborted) return; - if ((error as { name?: string }).name === "SessionTimeoutException") { - sessionTimedOut = true; - } else { - throw error; - } - } - - // A stream that ended without timing out was closed deliberately - // (server-side or by the caller); only a timeout warrants a reconnect. - if (!sessionTimedOut) return; - } - } - - /** - * Searches a runtime's log group over a closed time window via - * FilterLogEvents, paginating to completion and yielding events oldest to - * newest. `limit` caps the total number of events yielded. - */ - async *searchRuntimeLogs( - input: SearchRuntimeLogsInput, - options: CoreOptions, - signal?: AbortSignal, - ): AsyncGenerator { - const logGroupName = runtimeLogGroup(input.runtimeId, DEFAULT_ENDPOINT_QUALIFIER); - const logs = this.clients.logs(toClientConfig(options)); - - let nextToken: string | undefined; - let yielded = 0; - do { - let response; - try { - response = await logs.send( - new FilterLogEventsCommand({ - logGroupName, - startTime: input.startTimeMs, - endTime: input.endTimeMs, - ...(input.filterPattern ? { filterPattern: input.filterPattern } : {}), - ...(nextToken ? { nextToken } : {}), - // FilterLogEvents accepts at most 10k events per page. - ...(input.limit ? { limit: Math.min(input.limit - yielded, 10_000) } : {}), - }), - { abortSignal: signal }, - ); - } catch (error) { - if (error instanceof ResourceNotFoundException) { - throw missingLogGroupError(input.runtimeId, logGroupName, error); - } - throw error; - } - - for (const event of response.events ?? []) { - if (input.limit !== undefined && yielded >= input.limit) return; - yield { timestamp: event.timestamp ?? Date.now(), message: event.message ?? "" }; - yielded++; - } - nextToken = response.nextToken; - } while (nextToken && (input.limit === undefined || yielded < input.limit)); - } - /** * Lists the runtime's recent traces by aggregating its telemetry records with * a Logs Insights `stats … by traceId` query (mirrors the old CLI's diff --git a/src/core/observability/client.ts b/src/core/observability/client.ts new file mode 100644 index 000000000..c648121fa --- /dev/null +++ b/src/core/observability/client.ts @@ -0,0 +1,42 @@ +import type { CoreOptions } from "../types"; +import { CloudWatchClient } from "./cloudWatchClient"; +import type { + CloudWatchLogEvent, + InsightsQuery, + InsightsQueryRow, + LogSearchQuery, + LogSource, + LogTailQuery, +} from "./types"; + +/** Shared observability API over explicit CloudWatch log-group targets. */ +export class ObservabilityClient { + constructor(private readonly cloudWatch: CloudWatchClient) {} + + async *searchLogs( + source: LogSource, + query: LogSearchQuery, + options: CoreOptions, + signal?: AbortSignal, + ): AsyncGenerator { + yield* this.cloudWatch.searchLogs(source, query, options, signal); + } + + async *tailLogs( + source: LogSource, + query: LogTailQuery, + options: CoreOptions, + signal: AbortSignal, + ): AsyncGenerator { + yield* this.cloudWatch.tailLogs(source, query, options, signal); + } + + queryLogs( + source: LogSource, + query: InsightsQuery, + options: CoreOptions, + signal?: AbortSignal, + ): Promise { + return this.cloudWatch.queryLogs(source, query, options, signal); + } +} diff --git a/src/core/observability/cloudWatchClient.test.ts b/src/core/observability/cloudWatchClient.test.ts new file mode 100644 index 000000000..1832c4e41 --- /dev/null +++ b/src/core/observability/cloudWatchClient.test.ts @@ -0,0 +1,319 @@ +import { describe, expect, test } from "bun:test"; +import { + DescribeLogGroupsCommand, + FilterLogEventsCommand, + GetQueryResultsCommand, + ResourceNotFoundException, + StartLiveTailCommand, + StartQueryCommand, + type CloudWatchLogsClient, + type StartLiveTailResponseStream, +} from "@aws-sdk/client-cloudwatch-logs"; +import type { ClientConfig } from "../types"; +import { CloudWatchClient } from "./cloudWatchClient"; +import type { CloudWatchLogEvent } from "./types"; + +const SOURCE = { + provider: "cloudwatch" as const, + logGroupName: "/aws/bedrock-agentcore/runtimes/runtime-1-DEFAULT", +}; +const OPTIONS = { + region: "us-west-2", + endpointUrl: "https://logs.test", +}; + +type Send = (command: unknown, options?: unknown) => Promise; + +function clientWith(send: Send) { + const configs: ClientConfig[] = []; + const logs = { send } as unknown as CloudWatchLogsClient; + const client = new CloudWatchClient({ + logs: (config) => { + configs.push(config); + return logs; + }, + }); + return { client, configs }; +} + +async function collect(records: AsyncIterable) { + const result: CloudWatchLogEvent[] = []; + for await (const record of records) result.push(record); + return result; +} + +describe("CloudWatchClient.searchLogs", () => { + test("paginates, preserves provider metadata, and uses the configured client", async () => { + const inputs: unknown[] = []; + const { client, configs } = clientWith(async (command) => { + expect(command).toBeInstanceOf(FilterLogEventsCommand); + const input = (command as FilterLogEventsCommand).input; + inputs.push(input); + if (input.nextToken === "page-2") { + return { + events: [ + { + timestamp: 3, + ingestionTime: 4, + logStreamName: "stream-b", + message: "three", + eventId: "event-3", + }, + ], + }; + } + return { + events: [ + { timestamp: 1, message: "one" }, + { timestamp: 2, message: "two" }, + ], + nextToken: "page-2", + }; + }); + + const records = await collect( + client.searchLogs( + SOURCE, + { + startTimeMs: 1_000, + endTimeMs: 2_000, + filterPattern: "ERROR database", + }, + OPTIONS, + ), + ); + + expect(configs).toEqual([{ region: "us-west-2", endpoint: "https://logs.test" }]); + expect(inputs).toEqual([ + { + logGroupName: SOURCE.logGroupName, + startTime: 1_000, + endTime: 2_000, + filterPattern: "ERROR database", + }, + { + logGroupName: SOURCE.logGroupName, + startTime: 1_000, + endTime: 2_000, + filterPattern: "ERROR database", + nextToken: "page-2", + }, + ]); + expect(records.map(({ timestamp, message }) => ({ timestamp, message }))).toEqual([ + { timestamp: new Date(1), message: "one" }, + { timestamp: new Date(2), message: "two" }, + { timestamp: new Date(3), message: "three" }, + ]); + expect(records[2]).toMatchObject({ + ingestionTime: new Date(4), + logStreamName: "stream-b", + }); + }); + + test("applies a total limit across CloudWatch pages", async () => { + const requestedLimits: (number | undefined)[] = []; + const { client } = clientWith(async (command) => { + const input = (command as FilterLogEventsCommand).input; + requestedLimits.push(input.limit); + return input.nextToken + ? { + events: [ + { timestamp: 3, message: "three" }, + { timestamp: 4, message: "four" }, + ], + } + : { + events: [ + { timestamp: 1, message: "one" }, + { timestamp: 2, message: "two" }, + ], + nextToken: "page-2", + }; + }); + + const records = await collect( + client.searchLogs(SOURCE, { startTimeMs: 1, endTimeMs: 2, limit: 3 }, OPTIONS), + ); + + expect(records.map((record) => record.message)).toEqual(["one", "two", "three"]); + expect(requestedLimits).toEqual([3, 1]); + }); + + test("translates a missing group into customer guidance", async () => { + const { client } = clientWith(async () => { + throw new ResourceNotFoundException({ + message: "missing", + $metadata: {}, + }); + }); + + await expect( + collect(client.searchLogs(SOURCE, { startTimeMs: 1, endTimeMs: 2 }, OPTIONS)), + ).rejects.toThrow( + `CloudWatch log group ${SOURCE.logGroupName} does not exist. ` + + "Has the resource been invoked or emitted logs yet?", + ); + }); +}); + +describe("CloudWatchClient.queryLogs", () => { + test("runs an Insights query and flattens result fields", async () => { + const { client } = clientWith(async (command) => { + if (command instanceof StartQueryCommand) return { queryId: "query-1" }; + expect(command).toBeInstanceOf(GetQueryResultsCommand); + return { + status: "Complete", + results: [ + [ + { field: "traceId", value: "trace-1" }, + { field: "spanCount", value: "3" }, + ], + ], + }; + }); + + await expect( + client.queryLogs( + SOURCE, + { + queryString: "fields traceId", + startTimeMs: 1_000, + endTimeMs: 2_999, + }, + OPTIONS, + ), + ).resolves.toEqual([{ traceId: "trace-1", spanCount: "3" }]); + }); + + test("translates a missing query log group", async () => { + const { client } = clientWith(async () => { + throw new ResourceNotFoundException({ message: "missing", $metadata: {} }); + }); + + await expect( + client.queryLogs( + SOURCE, + { queryString: "fields @message", startTimeMs: 1_000, endTimeMs: 2_000 }, + OPTIONS, + ), + ).rejects.toThrow("Has the resource been invoked or emitted logs yet?"); + }); +}); + +describe("CloudWatchClient.tailLogs", () => { + const GROUP_ARN = "arn:aws:logs:us-west-2:111122223333:log-group:" + SOURCE.logGroupName; + + type LiveTailEvent = Partial; + + function liveTailReader( + sessions: (LiveTailEvent[] | Error)[], + groups: { logGroupName?: string; logGroupArn?: string; arn?: string }[] = [ + { logGroupName: SOURCE.logGroupName, logGroupArn: GROUP_ARN }, + ], + ) { + const starts: unknown[] = []; + const { client } = clientWith(async (command) => { + if (command instanceof DescribeLogGroupsCommand) { + expect(command.input.logGroupNamePrefix).toBe(SOURCE.logGroupName); + return { logGroups: groups }; + } + expect(command).toBeInstanceOf(StartLiveTailCommand); + starts.push((command as StartLiveTailCommand).input); + const session = sessions[starts.length - 1] ?? []; + return { + responseStream: (async function* () { + if (session instanceof Error) throw session; + yield* session as StartLiveTailResponseStream[]; + })(), + }; + }); + return { client, starts }; + } + + function update(...messages: string[]): LiveTailEvent { + return { + sessionUpdate: { + sessionResults: messages.map((message, index) => ({ + timestamp: 1_000 + index, + message, + logStreamName: "stream-a", + })), + }, + }; + } + + test("resolves the exact ARN and yields Live Tail updates", async () => { + const { client, starts } = liveTailReader([[update("one", "two"), update("three")]]); + + const records = await collect( + client.tailLogs(SOURCE, { filterPattern: "ERROR" }, OPTIONS, new AbortController().signal), + ); + + expect(records.map((record) => record.message)).toEqual(["one", "two", "three"]); + expect(starts).toEqual([ + { + logGroupIdentifiers: [GROUP_ARN], + logEventFilterPattern: "ERROR", + }, + ]); + }); + + test("reconnects after the service times out a session", async () => { + const { client, starts } = liveTailReader([ + [ + update("one"), + { + SessionTimeoutException: { name: "SessionTimeoutException" }, + } as never, + ], + [update("two")], + ]); + + const records = await collect( + client.tailLogs(SOURCE, {}, OPTIONS, new AbortController().signal), + ); + + expect(records.map((record) => record.message)).toEqual(["one", "two"]); + expect(starts).toHaveLength(2); + }); + + test("stops cleanly when the caller aborts an active session", async () => { + const controller = new AbortController(); + const { client, starts } = liveTailReader([ + [ + update("one"), + { + SessionTimeoutException: { name: "SessionTimeoutException" }, + } as never, + ], + ]); + const messages: string[] = []; + + for await (const record of client.tailLogs(SOURCE, {}, OPTIONS, controller.signal)) { + messages.push(record.message); + controller.abort(); + } + + expect(messages).toEqual(["one"]); + expect(starts).toHaveLength(1); + }); + + test("strips the legacy ARN suffix", async () => { + const { client, starts } = liveTailReader( + [[]], + [{ logGroupName: SOURCE.logGroupName, arn: `${GROUP_ARN}:*` }], + ); + + await collect(client.tailLogs(SOURCE, {}, OPTIONS, new AbortController().signal)); + + expect(starts).toEqual([{ logGroupIdentifiers: [GROUP_ARN] }]); + }); + + test("fails before starting a session when the group is absent", async () => { + const { client } = liveTailReader([[]], []); + + await expect( + collect(client.tailLogs(SOURCE, {}, OPTIONS, new AbortController().signal)), + ).rejects.toThrow("Has the resource been invoked or emitted logs yet?"); + }); +}); diff --git a/src/core/observability/cloudWatchClient.ts b/src/core/observability/cloudWatchClient.ts new file mode 100644 index 000000000..c668e9152 --- /dev/null +++ b/src/core/observability/cloudWatchClient.ts @@ -0,0 +1,181 @@ +import { + DescribeLogGroupsCommand, + FilterLogEventsCommand, + ResourceNotFoundException, + StartLiveTailCommand, + type FilteredLogEvent, + type LiveTailSessionLogEvent, +} from "@aws-sdk/client-cloudwatch-logs"; +import { ResourceNotFoundError } from "../../errors"; +import type { AwsClients, CoreOptions } from "../types"; +import { toClientConfig } from "../utils"; +import { runInsightsQuery } from "./insights"; +import type { + CloudWatchLogEvent, + InsightsQuery, + InsightsQueryRow, + LogSearchQuery, + LogSource, + LogTailQuery, +} from "./types"; + +export class CloudWatchClient { + constructor(private readonly clients: Pick) {} + + async *searchLogs( + source: LogSource, + query: LogSearchQuery, + options: CoreOptions, + signal?: AbortSignal, + ): AsyncGenerator { + if (query.limit !== undefined && query.limit <= 0) return; + + const logs = this.clients.logs(toClientConfig(options)); + let nextToken: string | undefined; + let yielded = 0; + + do { + const requestToken = nextToken; + let response; + try { + response = await logs.send( + new FilterLogEventsCommand({ + logGroupName: source.logGroupName, + startTime: query.startTimeMs, + endTime: query.endTimeMs, + ...(query.filterPattern ? { filterPattern: query.filterPattern } : {}), + ...(requestToken ? { nextToken: requestToken } : {}), + ...(query.limit ? { limit: Math.min(query.limit - yielded, 10_000) } : {}), + }), + { abortSignal: signal }, + ); + } catch (error) { + if (error instanceof ResourceNotFoundException) { + throw missingLogGroupError(source, error); + } + throw error; + } + + for (const event of response.events ?? []) { + if (query.limit !== undefined && yielded >= query.limit) return; + yield toCloudWatchLogEvent(event); + yielded++; + } + + nextToken = response.nextToken; + if (nextToken === requestToken) return; + } while (nextToken && (query.limit === undefined || yielded < query.limit)); + } + + async *tailLogs( + source: LogSource, + query: LogTailQuery, + options: CoreOptions, + signal: AbortSignal, + ): AsyncGenerator { + const logs = this.clients.logs(toClientConfig(options)); + const described = await logs.send( + new DescribeLogGroupsCommand({ logGroupNamePrefix: source.logGroupName }), + { abortSignal: signal }, + ); + const group = (described.logGroups ?? []).find( + (candidate) => candidate.logGroupName === source.logGroupName, + ); + // The legacy ARN field includes a suffix that StartLiveTail rejects. + const logGroupArn = group?.logGroupArn ?? group?.arn?.replace(/:\*$/, ""); + if (!logGroupArn) { + throw missingLogGroupError(source); + } + + while (!signal.aborted) { + let response; + try { + response = await logs.send( + new StartLiveTailCommand({ + logGroupIdentifiers: [logGroupArn], + ...(query.filterPattern ? { logEventFilterPattern: query.filterPattern } : {}), + }), + { abortSignal: signal }, + ); + } catch (error) { + if (signal.aborted) return; + throw error; + } + if (!response.responseStream) return; + + let sessionTimedOut = false; + try { + for await (const event of response.responseStream) { + if (signal.aborted) return; + for (const logEvent of event.sessionUpdate?.sessionResults ?? []) { + yield toCloudWatchLogEvent(logEvent); + } + if (event.SessionTimeoutException) { + sessionTimedOut = true; + break; + } + } + } catch (error) { + if (signal.aborted) return; + if ((error as { name?: string }).name === "SessionTimeoutException") { + sessionTimedOut = true; + } else { + throw error; + } + } + + if (!sessionTimedOut) return; + } + } + + async queryLogs( + source: LogSource, + query: InsightsQuery, + options: CoreOptions, + signal?: AbortSignal, + ): Promise { + const logs = this.clients.logs(toClientConfig(options)); + try { + const rows = await runInsightsQuery( + logs, + [source.logGroupName], + query.queryString, + Math.floor(query.startTimeMs / 1000), + Math.floor(query.endTimeMs / 1000), + query.rowLimit, + signal, + ); + return rows.map((row) => { + const result: InsightsQueryRow = {}; + for (const field of row) { + if (field.field && field.value !== undefined) result[field.field] = field.value; + } + return result; + }); + } catch (error) { + if (error instanceof ResourceNotFoundException) { + throw missingLogGroupError(source, error); + } + throw error; + } + } +} + +function toCloudWatchLogEvent( + event: FilteredLogEvent | LiveTailSessionLogEvent, +): CloudWatchLogEvent { + return { + timestamp: new Date(event.timestamp ?? Date.now()), + message: event.message ?? "", + ...(event.ingestionTime !== undefined ? { ingestionTime: new Date(event.ingestionTime) } : {}), + ...(event.logStreamName ? { logStreamName: event.logStreamName } : {}), + }; +} + +function missingLogGroupError(source: LogSource, cause?: unknown): ResourceNotFoundError { + return new ResourceNotFoundError( + `CloudWatch log group ${source.logGroupName} does not exist. ` + + "Has the resource been invoked or emitted logs yet?", + { cause, meta: { logGroupName: source.logGroupName } }, + ); +} diff --git a/src/core/observability/index.ts b/src/core/observability/index.ts new file mode 100644 index 000000000..bcf839bb9 --- /dev/null +++ b/src/core/observability/index.ts @@ -0,0 +1,16 @@ +export { CloudWatchClient } from "./cloudWatchClient"; +export { ObservabilityClient } from "./client"; +export { + INSIGHTS_MAX_ROWS, + runInsightsQuery, + sanitizeQueryValue, + type InsightsRowLimit, +} from "./insights"; +export type { + CloudWatchLogEvent, + InsightsQuery, + InsightsQueryRow, + LogSearchQuery, + LogSource, + LogTailQuery, +} from "./types"; diff --git a/src/core/observability/insights.test.ts b/src/core/observability/insights.test.ts new file mode 100644 index 000000000..603b4124d --- /dev/null +++ b/src/core/observability/insights.test.ts @@ -0,0 +1,83 @@ +import { describe, expect, test } from "bun:test"; +import { + GetQueryResultsCommand, + StartQueryCommand, + type CloudWatchLogsClient, +} from "@aws-sdk/client-cloudwatch-logs"; +import { CloudWatchQueryError, InputValidationError, ResultTruncationError } from "../../errors"; +import { runInsightsQuery, sanitizeQueryValue } from "./insights"; + +type Send = (command: unknown) => Promise; + +function fakeLogs(send: Send): CloudWatchLogsClient { + return { send } as unknown as CloudWatchLogsClient; +} + +function row(field: string, value: string) { + return [{ field, value }]; +} + +describe("runInsightsQuery", () => { + test("waits for completion and drains every result page", async () => { + const logs = fakeLogs(async (command) => { + if (command instanceof StartQueryCommand) { + expect(command.input).toEqual({ + logGroupNames: ["/aws/group-a", "/aws/group-b"], + queryString: "fields @message", + startTime: 100, + endTime: 200, + }); + return { queryId: "q-1" }; + } + const input = (command as GetQueryResultsCommand).input; + if (input.nextToken === "page-2") { + return { status: "Complete", results: [row("@message", "second")] }; + } + return { + status: "Complete", + results: [row("@message", "first")], + nextToken: "page-2", + }; + }); + + await expect( + runInsightsQuery(logs, ["/aws/group-a", "/aws/group-b"], "fields @message", 100, 200), + ).resolves.toEqual([row("@message", "first"), row("@message", "second")]); + }); + + test("throws a typed error for terminal failure states", async () => { + const logs = fakeLogs(async (command) => + command instanceof StartQueryCommand ? { queryId: "q-2" } : { status: "Failed" }, + ); + + await expect(runInsightsQuery(logs, ["/aws/g"], "q", 0, 1)).rejects.toThrow( + CloudWatchQueryError, + ); + }); + + test("applies caller-provided row-ceiling errors", async () => { + const logs = fakeLogs(async (command) => + command instanceof StartQueryCommand + ? { queryId: "q-3" } + : { status: "Complete", results: [row("@message", "a"), row("@message", "b")] }, + ); + + await expect( + runInsightsQuery(logs, ["/aws/g"], "q", 0, 1, { + maxRows: 2, + buildError: (maxRows) => new ResultTruncationError(`hit ceiling ${maxRows}`), + }), + ).rejects.toThrow("hit ceiling 2"); + await expect( + runInsightsQuery(logs, ["/aws/g"], "q", 0, 1, { + maxRows: 1, + buildError: () => new InputValidationError("narrow the scope"), + }), + ).rejects.toThrow(InputValidationError); + }); +}); + +test("sanitizeQueryValue strips quotes from interpolated values", () => { + expect(sanitizeQueryValue("abc'| drop '123")).toBe("abc| drop 123"); + expect(sanitizeQueryValue("clean-id")).toBe("clean-id"); +}); diff --git a/src/core/observability/insights.ts b/src/core/observability/insights.ts new file mode 100644 index 000000000..7b2fe7967 --- /dev/null +++ b/src/core/observability/insights.ts @@ -0,0 +1,81 @@ +import { + GetQueryResultsCommand, + StartQueryCommand, + type CloudWatchLogsClient, + type ResultField, +} from "@aws-sdk/client-cloudwatch-logs"; +import { CloudWatchQueryError, ResultTruncationError, type AgentCoreCLIError } from "../../errors"; + +export const INSIGHTS_MAX_ROWS = 100_000; + +export interface InsightsRowLimit { + maxRows: number; + buildError: (maxRows: number) => AgentCoreCLIError; +} + +const DEFAULT_ROW_LIMIT: InsightsRowLimit = { + maxRows: INSIGHTS_MAX_ROWS, + buildError: (maxRows) => + new ResultTruncationError( + `CloudWatch Logs Insights returned too many rows (>= ${maxRows}); narrow the time window`, + ), +}; + +export function sanitizeQueryValue(value: string): string { + return value.replace(/'/g, ""); +} + +export async function runInsightsQuery( + logs: CloudWatchLogsClient, + logGroupNames: string[], + queryString: string, + startSec: number, + endSec: number, + rowLimit: InsightsRowLimit = DEFAULT_ROW_LIMIT, + signal?: AbortSignal, +): Promise { + const started = await logs.send( + new StartQueryCommand({ + logGroupNames, + queryString, + startTime: startSec, + endTime: endSec, + }), + { abortSignal: signal }, + ); + const queryId = started.queryId; + + let status = "Running"; + for (let i = 0; i < 300 && status !== "Complete"; i++) { + const result = await logs.send(new GetQueryResultsCommand({ queryId }), { + abortSignal: signal, + }); + status = result.status ?? "Unknown"; + if (status === "Failed" || status === "Cancelled" || status === "Timeout") { + throw new CloudWatchQueryError(`CloudWatch Logs Insights query ${status.toLowerCase()}`, { + meta: { queryId, status }, + }); + } + if (status !== "Complete") await new Promise((resolve) => setTimeout(resolve, 1000)); + } + if (status !== "Complete") { + throw new CloudWatchQueryError("CloudWatch Logs Insights query did not finish in time", { + meta: { queryId, status }, + }); + } + + const rows: ResultField[][] = []; + let nextToken: string | undefined; + do { + const result = await logs.send(new GetQueryResultsCommand({ queryId, nextToken }), { + abortSignal: signal, + }); + rows.push(...(result.results ?? [])); + nextToken = result.nextToken; + } while (nextToken); + + if (rows.length >= rowLimit.maxRows) { + throw rowLimit.buildError(rowLimit.maxRows); + } + return rows; +} diff --git a/src/core/observability/types.ts b/src/core/observability/types.ts new file mode 100644 index 000000000..9e2d3e6ed --- /dev/null +++ b/src/core/observability/types.ts @@ -0,0 +1,34 @@ +import type { InsightsRowLimit } from "./insights"; + +/** Explicit CloudWatch Logs location selected by a primitive handler. */ +export type LogSource = { + logGroupName: string; +}; + +/** CloudWatch log event normalized at the AWS client boundary. */ +export type CloudWatchLogEvent = { + timestamp: Date; + message: string; + ingestionTime?: Date; + logStreamName?: string; +}; + +export type LogSearchQuery = { + startTimeMs: number; + endTimeMs: number; + filterPattern?: string; + limit?: number; +}; + +export type LogTailQuery = { + filterPattern?: string; +}; + +export type InsightsQuery = { + queryString: string; + startTimeMs: number; + endTimeMs: number; + rowLimit?: InsightsRowLimit; +}; + +export type InsightsQueryRow = Record; diff --git a/src/handlers/observability/filterPattern.test.ts b/src/handlers/observability/filterPattern.test.ts new file mode 100644 index 000000000..2e14da905 --- /dev/null +++ b/src/handlers/observability/filterPattern.test.ts @@ -0,0 +1,12 @@ +import { describe, expect, test } from "bun:test"; +import { buildFilterPattern } from "./filterPattern"; + +describe("buildFilterPattern", () => { + test("returns no pattern without filters", () => { + expect(buildFilterPattern({})).toBeUndefined(); + }); + + test("combines normalized level and query filters", () => { + expect(buildFilterPattern({ level: "error", query: '"timed out"' })).toBe('ERROR "timed out"'); + }); +}); diff --git a/src/handlers/observability/filterPattern.ts b/src/handlers/observability/filterPattern.ts new file mode 100644 index 000000000..58660c932 --- /dev/null +++ b/src/handlers/observability/filterPattern.ts @@ -0,0 +1,20 @@ +export const LOG_LEVELS = ["error", "warn", "info", "debug"] as const; + +export type LogLevel = (typeof LOG_LEVELS)[number]; + +const LEVEL_PATTERNS: Record = { + error: "ERROR", + warn: "WARN", + info: "INFO", + debug: "DEBUG", +}; + +export function buildFilterPattern(options: { + level?: LogLevel; + query?: string; +}): string | undefined { + const parts: string[] = []; + if (options.level) parts.push(LEVEL_PATTERNS[options.level]); + if (options.query) parts.push(options.query); + return parts.length > 0 ? parts.join(" ") : undefined; +} diff --git a/src/handlers/observability/logs.ts b/src/handlers/observability/logs.ts new file mode 100644 index 000000000..4a149012a --- /dev/null +++ b/src/handlers/observability/logs.ts @@ -0,0 +1,136 @@ +import z from "zod"; +import type { + CloudWatchLogEvent, + LogSearchQuery, + LogTailQuery, +} from "../../core/observability/index"; +import { InputValidationError } from "../../errors"; +import type { AppIO } from "../../io"; +import { createHandler, flag, type Context, type Flag } from "../../router"; +import { withUserCancellation } from "../../runnable"; +import { JsonRendererKey } from "../../tui"; +import { JsonKey } from "../keys"; +import { buildFilterPattern, LOG_LEVELS } from "./filterPattern"; +import { resolveTimeWindow } from "./time"; +import type { ResourceFlagValues } from "./types"; + +const DEFAULT_SEARCH_WINDOW_MS = 3_600_000; + +const levelSchema = z + .preprocess( + (value) => (typeof value === "string" ? value.toLowerCase() : value), + z.enum(LOG_LEVELS), + ) + .optional(); + +const logFlags = [ + flag( + "since", + 'search window start: "5m", "1h", ISO 8601, epoch ms, or "now"', + z.string().min(1).optional(), + ), + flag( + "until", + 'search window end: "5m", "1h", ISO 8601, epoch ms, or "now"', + z.string().min(1).optional(), + ), + flag("tail", "tail new log records", z.boolean().default(false)), + flag("level", `filter by log level (${LOG_LEVELS.join(", ")})`, levelSchema), + flag("query", "CloudWatch Logs filter pattern", z.string().optional()), + flag( + "limit", + "maximum number of log records to return in search mode", + z.number().int().positive().optional(), + ), +] as const; + +type LogFlagValues = ResourceFlagValues; + +export type LogsReadRequest = + | { + mode: "search"; + query: LogSearchQuery; + } + | { + mode: "tail"; + query: LogTailQuery; + }; + +export type LogsReadResult = { + events: AsyncIterable; + announcement?: string; +}; + +export function createLogsHandler[]>( + io: AppIO, + config: { + description: string; + flags: F; + read( + ctx: Context, + values: ResourceFlagValues & LogFlagValues, + request: LogsReadRequest, + signal: AbortSignal, + ): LogsReadResult | Promise; + }, +) { + const flags = [...config.flags, ...logFlags] as const; + + return createHandler({ + name: "logs", + description: config.description, + flags, + handle: async (ctx, values) => { + const parsed = values as unknown as ResourceFlagValues & LogFlagValues; + const searchMode = parsed.since !== undefined || parsed.until !== undefined; + if (parsed.tail && searchMode) { + throw new InputValidationError("--tail cannot be combined with --since or --until"); + } + if (!searchMode && parsed.limit !== undefined) { + throw new InputValidationError( + "--limit applies to search mode; add --since and/or --until", + ); + } + + const filterPattern = buildFilterPattern({ + level: parsed.level, + query: parsed.query, + }); + const { startTimeMs, endTimeMs } = resolveTimeWindow({ + since: parsed.since, + until: parsed.until, + defaultWindowMs: DEFAULT_SEARCH_WINDOW_MS, + }); + + const json = ctx.require(JsonKey); + const renderer = ctx.require(JsonRendererKey); + const writeEvent = (event: CloudWatchLogEvent) => { + if (json) { + renderer.renderJsonLine(event); + } else { + io.stdout.write(`${event.timestamp.toISOString()} ${event.message.trimEnd()}\n`); + } + }; + + await withUserCancellation(async (signal) => { + const request: LogsReadRequest = searchMode + ? { + mode: "search", + query: { + startTimeMs, + endTimeMs, + filterPattern, + limit: parsed.limit, + }, + } + : { + mode: "tail", + query: { filterPattern }, + }; + const result = await config.read(ctx, parsed, request, signal); + if (result.announcement) io.stderr.write(`${result.announcement}\n`); + for await (const event of result.events) writeEvent(event); + }); + }, + }); +} diff --git a/src/handlers/observability/time.test.ts b/src/handlers/observability/time.test.ts new file mode 100644 index 000000000..f47f85cdd --- /dev/null +++ b/src/handlers/observability/time.test.ts @@ -0,0 +1,49 @@ +import { describe, expect, test } from "bun:test"; +import { InputValidationError } from "../../errors"; +import { parseTimeString, resolveTimeWindow } from "./time"; + +describe("parseTimeString", () => { + const now = () => 1_700_000_000_000; + + test("parses relative, epoch, ISO, and now values", () => { + expect(parseTimeString("30s", now)).toBe(1_699_999_970_000); + expect(parseTimeString("5m", now)).toBe(1_699_999_700_000); + expect(parseTimeString("1h", now)).toBe(1_699_996_400_000); + expect(parseTimeString("2d", now)).toBe(1_699_827_200_000); + expect(parseTimeString("1709391000000", now)).toBe(1_709_391_000_000); + expect(parseTimeString("2026-03-02T14:30:00Z", now)).toBe(Date.parse("2026-03-02T14:30:00Z")); + expect(parseTimeString("now", now)).toBe(1_700_000_000_000); + }); + + test("rejects empty and invalid values with typed guidance", () => { + expect(() => parseTimeString(" ", now)).toThrow(InputValidationError); + expect(() => parseTimeString("5x", now)).toThrow('Invalid time string: "5x"'); + }); +}); + +describe("resolveTimeWindow", () => { + test("derives the default start from a historical end", () => { + expect( + resolveTimeWindow( + { + until: "1709391000000", + defaultWindowMs: 3_600_000, + }, + () => 1_800_000_000_000, + ), + ).toEqual({ + startTimeMs: 1_709_387_400_000, + endTimeMs: 1_709_391_000_000, + }); + }); + + test("rejects an inverted window", () => { + expect(() => + resolveTimeWindow({ + since: "1709391000000", + until: "1709381000000", + defaultWindowMs: 3_600_000, + }), + ).toThrow("--since must resolve to a time before --until"); + }); +}); diff --git a/src/handlers/observability/time.ts b/src/handlers/observability/time.ts new file mode 100644 index 000000000..ed0999f21 --- /dev/null +++ b/src/handlers/observability/time.ts @@ -0,0 +1,51 @@ +import { InputValidationError } from "../../errors"; + +const RELATIVE_DURATION_RE = /^(\d+)([smhd])$/; + +const UNIT_TO_MS: Record = { + s: 1_000, + m: 60_000, + h: 3_600_000, + d: 86_400_000, +}; + +export function parseTimeString(input: string, now: () => number = Date.now): number { + const trimmed = input.trim(); + if (trimmed === "") { + throw new InputValidationError("Time string cannot be empty"); + } + if (trimmed === "now") return now(); + + const relative = RELATIVE_DURATION_RE.exec(trimmed); + if (relative) { + return now() - parseInt(relative[1]!, 10) * UNIT_TO_MS[relative[2]!]!; + } + if (/^\d{13,}$/.test(trimmed)) return parseInt(trimmed, 10); + + const timestamp = Date.parse(trimmed); + if (!Number.isNaN(timestamp)) return timestamp; + + throw new InputValidationError( + `Invalid time string: "${input}". Use relative durations (5m, 1h, 2d), ` + + 'ISO 8601, epoch ms, or "now".', + ); +} + +export function resolveTimeWindow( + input: { + since?: string; + until?: string; + defaultWindowMs: number; + }, + now: () => number = Date.now, +): { startTimeMs: number; endTimeMs: number } { + const referenceTime = now(); + const parse = (value: string) => parseTimeString(value, () => referenceTime); + const endTimeMs = input.until === undefined ? referenceTime : parse(input.until); + const startTimeMs = + input.since === undefined ? endTimeMs - input.defaultWindowMs : parse(input.since); + if (startTimeMs > endTimeMs) { + throw new InputValidationError("--since must resolve to a time before --until"); + } + return { startTimeMs, endTimeMs }; +} diff --git a/src/handlers/observability/types.ts b/src/handlers/observability/types.ts new file mode 100644 index 000000000..8374ae057 --- /dev/null +++ b/src/handlers/observability/types.ts @@ -0,0 +1,6 @@ +import type z from "zod"; +import type { Flag } from "../../router"; + +export type ResourceFlagValues[]> = { + [E in F[number] as E["name"]]: E extends Flag ? z.infer> : never; +}; diff --git a/src/handlers/runtime/__fixtures__/FilterLogEventsCommand.17df600aa3fcb910.json b/src/handlers/runtime/__fixtures__/FilterLogEventsCommand.17df600aa3fcb910.json new file mode 100644 index 000000000..7701503fe --- /dev/null +++ b/src/handlers/runtime/__fixtures__/FilterLogEventsCommand.17df600aa3fcb910.json @@ -0,0 +1,13 @@ +{ + "events": [ + { + "logStreamName": "2026/08/12/[runtime-logs-67ebf93b-65e3-4127-9e13-483b239f256a]c7aba76f-c59f-4c8f-9e59-2872aea6dc45", + "timestamp": 1786555392053, + "message": "{\"timestamp\": \"2026-08-12T17:23:12.053Z\", \"level\": \"INFO\", \"message\": \"Returning streaming response (generator) (0.000s)\", \"logger\": \"bedrock_agentcore.app\", \"requestId\": \"37dcffe9-22bd-4a40-a18d-6cfbcc14a6d1\", \"sessionId\": \"67ebf93b-65e3-4127-9e13-483b239f256a\"}", + "ingestionTime": 1786555392873, + "eventId": "39841516581234934748312552609710478377045077922388246529" + } + ], + "searchedLogStreams": [], + "nextToken": "Bxkq6kVGFtq2y_MoigeqscPOdhXVbhiVtLoAmXb5jCreh6VSmn_zY7_b0sChd8dRESndx7N3wXVbpuIqdmNLkJpOqr-yUSOaZWF4_SfBMCKpa712QZRkZMBvlz6Zlf_KFhA4JI0um3l4ZnBohPfJ-EiDg23EyMn3LuVmDGPyslZHBzYyveSO7ePjizO2a8lydQgGIP47tglwQHwmaN_ou7RUU_APNaAohfFSoilGzq79ZmPOtzMAZIAsPXBKlG1gwreo2MRY9R7z7CBfdwdgcppoU9xXT5leGBh6fURYH-UD6hH7zSo5D_5VTFHC_5EXh0nXzUGI--D4-ACtc1cN5znhgo_aO3yqg2X6JVrLiZc" +} \ No newline at end of file diff --git a/src/handlers/runtime/__fixtures__/logs-search-json.golden.json b/src/handlers/runtime/__fixtures__/logs-search-json.golden.json new file mode 100644 index 000000000..384668e15 --- /dev/null +++ b/src/handlers/runtime/__fixtures__/logs-search-json.golden.json @@ -0,0 +1,6 @@ +{ + "timestamp": "2026-08-12T17:23:12.053Z", + "message": "{\"timestamp\": \"2026-08-12T17:23:12.053Z\", \"level\": \"INFO\", \"message\": \"Returning streaming response (generator) (0.000s)\", \"logger\": \"bedrock_agentcore.app\", \"requestId\": \"37dcffe9-22bd-4a40-a18d-6cfbcc14a6d1\", \"sessionId\": \"67ebf93b-65e3-4127-9e13-483b239f256a\"}", + "ingestionTime": "2026-08-12T17:23:12.873Z", + "logStreamName": "2026/08/12/[runtime-logs-67ebf93b-65e3-4127-9e13-483b239f256a]c7aba76f-c59f-4c8f-9e59-2872aea6dc45" +} \ No newline at end of file diff --git a/src/handlers/runtime/__fixtures__/logs-search.golden.txt b/src/handlers/runtime/__fixtures__/logs-search.golden.txt new file mode 100644 index 000000000..fef51377a --- /dev/null +++ b/src/handlers/runtime/__fixtures__/logs-search.golden.txt @@ -0,0 +1 @@ +2026-08-12T17:23:12.053Z {"timestamp": "2026-08-12T17:23:12.053Z", "level": "INFO", "message": "Returning streaming response (generator) (0.000s)", "logger": "bedrock_agentcore.app", "requestId": "37dcffe9-22bd-4a40-a18d-6cfbcc14a6d1", "sessionId": "67ebf93b-65e3-4127-9e13-483b239f256a"} \ No newline at end of file diff --git a/src/handlers/runtime/index.tsx b/src/handlers/runtime/index.tsx index 471af6aba..4fdb1a982 100644 --- a/src/handlers/runtime/index.tsx +++ b/src/handlers/runtime/index.tsx @@ -12,19 +12,15 @@ import { createRuntimeTracesHandler } from "./traces"; import { createRuntimeVersionHandler } from "./version"; export function createRuntimeHandler(core: Core, io: AppIO): Router { - return ( - new Router("runtime", "inspect AgentCore Runtimes") - .use(withTuiOnEmptyFlagsAndArgs(core, io)) - .default(renderTui(core, io)) - // logs and traces are headless-only: a bare `runtime logs` means "follow - // the project runtime's logs", so neither may fall into the TUI. - .supportedTuiCommands("get", "list", "invoke", "version", "endpoint") - .handler(createGetRuntimeHandler(core)) - .handler(createListRuntimesHandler(core)) - .handler(createInvokeRuntimeHandler(core, io)) - .handler(createRuntimeVersionHandler(core, io)) - .handler(createRuntimeEndpointHandler(core, io)) - .handler(createRuntimeLogsHandler(core, io)) - .handler(createRuntimeTracesHandler(core, io)) - ); + return new Router("runtime", "inspect AgentCore Runtimes") + .use(withTuiOnEmptyFlagsAndArgs(core, io)) + .default(renderTui(core, io)) + .supportedTuiCommands("get", "list", "invoke", "version", "endpoint") + .handler(createGetRuntimeHandler(core)) + .handler(createListRuntimesHandler(core)) + .handler(createInvokeRuntimeHandler(core, io)) + .handler(createRuntimeVersionHandler(core, io)) + .handler(createRuntimeEndpointHandler(core, io)) + .handler(createRuntimeLogsHandler(core, io)) + .handler(createRuntimeTracesHandler(core, io)); } diff --git a/src/handlers/runtime/logs/filterPattern.test.ts b/src/handlers/runtime/logs/filterPattern.test.ts deleted file mode 100644 index 3cc2b6ee2..000000000 --- a/src/handlers/runtime/logs/filterPattern.test.ts +++ /dev/null @@ -1,23 +0,0 @@ -import { describe, expect, test } from "bun:test"; -import { buildFilterPattern } from "./filterPattern"; - -describe("buildFilterPattern", () => { - test("returns undefined when neither level nor query is set", () => { - expect(buildFilterPattern({})).toBeUndefined(); - }); - - test("maps each level to its uppercase token", () => { - expect(buildFilterPattern({ level: "error" })).toBe("ERROR"); - expect(buildFilterPattern({ level: "warn" })).toBe("WARN"); - expect(buildFilterPattern({ level: "info" })).toBe("INFO"); - expect(buildFilterPattern({ level: "debug" })).toBe("DEBUG"); - }); - - test("passes the query through as-is", () => { - expect(buildFilterPattern({ query: '"timed out"' })).toBe('"timed out"'); - }); - - test("combines level and query with a space (implicit AND)", () => { - expect(buildFilterPattern({ level: "error", query: "database" })).toBe("ERROR database"); - }); -}); diff --git a/src/handlers/runtime/logs/filterPattern.ts b/src/handlers/runtime/logs/filterPattern.ts deleted file mode 100644 index b8eb156e5..000000000 --- a/src/handlers/runtime/logs/filterPattern.ts +++ /dev/null @@ -1,31 +0,0 @@ -// CloudWatch Logs filter-pattern assembly for `runtime logs`, ported from the -// old CLI's src/cli/commands/logs/filter-pattern.ts. - -export const LOG_LEVELS = ["error", "warn", "info", "debug"] as const; - -export type LogLevel = (typeof LOG_LEVELS)[number]; - -// Runtime log lines embed their level as uppercase text (ERROR, WARN, ...), so -// a level filter is just that token in the pattern. -const LEVEL_MAP: Record = { - error: "ERROR", - warn: "WARN", - info: "INFO", - debug: "DEBUG", -}; - -/** - * Builds a CloudWatch Logs filter pattern from the --level and --query options. - * The level maps to its uppercase token; the query passes through as-is; both - * combine with a space, which CloudWatch treats as an implicit AND. Returns - * undefined when neither is set (no server-side filtering). - */ -export function buildFilterPattern(options: { - level?: LogLevel; - query?: string; -}): string | undefined { - const parts: string[] = []; - if (options.level) parts.push(LEVEL_MAP[options.level]); - if (options.query) parts.push(options.query); - return parts.length > 0 ? parts.join(" ") : undefined; -} diff --git a/src/handlers/runtime/logs/index.tsx b/src/handlers/runtime/logs/index.tsx index c35346e67..f06249b33 100644 --- a/src/handlers/runtime/logs/index.tsx +++ b/src/handlers/runtime/logs/index.tsx @@ -1,28 +1,12 @@ -import z from "zod"; -import { parseTimeString } from "../../../core/observability"; -import { InputValidationError } from "../../../errors"; +import { DEFAULT_ENDPOINT_QUALIFIER, runtimeLogGroup } from "../../../core/observability"; import type { AppIO } from "../../../io"; -import { createHandler, flag } from "../../../router"; -import { withUserCancellation } from "../../../runnable"; -import { JsonRendererKey } from "../../../tui"; -import { JsonKey } from "../../keys"; +import { flag } from "../../../router"; +import { createLogsHandler } from "../../observability/logs"; import type { Core } from "../../types"; +import { coreOptsFromCtx } from "../../utils"; import { runtimeIdSchema } from "../invoke/request"; -import { resolveRuntimeTarget } from "../resolveRuntimeTarget"; -import type { RuntimeLogEvent } from "../types"; -import { buildFilterPattern, LOG_LEVELS } from "./filterPattern"; -// Search mode's default window when only one bound is given: the last hour. -const DEFAULT_SEARCH_WINDOW_MS = 3_600_000; - -const levelSchema = z - .preprocess( - (value) => (typeof value === "string" ? value.toLowerCase() : value), - z.enum(LOG_LEVELS), - ) - .optional(); - -const timeSchema = z.string().min(1).optional(); +const runtimeFlags = [flag("id", "the ID of the Runtime", runtimeIdSchema)] as const; /** * `runtime logs` follows a deployed runtime's CloudWatch log group live @@ -31,86 +15,24 @@ const timeSchema = z.string().min(1).optional(); * status (130). */ export const createRuntimeLogsHandler = (core: Core, io: AppIO) => - createHandler({ - name: "logs", + createLogsHandler(io, { description: "stream or search a Runtime's logs", - flags: [ - flag( - "id", - "the ID of the Runtime (defaults to the project's deployed runtime)", - runtimeIdSchema.optional(), - ), - flag( - "since", - 'search window start: "5m", "1h", "2d", ISO 8601, epoch ms, or "now" (default 1h ago; enables search mode)', - timeSchema, - ), - flag( - "until", - 'search window end: "5m", "1h", "2d", ISO 8601, epoch ms, or "now" (default now; enables search mode)', - timeSchema, - ), - flag("level", `filter by log level (${LOG_LEVELS.join(", ")})`, levelSchema), - flag("query", "server-side text filter", z.string().optional()), - flag( - "limit", - "maximum number of log lines to return (search mode)", - z.number().int().positive().optional(), - ), - ], - handle: async (ctx, flags) => { - const json = ctx.require(JsonKey); - const renderer = ctx.require(JsonRendererKey); + flags: runtimeFlags, + read: (ctx, flags, request, signal) => { + const options = coreOptsFromCtx(ctx); + const source = { + logGroupName: runtimeLogGroup(flags.id, DEFAULT_ENDPOINT_QUALIFIER), + }; - // --since / --until switch from live tail to a bounded search. - const searchMode = flags.since !== undefined || flags.until !== undefined; - if (!searchMode && flags.limit !== undefined) { - throw new InputValidationError( - "--limit applies to search mode; add --since and/or --until", - ); + if (request.mode === "search") { + return { + events: core.observability.searchLogs(source, request.query, options, signal), + }; } - const filterPattern = buildFilterPattern({ level: flags.level, query: flags.query }); - const startTimeMs = - flags.since !== undefined - ? parseTimeString(flags.since) - : Date.now() - DEFAULT_SEARCH_WINDOW_MS; - const endTimeMs = flags.until !== undefined ? parseTimeString(flags.until) : Date.now(); - const writeEvent = (event: RuntimeLogEvent) => { - const timestamp = new Date(event.timestamp).toISOString(); - if (json) { - renderer.renderJsonLine({ timestamp, message: event.message }); - } else { - io.stdout.write(`${timestamp} ${event.message.trimEnd()}\n`); - } + return { + events: core.observability.tailLogs(source, request.query, options, signal), + announcement: `Streaming logs for runtime ${flags.id}... (Ctrl+C to stop)`, }; - - await withUserCancellation(async (signal) => { - const target = await resolveRuntimeTarget(core, ctx, flags.id); - - if (searchMode) { - const events = core.observability.searchRuntimeLogs( - { - runtimeId: target.runtimeId, - startTimeMs, - endTimeMs, - filterPattern, - limit: flags.limit, - }, - target.options, - signal, - ); - for await (const event of events) writeEvent(event); - return; - } - - io.stderr.write(`Streaming logs for runtime ${target.runtimeId}... (Ctrl+C to stop)\n`); - const events = core.observability.streamRuntimeLogs( - { runtimeId: target.runtimeId, filterPattern }, - target.options, - signal, - ); - for await (const event of events) writeEvent(event); - }); }, }); diff --git a/src/handlers/runtime/logs/logs.test.tsx b/src/handlers/runtime/logs/logs.test.tsx index cf3c7d2e4..91e41cbd4 100644 --- a/src/handlers/runtime/logs/logs.test.tsx +++ b/src/handlers/runtime/logs/logs.test.tsx @@ -5,13 +5,11 @@ import { join } from "node:path"; import { createSilentLogger, TestCoreClient, testIO } from "../../../testing"; import { TestGlobalConfigAccessor } from "../../../testing/globalConfig"; import { createRootHandler } from "../../index"; -import type { SearchRuntimeLogsInput, StreamRuntimeLogsInput } from "../types"; +import type { LogSource } from "../../../core/observability/index"; const REGION = "us-west-2"; -// Fixed epoch bounds keep the tests clock-independent. const SINCE_MS = 1_709_391_000_000; -const UNTIL_MS = 1_709_394_600_000; function testLogsCommand() { const core = new TestCoreClient(); @@ -25,88 +23,11 @@ function testLogsCommand() { return { core, io, - route: (args: string[]) => root.route(["node", "agentcore", ...args, "--region", REGION]), + route: (args: string[]) => root.route(["bun", "agentcore", ...args, "--region", REGION]), }; } describe("runtime logs", () => { - test("searches when --since/--until are given and renders human lines", async () => { - const { core, io, route } = testLogsCommand(); - core.observability.logEvents = [ - { timestamp: SINCE_MS, message: "hello world\n" }, - { timestamp: SINCE_MS + 1_000, message: "second line" }, - ]; - - await route([ - "runtime", - "logs", - "--id", - "my_agent-AbC123XyZ9", - "--since", - `${SINCE_MS}`, - "--until", - `${UNTIL_MS}`, - ]); - - expect(core.observability.calls).toHaveLength(1); - const call = core.observability.calls[0]!; - expect(call.method).toBe("searchRuntimeLogs"); - expect(call.args[0] as SearchRuntimeLogsInput).toEqual({ - runtimeId: "my_agent-AbC123XyZ9", - startTimeMs: SINCE_MS, - endTimeMs: UNTIL_MS, - filterPattern: undefined, - limit: undefined, - }); - expect(call.args[1]).toEqual({ region: REGION, endpointUrl: undefined }); - - // Human mode: ` ` with the trailing newline normalized. - expect(io.stdout()).toBe( - "2024-03-02T14:50:00.000Z hello world\n2024-03-02T14:50:01.000Z second line", - ); - }); - - test("--json emits one JSON object per event (JSON Lines)", async () => { - const { core, io, route } = testLogsCommand(); - core.observability.logEvents = [{ timestamp: SINCE_MS, message: "hello" }]; - - await route([ - "runtime", - "logs", - "--id", - "my_agent-AbC123XyZ9", - "--since", - `${SINCE_MS}`, - "--json", - ]); - - expect(io.stdout()).toBe('{"timestamp":"2024-03-02T14:50:00.000Z","message":"hello"}'); - }); - - test("level and query compose into a CloudWatch filter pattern", async () => { - const { core, route } = testLogsCommand(); - - await route([ - "runtime", - "logs", - "--id", - "rt-1", - "--since", - "1709391000000", - "--level", - "ERROR", - "--query", - "database", - "--limit", - "25", - ]); - - const input = core.observability.calls[0]!.args[0] as SearchRuntimeLogsInput; - // --level is case-insensitive, like the old CLI. - expect(input.filterPattern).toBe("ERROR database"); - expect(input.limit).toBe(25); - }); - test("rejects an invalid --level", async () => { const { route } = testLogsCommand(); @@ -117,15 +38,17 @@ describe("runtime logs", () => { test("follows by default, announcing the stream on stderr", async () => { const { core, io, route } = testLogsCommand(); - core.observability.logEvents = [{ timestamp: SINCE_MS, message: "tailed" }]; + core.observability.logEvents = [{ timestamp: new Date(SINCE_MS), message: "tailed" }]; await route(["runtime", "logs", "--id", "my_agent-AbC123XyZ9"]); expect(core.observability.calls).toHaveLength(1); const call = core.observability.calls[0]!; - expect(call.method).toBe("streamRuntimeLogs"); - expect(call.args[0] as StreamRuntimeLogsInput).toEqual({ - runtimeId: "my_agent-AbC123XyZ9", + expect(call.method).toBe("tailLogs"); + expect(call.args[0] as LogSource).toEqual({ + logGroupName: "/aws/bedrock-agentcore/runtimes/my_agent-AbC123XyZ9-DEFAULT", + }); + expect(call.args[1]).toEqual({ filterPattern: undefined, }); expect(io.stderr()).toContain( @@ -142,18 +65,18 @@ describe("runtime logs", () => { ); }); - test("rejects an unparseable --since", async () => { + test("rejects --tail with a bounded search", async () => { const { route } = testLogsCommand(); await expect( - route(["runtime", "logs", "--id", "rt-1", "--since", "yesterday-ish"]), - ).rejects.toThrow('Invalid time string: "yesterday-ish"'); + route(["runtime", "logs", "--id", "rt-1", "--tail", "--since", "1h"]), + ).rejects.toThrow("--tail cannot be combined with --since or --until"); }); - test("auto-resolves the project's deployed runtime when --id is omitted", async () => { + test("requires --id even when invoked inside a project", async () => { const { core, route } = testLogsCommand(); - // A minimal-but-valid project for the on-disk project resolution. + // An imperative Runtime command must not fall back to project resolution. const root = mkdtempSync(join(tmpdir(), "logs-project-")); mkdirSync(join(root, "agentcore"), { recursive: true }); writeFileSync( @@ -164,21 +87,13 @@ describe("runtime logs", () => { const previousCwd = process.cwd(); process.chdir(root); try { - await route(["runtime", "logs", "--since", `${SINCE_MS}`]); + await expect(route(["runtime", "logs", "--since", `${SINCE_MS}`])).rejects.toThrow( + "required option '--id ' not specified", + ); } finally { process.chdir(previousCwd); } - const [resolveCall, searchCall] = core.observability.calls; - expect(resolveCall!.method).toBe("resolveDeployedRuntime"); - expect(resolveCall!.args[1]).toBe("default"); - expect(searchCall!.method).toBe("searchRuntimeLogs"); - // The stubbed deployed runtime (and its target region) win over --region. - expect((searchCall!.args[0] as SearchRuntimeLogsInput).runtimeId).toBe( - "project_runtime-0000000000", - ); - expect(searchCall!.args[1]).toMatchObject({ - region: core.observability.resolveDeployedRuntimeResponse.region, - }); + expect(core.observability.calls).toHaveLength(0); }); }); diff --git a/src/handlers/runtime/runtime.test.tsx b/src/handlers/runtime/runtime.test.tsx index e11d67373..6c856ca7d 100644 --- a/src/handlers/runtime/runtime.test.tsx +++ b/src/handlers/runtime/runtime.test.tsx @@ -19,8 +19,30 @@ const FIXTURES = join(import.meta.dir, "__fixtures__"); // Runtime pagination. Page-two requests use the token returned by page one. // Record with AWS_PROFILE=e2e-test RECORD=1 bun test src/handlers/runtime/runtime.test.tsx. const FIXTURE_RUNTIME_ID = "agentcore_cli_runtime_read_only_fixture-wZ7V4Q6vhx"; + +// Log search pins a separate Runtime invocation and fixed window. Re-recording +// requires these events to remain within that log group's retention period. +const FIXTURE_LOG_RUNTIME_ID = "asdf_MyAgent-3s5axvBC6Q"; +const FIXTURE_LOG_SESSION_ID = "67ebf93b-65e3-4127-9e13-483b239f256a"; +const LOG_WINDOW_START = "2026-08-12T00:00:00Z"; +const LOG_WINDOW_END = "2026-08-13T00:00:00Z"; const MISSING_RUNTIME_ID = "missing_runtime-0000000000"; +const LOG_SEARCH_ARGS = [ + "runtime", + "logs", + "--id", + FIXTURE_LOG_RUNTIME_ID, + "--since", + LOG_WINDOW_START, + "--until", + LOG_WINDOW_END, + "--query", + `"${FIXTURE_LOG_SESSION_ID}"`, + "--limit", + "1", +]; + function createFixtureCore(): CoreClient { const { createControlClient, createDataClient, createIamClient, createLogsClient } = fixtureFactories(FIXTURES); @@ -129,6 +151,23 @@ describe("runtime TUI dispatch", () => { }); describe("runtime read-only commands", () => { + test("searches recorded Runtime logs through the real CLI path", async () => { + const stdout = await run(LOG_SEARCH_ARGS); + + matchGolden(FIXTURES, "logs-search.golden.txt", stdout); + expect(stdout).toContain(FIXTURE_LOG_SESSION_ID); + }); + + test("renders recorded Runtime logs as JSON Lines", async () => { + const stdout = await run([...LOG_SEARCH_ARGS, "--json"]); + + matchGolden(FIXTURES, "logs-search-json.golden.json", stdout); + expect(JSON.parse(stdout)).toMatchObject({ + timestamp: "2026-08-12T17:23:12.053Z", + message: expect.any(String), + }); + }); + test("gets a Runtime whose ID exceeds the 48-character Runtime name limit", async () => { expect(FIXTURE_RUNTIME_ID.length).toBeGreaterThan(48); diff --git a/src/handlers/runtime/types.tsx b/src/handlers/runtime/types.tsx index 45dcb5378..35df3e928 100644 --- a/src/handlers/runtime/types.tsx +++ b/src/handlers/runtime/types.tsx @@ -5,6 +5,14 @@ import type { ListAgentRuntimesResponse, ListAgentRuntimeVersionsResponse, } from "@aws-sdk/client-bedrock-agentcore-control"; +import type { + CloudWatchLogEvent, + InsightsQuery, + InsightsQueryRow, + LogSearchQuery, + LogSource, + LogTailQuery, +} from "../../core/observability/types"; import type { CoreOptions } from "../../core/types"; import type { Project } from "../project/types"; @@ -82,13 +90,6 @@ export interface CoreRuntimeClient { ): Promise; } -/** One CloudWatch log event from a runtime's log group. */ -export type RuntimeLogEvent = { - /** Epoch milliseconds. */ - timestamp: number; - message: string; -}; - /** A project runtime resolved live from its CloudFormation stack outputs. */ export type DeployedRuntime = { runtimeId: string; @@ -98,24 +99,6 @@ export type DeployedRuntime = { targetName: string; }; -export type StreamRuntimeLogsInput = { - runtimeId: string; - /** CloudWatch Logs filter pattern applied server-side. */ - filterPattern?: string; -}; - -export type SearchRuntimeLogsInput = { - runtimeId: string; - /** Window start, epoch milliseconds (inclusive). */ - startTimeMs: number; - /** Window end, epoch milliseconds (inclusive). */ - endTimeMs: number; - /** CloudWatch Logs filter pattern applied server-side. */ - filterPattern?: string; - /** Maximum number of events to yield. */ - limit?: number; -}; - /** One trace aggregated from a runtime's telemetry, newest first. */ export type TraceSummary = { traceId: string; @@ -153,18 +136,24 @@ export type GetRuntimeTraceInput = { export interface CoreObservabilityClient { resolveDeployedRuntime(project: Project, targetName: string): Promise; - /** Live-tails the runtime's log group until `signal` aborts. */ - streamRuntimeLogs( - input: StreamRuntimeLogsInput, + searchLogs( + source: LogSource, + query: LogSearchQuery, + options: CoreOptions, + signal?: AbortSignal, + ): AsyncIterable; + tailLogs( + source: LogSource, + query: LogTailQuery, options: CoreOptions, signal: AbortSignal, - ): AsyncGenerator; - /** Searches the runtime's log group over a time window, oldest to newest. */ - searchRuntimeLogs( - input: SearchRuntimeLogsInput, + ): AsyncIterable; + queryLogs( + source: LogSource, + query: InsightsQuery, options: CoreOptions, signal?: AbortSignal, - ): AsyncGenerator; + ): 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. */ diff --git a/src/testing/TestCoreClient.tsx b/src/testing/TestCoreClient.tsx index 7cbe6b005..2451a5fe6 100644 --- a/src/testing/TestCoreClient.tsx +++ b/src/testing/TestCoreClient.tsx @@ -133,6 +133,14 @@ import type { UpdateOauth2CredentialProviderInput, } from "../handlers/identity/types"; import type { CoreMemoryClient } from "../handlers/memory/types"; +import type { + CloudWatchLogEvent, + InsightsQuery, + InsightsQueryRow, + LogSearchQuery, + LogSource, + LogTailQuery, +} from "../core/observability/types"; import type { CoreObservabilityClient, CoreRuntimeClient, @@ -141,9 +149,6 @@ import type { ListRuntimeTracesInput, RuntimeInvokeRequest, RuntimeInvokeResponse, - RuntimeLogEvent, - SearchRuntimeLogsInput, - StreamRuntimeLogsInput, TraceRecord, TraceSummary, } from "../handlers/runtime/types"; @@ -2330,7 +2335,8 @@ export class TestObservabilityClient implements CoreObservabilityClient { stackName: "AgentCore-project-default", targetName: "default", }; - logEvents: RuntimeLogEvent[] = []; + logEvents: CloudWatchLogEvent[] = []; + queryRows: InsightsQueryRow[] = []; async resolveDeployedRuntime(project: Project, targetName: string): Promise { this.calls.push({ method: "resolveDeployedRuntime", args: [project, targetName] }); @@ -2338,24 +2344,37 @@ export class TestObservabilityClient implements CoreObservabilityClient { return this.resolveDeployedRuntimeResponse; } - async *streamRuntimeLogs( - input: StreamRuntimeLogsInput, + async *searchLogs( + source: LogSource, + query: LogSearchQuery, + options: CoreOptions, + signal?: AbortSignal, + ): AsyncGenerator { + this.calls.push({ method: "searchLogs", args: [source, query, options, signal] }); + if (this.error) throw this.error; + yield* this.logEvents; + } + + async *tailLogs( + source: LogSource, + query: LogTailQuery, options: CoreOptions, signal: AbortSignal, - ): AsyncGenerator { - this.calls.push({ method: "streamRuntimeLogs", args: [input, options, signal] }); + ): AsyncGenerator { + this.calls.push({ method: "tailLogs", args: [source, query, options, signal] }); if (this.error) throw this.error; yield* this.logEvents; } - async *searchRuntimeLogs( - input: SearchRuntimeLogsInput, + async queryLogs( + source: LogSource, + query: InsightsQuery, options: CoreOptions, signal?: AbortSignal, - ): AsyncGenerator { - this.calls.push({ method: "searchRuntimeLogs", args: [input, options, signal] }); + ): Promise { + this.calls.push({ method: "queryLogs", args: [source, query, options, signal] }); if (this.error) throw this.error; - yield* this.logEvents; + return this.queryRows; } traceSummaries: TraceSummary[] = [];