diff --git a/src/core/src/workflows/http_trigger/enqueue.rs b/src/core/src/workflows/http_trigger/enqueue.rs index c01c378a1d..7581d6c61b 100644 --- a/src/core/src/workflows/http_trigger/enqueue.rs +++ b/src/core/src/workflows/http_trigger/enqueue.rs @@ -102,6 +102,8 @@ pub fn hook_audit_message( trace_id: None, }, delegated: None, + fleet_id: None, + target: None, } } @@ -128,6 +130,8 @@ pub fn reveal_audit_message( trace_id: None, }, delegated: None, + fleet_id: None, + target: None, } } diff --git a/src/infra/src/table/fleet_rollouts.rs b/src/infra/src/table/fleet_rollouts.rs index 0c1a05dd61..a9623dd452 100644 --- a/src/infra/src/table/fleet_rollouts.rs +++ b/src/infra/src/table/fleet_rollouts.rs @@ -184,9 +184,10 @@ pub async fn guard_with( Ok(res.rows_affected) } -/// Writes status, from_version, phases, failure, events and gates of a row the caller fenced. +/// Writes the strategy, status, from_version, phases, failure, events and gates of a fenced row. pub async fn save_state_with(conn: &C, row: &Model) -> Result<(), Error> { Entity::update_many() + .col_expr(Column::Strategy, Expr::value(row.strategy.clone())) .col_expr(Column::Status, Expr::value(row.status.clone())) .col_expr(Column::FromVersion, Expr::value(row.from_version)) .col_expr(Column::Phases, Expr::value(row.phases.clone())) @@ -505,6 +506,7 @@ mod tests { .unwrap(); assert_eq!(members_with(&db, "r1").await.unwrap()[1].state, "running"); + row.strategy = serde_json::json!({"verify": true}).to_string(); row.status = "halted".to_string(); row.from_version = Some(7); row.failure = diff --git a/web/src/components/fleet/FleetFlowOverview.spec.ts b/web/src/components/fleet/FleetFlowOverview.spec.ts index 714d31c77f..ea5abde5d2 100644 --- a/web/src/components/fleet/FleetFlowOverview.spec.ts +++ b/web/src/components/fleet/FleetFlowOverview.spec.ts @@ -193,10 +193,10 @@ afterEach(() => { }); describe("FleetFlowOverview map", () => { - it("draws the four columns fleet, receiver, exporter, destination", async () => { + it("draws the three columns fleet, receiver, destination", async () => { serve(basePipelines()); await mountOverview(); - const columns = ["f:", "r:", "e:", "d:"].map((prefix) => + const columns = ["f:", "r:", "d:"].map((prefix) => nodes() .filter((n) => n.id.startsWith(prefix)) .map((n) => n.id), @@ -204,44 +204,44 @@ describe("FleetFlowOverview map", () => { expect(columns).toEqual([ ["f:prod", "f:edge"], ["r:prod|filelog", "r:prod|hostmetrics", "r:edge|otlp"], - ["e:prod|otlphttp/o2", "e:prod|kafka/bus", "e:edge|otlphttp/o2"], [`d:${D1.id}`, `d:${D2.id}`], ]); const x = (id: string) => nodes().find((n) => n.id === id).position.x; expect(x("f:prod")).toBeLessThan(x("r:prod|filelog")); - expect(x("r:prod|filelog")).toBeLessThan(x("e:prod|kafka/bus")); - expect(x("e:prod|kafka/bus")).toBeLessThan(x(`d:${D1.id}`)); + expect(x("r:prod|filelog")).toBeLessThan(x(`d:${D1.id}`)); }); it("colours edges by signal and sizes them by rate, never by severity", async () => { serve(basePipelines()); await mountOverview(); - const metrics = edge(`ed:prod|otlphttp/o2|${D1.id}|metrics`).data; - const logs = edge(`ed:prod|otlphttp/o2|${D1.id}|logs`).data; + const metrics = edge(`rd:prod|hostmetrics|${D1.id}|metrics`).data; + const logs = edge(`rd:prod|filelog|${D1.id}|logs`).data; expect(metrics.signal).toBe("metrics"); expect(logs.signal).toBe("logs"); expect(metrics.width).toBeGreaterThan(logs.width); expect(Object.keys(logs)).not.toContain("severity"); }); - it("branches the failing exporter off with a dashed send-failed line and rings it", async () => { + it("branches the failing destination off with a dashed send-failed line and rings it", async () => { serve(basePipelines()); await mountOverview(); - const branch = edge("br:e:prod|otlphttp/o2"); - expect(branch.source).toBe("e:prod|otlphttp/o2"); - expect(branch.data).toMatchObject({ branch: true, label: "Send failed 3.3%" }); + const branch = edge(`br:d:${D1.id}`); + expect(branch.source).toBe(`d:${D1.id}`); + expect(branch.data).toMatchObject({ branch: true, label: "Send failed 20%" }); expect(edges().filter((e) => e.data.branch)).toHaveLength(1); - const ring = (id: string) => nodes().find((n) => n.id === id).data.ring; - expect(ring("e:prod|otlphttp/o2")).toBe("error"); - expect(ring("e:prod|kafka/bus")).toBe("warning"); - expect(ring("e:edge|otlphttp/o2")).toBeUndefined(); + const data = (id: string) => nodes().find((n) => n.id === id).data; + expect(data(`d:${D1.id}`)).toMatchObject({ ring: "error", outbound: true }); + expect(data(`d:${D2.id}`)).toMatchObject({ ring: "warning", outbound: false }); + const sink = nodes().find((n) => n.id === `x:d:${D1.id}`); + expect(sink.ariaLabel).toBe("o2.acme.io"); + expect(sink.data.label).toBe("o2.acme.io"); }); - it("opens the failing exporter's details from the branch", async () => { + it("opens the failing destination's details from the branch", async () => { serve(basePipelines()); await mountOverview(); - await clickEdge("br:e:prod|otlphttp/o2"); - expect(details().props("title")).toBe("otlphttp/o2"); + await clickEdge(`br:d:${D1.id}`); + expect(details().props("title")).toBe("o2.acme.io"); expect( details() .props("issues") @@ -259,13 +259,13 @@ describe("FleetFlowOverview map", () => { expect(nodes().find((n) => n.id === "f:edge").data.selected).toBe(true); }); - it("drops the exporters of an unchecked destination from the map", async () => { + it("drops an unchecked destination and its lines from the map", async () => { serve(basePipelines()); await mountOverview(); lists()[1].vm.$emit("update:modelValue", [D1.id]); await flushPromises(); expect(nodes().some((n) => n.id === `d:${D2.id}`)).toBe(false); - expect(nodes().some((n) => n.id === "e:prod|kafka/bus")).toBe(false); + expect(edges().some((e) => e.target === `d:${D2.id}`)).toBe(false); }); }); diff --git a/web/src/components/fleet/FleetFlowOverview.vue b/web/src/components/fleet/FleetFlowOverview.vue index 51bdf1a2d9..5665a2be13 100644 --- a/web/src/components/fleet/FleetFlowOverview.vue +++ b/web/src/components/fleet/FleetFlowOverview.vue @@ -346,9 +346,9 @@ const DESTINATION_ICON: Record = { awss3: "inventory-2", }; -const COLUMN_ICON: Record = { f: "dns", r: "input", e: "send" }; +const COLUMN_ICON: Record = { f: "dns", r: "input" }; -// An exporter failing at least this share of its items, or a receiver refusing it, is rung like its issue. +// A destination failing at least this share of its items, or a receiver refusing it, is rung like its issue. const ISSUE_RATIO = 0.01; const QUEUE_WARNING = 0.8; @@ -574,13 +574,13 @@ function nodeSignal(n: MapNode): FlowSignal { function nodeLabel(n: MapNode): I18nText { if (n.column === "fleet") return raw(fleetName(n.name)); - if (n.column === "destination") return destinationLabel(n.name); + if (n.column === "destination" || n.column === "sink") return destinationLabel(n.name); return raw(n.name); } function nodeRing(n: MapNode): PathSeverity | undefined { - if (n.column === "exporter") { - if (ratio(n.failed, n.rate) >= ISSUE_RATIO || n.dropped > 0) return "error"; + if (n.column === "destination") { + if (n.failedRatio >= ISSUE_RATIO || n.dropped > 0) return "error"; if ((n.queue ?? 0) >= QUEUE_WARNING) return "warning"; } if (n.column === "receiver" && ratio(n.refused, n.rate) >= ISSUE_RATIO) return "warning"; @@ -623,7 +623,7 @@ function nodeData(n: MapNode): FleetFlowNodeData { compact: !destination, selected: sel?.kind === "node" && sel.id === n.id, inbound: true, - outbound: !destination, + outbound: !destination || n.failed > 0 || n.dropped > 0, ring: nodeRing(n), }; } diff --git a/web/src/components/fleet/pipelineGraph.spec.ts b/web/src/components/fleet/pipelineGraph.spec.ts index 33783814ce..cb3ade8d77 100644 --- a/web/src/components/fleet/pipelineGraph.spec.ts +++ b/web/src/components/fleet/pipelineGraph.spec.ts @@ -65,57 +65,112 @@ const traces = pipeline({ const byId = (list: T[]) => new Map(list.map((x) => [x.id, x])); describe("buildPipelineMap", () => { - it("lays out four columns: fleet, receiver, exporter, destination", () => { + it("lays out three columns: fleet, receiver, destination", () => { const map = buildPipelineMap([logs, traces], layout); - const columns = ["fleet", "receiver", "exporter", "destination"].map((c) => + const columns = ["fleet", "receiver", "destination"].map((c) => map.nodes.filter((n) => n.column === c).map((n) => n.id), ); - expect(columns).toEqual([ - ["f:prod"], - ["r:prod|filelog", "r:prod|otlp"], - ["e:prod|otlphttp", "e:prod|s3"], - ["d:d_o2", "d:d_s3"], - ]); + expect(columns).toEqual([["f:prod"], ["r:prod|filelog", "r:prod|otlp"], ["d:d_o2", "d:d_s3"]]); const xs = columns.map((ids) => byId(map.nodes).get(ids[0])!.x); expect(xs).toEqual([...xs].sort((a, b) => a - b)); - expect(new Set(xs).size).toBe(4); + expect(new Set(xs).size).toBe(3); }); - it("splits an exporter's rate over the receivers by their share of the pipeline input", () => { + it("splits a destination's rate over the receivers by their share of the pipeline input", () => { const edges = byId(buildPipelineMap([logs], layout).edges); - expect(edges.get("re:prod|filelog|otlphttp|logs")?.rate).toBe(150); - expect(edges.get("re:prod|otlp|otlphttp|logs")?.rate).toBe(50); - expect(edges.get("re:prod|filelog|s3|logs")?.rate).toBe(300); + expect(edges.get("rd:prod|filelog|d_o2|logs")?.rate).toBe(150); + expect(edges.get("rd:prod|otlp|d_o2|logs")?.rate).toBe(50); + expect(edges.get("rd:prod|filelog|d_s3|logs")?.rate).toBe(300); }); - it("keeps one line per signal, so a shared receiver and exporter carry each signal apart", () => { + it("merges every exporter sending to one destination, across fleets, into one node", () => { + const prod = pipeline({ + fleet: "prod", + name: "logs", + receivers: [{ name: "otlp", rate: 100, refused: 0 }], + exporters: [ + exporter("otlphttp/a", "d_o2", 60, { failed: 10, queue: 0.2 }), + exporter("otlphttp/b", "d_o2", 40), + ], + }); + const edge = pipeline({ + fleet: "edge", + name: "logs", + receivers: [{ name: "otlp", rate: 30, refused: 0 }], + exporters: [exporter("otlphttp/c", "d_o2", 30, { dropped: 5, queue: 0.9 })], + }); + const map = buildPipelineMap([prod, edge], layout); + expect(map.nodes.filter((n) => n.column === "destination").map((n) => n.id)).toEqual([ + "d:d_o2", + ]); + expect(byId(map.nodes).get("d:d_o2")).toMatchObject({ + rate: 130, + failed: 10, + dropped: 5, + queue: 0.9, + pipelines: ["prod/logs", "edge/logs"], + }); + const edges = byId(map.edges); + expect(edges.get("rd:prod|otlp|d_o2|logs")?.rate).toBe(100); + expect(edges.get("rd:edge|otlp|d_o2|logs")?.rate).toBe(30); + }); + + it("keeps one line per signal, so a shared receiver and destination carry each signal apart", () => { const map = buildPipelineMap([logs, traces], layout); const edges = byId(map.edges); expect(edges.get("fr:prod|otlp|logs")?.rate).toBe(100); expect(edges.get("fr:prod|otlp|traces")?.rate).toBe(80); - expect(edges.get("ed:prod|otlphttp|d_o2|logs")?.signal).toBe("logs"); - expect(edges.get("ed:prod|otlphttp|d_o2|traces")?.signal).toBe("traces"); - expect(byId(map.nodes).get("e:prod|otlphttp")?.rate).toBe(280); + expect(edges.get("rd:prod|otlp|d_o2|logs")?.signal).toBe("logs"); + expect(edges.get("rd:prod|otlp|d_o2|traces")?.signal).toBe("traces"); + expect(byId(map.nodes).get("d:d_o2")?.rate).toBe(280); expect(byId(map.nodes).get("r:prod|otlp")?.rate).toBe(180); - expect(byId(map.nodes).get("e:prod|otlphttp")?.pipelines).toEqual(["prod/logs", "prod/traces"]); + expect(byId(map.nodes).get("d:d_o2")?.pipelines).toEqual(["prod/logs", "prod/traces"]); }); - it("draws a failure branch off the failing exporter with its failed share", () => { + it("draws a failure branch off the failing destination with its failed share", () => { const map = buildPipelineMap([logs], layout); const branch = map.edges.find((e) => e.branch); expect(branch).toMatchObject({ - id: "br:e:prod|otlphttp", - source: "e:prod|otlphttp", - target: "x:e:prod|otlphttp", + id: "br:d:d_o2", + source: "d:d_o2", + target: "x:d:d_o2", branch: { kind: "failed", ratio: 0.2 }, }); expect(map.edges.filter((e) => e.branch)).toHaveLength(1); - const sink = byId(map.nodes).get("x:e:prod|otlphttp")!; - const from = byId(map.nodes).get("e:prod|otlphttp")!; + const sink = byId(map.nodes).get("x:d:d_o2")!; + const from = byId(map.nodes).get("d:d_o2")!; expect(sink.x).toBeGreaterThan(from.x); expect(sink.y).toBeGreaterThan(from.y); }); + it("keeps a failure branch and its label inside the canvas, past the last column", () => { + const failing = buildPipelineMap([logs], layout).nodes; + const healthy = buildPipelineMap([traces], layout).nodes; + const x = (nodes: typeof failing, id: string) => byId(nodes).get(id)!.x; + expect(x(healthy, "d:d_o2") + layout.nodeWidth).toBe(layout.width - 16); + const sink = x(failing, "x:d:d_o2"); + expect(sink).toBeGreaterThan(x(failing, "d:d_o2") + layout.nodeWidth); + expect(sink + 120).toBeLessThanOrEqual(layout.width); + }); + + it("reports the worst exporter's share, so a busy healthy fleet cannot hide a failing one", () => { + const failing = pipeline({ + fleet: "edge", + name: "logs", + receivers: [{ name: "otlp", rate: 100, refused: 0 }], + exporters: [exporter("otlphttp/a", "d_o2", 50, { failed: 50 })], + }); + const healthy = pipeline({ + fleet: "prod", + name: "logs", + receivers: [{ name: "otlp", rate: 100000, refused: 0 }], + exporters: [exporter("otlphttp/b", "d_o2", 100000)], + }); + const map = buildPipelineMap([failing, healthy], layout); + expect(byId(map.nodes).get("d:d_o2")?.failedRatio).toBe(0.5); + expect(map.edges.find((e) => e.branch)?.branch).toEqual({ kind: "failed", ratio: 0.5 }); + }); + it("names a drop-only branch as dropped", () => { const dropping = pipeline({ fleet: "edge", @@ -145,8 +200,8 @@ describe("buildPipelineMap", () => { }); const edges = byId(buildPipelineMap([silent], layout).edges); expect(edges.get("fr:edge|a|logs")?.rate).toBeNull(); - expect(edges.get("re:edge|a|otlp|logs")?.rate).toBe(5); - expect(edges.get("re:edge|a|debug|logs")?.rate).toBeNull(); + expect(edges.get("rd:edge|a|d_o2|logs")?.rate).toBe(5); + expect(edges.get("rd:edge|a|d_dbg|logs")?.rate).toBeNull(); }); }); diff --git a/web/src/components/fleet/pipelineGraph.ts b/web/src/components/fleet/pipelineGraph.ts index a3673730a0..a5b674bd52 100644 --- a/web/src/components/fleet/pipelineGraph.ts +++ b/web/src/components/fleet/pipelineGraph.ts @@ -17,7 +17,7 @@ import type { FleetPipeline } from "@/services/fleet"; import { ratio } from "./flowIssues"; import type { Telemetry } from "./flowTypes"; -export type MapColumn = "fleet" | "receiver" | "exporter" | "destination" | "sink"; +export type MapColumn = "fleet" | "receiver" | "destination" | "sink"; export interface MapNode { id: string; @@ -30,6 +30,8 @@ export interface MapNode { dropped: number; refused: number; queue: number | null; + failedRatio: number; + droppedRatio: number; pipelines: string[]; x: number; y: number; @@ -57,9 +59,12 @@ export interface PipelineMap { edges: MapEdge[]; } -const COLUMNS: MapColumn[] = ["fleet", "receiver", "exporter", "destination"]; +const COLUMNS: MapColumn[] = ["fleet", "receiver", "destination"]; const MIN_COLUMN_GAP = 168; const PAD = 16; +// The last column has no neighbour, so its failure branch and the label after it need their own room. +const BRANCH_ROOM = 168; +const BRANCH_REACH = 40; export const MIN_WIDTH = 1.5; export const MAX_WIDTH = 8; @@ -99,6 +104,8 @@ class MapBuilder { dropped: 0, refused: 0, queue: null, + failedRatio: 0, + droppedRatio: 0, pipelines: [], x: 0, y: 0, @@ -121,24 +128,17 @@ class MapBuilder { if (!n.pipelines.includes(p.id)) n.pipelines.push(p.id); } - addExporters(p: FleetPipeline) { + addDestinations(p: FleetPipeline) { for (const ex of p.exporters) { - const en = this.node(`e:${p.fleet}|${ex.name}`, "exporter", p.fleet, ex.name); - this.touch(en, p); - en.rate = addRate(en.rate, ex.rate); - en.failed += ex.failed; - en.dropped += ex.dropped; - en.queue = maxRate(en.queue, ex.queue); const dn = this.node(`d:${ex.destination}`, "destination", p.fleet, ex.destination); this.touch(dn, p); dn.rate = addRate(dn.rate, ex.rate); - const ed = this.edge( - `ed:${p.fleet}|${ex.name}|${ex.destination}|${p.signal}`, - en.id, - dn.id, - p, - ); - ed.rate = addRate(ed.rate, ex.rate); + dn.failed += ex.failed; + dn.dropped += ex.dropped; + dn.queue = maxRate(dn.queue, ex.queue); + // Per exporter, because a destination is shared and a busy healthy fleet must not dilute a failing one. + dn.failedRatio = Math.max(dn.failedRatio, ratio(ex.failed, ex.rate)); + dn.droppedRatio = Math.max(dn.droppedRatio, ratio(ex.dropped, ex.rate)); } } @@ -156,31 +156,31 @@ class MapBuilder { fr.rate = maxRate(fr.rate, r.rate); const share = receiverShare(p, r.rate); for (const ex of p.exporters) { - const id = `re:${p.fleet}|${r.name}|${ex.name}|${p.signal}`; - const re = this.edge(id, rn.id, `e:${p.fleet}|${ex.name}`, p); - re.rate = addRate(re.rate, ex.rate === null ? null : ex.rate * share); + const id = `rd:${p.fleet}|${r.name}|${ex.destination}|${p.signal}`; + const rd = this.edge(id, rn.id, `d:${ex.destination}`, p); + rd.rate = addRate(rd.rate, ex.rate === null ? null : ex.rate * share); } } } addBranches() { - for (const en of [...this.nodes.values()].filter((n) => n.column === "exporter")) { - if (en.failed <= 0 && en.dropped <= 0) continue; - const failed = en.failed > 0; - const sink = this.node(`x:${en.id}`, "sink", en.fleet, en.name); - sink.pipelines = [...en.pipelines]; - sink.rate = failed ? en.failed : en.dropped; - const signal = [...this.edges.values()].find((e) => e.target === en.id)?.signal ?? "logs"; - this.edges.set(`br:${en.id}`, { - id: `br:${en.id}`, - source: en.id, + for (const dn of [...this.nodes.values()].filter((n) => n.column === "destination")) { + if (dn.failed <= 0 && dn.dropped <= 0) continue; + const failed = dn.failed > 0; + const sink = this.node(`x:${dn.id}`, "sink", dn.fleet, dn.name); + sink.pipelines = [...dn.pipelines]; + sink.rate = failed ? dn.failed : dn.dropped; + const signal = [...this.edges.values()].find((e) => e.target === dn.id)?.signal ?? "logs"; + this.edges.set(`br:${dn.id}`, { + id: `br:${dn.id}`, + source: dn.id, target: sink.id, signal, rate: sink.rate, - pipelines: [...en.pipelines], + pipelines: [...dn.pipelines], branch: { kind: failed ? "failed" : "dropped", - ratio: ratio(failed ? en.failed : en.dropped, en.rate), + ratio: failed ? dn.failedRatio : dn.droppedRatio, }, }); } @@ -188,8 +188,9 @@ class MapBuilder { } function place(nodes: MapNode[], layout: MapLayout) { + const room = nodes.some((n) => n.column === "sink") ? BRANCH_ROOM : 0; const gap = Math.max( - (layout.width - 2 * PAD - layout.nodeWidth) / (COLUMNS.length - 1), + (layout.width - 2 * PAD - layout.nodeWidth - room) / (COLUMNS.length - 1), MIN_COLUMN_GAP, ); const columns = COLUMNS.map((c) => nodes.filter((n) => n.column === c)); @@ -206,19 +207,19 @@ function place(nodes: MapNode[], layout: MapLayout) { for (const sink of nodes.filter((n) => n.column === "sink")) { const from = byId.get(sink.id.slice(2)); if (!from) continue; - sink.x = from.x + layout.nodeWidth / 2 + gap / 2; + sink.x = from.x + layout.nodeWidth + BRANCH_REACH; sink.y = from.y + layout.row * 0.7; } } -/** Fleet → receiver → exporter → destination; nodes keep the order of the pipelines they first appear in. */ +/** Fleet → receiver → destination; nodes keep the order of the pipelines they first appear in. */ export function buildPipelineMap(pipelines: FleetPipeline[], layout: MapLayout): PipelineMap { const b = new MapBuilder(); for (const p of pipelines) { const fleetNode = b.node(`f:${p.fleet}`, "fleet", p.fleet, p.fleet); b.touch(fleetNode, p); fleetNode.rate = addRate(fleetNode.rate, p.out_rate); - b.addExporters(p); + b.addDestinations(p); } for (const p of pipelines) b.addReceivers(p, b.node(`f:${p.fleet}`, "fleet", p.fleet, p.fleet)); b.addBranches();