From 9f373b7b4e538ee89e6e5e0383ec5a67df4e33a9 Mon Sep 17 00:00:00 2001 From: Oreofe Date: Sun, 30 Aug 2026 15:55:07 +0100 Subject: [PATCH] feat(metrics): record stream, queue and per-shard counters (#44) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Several metric families were registered and exported but never written, so they scraped as 0 forever. That is worse than being absent: a dashboard built on flo_stream_append_records_total{topic="orders"} reads "no traffic" on a stream actively taking writes. Wired the write paths to the record* methods that already existed: - stream append — recordAppend with the batch's logical record count, so a three-record batch counts three rather than one entry - stream read — recordRead, likewise summing batch contents, and recordEmptyRead when a read returns nothing - queue enqueue/dequeue — recordEnqueue, recordDequeue, recordEmptyDequeue - per-shard — shardMetrics() had no callers at all, so every flo_shard_* row exported 0 while its global flo_* equivalent moved. The shard now resolves its metrics once at wire-up and records alongside the global counters. registerStream and registerQueue already return the per-entity metrics and the handlers already called them, discarding the result — so most of this is capturing a pointer that was being thrown away. ## Tests Four e2e tests asserting values, not presence: append three records and assert the counter says 3, read them back and assert the read counter says 3, enqueue two and dequeue and assert both, and assert per-shard commands are non-zero. A parser pulls the numeric value out of the exposition line rather than substring-matching, so a wrong number fails instead of passing on the metric name appearing. All four fail with the instrumentation stashed — verified rather than assumed, since a metrics test that passes uninstrumented is exactly what let this hide. ## Not in this change - processing and workflow: neither handler has a metrics_registry field and the shard does not wire one, so those need plumbing first rather than a call at an existing site. Both keep separate module-local metrics that are unrelated to the registry families. - KVMetrics and TieredLogMetrics: exportPrometheus does not emit them and registerTieredLog has no callers. Whether to export or delete them is a product decision, not a mechanical fix. The note in docs/deployment/clustering.mdx covers all of these families, so it stays until the rest land. --- src/node/shard.zig | 15 +++++ src/queue/handler.zig | 16 ++++- src/stream/handler.zig | 19 +++++- tests/e2e/metrics_values_test.zig | 107 ++++++++++++++++++++++++++++++ tests/e2e/mod.zig | 1 + 5 files changed, 154 insertions(+), 4 deletions(-) create mode 100644 tests/e2e/metrics_values_test.zig diff --git a/src/node/shard.zig b/src/node/shard.zig index d480da5..0de2c83 100644 --- a/src/node/shard.zig +++ b/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); } diff --git a/src/queue/handler.zig b/src/queue/handler.zig index 5c3d314..9e0f622 100644 --- a/src/queue/handler.zig +++ b/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 } }; } diff --git a/src/stream/handler.zig b/src/stream/handler.zig index d7e14d0..2bbcfda 100644 --- a/src/stream/handler.zig +++ b/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, diff --git a/tests/e2e/metrics_values_test.zig b/tests/e2e/metrics_values_test.zig new file mode 100644 index 0000000..60ec8b4 --- /dev/null +++ b/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); +} diff --git a/tests/e2e/mod.zig b/tests/e2e/mod.zig index 5c61aa3..bc37f1d 100644 --- a/tests/e2e/mod.zig +++ b/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");