fix(fleet): persist the rollout strategy, three-column pipeline map, enterprise build fix (#16365)

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,
     }
 }
 

src/infra/src/table/fleet_rollouts.rs

@@ -184,9 +184,10 @@ pub async fn guard_with<C: ConnectionTrait>(
     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<C: ConnectionTrait>(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 =

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);
   });
 });
 

web/src/components/fleet/FleetFlowOverview.vue

@@ -346,9 +346,9 @@ const DESTINATION_ICON: Record<string, IconName> = {
   awss3: "inventory-2",
 };
 
-const COLUMN_ICON: Record<string, IconName> = { f: "dns", r: "input", e: "send" };
+const COLUMN_ICON: Record<string, IconName> = { 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),
   };
 }

web/src/components/fleet/pipelineGraph.spec.ts

@@ -65,57 +65,112 @@ const traces = pipeline({
 const byId = <T extends { id: string }>(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();
   });
 });
 

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();