diff --git a/web/src/composables/useStreamingSearch.spec.ts b/web/src/composables/useStreamingSearch.spec.ts index 5d0b52cd19..76e2799541 100644 --- a/web/src/composables/useStreamingSearch.spec.ts +++ b/web/src/composables/useStreamingSearch.spec.ts @@ -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, "bad gateway", "HTTP 502"], + [403, "forbidden", "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", () => { diff --git a/web/src/composables/useStreamingSearch.ts b/web/src/composables/useStreamingSearch.ts index 6b23c73ec6..74d6c83944 100644 --- a/web/src/composables/useStreamingSearch.ts +++ b/web/src/composables/useStreamingSearch.ts @@ -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)); } } }; diff --git a/web/src/services/kubernetes.queries.spec.ts b/web/src/services/kubernetes.queries.spec.ts index ed525f9490..60dd06dba6 100644 --- a/web/src/services/kubernetes.queries.spec.ts +++ b/web/src/services/kubernetes.queries.spec.ts @@ -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 }); }); }); diff --git a/web/src/services/kubernetes.queries.ts b/web/src/services/kubernetes.queries.ts index adb5e65cd7..55cf14e84c 100644 --- a/web/src/services/kubernetes.queries.ts +++ b/web/src/services/kubernetes.queries.ts @@ -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) { + const { fetchQueryDataWithHttpStream } = useHttpStreaming(); + const { traceId } = generateTraceContext(); + return new Promise((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 => { - 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 => + streamSqlHits(org, { + sql, + start_time: startUs, + end_time: endUs, + from: 0, + ...(size ? { size } : {}), + }), staleTime: LIVE_STALE_TIME, }); diff --git a/web/src/views/Infrastructure/kubernetes2/KubernetesPage.spec.ts b/web/src/views/Infrastructure/kubernetes2/KubernetesPage.spec.ts index 3e76f49a1e..e6d80dfd3b 100644 --- a/web/src/views/Infrastructure/kubernetes2/KubernetesPage.spec.ts +++ b/web/src/views/Infrastructure/kubernetes2/KubernetesPage.spec.ts @@ -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))]; diff --git a/web/src/views/Infrastructure/kubernetes2/__fixtures__/streamSearchMock.ts b/web/src/views/Infrastructure/kubernetes2/__fixtures__/streamSearchMock.ts new file mode 100644 index 0000000000..06f9eef14d --- /dev/null +++ b/web/src/views/Infrastructure/kubernetes2/__fixtures__/streamSearchMock.ts @@ -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 . + +// Adapts a unary `search({ query: { query } }) -> { data: { hits } }` mock to the _search_stream composable. +export function streamSearchApi(search: (args: any) => Promise) { + 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: () => {}, + }; +} diff --git a/web/src/views/Infrastructure/kubernetes2/useKubernetesInventory.spec.ts b/web/src/views/Infrastructure/kubernetes2/useKubernetesInventory.spec.ts index f4b2f641af..a4eb304c94 100644 --- a/web/src/views/Infrastructure/kubernetes2/useKubernetesInventory.spec.ts +++ b/web/src/views/Infrastructure/kubernetes2/useKubernetesInventory.spec.ts @@ -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((r) => (release = r)); + return { promise, release }; + }; + const holdPromql = (gate: Promise) => + 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 = 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((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((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((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((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((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((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((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 = 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((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((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((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(), diff --git a/web/src/views/Infrastructure/kubernetes2/useKubernetesInventory.ts b/web/src/views/Infrastructure/kubernetes2/useKubernetesInventory.ts index 7575caa185..36d99c3f96 100644 --- a/web/src/views/Infrastructure/kubernetes2/useKubernetesInventory.ts +++ b/web/src/views/Infrastructure/kubernetes2/useKubernetesInventory.ts @@ -138,6 +138,8 @@ export function useKubernetesInventory( const sql = shallowRef>(new Map()); // Kept apart from sql so the drawer events landing does not rebuild the inventory. const detailEventHits = shallowRef(null); + // Off the critical path: the sidebar badge fills in after the page has rendered. + const eventCountHits = shallowRef(undefined); const failed = shallowRef>(new Map()); const pageError = ref(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(null); const loadedObjects = ref(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, + ) => { + 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([...next.ids, ...next.requests.map((r) => r.name)]); + const drop = new Set(["O1obj", "DE", ...old.ids, ...old.requests.map((r) => r.name)]); + for (const name of keep) drop.delete(name); + const without = (m: Map) => + new Map([...m].filter(([k]) => !drop.has(k))); + detailEventHits.value = null; + const commitScope = (r: QueryResults, q: Map, f: Map) => { + 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(); @@ -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()]; }