fix(infra): load the Kubernetes explorer faster and read its SQL via search_stream (#15911)

This commit is contained in:
Bhargav
2026-10-28 05:03:30 +00:00
committed by GitHub
parent a135552022
commit 00d4cb9e8e
8 changed files with 1046 additions and 55 deletions
+65 -1
View File
@@ -299,11 +299,61 @@ describe("useHttpStreaming", () => {
expect(mockHandlers.error).toHaveBeenCalledTimes(2);
expect(mockHandlers.error).toHaveBeenCalledWith(mockData, {
content: { trace_id: mockData.traceId },
content: { trace_id: mockData.traceId, message: "Network failure" },
type: "error",
});
});
it.each([
[502, "<html>bad gateway</html>", "HTTP 502"],
[403, "<html>forbidden</html>", "HTTP 403"],
])(
"passes status %i and a non-empty message of a non-JSON Response to the error handler",
async (status, body, message) => {
vi.spyOn(console, "error").mockImplementation(() => {});
mockFetch.mockResolvedValueOnce(new Response(body, { status }));
httpStreaming.traceMap.value[mockData.traceId] = {
data: [],
error: [mockHandlers.error],
complete: [],
abortController: new AbortController(),
};
await httpStreaming.fetchQueryDataWithHttpStream(mockData, mockHandlers);
await flushPromises();
expect(mockHandlers.error).toHaveBeenCalledWith(
mockData,
expect.objectContaining({
type: "error",
content: expect.objectContaining({ status, message }),
}),
);
},
);
it("passes the message of a rejected fetch to the error handler", async () => {
vi.spyOn(console, "error").mockImplementation(() => {});
mockFetch.mockRejectedValueOnce(new TypeError("Failed to fetch"));
httpStreaming.traceMap.value[mockData.traceId] = {
data: [],
error: [mockHandlers.error],
complete: [],
abortController: new AbortController(),
};
await httpStreaming.fetchQueryDataWithHttpStream(mockData, mockHandlers);
await flushPromises();
expect(mockHandlers.error).toHaveBeenCalledWith(
mockData,
expect.objectContaining({
type: "error",
content: expect.objectContaining({ message: "Failed to fetch" }),
}),
);
});
// TODO: Fix Worker mocking
it.skip("should cancel the stream and clean up resources", async () => {
const traceId = "0534c2294079426a86bd88b137752535";
@@ -658,6 +708,20 @@ describe("useHttpStreaming", () => {
const { err } = await failRead(new Error("worker exploded"));
expect(err).toHaveBeenCalledWith("Error reading stream:", expect.any(Error));
});
it("passes the read failure message to the error handler", async () => {
handlers.error.mockClear();
await failRead(new TypeError("Failed to fetch"));
expect(handlers.error.mock.calls[0][1].content.message).toBe("Failed to fetch");
});
it("keeps an own enumerable status on an Error in the error content", async () => {
handlers.error.mockClear();
await failRead(Object.assign(new Error("denied"), { status: 403 }));
const content = handlers.error.mock.calls[0][1].content;
expect(content.status).toBe(403);
expect(content.message).toBe("denied");
});
});
describe("WebSocket response converters", () => {
+10 -2
View File
@@ -38,6 +38,14 @@ const NETWORK_FAILURE_PATTERN = /network error|Load failed|Failed to fetch|Netwo
const isNetworkFailure = (error: unknown): boolean =>
error instanceof TypeError && NETWORK_FAILURE_PATTERN.test(error.message);
// Status and message sit on prototypes of a Response or Error, which the error spread would drop.
const withErrorFields = (error: unknown): unknown =>
error instanceof Response
? { status: error.status, message: error.statusText || `HTTP ${error.status}` }
: error instanceof Error
? { ...error, message: error.message }
: error;
// Type definitions similar to WebSocket but for HTTP/2 streaming
type StreamHandler = (data: any, traceId: string) => void;
type ErrorHandler = (error: any, traceId: string) => void;
@@ -430,7 +438,7 @@ const useHttpStreaming = () => {
// A dropped connection (offline, sleep, tab freeze) is not an app fault.
if (isNetworkFailure(error)) console.warn("Error reading stream:", error);
else console.error("Error reading stream:", error);
onError(traceId, error);
onError(traceId, withErrorFields(error));
}
} finally {
// Always release the reader lock to prevent resource leaks
@@ -462,7 +470,7 @@ const useHttpStreaming = () => {
sessionStorage.clear();
window.location.reload();
} else {
onError(traceId, error);
onError(traceId, withErrorFields(error));
}
}
};
+89 -16
View File
@@ -20,10 +20,29 @@ import { k8sKeys } from "./kubernetes.querykeys";
import { LIVE_STALE_TIME } from "@/composables/query/cachePolicy";
import { queryClient } from "@/composables/query/queryClient";
vi.mock("./search", () => ({ default: { metrics_query: vi.fn(), search: vi.fn() } }));
const stream = vi.hoisted(() => ({
fetch: vi.fn(),
}));
vi.mock("./search", () => ({ default: { metrics_query: vi.fn() } }));
vi.mock("@/composables/useStreamingSearch", () => ({
default: () => ({
fetchQueryDataWithHttpStream: stream.fetch,
}),
}));
const metricsQuery = vi.mocked(searchService.metrics_query);
const search = vi.mocked(searchService.search);
const hitsChunk = (hits: any[]) => ({
type: "search_response_hits",
content: { results: { hits } },
});
const streamOnce = (chunks: any[]) =>
stream.fetch.mockImplementation(async (req: any, handlers: any) => {
chunks.forEach((c) => handlers.data(req, c));
handlers.complete(req, {});
});
const ORG = "org1";
const END = 1_700_000_045_000_000;
@@ -35,7 +54,7 @@ describe("kubernetes queries", () => {
metricsQuery.mockResolvedValue({
data: { data: { result: [{ metric: { node: "n1" }, value: [1, "2"] }] } },
} as any);
search.mockResolvedValue({ data: { hits: [{ n: 1 }] } } as any);
streamOnce([hitsChunk([{ n: 1 }])]);
});
it("roots every key under the org's kubernetes scope", () => {
@@ -58,19 +77,73 @@ describe("kubernetes queries", () => {
});
});
it("returns SQL hits and sends the exact window and size", async () => {
const options = k8sSqlQuery(ORG, "W", "SELECT 1", START, END, true, 20001);
expect(await (options.queryFn as any)()).toEqual([{ n: 1 }]);
const [args, type] = search.mock.calls[0] as any[];
expect(type).toBe("ui");
expect(args.org_identifier).toBe(ORG);
expect(args.page_type).toBe("logs");
expect(args.query.query).toMatchObject({
sql: "SELECT 1",
start_time: START,
end_time: END,
from: 0,
size: 20001,
describe("SQL over the search stream", () => {
const run = (size?: number) =>
(k8sSqlQuery(ORG, "W", "SELECT 1", START, END, true, size).queryFn as any)();
it("returns the hits and sends the exact window and size", async () => {
expect(await run(20001)).toEqual([{ n: 1 }]);
const [req] = stream.fetch.mock.calls[0];
expect(req).toMatchObject({
type: "search",
pageType: "logs",
org_id: ORG,
searchType: "ui",
});
expect(req.traceId).toEqual(expect.any(String));
expect(req.queryReq.query).toEqual({
sql: "SELECT 1",
start_time: START,
end_time: END,
from: 0,
size: 20001,
});
});
it("omits size when none is given", async () => {
await run();
expect(stream.fetch.mock.calls[0][0].queryReq.query).not.toHaveProperty("size");
});
it("appends the hits of successive non-aggregate chunks", async () => {
streamOnce([hitsChunk([{ n: 1 }]), hitsChunk([{ n: 2 }, { n: 3 }])]);
expect(await run()).toEqual([{ n: 1 }, { n: 2 }, { n: 3 }]);
});
it("ignores chunks that carry no hits", async () => {
streamOnce([{ type: "search_response_metadata", content: {} }, hitsChunk([{ n: 1 }])]);
expect(await run()).toEqual([{ n: 1 }]);
});
it("rejects with the status when the handler errors", async () => {
stream.fetch.mockImplementation(async (req: any, handlers: any) =>
handlers.error(req, { content: { message: "forbidden", status: 403 } }),
);
const err: any = await run().catch((e) => e);
expect(err).toBeInstanceOf(Error);
expect(err.message).toBe("forbidden");
expect(err.status).toBe(403);
});
it("falls back to a non-empty message when the error payload message is empty", async () => {
stream.fetch.mockImplementation(async (req: any, handlers: any) =>
handlers.error(req, { content: { message: "", status: 502 } }),
);
const err: any = await run().catch((e) => e);
expect(err.message).toBe("search stream failed");
expect(err.status).toBe(502);
});
it("reads the status from content.code when content.status is absent", async () => {
stream.fetch.mockImplementation(async (req: any, handlers: any) =>
handlers.error(req, { content: { message: "denied", code: 403 } }),
);
await expect(run()).rejects.toMatchObject({ message: "denied", status: 403 });
});
it("rejects with a null status when the error carries none", async () => {
stream.fetch.mockImplementation(async (req: any, handlers: any) => handlers.error(req, {}));
await expect(run()).rejects.toMatchObject({ status: null });
});
});
+43 -19
View File
@@ -15,6 +15,8 @@
import { queryOptions } from "@tanstack/vue-query";
import searchService from "./search";
import useHttpStreaming from "@/composables/useStreamingSearch";
import { generateTraceContext } from "@/utils/auth";
import { k8sKeys } from "./kubernetes.querykeys";
import { LIVE_STALE_TIME } from "@/composables/query/cachePolicy";
import { quantizeRange } from "@/composables/query/queryClient";
@@ -24,6 +26,39 @@ import { parseVector, type Series } from "@/views/Infrastructure/kubernetes2/kub
const timeKey = (startUs: number, endUs: number, relative: boolean) =>
relative ? quantizeRange(startUs, endUs) : { start: startUs, end: endUs };
// The stream reports errors as { content }; callers read status and message off the error itself.
const streamError = (res: any) =>
Object.assign(new Error(res?.content?.message || res?.content?.error || "search stream failed"), {
status: res?.content?.status ?? res?.content?.code ?? null,
});
// Every k8s query is complex, so the server runs it as one partition and its hits are appended.
function streamSqlHits(org: string, query: Record<string, unknown>) {
const { fetchQueryDataWithHttpStream } = useHttpStreaming();
const { traceId } = generateTraceContext();
return new Promise<any[]>((resolve, reject) => {
const hits: any[] = [];
fetchQueryDataWithHttpStream(
{
queryReq: { query },
type: "search",
traceId,
org_id: org,
pageType: "logs",
searchType: "ui",
},
{
data: (_req: unknown, res: any) => {
if (res?.type === "search_response_hits")
hits.push(...(res.content?.results?.hits ?? []));
},
error: (_req: unknown, err: any) => reject(streamError(err)),
complete: () => resolve(hits),
},
);
});
}
export const k8sInstantQuery = (
org: string,
id: string,
@@ -56,24 +91,13 @@ export const k8sSqlQuery = (
) =>
queryOptions({
queryKey: k8sKeys.sql(org, name, sql, timeKey(startUs, endUs, relative)),
queryFn: async (): Promise<any[]> => {
const res: any = await searchService.search(
{
org_identifier: org,
query: {
query: {
sql,
start_time: startUs,
end_time: endUs,
from: 0,
...(size ? { size } : {}),
},
},
page_type: "logs",
},
"ui",
);
return res?.data?.hits ?? [];
},
queryFn: (): Promise<any[]> =>
streamSqlHits(org, {
sql,
start_time: startUs,
end_time: endUs,
from: 0,
...(size ? { size } : {}),
}),
staleTime: LIVE_STALE_TIME,
});
@@ -38,12 +38,17 @@ vi.mock("@/lib/vue-shortcut-manager", async (importOriginal) => ({
shortcuts.defs = defs;
},
}));
vi.mock("@/services/search", () => ({ default: { metrics_query: vi.fn(), search: vi.fn() } }));
vi.mock("@/services/search", () => ({ default: { metrics_query: vi.fn() } }));
vi.mock("@/composables/useStreamingSearch", async () => {
const { streamSearchApi } = await import("./__fixtures__/streamSearchMock");
return { default: () => streamSearchApi((args) => streamSearch(args)) };
});
vi.mock("@/lib/feedback/Toast/useToast", () => ({ toast: vi.fn() }));
vi.mock("@/composables/useStreams", () => ({ default: () => ({ getStreams, getStream }) }));
const metricsQuery = vi.mocked(searchService.metrics_query);
const search = vi.mocked(searchService.search);
const search = vi.hoisted(() => vi.fn());
const streamSearch = (args: any) => search(args);
const ALL_IDS = Object.keys(QUERY_STREAM) as QueryId[];
const ALL_STREAMS = [...new Set(Object.values(QUERY_STREAM))];
@@ -0,0 +1,34 @@
// Copyright 2026 OpenObserve Inc.
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
// Adapts a unary `search({ query: { query } }) -> { data: { hits } }` mock to the _search_stream composable.
export function streamSearchApi(search: (args: any) => Promise<any>) {
return {
fetchQueryDataWithHttpStream: async (req: any, handlers: any) => {
try {
const res: any = await search({ query: { query: req.queryReq.query } });
handlers.data(req, {
type: "search_response_hits",
content: { results: { hits: res?.data?.hits ?? [] } },
});
handlers.complete(req, {});
} catch (e: any) {
const status = e?.response?.status ?? e?.status ?? null;
handlers.error(req, { content: { message: e?.message, status } });
}
},
cancelStreamQueryBasedOnRequestId: () => {},
};
}
@@ -26,7 +26,11 @@ import { parseUrlState, type K8sUrlState } from "./kubernetesUrlState";
const getStreams = vi.fn();
const getStream = vi.fn();
vi.mock("@/services/search", () => ({ default: { metrics_query: vi.fn(), search: vi.fn() } }));
vi.mock("@/services/search", () => ({ default: { metrics_query: vi.fn() } }));
vi.mock("@/composables/useStreamingSearch", async () => {
const { streamSearchApi } = await import("./__fixtures__/streamSearchMock");
return { default: () => streamSearchApi((args) => streamSearch(args)) };
});
vi.mock("@/composables/useStreams", () => ({ default: () => ({ getStreams, getStream }) }));
vi.mock("./kubernetesObjects", async (importOriginal) => {
const actual: any = await importOriginal();
@@ -34,7 +38,8 @@ vi.mock("./kubernetesObjects", async (importOriginal) => {
});
const metricsQuery = vi.mocked(searchService.metrics_query);
const search = vi.mocked(searchService.search);
const search = vi.hoisted(() => vi.fn());
const streamSearch = (args: any) => search(args);
const END = 1_700_000_000_000_000;
const START = END - 3_600_000_000;
@@ -465,7 +470,7 @@ describe("useKubernetesInventory", () => {
expect(counts).toHaveLength(1);
expect(counts[0].sql).toContain("k8s_cluster = 'prod'");
expect(counts[0].start_time).toBe(START);
expect(inv.eventCounts.value).toEqual({ total: 14, warnings: 8 });
await vi.waitFor(() => expect(inv.eventCounts.value).toEqual({ total: 14, warnings: 8 }));
});
it("stays null when the count returns no row, and is not sent for unscoped events", async () => {
@@ -615,6 +620,34 @@ describe("useKubernetesInventory", () => {
expect(inv.banners.value.map((b) => b.id)).toContain("partial-failure");
});
it.each([
["W", "body_object_type = 'Warning'"],
["O", "k8s_resource_name"],
])(
"banners a partial failure, not a page error, when the early %s query rejects",
async (name, marker) => {
fixture = { ...CL(), P1: [ksm({ namespace: "a", pod: "p", uid: "u", phase: "Running" })] };
search.mockImplementation((async (args: any) => {
if (args.query.query.sql.includes(marker)) throw new Error(`sql ${name} down`);
return { data: { hits: [] } };
}) as any);
const { inv } = await setup({ view: "pods", cluster: "prod" });
await inv.load();
expect(inv.pageError.value).toBeNull();
expect([...inv.failed.value.keys()].some((k) => k.startsWith(`${name}:`))).toBe(true);
expect(inv.banners.value.map((b) => b.id)).toContain("partial-failure");
},
);
it("shows a non-empty page error when every query rejects with an empty message", async () => {
metricsQuery.mockRejectedValue(new Error(""));
search.mockRejectedValue(new Error(""));
const { inv } = await setup({ view: "pods", cluster: "prod" });
await inv.load();
expect(typeof inv.pageError.value).toBe("string");
expect(inv.pageError.value).not.toBe("");
});
it("shows one page error when every query is rejected", async () => {
metricsQuery.mockRejectedValue(new Error("all down"));
search.mockRejectedValue(new Error("all down"));
@@ -623,6 +656,28 @@ describe("useKubernetesInventory", () => {
expect(inv.pageError.value).toBe("all down");
});
it("shows a page error, not partial-failure, when every list query rejects but the drawer object read succeeds", async () => {
metricsQuery.mockRejectedValue(new Error("all down"));
search.mockImplementation((async (args: any) => {
const q = args.query.query.sql;
if (q.includes("'horizontalpodautoscalers'") && q.endsWith("LIMIT 1"))
return { data: { hits: [] } };
throw new Error("all down");
}) as any);
const { inv } = await setup({ view: "hpas", cluster: "prod", details: "hpa/prod/m/h" });
await inv.load();
expect(inv.pageError.value).toBe("all down");
expect(inv.banners.value.map((b) => b.id)).not.toContain("partial-failure");
});
it("shows a page error when every query rejects and no cluster is in the URL", async () => {
metricsQuery.mockRejectedValue(new Error("all down"));
search.mockRejectedValue(new Error("all down"));
const { inv } = await setup({ view: "hpas", details: "hpa/prod/m/h" });
await inv.load();
expect(inv.pageError.value).toBe("all down");
});
it("reports forbidden when the view's anchor query is rejected with 403", async () => {
metricsQuery.mockImplementation((async ({ query }: { query: string }) => {
const id = idOf(query);
@@ -827,6 +882,577 @@ describe("useKubernetesInventory", () => {
});
});
describe("SQL overlaps the cluster queries", () => {
const POD_ROW = (pod: string) =>
ksm({ namespace: "a", pod, uid: `u-${pod}`, phase: "Running" });
const deferred = () => {
let release!: () => void;
const promise = new Promise<void>((r) => (release = r));
return { promise, release };
};
const holdPromql = (gate: Promise<void>) =>
metricsQuery.mockImplementation((async ({ query }: { query: string }) => {
const rows = fixture[idOf(query)];
await gate;
return vector(rows);
}) as any);
const objectCalls = (kind: string) =>
sentSql().filter((q: any) => q.sql.includes(`'${kind}'`) && q.sql.endsWith("LIMIT 1"));
const tick = () => new Promise((r) => setTimeout(r, 20));
it("issues SQL before the cluster PromQL resolves when a cluster is in the URL", async () => {
const gate = deferred();
holdPromql(gate.promise);
const { inv } = await setup({ view: "pods", cluster: "prod" });
const done = inv.load();
await vi.waitFor(() => expect(search).toHaveBeenCalled());
expect(sentSql().some((q: any) => q.sql.includes("count(*) AS total"))).toBe(false);
expect(inv.loading.value).toBe(true);
gate.release();
await done;
expect(inv.loading.value).toBe(false);
});
it("waits for the cluster PromQL before any SQL when no cluster is in the URL", async () => {
const gate = deferred();
holdPromql(gate.promise);
const { inv } = await setup({ view: "pods" });
const done = inv.load();
await vi.waitFor(() => expect(metricsQuery).toHaveBeenCalled());
await tick();
expect(search).not.toHaveBeenCalled();
gate.release();
await done;
expect(search).toHaveBeenCalled();
});
it("clears loading before the event counts land, then fills eventCounts", async () => {
const ec = deferred();
search.mockImplementation((async (args: any) => {
if (!args.query.query.sql.includes("count(*) AS total")) return { data: { hits: [] } };
await ec.promise;
return { data: { hits: [{ total: 5, warnings: "2" }] } };
}) as any);
const { inv } = await setup({ view: "pods", cluster: "prod" });
await inv.load();
expect(inv.loading.value).toBe(false);
expect(inv.loaded.value).toBe(true);
expect(inv.eventCounts.value).toBeNull();
ec.release();
await vi.waitFor(() => expect(inv.eventCounts.value).toEqual({ total: 5, warnings: 2 }));
});
it("requests a non-pod drawer's object once, before the cluster PromQL resolves", async () => {
const gate = deferred();
holdPromql(gate.promise);
const { inv } = await setup({ view: "hpas", cluster: "prod", details: "hpa/prod/m/h" });
const done = inv.load();
await vi.waitFor(() => expect(objectCalls("horizontalpodautoscalers")).toHaveLength(1));
gate.release();
await done;
expect(objectCalls("horizontalpodautoscalers")).toHaveLength(1);
});
it("still reads a pod drawer's object after the main load, once its uid is known", async () => {
const gate = deferred();
holdPromql(gate.promise);
fixture = { ...CL(), P1: [POD_ROW("p")] };
const { inv } = await setup({ view: "pods", cluster: "prod", details: "pod/prod/a/p" });
const done = inv.load();
await vi.waitFor(() => expect(search).toHaveBeenCalled());
expect(objectCalls("pods")).toHaveLength(0);
gate.release();
await done;
expect(objectCalls("pods")).toHaveLength(1);
expect(objectCalls("pods")[0].sql).toContain("'u-p'");
});
it("drops the late results of a superseded load", async () => {
const gate = deferred();
fixture = { ...CL(), P1: [POD_ROW("old")] };
holdPromql(gate.promise);
const { inv, time } = await setup({ view: "pods", cluster: "prod" }, { relative: false });
const first = inv.load();
await vi.waitFor(() => expect(metricsQuery).toHaveBeenCalled());
fixture = { ...CL(), P1: [POD_ROW("new")] };
holdPromql(Promise.resolve());
time.value = { start: START, end: END + HOUR, relative: false };
await inv.load();
gate.release();
await first;
await tick();
expect(inv.inventory.value.pods.map((p: any) => p.name)).toEqual(["new"]);
expect(inv.loading.value).toBe(false);
});
});
describe("event count badge across loads", () => {
const isEc = (args: any) => args.query.query.sql.includes("count(*) AS total");
let ecGate: Promise<void> = Promise.resolve();
let ecTotal = 0;
let ecFail = false;
const wireEc = () =>
search.mockImplementation((async (args: any) => {
if (!isEc(args)) return { data: { hits: [] } };
const total = ecTotal;
await ecGate;
if (ecFail) throw Object.assign(new Error("ec down"), { response: { status: 500 } });
return { data: { hits: [{ total, warnings: "0" }] } };
}) as any);
const later = (t: K8sTime): K8sTime => ({ ...t, end: t.end + HOUR });
beforeEach(() => {
ecGate = Promise.resolve();
ecTotal = 0;
ecFail = false;
wireEc();
});
it("keeps the previous count on a same-cluster refresh until the new one lands, even if it belongs to the earlier window", async () => {
ecTotal = 3;
const { inv, time } = await setup({ view: "pods", cluster: "prod" }, { relative: false });
await inv.load();
await vi.waitFor(() => expect(inv.eventCounts.value).toEqual({ total: 3, warnings: 0 }));
let release!: () => void;
ecGate = new Promise<void>((r) => (release = r));
ecTotal = 7;
time.value = later(time.value);
await inv.load();
expect(inv.eventCounts.value).toEqual({ total: 3, warnings: 0 });
release();
await vi.waitFor(() => expect(inv.eventCounts.value).toEqual({ total: 7, warnings: 0 }));
});
it("clears the count when the cluster changes", async () => {
ecTotal = 3;
const { inv, state } = await setup({ view: "pods", cluster: "prod" });
await inv.load();
await vi.waitFor(() => expect(inv.eventCounts.value).not.toBeNull());
let release!: () => void;
ecGate = new Promise<void>((r) => (release = r));
ecTotal = 9;
state.value = parseUrlState({ view: "pods", cluster: "dev" });
await inv.load();
expect(inv.eventCounts.value).toBeNull();
release();
await vi.waitFor(() => expect(inv.eventCounts.value).toEqual({ total: 9, warnings: 0 }));
});
it("clears the count when a load issues no count request", async () => {
ecTotal = 3;
const { inv, time } = await setup({ view: "pods", cluster: "prod" }, { relative: false });
await inv.load();
await vi.waitFor(() => expect(inv.eventCounts.value).not.toBeNull());
getStream.mockResolvedValue({ schema: [] });
await inv.loadStreams();
search.mockClear();
time.value = later(time.value);
await inv.load();
expect(sentSql().some((q: any) => q.sql.includes("count(*) AS total"))).toBe(false);
expect(inv.eventCounts.value).toBeNull();
});
it("keeps an in-flight count across a drawer open and close", async () => {
ecTotal = 5;
let release!: () => void;
ecGate = new Promise<void>((r) => (release = r));
const { inv, state } = await setup({ view: "pods", cluster: "prod" });
await inv.load();
state.value = parseUrlState({ view: "pods", cluster: "prod", details: "pod/prod/a/p" });
await inv.load();
state.value = parseUrlState({ view: "pods", cluster: "prod" });
await inv.load();
expect(inv.eventCounts.value).toBeNull();
release();
await vi.waitFor(() => expect(inv.eventCounts.value).toEqual({ total: 5, warnings: 0 }));
});
it("drops a late count from an older full load", async () => {
ecTotal = 1;
let releaseOld!: () => void;
ecGate = new Promise<void>((r) => (releaseOld = r));
const { inv, time } = await setup({ view: "pods", cluster: "prod" }, { relative: false });
await inv.load();
ecTotal = 8;
let releaseNew!: () => void;
ecGate = new Promise<void>((r) => (releaseNew = r));
time.value = later(time.value);
await inv.load();
releaseOld();
await new Promise((r) => setTimeout(r, 10));
expect(inv.eventCounts.value).toBeNull();
releaseNew();
await vi.waitFor(() => expect(inv.eventCounts.value).toEqual({ total: 8, warnings: 0 }));
});
it("drops a late count after reset", async () => {
ecTotal = 4;
let release!: () => void;
ecGate = new Promise<void>((r) => (release = r));
const { inv } = await setup({ view: "pods", cluster: "prod" });
await inv.load();
inv.reset();
release();
await new Promise((r) => setTimeout(r, 10));
expect(inv.eventCounts.value).toBeNull();
});
it("records an EC failure and banners it without delaying loading", async () => {
ecFail = true;
let release!: () => void;
ecGate = new Promise<void>((r) => (release = r));
const { inv } = await setup({ view: "pods", cluster: "prod" });
await inv.load();
expect(inv.loading.value).toBe(false);
expect(inv.failed.value.has("EC")).toBe(false);
release();
await vi.waitFor(() => expect(inv.failed.value.get("EC")).toBe(500));
expect(inv.eventCounts.value).toBeNull();
expect(inv.banners.value.map((b) => b.id)).toContain("partial-failure");
});
});
describe("drawer open and close skip the page load", () => {
const POD = (pod: string, uid: string) => ksm({ namespace: "a", pod, uid, phase: "Running" });
const objectFor = (uid: string) => ({
uid,
event_name: "p",
k8s_namespace_name: "a",
body_type: "MODIFIED",
body_object_metadata: '{"labels":{"app":"x"}}',
body_object_spec: "{}",
body_object_status: "{}",
});
const seed = () => {
fixture = { ...CL(), P1: [POD("p", "u1"), POD("q", "u2")] };
sqlHits = (sql) => {
if (sql.endsWith("LIMIT 100")) {
return [
{
event_name: "e",
body_object_note: "for drawer",
body_object_regarding: '{"kind":"Pod","name":"p"}',
},
];
}
if (sql.endsWith("LIMIT 1")) return [objectFor("u1")];
if (sql.includes("'Warning'")) {
return [
{ kind: "Pod", name: "p", namespace: "a", uid: "u1", last_seen: END - 1, reason: "R" },
];
}
return [];
};
};
const open = (view: string, details: string) => ({ view, details, cluster: "prod" });
const podView = { view: "nodes", cluster: "prod" };
const loadThenClear = async (query: Record<string, string> = podView) => {
seed();
const ctx = await setup(query);
await ctx.inv.load();
metricsQuery.mockClear();
search.mockClear();
return ctx;
};
it("sends only the missing detail ids and SQL, and never raises loading", async () => {
const { inv, state } = await loadThenClear();
const seen: boolean[] = [];
state.value = parseUrlState(open("nodes", "pod/prod/a/p"));
const done = inv.load();
seen.push(inv.loading.value);
await done;
seen.push(inv.loading.value);
expect(seen).toEqual([false, false]);
const clusterOrView = sentIds().filter((id) => CLUSTER_QUERIES.includes(id) || id === "NS1");
expect(clusterOrView).toEqual([]);
expect(sentIds().length).toBeGreaterThan(0);
const sqls = sentSql().map((q: any) => q.sql);
expect(sqls.some((q: string) => q.includes("'Warning'") && q.includes("'Pod'"))).toBe(true);
expect(sqls.some((q: string) => q.includes("'nodes'") && !q.endsWith("LIMIT 1"))).toBe(false);
expect(sqls.some((q: string) => q.endsWith("LIMIT 1"))).toBe(true);
expect(sqls.some((q: string) => q.endsWith("LIMIT 100"))).toBe(true);
expect(inv.detailLoading.value).toBe(false);
expect(inv.pageError.value).toBeNull();
});
it("produces the same drawer data as the full path for the same state", async () => {
const { inv, state } = await loadThenClear();
state.value = parseUrlState(open("nodes", "pod/prod/a/p"));
await inv.load();
const full = await setup(open("nodes", "pod/prod/a/p"));
await full.inv.load();
const shape = (i: typeof inv) =>
i.inventory.value.pods.map((p) => [p.name, p.warnings, p.object?.metadata.labels]);
expect(shape(inv)).toEqual(shape(full.inv));
expect(shape(inv).find((r) => r[0] === "p")?.[2]).toEqual({ app: "x" });
expect(inv.detailEvents.value?.map((e) => e.note)).toEqual(["for drawer"]);
expect(inv.detailEvents.value).toEqual(full.inv.detailEvents.value);
expect(inv.warnings.value.rows).toEqual(full.inv.warnings.value.rows);
expect([...inv.sql.value.keys()].sort()).toEqual([...full.inv.sql.value.keys()].sort());
expect([...inv.results.value.keys()].sort()).toEqual(
[...full.inv.results.value.keys()].sort(),
);
expect(inv.detailObserved.value).toBe(true);
});
it("closing sends nothing, clears the drawer events and drops drawer-only data", async () => {
seed();
const { inv, state } = await setup(open("nodes", "pod/prod/a/p"));
await inv.load();
expect(inv.detailEvents.value).not.toBeNull();
metricsQuery.mockClear();
search.mockClear();
state.value = parseUrlState(podView);
await inv.load();
expect(metricsQuery).not.toHaveBeenCalled();
expect(search).not.toHaveBeenCalled();
expect(inv.loading.value).toBe(false);
expect(inv.detailLoading.value).toBe(false);
expect(inv.detailEvents.value).toBeNull();
expect(inv.sql.value.has("O1obj")).toBe(false);
expect(inv.sql.value.has("W:Pod")).toBe(false);
const plain = await setup(podView);
await plain.inv.load();
expect([...inv.results.value.keys()].sort()).toEqual(
[...plain.inv.results.value.keys()].sort(),
);
expect(inv.warnings.value.rows).toEqual(plain.inv.warnings.value.rows);
});
it("switching A to B fetches only B's object and events", async () => {
seed();
const { inv, state } = await setup(open("nodes", "pod/prod/a/p"));
await inv.load();
metricsQuery.mockClear();
search.mockClear();
state.value = parseUrlState(open("nodes", "pod/prod/a/q"));
await inv.load();
expect(metricsQuery).not.toHaveBeenCalled();
const sqls = sentSql().map((q: any) => q.sql);
expect(sqls.some((q: string) => q.includes("'Warning'"))).toBe(false);
expect(sqls.filter((q: string) => q.endsWith("LIMIT 1"))).toHaveLength(1);
expect(sqls.filter((q: string) => q.endsWith("LIMIT 100"))).toHaveLength(1);
expect(inv.loading.value).toBe(false);
});
it.each([
["time end", (s: any) => (s.time.value = { start: START, end: END + 1, relative: true })],
["view", (s: any) => (s.state.value = parseUrlState({ view: "pods", cluster: "prod" }))],
["cluster", (s: any) => (s.state.value = parseUrlState({ view: "nodes", cluster: "dev" }))],
[
"namespaces",
(s: any) =>
(s.state.value = parseUrlState({ view: "nodes", cluster: "prod", namespace: "a" })),
],
])("falls back to the full path when the %s changes", async (_n, change) => {
const ctx = await loadThenClear({ view: "nodes", cluster: "prod", details: "pod/prod/a/p" });
queryClient.clear();
change(ctx);
const done = ctx.inv.load();
expect(ctx.inv.loading.value).toBe(true);
await done;
expect(sentIds().some((id) => CLUSTER_QUERIES.includes(id))).toBe(true);
});
it("falls back to the full path on force and after a page error", async () => {
const forced = await loadThenClear();
forced.state.value = parseUrlState(open("nodes", "pod/prod/a/p"));
const done = forced.inv.load({ force: true });
expect(forced.inv.loading.value).toBe(true);
await done;
expect(sentIds().some((id) => CLUSTER_QUERIES.includes(id))).toBe(true);
queryClient.clear();
const failedPage = await setup(podView);
reject = [...CLUSTER_QUERIES];
await failedPage.inv.load();
expect(failedPage.inv.pageError.value).not.toBeNull();
reject = [];
queryClient.clear();
metricsQuery.mockClear();
failedPage.state.value = parseUrlState(open("nodes", "pod/prod/a/p"));
await failedPage.inv.load();
expect(sentIds().some((id) => CLUSTER_QUERIES.includes(id))).toBe(true);
expect(failedPage.inv.pageError.value).toBeNull();
});
it("records a drawer fetch failure as a partial failure without a page error", async () => {
const { inv, state } = await loadThenClear();
search.mockImplementation((async (args: any) => {
if (args.query.query.sql.endsWith("LIMIT 100")) {
throw Object.assign(new Error("no"), { response: { status: 500 } });
}
return { data: { hits: sqlHits(args.query.query.sql) } };
}) as any);
state.value = parseUrlState(open("nodes", "pod/prod/a/p"));
await inv.load();
expect(inv.failed.value.get("DE")).toBe(500);
expect(inv.banners.value.map((b) => b.id)).toContain("partial-failure");
expect(inv.pageError.value).toBeNull();
expect(inv.inventory.value.pods.length).toBeGreaterThan(0);
});
it("clears a detail failure the drawer retry fixes and keeps one that fails again", async () => {
seed();
const warningFails = { on: true };
search.mockImplementation((async (args: any) => {
const sql = args.query.query.sql;
if (warningFails.on && sql.includes("'Warning'") && sql.includes("'Pod'")) {
throw Object.assign(new Error("no"), { response: { status: 500 } });
}
return { data: { hits: sqlHits(sql) } };
}) as any);
const { inv, state } = await setup(open("nodes", "pod/prod/a/p"));
await inv.load();
expect(inv.failed.value.get("W:Pod")).toBe(500);
expect(inv.banners.value.map((b) => b.id)).toContain("partial-failure");
state.value = parseUrlState(open("nodes", "pod/prod/a/q"));
await inv.load();
expect(inv.failed.value.get("W:Pod")).toBe(500);
warningFails.on = false;
state.value = parseUrlState(podView);
await inv.load();
state.value = parseUrlState(open("nodes", "pod/prod/a/p"));
await inv.load();
expect(inv.failed.value.has("W:Pod")).toBe(false);
expect(inv.banners.value.map((b) => b.id)).not.toContain("partial-failure");
});
it("drops a superseded drawer load", async () => {
const { inv, state } = await loadThenClear();
let release!: () => void;
const hold = new Promise<void>((r) => (release = r));
search.mockImplementation((async (args: any) => {
if (args.query.query.sql.endsWith("LIMIT 100")) await hold;
return { data: { hits: sqlHits(args.query.query.sql) } };
}) as any);
state.value = parseUrlState(open("nodes", "pod/prod/a/p"));
const first = inv.load();
await vi.waitFor(() =>
expect(sentSql().some((q: any) => q.sql.endsWith("LIMIT 100"))).toBe(true),
);
state.value = parseUrlState(podView);
await inv.load();
release();
await first;
expect(inv.detailEvents.value).toBeNull();
expect(inv.detailLoading.value).toBe(false);
});
it("treats a close before the first batch lands as a drawer toggle, then reopens fine", async () => {
const { inv, state } = await loadThenClear();
let release!: () => void;
const hold = new Promise<void>((r) => (release = r));
search.mockImplementation((async (args: any) => {
if (args.query.query.sql.includes("'Warning'")) await hold;
return { data: { hits: sqlHits(args.query.query.sql) } };
}) as any);
const seen: boolean[] = [];
state.value = parseUrlState(open("nodes", "pod/prod/a/p"));
const first = inv.load();
seen.push(inv.loading.value);
await vi.waitFor(() =>
expect(sentSql().some((q: any) => q.sql.includes("'Warning'"))).toBe(true),
);
metricsQuery.mockClear();
state.value = parseUrlState(podView);
const second = inv.load();
seen.push(inv.loading.value);
await second;
seen.push(inv.loading.value);
expect(sentIds().filter((id) => CLUSTER_QUERIES.includes(id) || id === "NS1")).toEqual([]);
release();
await first;
seen.push(inv.loading.value);
expect(seen).toEqual([false, false, false, false]);
expect(inv.detailLoading.value).toBe(false);
state.value = parseUrlState(open("nodes", "pod/prod/a/p"));
await inv.load();
expect(inv.loading.value).toBe(false);
expect(inv.detailEvents.value?.map((e) => e.note)).toEqual(["for drawer"]);
expect(inv.pageError.value).toBeNull();
});
it("reports a page error when every drawer request fails, and force-load recovers", async () => {
const { inv, state } = await loadThenClear();
reject = ALL_IDS.filter((id) => !CLUSTER_QUERIES.includes(id) && id !== "NS1");
search.mockImplementation((async () => {
throw Object.assign(new Error("no"), { response: { status: 500 } });
}) as any);
state.value = parseUrlState(open("nodes", "pod/prod/a/p"));
await inv.load();
expect(inv.pageError.value).not.toBeNull();
expect(inv.detailLoading.value).toBe(false);
reject = [];
seed();
search.mockImplementation((async (args: any) => ({
data: { hits: sqlHits(args.query.query.sql) },
})) as any);
metricsQuery.mockClear();
queryClient.clear();
await inv.load({ force: true });
expect(sentIds().some((id) => CLUSTER_QUERIES.includes(id))).toBe(true);
expect(inv.pageError.value).toBeNull();
expect(inv.detailLoading.value).toBe(false);
expect(inv.detailEvents.value?.map((e) => e.note)).toEqual(["for drawer"]);
});
it("only banners when the drawer fails partially", async () => {
const { inv, state } = await loadThenClear();
search.mockImplementation((async (args: any) => {
const sql = args.query.query.sql;
if (sql.endsWith("LIMIT 1")) {
throw Object.assign(new Error("no"), { response: { status: 500 } });
}
return { data: { hits: sqlHits(sql) } };
}) as any);
state.value = parseUrlState(open("nodes", "pod/prod/a/p"));
await inv.load();
expect(inv.pageError.value).toBeNull();
expect(inv.detailLoading.value).toBe(false);
expect(inv.banners.value.map((b) => b.id)).toContain("partial-failure");
});
it("takes the full path for a drawer toggle while a full load is in flight", async () => {
const { inv, state } = await loadThenClear();
let release!: () => void;
const hold = new Promise<void>((r) => (release = r));
const base = metricsQuery.getMockImplementation()!;
metricsQuery.mockImplementation((async (args: any) => {
await hold;
return base(args);
}) as any);
queryClient.clear();
const forced = inv.load({ force: true });
expect(inv.loading.value).toBe(true);
state.value = parseUrlState(open("nodes", "pod/prod/a/p"));
const toggled = inv.load();
expect(inv.loading.value).toBe(true);
release();
await Promise.all([forced, toggled]);
expect(inv.loading.value).toBe(false);
expect(inv.detailLoading.value).toBe(false);
expect(inv.pageError.value).toBeNull();
expect(inv.detailEvents.value?.map((e) => e.note)).toEqual(["for drawer"]);
});
it("fetches only the missing detail ids when no events SQL is planned", async () => {
seed();
const { inv, state } = await setup(podView, { logs: [] });
await inv.load();
metricsQuery.mockClear();
search.mockClear();
state.value = parseUrlState(open("nodes", "pod/prod/a/p"));
await inv.load();
expect(search).not.toHaveBeenCalled();
expect(sentIds().length).toBeGreaterThan(0);
expect(sentIds().filter((id) => CLUSTER_QUERIES.includes(id) || id === "NS1")).toEqual([]);
expect(inv.loading.value).toBe(false);
expect(inv.detailLoading.value).toBe(false);
expect(inv.pageError.value).toBeNull();
});
});
it("lists namespace options from NS1 plus the current selection", async () => {
fixture = {
...CL(),
@@ -138,6 +138,8 @@ export function useKubernetesInventory(
const sql = shallowRef<Map<string, any[]>>(new Map());
// Kept apart from sql so the drawer events landing does not rebuild the inventory.
const detailEventHits = shallowRef<any[] | null>(null);
// Off the critical path: the sidebar badge fills in after the page has rendered.
const eventCountHits = shallowRef<any[] | undefined>(undefined);
const failed = shallowRef<Map<string, number | null>>(new Map());
const pageError = ref<string | null>(null);
const loading = ref(false);
@@ -151,13 +153,18 @@ export function useKubernetesInventory(
end: number;
details: DetailsRef | null;
viewKey: string;
scope: string;
} | null>(null);
// What each load's object query depends on: a refresh, a new time or another scope changes it.
const pendingObjects = ref<string | null>(null);
const loadedObjects = ref<string | null>(null);
// metrics_query and search take no AbortSignal, so superseded responses are dropped by generation.
// Nothing cancels a superseded query, so its response is dropped by generation.
let generation = 0;
// Separate from generation so a drawer toggle does not discard the deferred event count.
let countsGeneration = 0;
// The drawer a details-only load is fetching, so closing it before it lands is still a drawer toggle.
let detailsTarget: DetailsRef | null | undefined;
let streamsGeneration = 0;
const has = (stream: string) => metricStreams.value?.has(stream) ?? false;
@@ -250,6 +257,8 @@ export function useKubernetesInventory(
// An org switch must never query or render with the previous org's streams or rows.
const reset = () => {
generation++;
countsGeneration++;
detailsTarget = undefined;
streamsGeneration++;
metricStreams.value = null;
logStreams.value = null;
@@ -260,6 +269,7 @@ export function useKubernetesInventory(
results.value = new Map();
sql.value = new Map();
detailEventHits.value = null;
eventCountHits.value = undefined;
failed.value = new Map();
pageError.value = null;
loading.value = false;
@@ -299,7 +309,7 @@ export function useKubernetesInventory(
const name = i < send.length ? send[i] : requests[i - send.length].name;
if (outcome.status === "rejected") {
out.failed.set(name, statusOf(outcome.reason));
out.firstError ??= outcome.reason?.message ?? String(outcome.reason);
out.firstError ||= outcome.reason?.message || String(outcome.reason);
} else if (i < send.length) out.results.set(send[i], outcome.value as Series[]);
else out.sql.set(name, outcome.value as any[]);
});
@@ -421,12 +431,113 @@ export function useKubernetesInventory(
];
};
// A drawer open/close/switch on a committed page must not redo the page's queries.
const isDetailsOnly = (s: K8sUrlState, t: K8sTime, cluster: string | null) => {
const prior = loadedFor.value;
if (loading.value || !loaded.value || pageError.value !== null || !prior) return false;
if (prior.end !== t.end || prior.cluster !== cluster) return false;
if (prior.scope !== scopeKeyOf(s, cluster)) return false;
const shown = detailsTarget === undefined ? prior.details : detailsTarget;
return encodeDetailsOrNull(shown) !== encodeDetailsOrNull(s.details);
};
const loadDetailsOnly = async (
gen: number,
org: string,
t: K8sTime,
s: K8sUrlState,
cluster: string | null,
prior: NonNullable<typeof loadedFor.value>,
) => {
const plan = (d: DetailsRef | null) => {
const at = { ...s, details: d };
return {
ids: neededIds(at),
requests: sqlRequests(at, t, cluster).filter(notCounts),
};
};
detailsTarget = s.details;
const old = plan(prior.details);
const next = plan(s.details);
const keep = new Set<string>([...next.ids, ...next.requests.map((r) => r.name)]);
const drop = new Set<string>(["O1obj", "DE", ...old.ids, ...old.requests.map((r) => r.name)]);
for (const name of keep) drop.delete(name);
const without = <K extends string, V>(m: Map<K, V>) =>
new Map([...m].filter(([k]) => !drop.has(k)));
detailEventHits.value = null;
const commitScope = (r: QueryResults, q: Map<string, any[]>, f: Map<string, number | null>) => {
results.value = r;
sql.value = q;
failed.value = f;
loadedFor.value = { ...prior, details: s.details, viewKey: viewKeyOf(s) };
};
if (!s.details) {
commitScope(without(results.value), without(sql.value), without(failed.value));
detailLoading.value = false;
return;
}
detailLoading.value = true;
const missingIds = next.ids.filter((id) => !results.value.has(id));
const missingSql = next.requests.filter((r) => !sql.value.has(r.name));
const earlyObject =
s.details.kind !== "pod"
? settle(org, t, [], null, detailRequests(s, t, cluster, null, "object"))
: null;
const extra = await settle(
org,
t,
missingIds,
matchersFor(cluster, clusterResults.value),
missingSql,
);
if (gen !== generation) return;
commitScope(
new Map([...without(results.value), ...extra.results]),
new Map([...without(sql.value), ...extra.sql]),
new Map([
...[...without(failed.value)].filter(
([k]) => !extra.results.has(k as QueryId) && !extra.sql.has(k),
),
...extra.failed,
]),
);
const detail = s.details;
const rowOf = () =>
findRow(
inventory.value,
detail.kind,
rowKey(detail.kind, detail.cluster, detail.namespace, detail.name),
);
const obj =
(await earlyObject) ??
(await settle(org, t, [], null, detailRequests(s, t, cluster, rowOf(), "object")));
if (gen !== generation) return;
mergeSql(obj);
const events = await settle(org, t, [], null, detailRequests(s, t, cluster, rowOf(), "events"));
if (gen !== generation) return;
detailEventHits.value = events.sql.get("DE") ?? null;
if (events.failed.size) failed.value = new Map([...failed.value, ...events.failed]);
const parts = [extra, obj, events];
const attempted = parts.reduce((n, p) => n + p.attempted, 0);
// Every drawer request failing is an outage, which the page reports with its retry like a full load does.
if (attempted > 0 && parts.reduce((n, p) => n + p.failed.size, 0) === attempted) {
fail(parts.find((p) => p.firstError)?.firstError ?? null);
return;
}
detailLoading.value = false;
};
const load = async ({ force = false }: { force?: boolean } = {}) => {
if (detection.value !== "detected") return;
const gen = ++generation;
const org = orgId();
const t = time();
const s = state();
const resolved = s.cluster ?? clusters.value[0] ?? null;
const detailsOnly = !force && isDetailsOnly(s, t, resolved);
const gen = ++generation;
if (detailsOnly) return loadDetailsOnly(gen, org, t, s, resolved, loadedFor.value!);
const countsGen = ++countsGeneration;
detailsTarget = undefined;
loading.value = true;
detailLoading.value = !!s.details;
if (force) {
@@ -442,6 +553,14 @@ export function useKubernetesInventory(
refreshNonce.value,
]);
pendingObjects.value = objectsKey;
// SQL needs only the cluster name, so a cluster already in the URL lets it overlap the cluster queries.
const planned = s.cluster ? sqlRequests(s, t, s.cluster) : null;
const early = planned ? settle(org, t, [], null, planned.filter(notCounts)) : null;
// Kept out of the page batch so a drawer read cannot mask every list query failing.
const earlyObject =
s.cluster && s.details && s.details.kind !== "pod"
? settle(org, t, [], null, detailRequests(s, t, s.cluster, null, "object"))
: null;
const cl = await settle(
org,
t,
@@ -467,13 +586,15 @@ export function useKubernetesInventory(
].sort((a, b) => a.localeCompare(b));
}
const cluster = s.cluster ?? clusters.value[0] ?? null;
const main = await settle(
org,
t,
neededIds(s),
matchersFor(cluster, cl.results),
sqlRequests(s, t, cluster),
);
const requests = planned ?? sqlRequests(s, t, cluster);
const blocking = requests.filter(notCounts);
// Same cluster as the early batch, so its SQL is already in flight and only the PromQL is left.
const main = early
? mergeSettled(
await settle(org, t, neededIds(s), matchersFor(cluster, cl.results), []),
await early,
)
: await settle(org, t, neededIds(s), matchersFor(cluster, cl.results), blocking);
if (gen !== generation) return;
if (main.attempted > 0 && main.failed.size === main.attempted) {
fail(main.firstError);
@@ -481,6 +602,15 @@ export function useKubernetesInventory(
}
commit(main, cluster, t, s);
loadedObjects.value = objectsKey;
const counts = requests.filter((r) => r.name === "EC");
if (!counts.length) eventCountHits.value = undefined;
else {
void settle(org, t, [], null, counts).then((ec) => {
if (countsGen !== countsGeneration) return;
eventCountHits.value = ec.sql.get("EC");
if (ec.failed.size) failed.value = new Map([...failed.value, ...ec.failed]);
});
}
const detail = s.details;
if (detail) {
const row = findRow(
@@ -488,7 +618,10 @@ export function useKubernetesInventory(
detail.kind,
rowKey(detail.kind, detail.cluster, detail.namespace, detail.name),
);
const obj = await settle(org, t, [], null, detailRequests(s, t, cluster, row, "object"));
// A non-pod object needs no row, so the early batch has already read it.
const obj =
(await earlyObject) ??
(await settle(org, t, [], null, detailRequests(s, t, cluster, row, "object")));
if (gen !== generation) return;
mergeSql(obj);
const enriched = findRow(
@@ -515,6 +648,7 @@ export function useKubernetesInventory(
results.value = new Map();
sql.value = new Map();
detailEventHits.value = null;
eventCountHits.value = undefined;
failed.value = new Map();
loading.value = false;
detailLoading.value = false;
@@ -525,12 +659,15 @@ export function useKubernetesInventory(
results.value = main.results;
sql.value = main.sql;
detailEventHits.value = null;
// The last count stays until the new one lands; another cluster's count never shows.
if (loadedFor.value?.cluster !== cluster) eventCountHits.value = undefined;
failed.value = main.failed;
loadedFor.value = {
cluster,
end: t.end,
details: s.details,
viewKey: viewKeyOf(s),
scope: scopeKeyOf(s, cluster),
};
lastUpdatedAt.value = main.updatedAt;
loading.value = false;
@@ -560,7 +697,7 @@ export function useKubernetesInventory(
return { rows: [...seen.values()], truncated };
});
const eventCounts = computed(() => parseEventCounts(sql.value.get("EC")));
const eventCounts = computed(() => parseEventCounts(eventCountHits.value));
// Parsed once per hits array: the inventory, banners and drawer all read them, across drawer merges.
const parsed = new WeakMap<any[], ObjectRecord[]>();
@@ -725,6 +862,26 @@ function viewKeyOf(s: K8sUrlState) {
return JSON.stringify([s.view, s.cluster, s.namespaces, s.entity, byNode]);
}
// The drawer can set the cluster the page defaulted to, which is not a scope change.
function scopeKeyOf(s: K8sUrlState, cluster: string | null) {
return viewKeyOf({ ...s, cluster });
}
const encodeDetailsOrNull = (d: DetailsRef | null) => (d ? encodeDetails(d) : null);
const notCounts = (r: SqlRequest) => r.name !== "EC";
function mergeSettled(a: Settled, b: Settled): Settled {
return {
results: new Map([...a.results, ...b.results]),
sql: new Map([...a.sql, ...b.sql]),
failed: new Map([...a.failed, ...b.failed]),
firstError: a.firstError ?? b.firstError,
attempted: a.attempted + b.attempted,
updatedAt: Math.max(a.updatedAt, b.updatedAt),
};
}
function dedupe(requests: SqlRequest[]): SqlRequest[] {
return [...new Map(requests.map((r) => [r.name, r])).values()];
}