feat(metrics): record stream, queue and per-shard counters (#44)

src/node/shard.zig

@@ -93,6 +93,7 @@ const Coordinator = @import("../cluster/coordinator.zig").Coordinator;
 const NodeId = @import("../raft/node.zig").NodeId;
 pub const run_id_mod = @import("run_id.zig");
 const MetricsRegistry = @import("../metrics/registry.zig").MetricsRegistry;
+const ShardMetrics = @import("../metrics/registry.zig").ShardMetrics;
 
 /// Maximum single-request size we handle on the stack.
 const MAX_REQUEST_SIZE = 256 * 1024; // 256 KB
@@ -194,6 +195,9 @@ pub const Shard = struct {
     /// Raft network reference (set by runtime for shard 0, null otherwise).
     raft_network: ?*RaftNetwork,
 
+    /// Per-shard counters, mirroring the global server ones for attribution.
+    shard_metrics: ?*ShardMetrics,
+
     /// This node's cluster-wide node ID, set by the runtime at start. Distinct
     /// from `raft_node.id`, which identifies the per-shard Raft group.
     cluster_node_id: u32,
@@ -520,6 +524,7 @@ pub const Shard = struct {
             .shard_data_dir = shard_data_dir,
             .pipe_registered = false,
             .raft_network = null,
+            .shard_metrics = null,
             .cluster_node_id = 0,
             .waiter_pool = WaiterPool.init(),
             .task_scheduler = TaskScheduler.init(),
@@ -602,6 +607,11 @@ pub const Shard = struct {
     /// Called by runtime after the registry is created.
     pub fn setMetricsRegistry(self: *Shard, registry: *MetricsRegistry) void {
         self.metrics_registry = registry;
+        // Resolved once: the shard table is sized by `initShards` before this
+        // runs, and the pointer is stable for the registry's lifetime. Without
+        // it every `flo_shard_*` series exports 0 while the global `flo_*`
+        // equivalents move, which reads as "this shard is idle".
+        self.shard_metrics = registry.shardMetrics(self.id);
         self.stream_handler.metrics_registry = registry;
         self.queue_handler.metrics_registry = registry;
         self.kv_handler.metrics_registry = registry;
@@ -746,6 +756,7 @@ pub const Shard = struct {
 
         try self.connections.put(self.allocator, fd, conn);
         if (self.metrics_registry) |m| m.server.connectionOpened();
+        if (self.shard_metrics) |sm| sm.connectionOpened();
         return conn;
     }
 
@@ -757,6 +768,7 @@ pub const Shard = struct {
             kv.value.deinit();
             self.allocator.destroy(kv.value);
             if (self.metrics_registry) |m| m.server.connectionClosed();
+            if (self.shard_metrics) |sm| sm.connectionClosed();
         }
     }
 
@@ -787,6 +799,7 @@ pub const Shard = struct {
         // Overview's commands_total + rps). Other server counters (connections,
         // bytes) are not yet wired — see the metrics gap log.
         if (self.metrics_registry) |m| m.server.recordCommand();
+        if (self.shard_metrics) |sm| sm.recordCommand();
 
         const op = req.header.op_code;
 
@@ -1512,6 +1525,7 @@ pub const Shard = struct {
         }
 
         if (self.metrics_registry) |m| m.server.recordBytesReceived(@intCast(n));
+        if (self.shard_metrics) |sm| sm.recordBytesReceived(@intCast(n));
 
         // Accumulate data in the read buffer
         _ = conn.read_buf.write(tmp_buf[0..n]);
@@ -1766,6 +1780,7 @@ pub const Shard = struct {
                 return;
             }
             if (self.metrics_registry) |m| m.server.recordBytesSent(@intCast(written));
+            if (self.shard_metrics) |sm| sm.recordBytesSent(@intCast(written));
             conn.consumeWritten(written);
         }
 

src/queue/handler.zig

@@ -317,9 +317,13 @@ pub const QueueHandler = struct {
             return .{ .err = .{ .code = .internal_error, .message = "UAL append failed" } };
         };
 
-        // Register in global metrics registry for dashboard/Prometheus
+        // Register in global metrics registry for dashboard/Prometheus, and
+        // record the enqueue against the per-queue metrics `registerQueue`
+        // returns.
         if (self.metrics_registry) |mr| {
-            _ = mr.registerQueue(req.namespace, req.key) catch {};
+            if (mr.registerQueue(req.namespace, req.key)) |qm| {
+                qm.recordEnqueue(1, req.value.len, false);
+            } else |_| {}
         }
 
         // Seq was assigned by the projection during apply
@@ -369,10 +373,18 @@ pub const QueueHandler = struct {
 
         // Persist auto-ack entries so dequeued messages don't reappear after restart.
         // In this simplified model, dequeue = consume (not a lease-based model).
+        var dequeued_bytes: usize = 0;
         for (results[0..actual]) |r| {
+            dequeued_bytes += r.payload.len;
             self.persistAck(ns_hash, r.seq);
         }
 
+        if (self.metrics_registry) |mr| {
+            if (mr.registerQueue(req.namespace, req.key)) |qm| {
+                if (actual == 0) qm.recordEmptyDequeue() else qm.recordDequeue(actual, dequeued_bytes);
+            } else |_| {}
+        }
+
         return .{ .queue_messages = .{ .data = data } };
     }
 

src/stream/handler.zig

@@ -433,9 +433,14 @@ pub const StreamHandler = struct {
         const ns_stream_name = ns_keys.qualifyKey(&ns_reg_buf, req.namespace, req.key) catch req.key;
         self.stream.registerStream(ns_stream_name) catch {};
 
-        // Register in global metrics registry for dashboard/Prometheus
+        // Register in global metrics registry for dashboard/Prometheus, and
+        // record the append against it. `registerStream` returns the per-stream
+        // metrics, so the counters cost one call next to the registration they
+        // already do.
         if (self.metrics_registry) |mr| {
-            _ = mr.registerStream(req.namespace, req.key, 0) catch {};
+            if (mr.registerStream(req.namespace, req.key, 0)) |sm| {
+                sm.recordAppend(stream_mod.batchRecordCount(payload_value), payload_value.len);
+            } else |_| {}
         }
 
         return .{ .stream_append_ok = .{
@@ -507,6 +512,16 @@ pub const StreamHandler = struct {
             return .{ .err = .{ .code = .internal_error, .message = "read serialization failed" } };
         };
 
+        if (self.metrics_registry) |mr| {
+            if (mr.registerStream(req.namespace, req.key, 0)) |sm| {
+                // `count` is append entries; sum their batch contents so the
+                // counter matches what the caller actually receives.
+                var records: u64 = 0;
+                for (buf[0..count]) |rec| records += rec.record_count;
+                if (records == 0) sm.recordEmptyRead() else sm.recordRead(records, data.len);
+            } else |_| {}
+        }
+
         const last_id = if (count > 0) buf[count - 1].id else StreamID.MIN;
         return .{ .stream_messages = .{
             .data = data,

tests/e2e/metrics_values_test.zig

@@ -0,0 +1,107 @@
+//! Metrics Value E2E Tests
+//!
+//! These assert the *numbers*, not that a family appears. A metric that is
+//! registered but never written scrapes as 0 forever, which reads as an idle
+//! node rather than an uninstrumented one — and a test that only checks the
+//! name is present passes in exactly that state.
+
+const std = @import("std");
+const testing = std.testing;
+const stdx = @import("stdx");
+
+/// Value of `metric{...labels...}` from a Prometheus exposition body, matching
+/// the first series whose line starts with `name{` and contains `needle`.
+fn seriesValue(body: []const u8, name: []const u8, needle: []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;
+        if (std.mem.indexOf(u8, line, needle) == null) continue;
+        const sp = std.mem.lastIndexOfScalar(u8, line, ' ') orelse continue;
+        return std.fmt.parseInt(u64, std.mem.trim(u8, line[sp + 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 },
+    });
+    defer ctx.deinit();
+
+    // One batch of three records.
+    try ctx.exec(&.{ "stream", "append", "orders", "a", "b", "c" });
+
+    var http = try ctx.createMetricsHttp();
+    defer http.deinit();
+    var resp = try http.get("/metrics");
+    defer resp.deinit();
+    try testing.expectEqual(@as(u16, 200), resp.status);
+
+    // Three records in one append op — not "the family exists".
+    try testing.expectEqual(@as(?u64, 3), seriesValue(resp.body, "flo_stream_append_records_total", "orders"));
+    try testing.expectEqual(@as(?u64, 1), seriesValue(resp.body, "flo_stream_append_ops_total", "orders"));
+
+    const bytes = seriesValue(resp.body, "flo_stream_append_bytes_total", "orders") orelse 0;
+    try testing.expect(bytes > 0);
+}
+
+test "e2e/metrics: stream read counters carry real values" {
+    var ctx = try stdx.testing.TestContext.initWithConfig(testing.allocator, .{
+        .server = .{ .metrics_enabled = true },
+    });
+    defer ctx.deinit();
+
+    try ctx.exec(&.{ "stream", "append", "reads", "a", "b", "c" });
+    try ctx.exec(&.{ "stream", "read", "reads" });
+
+    var http = try ctx.createMetricsHttp();
+    defer http.deinit();
+    var resp = try http.get("/metrics");
+    defer resp.deinit();
+
+    // The read returned three records, so the counter must say three — a batch
+    // counted as one entry would report 1 here.
+    try testing.expectEqual(@as(?u64, 3), seriesValue(resp.body, "flo_stream_read_records_total", "reads"));
+    try testing.expectEqual(@as(?u64, 1), seriesValue(resp.body, "flo_stream_read_ops_total", "reads"));
+}
+
+test "e2e/metrics: queue enqueue and dequeue counters carry real values" {
+    var ctx = try stdx.testing.TestContext.initWithConfig(testing.allocator, .{
+        .server = .{ .metrics_enabled = true },
+    });
+    defer ctx.deinit();
+
+    try ctx.exec(&.{ "queue", "enqueue", "jobs", "one" });
+    try ctx.exec(&.{ "queue", "enqueue", "jobs", "two" });
+    try ctx.exec(&.{ "queue", "dequeue", "jobs" });
+
+    var http = try ctx.createMetricsHttp();
+    defer http.deinit();
+    var resp = try http.get("/metrics");
+    defer resp.deinit();
+
+    try testing.expectEqual(@as(?u64, 2), seriesValue(resp.body, "flo_queue_enqueue_messages_total", "jobs"));
+    const dequeued = seriesValue(resp.body, "flo_queue_dequeue_messages_total", "jobs") orelse 0;
+    try testing.expect(dequeued >= 1);
+}
+
+test "e2e/metrics: per-shard counters are attributed, not left at zero" {
+    var ctx = try stdx.testing.TestContext.initWithConfig(testing.allocator, .{
+        .server = .{ .metrics_enabled = true },
+    });
+    defer ctx.deinit();
+
+    try ctx.exec(&.{ "kv", "set", "k", "v" });
+    try ctx.exec(&.{ "kv", "get", "k" });
+
+    var http = try ctx.createMetricsHttp();
+    defer http.deinit();
+    var resp = try http.get("/metrics");
+    defer resp.deinit();
+
+    // shardMetrics() had no callers, so every shard row exported 0 while the
+    // global flo_commands_total moved.
+    const cmds = seriesValue(resp.body, "flo_shard_commands_total", "shard") orelse 0;
+    try testing.expect(cmds > 0);
+}

tests/e2e/mod.zig

@@ -21,6 +21,7 @@ pub const processing_test = @import("processing_test.zig");
 pub const ts_test = @import("ts_test.zig");
 pub const dual_connection_test = @import("dual_connection_test.zig");
 pub const metrics_test = @import("metrics_test.zig");
+pub const metrics_values_test = @import("metrics_values_test.zig");
 pub const dashboard_streams_test = @import("dashboard_streams_test.zig");
 pub const dashboard_queues_test = @import("dashboard_queues_test.zig");
 pub const dashboard_actions_test = @import("dashboard_actions_test.zig");