feat(metrics): export KV counters and populate the tiered-log family (#44)

src/kv/handler.zig

@@ -254,6 +254,11 @@ pub const KVHandler = struct {
         const cmd_result = shard.kv_handler.*.handleCommand(req);
         defer shard.kv_handler.*.freeResult(cmd_result);
         log.debug("KV GET: key={s}, hit={}", .{ req.key, cmd_result != .kv_not_found });
+
+        if (shard.kv_handler.metrics_registry) |mr| {
+            if (mr.registerKVNamespace(req.namespace)) |km| km.recordGet() else |_| {}
+        }
+
         sendKVResponse(shard, conn, req.header.request_id, cmd_result);
     }
 
@@ -318,9 +323,12 @@ pub const KVHandler = struct {
         // Track namespace data for non-empty delete check
         shard.namespace_handler.markNamespaceHasData(req.namespace, shard);
 
-        // Register KV namespace in global metrics registry for dashboard/Prometheus
+        // Register KV namespace in global metrics registry for dashboard/Prometheus,
+        // and record the write against the per-namespace metrics it returns.
         if (shard.kv_handler.metrics_registry) |mr| {
-            _ = mr.registerKVNamespace(req.namespace) catch {};
+            if (mr.registerKVNamespace(req.namespace)) |km| {
+                km.recordSet(req.value.len, version == 1);
+            } else |_| {}
         }
 
         sendKVResponse(shard, conn, req.header.request_id, cmd_result);
@@ -386,6 +394,14 @@ pub const KVHandler = struct {
         shard.waiter_pool.notify(.kv_get, qkey, @import("../node/shard.zig").resolveKVWaiter, @ptrCast(shard));
 
         log.debug("KV DELETE: key={s}", .{req.key});
+
+        if (shard.kv_handler.metrics_registry) |mr| {
+            if (mr.registerKVNamespace(req.namespace)) |km| {
+                km.recordDelete();
+                km.decrementKeyCount();
+            } else |_| {}
+        }
+
         sendKVResponse(shard, conn, req.header.request_id, .ok);
     }
 

src/metrics/registry.zig

@@ -47,7 +47,15 @@ pub const KVMetrics = struct {
     }
 
     pub fn decrementKeyCount(self: *KVMetrics) void {
-        _ = self.key_count.fetchSub(1, .monotonic);
+        // Saturate at zero. Recovery rebuilds keys without going through
+        // recordSet, so key_count can be 0 while keys exist on disk — an
+        // unguarded fetchSub would wrap to u64 max and export a nonsense gauge.
+        var cur = self.key_count.load(.monotonic);
+        while (cur > 0) {
+            if (self.key_count.cmpxchgWeak(cur, cur - 1, .monotonic, .monotonic)) |actual| {
+                cur = actual;
+            } else return;
+        }
     }
 
     pub const Snapshot = struct {
@@ -1017,6 +1025,19 @@ pub const MetricsRegistry = struct {
             try writeStreamMetrics(writer, snapshot, labels);
         }
 
+        // Export KV metrics, one series set per namespace.
+        var kv_iter = self.kv_namespaces.valueIterator();
+        while (kv_iter.next()) |entry| {
+            const kv_snapshot = entry.metrics.snapshot();
+            const kv_labels = try std.fmt.allocPrint(
+                allocator,
+                "namespace=\"{s}\"",
+                .{entry.namespace},
+            );
+            defer allocator.free(kv_labels);
+            try writeKVMetrics(writer, kv_snapshot, kv_labels);
+        }
+
         // Export tiered log metrics
         var tiered_iter = self.tiered_logs.valueIterator();
         while (tiered_iter.next()) |entry| {
@@ -1198,6 +1219,32 @@ fn writeQueueMetrics(
 }
 
 /// Write tiered log metrics in Prometheus format
+fn writeKVMetrics(
+    writer: anytype,
+    snapshot: KVMetrics.Snapshot,
+    labels: []const u8,
+) !void {
+    try writer.print("# HELP flo_kv_keys Current keys stored in this namespace\n", .{});
+    try writer.print("# TYPE flo_kv_keys gauge\n", .{});
+    try writer.print("flo_kv_keys{{{s}}} {d}\n", .{ labels, snapshot.key_count });
+
+    try writer.print("# HELP flo_kv_get_ops_total Total KV get operations\n", .{});
+    try writer.print("# TYPE flo_kv_get_ops_total counter\n", .{});
+    try writer.print("flo_kv_get_ops_total{{{s}}} {d}\n", .{ labels, snapshot.get_ops_total });
+
+    try writer.print("# HELP flo_kv_set_ops_total Total KV set operations\n", .{});
+    try writer.print("# TYPE flo_kv_set_ops_total counter\n", .{});
+    try writer.print("flo_kv_set_ops_total{{{s}}} {d}\n", .{ labels, snapshot.set_ops_total });
+
+    try writer.print("# HELP flo_kv_delete_ops_total Total KV delete operations\n", .{});
+    try writer.print("# TYPE flo_kv_delete_ops_total counter\n", .{});
+    try writer.print("flo_kv_delete_ops_total{{{s}}} {d}\n", .{ labels, snapshot.delete_ops_total });
+
+    try writer.print("# HELP flo_kv_bytes_stored Approximate bytes stored in this namespace\n", .{});
+    try writer.print("# TYPE flo_kv_bytes_stored gauge\n", .{});
+    try writer.print("flo_kv_bytes_stored{{{s}}} {d}\n", .{ labels, snapshot.bytes_stored });
+}
+
 fn writeTieredLogMetrics(
     writer: anytype,
     snapshot: TieredLogMetrics.Snapshot,

src/node/shard.zig

@@ -613,6 +613,10 @@ pub const Shard = struct {
         // equivalents move, which reads as "this shard is idle".
         self.shard_metrics = registry.shardMetrics(self.id);
         self.stream_handler.metrics_registry = registry;
+        // Tier-hit counters are per log; resolve once so the read path avoids a
+        // registry lookup per record. Also the only caller of registerTieredLog,
+        // without which the flo_tiered_log_* family never appears at all.
+        self.stream_handler.tiered_metrics = registry.registerTieredLog(self.id) catch null;
         self.queue_handler.metrics_registry = registry;
         self.kv_handler.metrics_registry = registry;
     }

src/stream/handler.zig

@@ -60,6 +60,7 @@ const UAL = @import("../storage/ual/ual.zig").UAL;
 const persistence_mod = @import("../storage/persistence.zig");
 const ReplayRegistry = persistence_mod.ReplayRegistry;
 const MetricsRegistry = @import("../metrics/registry.zig").MetricsRegistry;
+const TieredLogMetrics = @import("../metrics/registry.zig").TieredLogMetrics;
 
 // ═══════════════════════════════════════════════════════════════════════════════
 // StreamHandler
@@ -77,6 +78,10 @@ pub const StreamHandler = struct {
     /// Global metrics registry (optional, set by runtime when dashboard is enabled).
     metrics_registry: ?*MetricsRegistry,
 
+    /// Tier-hit counters for this shard's log, resolved once at wire-up so the
+    /// read path does not do a registry lookup per record.
+    tiered_metrics: ?*TieredLogMetrics = null,
+
     /// Maximum number of messages in a single read response.
     const MAX_READ_BATCH: usize = 1000;
     const DEFAULT_READ_BATCH: usize = 100;
@@ -1589,15 +1594,20 @@ pub const StreamHandler = struct {
         // prefix so callers see the bare batch blob `unpackBatch` expects.
         if (self.partition.ual.read(ual_index)) |ual_entry| {
             if (ual_entry.commandPayload()) |cmd| {
+                if (self.tiered_metrics) |tm| tm.recordHotHit();
                 return .{ .payload = stream_mod.decodeAppendValue(cmd.value).payload, .tier = 0 };
             }
         }
         // Warm fallback — payload copied to partition warm store on apply()
         if (self.partition.readPayloadWarm(ual_index)) |raw| {
             if (entry_mod.CommandPayload.deserialize(raw)) |cmd| {
+                if (self.tiered_metrics) |tm| tm.recordWarmHit();
                 return .{ .payload = stream_mod.decodeAppendValue(cmd.value).payload, .tier = 1 };
             }
         }
+        // Neither tier had it. Cold is not consulted here, so cold_hits stays 0
+        // for this path by design rather than for want of instrumentation.
+        if (self.tiered_metrics) |tm| tm.recordMiss();
         return .{ .payload = "", .tier = 0 };
     }
 

tests/e2e/metrics_values_test.zig

@@ -176,3 +176,70 @@ test "e2e/metrics: processing submitted counter carries a real value" {
 
     try testing.expectEqual(@as(?u64, 1), scalarValue(resp.body, "flo_processing_jobs_submitted_total"));
 }
+
+test "e2e/metrics: kv counters carry real values" {
+    var ctx = try stdx.testing.TestContext.initWithConfig(testing.allocator, .{
+        .server = .{ .metrics_enabled = true },
+    });
+    defer ctx.deinit();
+
+    try ctx.exec(&.{ "kv", "set", "a", "1" });
+    try ctx.exec(&.{ "kv", "set", "b", "2" });
+    try ctx.exec(&.{ "kv", "get", "a" });
+    try ctx.exec(&.{ "kv", "delete", "b" });
+
+    var http = try ctx.createMetricsHttp();
+    defer http.deinit();
+    var resp = try http.get("/metrics");
+    defer resp.deinit();
+
+    // The family was not emitted by exportPrometheus at all before, so these
+    // series did not exist rather than reading zero.
+    try testing.expectEqual(@as(?u64, 2), seriesValue(resp.body, "flo_kv_set_ops_total", "default"));
+    try testing.expectEqual(@as(?u64, 1), seriesValue(resp.body, "flo_kv_get_ops_total", "default"));
+    try testing.expectEqual(@as(?u64, 1), seriesValue(resp.body, "flo_kv_delete_ops_total", "default"));
+}
+
+test "e2e/metrics: tiered log hit 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", "tiered", "a", "b", "c" });
+    try ctx.exec(&.{ "stream", "read", "tiered" });
+
+    var http = try ctx.createMetricsHttp();
+    defer http.deinit();
+    var resp = try http.get("/metrics");
+    defer resp.deinit();
+
+    // registerTieredLog had no callers, so the whole family was absent.
+    // A fresh append is served from the hot ring.
+    const hot = seriesValue(resp.body, "flo_tiered_log_hot_hits_total", "group_id") orelse 0;
+    try testing.expect(hot > 0);
+    try testing.expect(seriesValue(resp.body, "flo_tiered_log_reads_total", "group_id") != null);
+}
+
+test "e2e/metrics: kv key count does not wrap when deleting a recovered key" {
+    var ctx = try stdx.testing.TestContext.initWithConfig(testing.allocator, .{
+        .server = .{ .metrics_enabled = true, .durability = .sync },
+    });
+    defer ctx.deinit();
+
+    try ctx.exec(&.{ "kv", "set", "survivor", "v" });
+
+    // Recovery rebuilds keys without going through recordSet, so key_count is
+    // back to 0 here while the key exists. Deleting it used to wrap the gauge
+    // to u64 max.
+    try ctx.restartServer();
+    try ctx.exec(&.{ "kv", "delete", "survivor" });
+
+    var http = try ctx.createMetricsHttp();
+    defer http.deinit();
+    var resp = try http.get("/metrics");
+    defer resp.deinit();
+
+    const keys = seriesValue(resp.body, "flo_kv_keys", "default") orelse 0;
+    try testing.expect(keys < 1_000_000);
+}