feat: add component latency metrics end-to-end (#91)

* feat(agent): scrape component_latency_mean_seconds from Vector 0.54.0

* feat(db): add latencyMeanMs column to PipelineMetric

* feat: accept and store component latency in heartbeat pipeline

* feat(api): expose latency data in metrics and dashboard endpoints

* feat(ui): add formatLatency helper and Component Latency chart to dashboard

* feat(ui): add Component Latency chart to pipeline metrics page

* feat(ui): add latency to flow editor overlay and show-metrics panel

* feat(ui): add Avg Latency column to fleet node pipeline table

* feat: add latency_mean SLI metric type for pipeline health evaluation

* fix: use proper mean for latency aggregation and handle zero formatting

- Fix running-average bug in getNodePipelineRates that gave incorrect
  latency means when aggregating across >2 components
- Add early return for formatLatency(0) to display "0ms" instead of "0.000ms"

Author: TerrifiedBug

agent/internal/agent/heartbeat.go

@@ -42,14 +42,15 @@ func buildHeartbeat(sup *supervisor.Supervisor, vectorVersion string, deployment
 			// Map per-component metrics for editor node overlays
 			for _, cm := range sr.Components {
 				ps.ComponentMetrics = append(ps.ComponentMetrics, client.ComponentMetric{
-					ComponentID:     cm.ComponentID,
-					ComponentKind:   cm.ComponentKind,
-					ReceivedEvents:  cm.ReceivedEvents,
-					SentEvents:      cm.SentEvents,
-					ReceivedBytes:   cm.ReceivedBytes,
-					SentBytes:       cm.SentBytes,
-					ErrorsTotal:     cm.ErrorsTotal,
-					DiscardedEvents: cm.DiscardedEvents,
+					ComponentID:        cm.ComponentID,
+					ComponentKind:      cm.ComponentKind,
+					ReceivedEvents:     cm.ReceivedEvents,
+					SentEvents:         cm.SentEvents,
+					ReceivedBytes:      cm.ReceivedBytes,
+					SentBytes:          cm.SentBytes,
+					ErrorsTotal:        cm.ErrorsTotal,
+					DiscardedEvents:    cm.DiscardedEvents,
+					LatencyMeanSeconds: cm.LatencyMeanSeconds,
 				})
 			}
 

agent/internal/client/client.go

@@ -178,7 +178,8 @@ type ComponentMetric struct {
 	ReceivedBytes   int64  `json:"receivedBytes,omitempty"`
 	SentBytes       int64  `json:"sentBytes,omitempty"`
 	ErrorsTotal     int64  `json:"errorsTotal,omitempty"`
-	DiscardedEvents int64  `json:"discardedEvents,omitempty"`
+	DiscardedEvents    int64   `json:"discardedEvents,omitempty"`
+	LatencyMeanSeconds float64 `json:"latencyMeanSeconds,omitempty"`
 }
 
 // HostMetrics holds system-level metrics from the Vector host

agent/internal/metrics/scraper.go

@@ -28,7 +28,8 @@ type ComponentMetrics struct {
 	ReceivedBytes   int64
 	SentBytes       int64
 	ErrorsTotal     int64
-	DiscardedEvents int64
+	DiscardedEvents    int64
+	LatencyMeanSeconds float64 // mean event time in component (seconds)
 }
 
 // HostMetrics holds system-level metrics from Vector's host_metrics source.
@@ -137,6 +138,11 @@ func ScrapePrometheus(metricsPort int) ScrapeResult {
 			}
 			getOrCreate(componentMap, componentID, componentKind).DiscardedEvents += v
 
+		case "vector_component_latency_mean_seconds", "component_latency_mean_seconds":
+			if !isInternal {
+				getOrCreate(componentMap, componentID, componentKind).LatencyMeanSeconds = value
+			}
+
 		// Host metrics – use += to aggregate across CPU cores, devices, interfaces, etc.
 		case "host_memory_total_bytes":
 			sr.Host.MemoryTotalBytes += int64(value)

prisma/migrations/20260311000000_add_pipeline_latency_mean/migration.sql

@@ -0,0 +1,2 @@
+-- AlterTable: Add latencyMeanMs column to PipelineMetric
+ALTER TABLE "PipelineMetric" ADD COLUMN "latencyMeanMs" DOUBLE PRECISION;

prisma/schema.prisma

@@ -321,6 +321,7 @@ model PipelineMetric {
   bytesIn         BigInt   @default(0)
   bytesOut        BigInt   @default(0)
   utilization     Float    @default(0)
+  latencyMeanMs   Float?                    // pipeline-level weighted mean latency (ms)
 
   @@index([pipelineId, timestamp])
   @@index([timestamp])
@@ -330,7 +331,7 @@ model PipelineSli {
   id            String   @id @default(cuid())
   pipelineId    String
   pipeline      Pipeline @relation(fields: [pipelineId], references: [id], onDelete: Cascade)
-  metric        String   // "error_rate" | "throughput_floor" | "discard_rate"
+  metric        String   // "error_rate" | "throughput_floor" | "discard_rate" | "latency_mean"
   condition     String   // "lt" | "gt"
   threshold     Float
   windowMinutes Int      @default(5)

src/app/(dashboard)/fleet/[nodeId]/page.tsx

@@ -30,6 +30,7 @@ import {
   formatCount,
   formatBytes,
   formatBytesRate,
+  formatLatency,
 } from "@/lib/format";
 import { nodeStatusVariant, nodeStatusLabel, pipelineStatusVariant, pipelineStatusLabel } from "@/lib/status";
 
@@ -521,6 +522,7 @@ export default function NodeDetailPage() {
                   <TableHead className="text-right">Errors</TableHead>
                   <TableHead className="text-right">Bytes In</TableHead>
                   <TableHead className="text-right">Bytes Out</TableHead>
+                  <TableHead className="text-right">Avg Latency</TableHead>
                   <TableHead className="text-right">Uptime</TableHead>
                 </TableRow>
               </TableHeader>
@@ -557,6 +559,11 @@ export default function NodeDetailPage() {
                       <div>{formatBytes(ps.bytesOut)}</div>
                       {rates && <div className="text-xs text-muted-foreground">{formatBytesRate(rates.bytesOutRate)}</div>}
                     </TableCell>
+                    <TableCell className="text-right font-mono text-sm">
+                      {rates?.latencyMeanMs != null
+                        ? formatLatency(rates.latencyMeanMs)
+                        : "—"}
+                    </TableCell>
                     <TableCell className="text-right font-mono text-sm">
                       {formatUptime(ps.uptimeSeconds)}
                     </TableCell>

src/app/(dashboard)/page.tsx

@@ -15,6 +15,7 @@ import {
   Plus,
   Pencil,
   Trash2,
+  Timer,
 } from "lucide-react";
 import { Card, CardContent } from "@/components/ui/card";
 import { Skeleton } from "@/components/ui/skeleton";
@@ -29,7 +30,7 @@ import { MetricsSection } from "@/components/dashboard/metrics-section";
 import { MetricChart } from "@/components/dashboard/metric-chart";
 import { ViewBuilderDialog } from "@/components/dashboard/view-builder-dialog";
 import { CustomView } from "@/components/dashboard/custom-view";
-import { formatSI, formatBytesRate, formatEventsRate } from "@/lib/format";
+import { formatSI, formatBytesRate, formatEventsRate, formatLatency } from "@/lib/format";
 import { PageHeader } from "@/components/page-header";
 import { cn } from "@/lib/utils";
 
@@ -380,6 +381,15 @@ export default function DashboardPage() {
               timeRange={timeRange}
               height={200}
             />
+            <MetricChart
+              title="Component Latency"
+              icon={<Timer className="h-4 w-4" />}
+              data={chartData.data?.pipeline.latency ?? {}}
+              variant="area"
+              yFormatter={formatLatency}
+              timeRange={timeRange}
+              height={200}
+            />
           </MetricsSection>
 
           {/* System Metrics */}

src/app/(dashboard)/pipelines/[id]/metrics/page.tsx

@@ -115,6 +115,15 @@ export default function PipelineMetricsPage() {
               <MetricsChart rows={rows} dataKey="errors" height={220} />
             </CardContent>
           </Card>
+
+          <Card>
+            <CardHeader>
+              <CardTitle>Component Latency</CardTitle>
+            </CardHeader>
+            <CardContent>
+              <MetricsChart rows={rows} dataKey="latency" height={220} />
+            </CardContent>
+          </Card>
         </>
       )}
 

src/app/(dashboard)/pipelines/[id]/page.tsx

@@ -216,6 +216,7 @@ function PipelineBuilderInner({ pipelineId }: { pipelineId: string }) {
         ...(entry.kind === "TRANSFORM" ? { eventsInPerSec: latest.receivedEventsRate } : {}),
         status: eventsPerSec > 0 ? "healthy" : "degraded",
         samples: entry.samples,
+        latencyMs: latest.latencyMeanMs,
       });
     }
 

src/app/api/agent/heartbeat/route.ts

@@ -14,6 +14,23 @@ import { deliverToChannels } from "@/server/services/channels";
 import { DeploymentMode } from "@/generated/prisma";
 import { isVersionOlder } from "@/lib/version";
 
+/** Compute pipeline-level weighted mean latency (ms) from per-component metrics. */
+function computeWeightedLatency(
+  components?: Array<{ receivedEvents: number; sentEvents: number; latencyMeanSeconds?: number }>,
+): number | null {
+  if (!components || components.length === 0) return null;
+  let weightedSum = 0;
+  let totalEvents = 0;
+  for (const cm of components) {
+    if (cm.latencyMeanSeconds == null || cm.latencyMeanSeconds === 0) continue;
+    const events = cm.receivedEvents + cm.sentEvents;
+    weightedSum += cm.latencyMeanSeconds * 1000 * events; // convert seconds → ms
+    totalEvents += events;
+  }
+  if (totalEvents === 0) return null;
+  return weightedSum / totalEvents;
+}
+
 const heartbeatSchema = z.object({
   agentVersion: z.string().max(100).optional(),
   vectorVersion: z.string().max(100).optional(),
@@ -39,6 +56,7 @@ const heartbeatSchema = z.object({
       sentBytes: z.number().optional(),
       errorsTotal: z.number().optional(),
       discardedEvents: z.number().optional(),
+      latencyMeanSeconds: z.number().optional(), // NEW
     })).optional(),
     utilization: z.number().optional(),
     recentLogs: z.array(z.string()).optional(),
@@ -94,6 +112,7 @@ interface PipelineStatus {
     sentBytes?: number;
     errorsTotal?: number;
     discardedEvents?: number;
+    latencyMeanSeconds?: number; // NEW
   }>;
   utilization?: number;
   recentLogs?: string[];
@@ -319,6 +338,7 @@ export async function POST(request: Request) {
         bytesIn: BigInt(p.bytesIn ?? 0),
         bytesOut: BigInt(p.bytesOut ?? 0),
         utilization: p.utilization ?? 0,
+        latencyMeanMs: computeWeightedLatency(p.componentMetrics),
       }));
 
     if (metricsData.length > 0) {
@@ -336,6 +356,9 @@ export async function POST(request: Request) {
             sentEventsTotal: cm.sentEvents,
             receivedBytesTotal: cm.receivedBytes ?? 0,
             sentBytesTotal: cm.sentBytes ?? 0,
+            errorsTotal: cm.errorsTotal ?? 0,
+            discardedTotal: cm.discardedEvents ?? 0,
+            latencyMeanSeconds: cm.latencyMeanSeconds,
           });
         }
       }