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