diff --git a/src/kv/handler.zig b/src/kv/handler.zig index 21f5a7a..dc3ec8e 100644 --- a/src/kv/handler.zig +++ b/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); } diff --git a/src/metrics/registry.zig b/src/metrics/registry.zig index c353cfc..bacf0fa 100644 --- a/src/metrics/registry.zig +++ b/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, diff --git a/src/node/shard.zig b/src/node/shard.zig index 0de2c83..1385554 100644 --- a/src/node/shard.zig +++ b/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; } diff --git a/src/stream/handler.zig b/src/stream/handler.zig index 2bbcfda..3ad935b 100644 --- a/src/stream/handler.zig +++ b/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 }; } diff --git a/tests/e2e/metrics_values_test.zig b/tests/e2e/metrics_values_test.zig index a94b2d5..d8ea5a1 100644 --- a/tests/e2e/metrics_values_test.zig +++ b/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); +}