From ebd57d8ef51b5c9b4c74de671b609e655dd923ac Mon Sep 17 00:00:00 2001 From: Oreofe Date: Sun, 30 Aug 2026 17:44:15 +0100 Subject: [PATCH] feat(metrics): record workflow and processing lifecycle counters (#44) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Completes the instrumentation started in #64. Both families exported 0 regardless of traffic because neither handler ever touched the registry. Contrary to the note in #64, no plumbing was needed: every dispatch function in both handlers already receives `shard: *Shard`, which carries the registry. The counters go in at the points every transition already passes through: - workflow: `completeRun` is the single terminal transition, so completed / failed / cancelled / timed_out all record there; started in `handleStart`, signal delivery in `handleSignal`, and step execution at the four `step_completed` history sites (all four have `shard` in scope). - processing: submitted in `handleSubmit`, and `persistStatusChange` covers cancelled and stopped. ## A crash this surfaced Eight workflow unit tests began aborting. Their test shard is built as `var shard: Shard = undefined` with fields assigned one at a time, so `metrics_registry` held garbage and the new read dereferenced it. The tests passed before only because nothing in the workflow path read that field. Both fields are now assigned in the helper. This is the third instance of the same hazard today — `allocator.create` and `= undefined` both leave field defaults inapplicable, and the compiler cannot see it. ## Tests Two more value-asserting e2e tests: start two workflows and assert `flo_workflow_started_total` is 2; submit a job and assert `flo_processing_jobs_submitted_total` is 1. Both fail with the instrumentation stashed. test-unit, test-integration, and the metrics / workflow / processing e2e filters all pass. ## Still open on #44 KVMetrics and TieredLogMetrics remain unexported and unregistered. Whether to emit or delete them is a product decision, so #44 stays open for that alone. --- src/processing/handler.zig | 9 ++++ src/workflow/handler.zig | 21 +++++++++ tests/e2e/metrics_values_test.zig | 71 +++++++++++++++++++++++++++++++ 3 files changed, 101 insertions(+) diff --git a/src/processing/handler.zig b/src/processing/handler.zig index 6430129..958fed3 100644 --- a/src/processing/handler.zig +++ b/src/processing/handler.zig @@ -541,6 +541,8 @@ pub const ProcessingHandler = struct { // replay path so a RUNNING job resumes after restart (FLO-104). self.startPipelines(owned_id, &def); + if (shard.metrics_registry) |m| m.processing.recordSubmitted(); + // Return the job ID shard.sendOkResponse(conn, req.header.request_id, job_id); } @@ -859,6 +861,13 @@ pub const ProcessingHandler = struct { }; const value = &[_]u8{@intFromEnum(status)}; _ = persistence_mod.persistEntry(shard, entry_type, Flags.NONE, namespace, job_id, value) catch {}; + + // Both terminal transitions route through here. + if (shard.metrics_registry) |m| switch (status) { + .cancelled => m.processing.recordCancelled(), + .stopped => m.processing.recordCompleted(), + else => {}, + }; } /// Persist a processing_savepoint entry. Key = savepoint_id. diff --git a/src/workflow/handler.zig b/src/workflow/handler.zig index 94e1faa..0817235 100644 --- a/src/workflow/handler.zig +++ b/src/workflow/handler.zig @@ -795,6 +795,8 @@ pub const WorkflowHandler = struct { .history = .empty, }; + if (shard.metrics_registry) |m| m.workflow.recordStarted(); + // Add initial history event const evt_type = self.allocator.dupe(u8, "workflow_started") catch { self.freeRunRecord(&run); @@ -907,6 +909,7 @@ pub const WorkflowHandler = struct { // Add history event self.addHistoryEvent(run, "signal_received", signal_type, now_ms); + if (shard.metrics_registry) |m| m.workflow.recordSignalDelivered(); // If the run is waiting for this signal type, resume execution if (run.status == .waiting) { @@ -1565,6 +1568,7 @@ pub const WorkflowHandler = struct { } self.addHistoryEvent(run, "step_completed", step_label, now_ms); + if (shard.metrics_registry) |m| m.workflow.recordStepExecuted(); // Reset retry + poll counters on step transition run.retry_count = 0; @@ -1604,6 +1608,7 @@ pub const WorkflowHandler = struct { if (signal_found) { // Signal already received — follow "success" transition self.addHistoryEvent(run, "step_completed", step_label, now_ms); + if (shard.metrics_registry) |m| m.workflow.recordStepExecuted(); const transition = wait_step.getTransition(definition.StepOutcome.success) orelse { self.completeRun(shard, run_ns_key, run, .failed, "no success transition for wait step", now_ms); return; @@ -2372,6 +2377,7 @@ pub const WorkflowHandler = struct { run.status = .running; self.addHistoryEvent(run, "child_completed", outcome, now_ms); self.addHistoryEvent(run, "step_completed", step_label, now_ms); + if (shard.metrics_registry) |m| m.workflow.recordStepExecuted(); // Clear pending child state. if (run.pending_child_run_id_owned) |a| self.allocator.free(a); @@ -2444,6 +2450,7 @@ pub const WorkflowHandler = struct { run.status = .running; self.addHistoryEvent(run, "action_completed", outcome, now_ms); self.addHistoryEvent(run, "step_completed", step_label, now_ms); + if (shard.metrics_registry) |m| m.workflow.recordStepExecuted(); // Clear pending action state if (run.pending_action_run_id_owned) |a| self.allocator.free(a); @@ -2663,6 +2670,16 @@ pub const WorkflowHandler = struct { run.status = status; run.completed_at_ms = now_ms; + // Every terminal transition passes through here, so this is the one + // place the outcome counters need to be recorded. + if (shard.metrics_registry) |m| switch (status) { + .completed => m.workflow.recordCompleted(), + .failed => m.workflow.recordFailed(), + .cancelled => m.workflow.recordCancelled(), + .timed_out => m.workflow.recordTimedOut(), + else => {}, + }; + // Resolve explicit output mapping from definition (if declared) if (status == .completed) { self.resolveWorkflowOutput(run, run_ns_key); @@ -4086,6 +4103,10 @@ fn createTestShard(actions: *ActionsHandler) !Shard { shard.actions_handler = actions; shard.peer_shards = null; shard.peer_inboxes = null; + // `Shard = undefined` means field defaults do not apply — anything the code + // under test reads must be assigned here or it holds garbage. + shard.metrics_registry = null; + shard.shard_metrics = null; const raft_node = try std.testing.allocator.create(RaftNode); raft_node.* = try RaftNode.init(std.testing.allocator, 1, 0, 4096, .{}); try raft_node.bootstrap(); diff --git a/tests/e2e/metrics_values_test.zig b/tests/e2e/metrics_values_test.zig index 60ec8b4..a94b2d5 100644 --- a/tests/e2e/metrics_values_test.zig +++ b/tests/e2e/metrics_values_test.zig @@ -23,6 +23,17 @@ fn seriesValue(body: []const u8, name: []const u8, needle: []const u8) ?u64 { return null; } +/// Value of an unlabeled `name ` series. +fn scalarValue(body: []const u8, name: []const u8) ?u64 { + var lines = std.mem.splitScalar(u8, body, '\n'); + while (lines.next()) |line| { + if (!std.mem.startsWith(u8, line, name)) continue; + if (line.len <= name.len or line[name.len] != ' ') continue; + return std.fmt.parseInt(u64, std.mem.trim(u8, line[name.len + 1 ..], " \r"), 10) catch continue; + } + return null; +} + test "e2e/metrics: stream append counters carry real values" { var ctx = try stdx.testing.TestContext.initWithConfig(testing.allocator, .{ .server = .{ .metrics_enabled = true }, @@ -105,3 +116,63 @@ test "e2e/metrics: per-shard counters are attributed, not left at zero" { const cmds = seriesValue(resp.body, "flo_shard_commands_total", "shard") orelse 0; try testing.expect(cmds > 0); } + +test "e2e/metrics: workflow lifecycle counters carry real values" { + var ctx = try stdx.testing.TestContext.initWithConfig(testing.allocator, .{ + .server = .{ .metrics_enabled = true }, + }); + defer ctx.deinit(); + + try ctx.exec(&.{ "action", "register", "echo" }); + + const workflow_def = + \\kind: Workflow + \\name: metrics-wf + \\version: 1.0.0 + \\start.run: @actions/echo + \\start.transition.success: flo.Completed + \\start.transition.failure: flo.Failed + ; + const path = try stdx.testing.writeDottedToTempYaml(testing.allocator, workflow_def, "metrics-wf.yaml"); + defer stdx.testing.cleanupTempFile(testing.allocator, path); + + try ctx.exec(&.{ "workflow", "create", "-f", path }); + try ctx.exec(&.{ "workflow", "start", "metrics-wf", "{\"a\":1}" }); + try ctx.exec(&.{ "workflow", "start", "metrics-wf", "{\"a\":2}" }); + + var http = try ctx.createMetricsHttp(); + defer http.deinit(); + var resp = try http.get("/metrics"); + defer resp.deinit(); + + // Two runs started — the family exported 0 regardless of traffic before, + // because the workflow handler never touched the registry. + try testing.expectEqual(@as(?u64, 2), scalarValue(resp.body, "flo_workflow_started_total")); +} + +test "e2e/metrics: processing submitted counter carries a real value" { + var ctx = try stdx.testing.TestContext.initWithConfig(testing.allocator, .{ + .server = .{ .metrics_enabled = true }, + }); + defer ctx.deinit(); + + try ctx.exec(&.{ "stream", "append", "proc_src", "seed" }); + + const job = + \\kind: Processing + \\name: metrics-proc + \\sources.[0].stream.name: proc_src + \\sinks.[0].stream.name: proc_dst + ; + const path = try stdx.testing.writeDottedToTempYaml(testing.allocator, job, "metrics-proc.yaml"); + defer stdx.testing.cleanupTempFile(testing.allocator, path); + + try ctx.exec(&.{ "processing", "submit", path }); + + var http = try ctx.createMetricsHttp(); + defer http.deinit(); + var resp = try http.get("/metrics"); + defer resp.deinit(); + + try testing.expectEqual(@as(?u64, 1), scalarValue(resp.body, "flo_processing_jobs_submitted_total")); +}