From 7a12a6cf12b41b3c07c543a7bd0304a9dfc87895 Mon Sep 17 00:00:00 2001 From: zzstoatzz Date: Tue, 18 Aug 2026 03:14:53 -0500 Subject: [PATCH] per-key archive metrics + operator consumers dashboard MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit the key ring renders one prometheus series set per key (served bytes, requests, throttles, live bucket level, configured rate) appended to /metrics; counters survive ring reloads alongside the bucket. new grafana dashboard deploy/grafana/stream-consumers.json answers 'who is putting load on the system' per key — operator-facing. Co-Authored-By: Claude Fable 5 --- deploy/grafana/stream-consumers.json | 213 +++++++++++++++++++++++++++ src/internal/serve/api_keys.zig | 57 +++++++ src/internal/serve/server.zig | 8 +- src/main.zig | 1 + 4 files changed, 278 insertions(+), 1 deletion(-) create mode 100644 deploy/grafana/stream-consumers.json diff --git a/deploy/grafana/stream-consumers.json b/deploy/grafana/stream-consumers.json new file mode 100644 index 0000000..60dd8e9 --- /dev/null +++ b/deploy/grafana/stream-consumers.json @@ -0,0 +1,213 @@ +{ + "uid": "stream-consumers", + "title": "stream / archive consumers (operator)", + "tags": [ + "stream", + "operator" + ], + "timezone": "utc", + "refresh": "30s", + "time": { + "from": "now-24h", + "to": "now" + }, + "panels": [ + { + "id": 1, + "type": "timeseries", + "title": "throughput per key", + "description": "compressed archive bytes served per key", + "datasource": { + "type": "prometheus", + "uid": "prometheus" + }, + "gridPos": { + "h": 8, + "w": 8, + "x": 0, + "y": 0 + }, + "fieldConfig": { + "defaults": { + "custom": { + "lineWidth": 2, + "fillOpacity": 12 + }, + "unit": "Bps" + }, + "overrides": [] + }, + "targets": [ + { + "expr": "rate(stream_archive_key_served_bytes_total[5m])", + "legendFormat": "{{key}}", + "refId": "A" + } + ] + }, + { + "id": 2, + "type": "timeseries", + "title": "cumulative served per key", + "description": "", + "datasource": { + "type": "prometheus", + "uid": "prometheus" + }, + "gridPos": { + "h": 8, + "w": 8, + "x": 8, + "y": 0 + }, + "fieldConfig": { + "defaults": { + "custom": { + "lineWidth": 2, + "fillOpacity": 12 + }, + "unit": "bytes" + }, + "overrides": [] + }, + "targets": [ + { + "expr": "stream_archive_key_served_bytes_total", + "legendFormat": "{{key}}", + "refId": "A" + } + ] + }, + { + "id": 3, + "type": "timeseries", + "title": "throttles (429s) per key", + "description": "requests refused while the key's bucket was in deficit", + "datasource": { + "type": "prometheus", + "uid": "prometheus" + }, + "gridPos": { + "h": 8, + "w": 8, + "x": 16, + "y": 0 + }, + "fieldConfig": { + "defaults": { + "custom": { + "lineWidth": 2, + "fillOpacity": 12 + } + }, + "overrides": [] + }, + "targets": [ + { + "expr": "increase(stream_archive_key_throttled_total[10m])", + "legendFormat": "{{key}}", + "refId": "A" + } + ] + }, + { + "id": 4, + "type": "timeseries", + "title": "bucket headroom per key", + "description": "negative = in deficit (serving 429s)", + "datasource": { + "type": "prometheus", + "uid": "prometheus" + }, + "gridPos": { + "h": 8, + "w": 8, + "x": 0, + "y": 8 + }, + "fieldConfig": { + "defaults": { + "custom": { + "lineWidth": 2, + "fillOpacity": 12 + }, + "unit": "bytes" + }, + "overrides": [] + }, + "targets": [ + { + "expr": "stream_archive_key_bucket_bytes", + "legendFormat": "{{key}}", + "refId": "A" + } + ] + }, + { + "id": 5, + "type": "timeseries", + "title": "configured rate per key", + "description": "", + "datasource": { + "type": "prometheus", + "uid": "prometheus" + }, + "gridPos": { + "h": 8, + "w": 8, + "x": 8, + "y": 8 + }, + "fieldConfig": { + "defaults": { + "custom": { + "lineWidth": 2, + "fillOpacity": 12 + }, + "unit": "Bps" + }, + "overrides": [] + }, + "targets": [ + { + "expr": "stream_archive_key_rate_bps", + "legendFormat": "{{key}}", + "refId": "A" + } + ] + }, + { + "id": 6, + "type": "timeseries", + "title": "getBlock requests (all keys + fleet)", + "description": "", + "datasource": { + "type": "prometheus", + "uid": "prometheus" + }, + "gridPos": { + "h": 8, + "w": 8, + "x": 16, + "y": 8 + }, + "fieldConfig": { + "defaults": { + "custom": { + "lineWidth": 2, + "fillOpacity": 12 + } + }, + "overrides": [] + }, + "targets": [ + { + "expr": "rate(jetstream_getblock_requests_total[5m])", + "legendFormat": "{{result}}", + "refId": "A" + } + ] + } + ], + "schemaVersion": 39 +} \ No newline at end of file diff --git a/src/internal/serve/api_keys.zig b/src/internal/serve/api_keys.zig index 5082b7c..ba922ca 100644 --- a/src/internal/serve/api_keys.zig +++ b/src/internal/serve/api_keys.zig @@ -47,6 +47,11 @@ pub const Entry = struct { /// token bucket level; negative = in deficit bucket_bytes: i64, last_refill_us: i64, + // per-key observability (rendered into /metrics; carried across + // reloads with the bucket) + served_bytes_total: u64 = 0, + requests_total: u64 = 0, + throttled_total: u64 = 0, }; pub const KeyRing = struct { @@ -100,6 +105,7 @@ pub const KeyRing = struct { if (entry.rate_bps == 0) return null; refill(entry, now_us); if (entry.bucket_bytes >= 0) return null; + entry.throttled_total += 1; const deficit: u64 = @intCast(-entry.bucket_bytes); return @max(1, deficit / entry.rate_bps); } @@ -110,11 +116,32 @@ pub const KeyRing = struct { pub fn charge(self: *KeyRing, io: Io, entry: *Entry, bytes: u64, now_us: i64) void { self.mutex.lockUncancelable(io); defer self.mutex.unlock(io); + entry.requests_total += 1; + entry.served_bytes_total +|= bytes; if (entry.rate_bps == 0) return; refill(entry, now_us); entry.bucket_bytes -= @intCast(@min(bytes, std.math.maxInt(i64))); } + /// prometheus lines for every key: consumption, throttling, and live + /// bucket headroom. key names are operator-chosen and bounded, so the + /// label cardinality is too. + pub fn renderMetrics(self: *KeyRing, io: Io, w: anytype) void { + self.mutex.lockUncancelable(io); + defer self.mutex.unlock(io); + if (self.entries.items.len == 0) return; + w.print("# TYPE stream_archive_key_served_bytes_total counter\n", .{}) catch return; + for (self.entries.items) |e| w.print("stream_archive_key_served_bytes_total{{key=\"{s}\"}} {d}\n", .{ e.name, e.served_bytes_total }) catch return; + w.print("# TYPE stream_archive_key_requests_total counter\n", .{}) catch return; + for (self.entries.items) |e| w.print("stream_archive_key_requests_total{{key=\"{s}\"}} {d}\n", .{ e.name, e.requests_total }) catch return; + w.print("# TYPE stream_archive_key_throttled_total counter\n", .{}) catch return; + for (self.entries.items) |e| w.print("stream_archive_key_throttled_total{{key=\"{s}\"}} {d}\n", .{ e.name, e.throttled_total }) catch return; + w.print("# TYPE stream_archive_key_bucket_bytes gauge\n", .{}) catch return; + for (self.entries.items) |e| w.print("stream_archive_key_bucket_bytes{{key=\"{s}\"}} {d}\n", .{ e.name, e.bucket_bytes }) catch return; + w.print("# TYPE stream_archive_key_rate_bps gauge\n", .{}) catch return; + for (self.entries.items) |e| w.print("stream_archive_key_rate_bps{{key=\"{s}\"}} {d}\n", .{ e.name, e.rate_bps }) catch return; + } + fn refill(entry: *Entry, now_us: i64) void { const elapsed_us = now_us - entry.last_refill_us; entry.last_refill_us = now_us; @@ -192,6 +219,9 @@ pub const KeyRing = struct { if (std.crypto.timing_safe.eql([32]u8, old.key_hash, hash)) { entry.bucket_bytes = old.bucket_bytes; entry.last_refill_us = old.last_refill_us; + entry.served_bytes_total = old.served_bytes_total; + entry.requests_total = old.requests_total; + entry.throttled_total = old.throttled_total; break; } } @@ -263,3 +293,30 @@ test "bucket: burst allowed, deficit 429s, refill clears it" { ring.charge(io, open, 1 << 40, 5); try testing.expectEqual(@as(?u64, null), ring.precheck(io, open, 6)); } + +test "renderMetrics: one line per key per series, counters carry across reload" { + var threaded: Io.Threaded = .init(testing.allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + + var ring = KeyRing.init(testing.allocator, "/nonexistent"); + defer ring.deinit(); + try ring.loadFromSlice("evelyn:k1:16\nci:k2\n"); + ring.charge(io, &ring.entries.items[0], 1000, 10); + ring.charge(io, &ring.entries.items[0], 500, 20); + _ = ring.precheck(io, &ring.entries.items[0], 30); // no throttle: bucket positive + + var buf: [4096]u8 = undefined; + var w: std.Io.Writer = .fixed(&buf); + ring.renderMetrics(io, &w); + const out = w.buffered(); + try testing.expect(std.mem.indexOf(u8, out, "stream_archive_key_served_bytes_total{key=\"evelyn\"} 1500") != null); + try testing.expect(std.mem.indexOf(u8, out, "stream_archive_key_requests_total{key=\"evelyn\"} 2") != null); + try testing.expect(std.mem.indexOf(u8, out, "stream_archive_key_rate_bps{key=\"evelyn\"} 16000000") != null); + try testing.expect(std.mem.indexOf(u8, out, "stream_archive_key_served_bytes_total{key=\"ci\"} 0") != null); + + // reload keeps consumption history for surviving keys + try ring.loadFromSlice("evelyn:k1:8\n"); + try testing.expectEqual(@as(u64, 1500), ring.entries.items[0].served_bytes_total); + try testing.expectEqual(@as(u64, 2), ring.entries.items[0].requests_total); +} diff --git a/src/internal/serve/server.zig b/src/internal/serve/server.zig index 094b55a..b6be802 100644 --- a/src/internal/serve/server.zig +++ b/src/internal/serve/server.zig @@ -25,6 +25,7 @@ const repo_export = @import("repo_export.zig"); const repo_store = @import("../ingest/backfill/repo_store.zig"); const phase_mod = @import("../ingest/backfill/phase.zig"); const status_page = @import("status_page.zig"); +const api_keys = @import("api_keys.zig"); const tail_mod = @import("tail.zig"); const wire = @import("wire.zig"); const xrpcapi = @import("xrpcapi.zig"); @@ -138,6 +139,8 @@ pub const Hub = struct { /// that observes true may dereference them. Defaults true so test hubs /// that publish pointers before serving are unaffected. storage_ready: std.atomic.Value(bool) = .init(true), + /// per-key archive metering ring, rendered into /metrics when set + key_ring: ?*api_keys.KeyRing = null, pub fn deinit(self: *Hub) void { self.repo_action_limiter.deinit(self.allocator); @@ -1271,7 +1274,10 @@ fn respondMetrics(conn: *websocket.Conn, hub: *Hub, observation: *HttpObservatio .base = g.base, .tip = g.tip, }, now_s, process_metrics.collect(hub.io), if (hub.archive) |archive| disk_space.freeBytes(archive.dir) else null, readDurableRepos(hub)); - respondPartsMaybeHead(conn, "200 OK", "text/plain; version=0.0.4", out, "", head); + var key_buf: [16 * 1024]u8 = undefined; + var key_writer: Io.Writer = .fixed(&key_buf); + if (hub.key_ring) |ring| ring.renderMetrics(hub.io, &key_writer); + respondPartsMaybeHead(conn, "200 OK", "text/plain; version=0.0.4", out, key_writer.buffered(), head); } /// adapter the xrpc api writes through (json / raw+etag / xrpc error) diff --git a/src/main.zig b/src/main.zig index 31595ff..cc5a5e5 100644 --- a/src/main.zig +++ b/src/main.zig @@ -729,6 +729,7 @@ pub fn main(init: std.process.Init.Minimal) !void { defer if (key_ring) |*ring| ring.deinit(); if (key_ring) |*ring| { xrpc_api.key_ring = ring; + hub.key_ring = ring; log.info("archive key ring enabled: {s} (revocation = edit the file)", .{cfg.archive_api_keys_file}); } -- 2.51.2