Skip to content

feat: add shared observability abstractions and wire to runtime log handler #2169

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jariy17 merged 1 commit into from
Sep 2, 2026
+1,202 −743
Merged

feat: add shared observability abstractions and wire to runtime log handler #2169

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
292 changes: 0 additions & 292 deletions src/core/observability.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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";
Expand All @@ -24,7 +20,6 @@ import type { Project } from "../handlers/project/types";
import type { AwsClients } from "./types";
import {
ObservabilityClient,
parseTimeString,
runInsightsQuery,
runtimeLogGroup,
sanitizeQueryValue,
Expand All @@ -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<unknown>;

function fakeLogs(send: Send): CloudWatchLogsClient {
Expand Down Expand Up @@ -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<StartLiveTailResponseStream>;

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 }[][]) {
Expand Down
Loading
Loading