diff --git a/CHANGELOG.md b/CHANGELOG.md index 42ee027..ecdad1a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,12 @@ All notable changes to nxdns are recorded here. The format follows [Keep a Chang Sections are written by hand. Nothing here is generated from commit messages: the point of the file is to say what changed for an operator, which a commit subject rarely does. +## [0.0.22] - 2026-10-19 + +### Added + +- **`/metrics` reports the most concurrent work the server has seen.** Five gauges hold peaks since start: `nxdns_dns_in_flight_peak` (queries handled at once, every listener together), `nxdns_tcp_server_connections_peak`, `nxdns_dot_server_connections_peak` and `nxdns_doh_server_connections_peak` (connections one listener held at once), and `nxdns_querylog_queue_peak` (entries waiting for the query log writer at once). The memory budget work sizes its limits from them; "Measure performance" shows how to read them beside the process's resident memory, thread count and worker stacks. + ## [0.0.21] - 2026-09-12 ### Changed diff --git a/docs/how-to/measure-performance.md b/docs/how-to/measure-performance.md index 5ffa442..d7638e8 100644 --- a/docs/how-to/measure-performance.md +++ b/docs/how-to/measure-performance.md @@ -114,6 +114,26 @@ Read the client's own rate and the server's `/metrics` together: a load generato **Not verified on this host:** `dnsperf` is not installed here and the target platform is a Raspberry Pi 5, not this development machine. The command above is the shape of the measurement, not a transcript. +## Measure memory on a running server + +The bench measures one structure at a time. A running server's memory also depends on the most concurrent work it has seen since it started, so read it on the server itself, after it has served a normal day: + +```sh +pid=$(pidof nxdns) +grep -E 'VmRSS|VmHWM|Threads' /proc/$pid/status +curl -s http://127.0.0.1:8080/metrics | grep '^nxdns_.*_peak' +``` + +- `VmRSS` is what the process holds resident now. `VmHWM` is the most it has ever held, which is normally the last blocklist refresh. +- `Threads` is a high-water too: the I/O runtime adds a worker thread only when every existing worker is busy, and never retires one. The count is the largest the worker pool has been since start, plus the main thread. +- `nxdns_dns_in_flight_peak` is the most queries handled at once, every listener together. +- `nxdns_tcp_server_connections_peak`, `nxdns_dot_server_connections_peak` and `nxdns_doh_server_connections_peak` are the most connections one listener has held at once. TCP/53 has one listener per address family; the gauge is the larger of the two. +- `nxdns_querylog_queue_peak` is the most entries that have waited for the query log writer at once. It can overstate the true peak by the entries in transit at that instant, never understate it. + +Each worker thread's stack is its own anonymous mapping in `/proc/$pid/smaps`, directly above a one-page `---p` guard, and every worker's mapping has the same size. Its `Rss` is the stack that thread has actually touched. + +**Verified on this host** against a static x86_64 musl build under a burst of 300 UDP queries: the gauges and the stack mappings behave as described. Not yet run on the Pi. + ## Where to run it The target platform is a Raspberry Pi 5. Numbers from a development x86_64 box do not transfer — the Pi's Cortex-A76 is far slower — so a passing run here is evidence the harness works and a baseline for spotting regressions on the machine development happens on, and nothing more. Run `--assert` on the Pi, where the numbers mean something. diff --git a/specs/memory-budget.md b/specs/memory-budget.md index 8ba995f..282850b 100644 --- a/specs/memory-budget.md +++ b/specs/memory-budget.md @@ -31,13 +31,15 @@ Why freed small blocks stay resident: with libc linked, `malloc` is zig's `lib/c The admission budget, the listener caps, the stack size and the queue size are chosen from the Pi's real peaks, so those peaks become measurements before anything is resized: -- high-water of in-flight DNS queries; -- high-water of active stream connections, per listener (today `connections` counts accepts over time, not concurrency); -- high-water of concurrent `std.Io.Threaded` tasks; -- high-water of the query-log writer queue's occupancy (not arrivals per interval: a full batch flushes early, and a disk gate or a slow write lets backlog build past one interval); -- worker call-stack use, measured on representative workers. The signal stack (`std_options.signal_stack_size`) is a separate allocation for signal handling and stack traces and keeps its stdlib default unless its own use is measured. +- high-water of in-flight DNS queries: `nxdns_dns_in_flight_peak`, counted around `Handler.handle`, every listener together; +- high-water of active stream connections, per listener (today `connections` counts accepts over time, not concurrency): `nxdns_tcp_server_connections_peak` (the larger of the two address families), `nxdns_dot_server_connections_peak`, `nxdns_doh_server_connections_peak`; +- high-water of the `std.Io.Threaded` worker pool: the process thread count. `Threaded` spawns a worker only when every worker is busy and never retires one (`Threaded.zig:2110`), so the count is that high-water already and needs no code. It counts workers, not tasks: `async` work can run inline or wait in the run queue; +- high-water of the query-log writer queue's occupancy (not arrivals per interval: a full batch flushes early, and a disk gate or a slow write lets backlog build past one interval): `nxdns_querylog_queue_peak`. `std.Io.Queue` exposes no length, so a count beside it is raised before each insert and lowered after each removal; it can overstate the peak by the entries in transit, never understate it; +- worker call-stack use, read from `/proc//smaps`: each worker stack is an anonymous mapping directly above its own guard page, all of one size, and its `Rss` is what the thread has touched and still holds resident. That finds the expensive paths; it is not a depth bound. Before item 3 lowers `stack_size`, each expensive path gets a measured stack high-water, and the size keeps headroom above the deepest one. The signal stack (`std_options.signal_stack_size`) is a separate allocation for signal handling and stack traces and keeps its stdlib default unless its own use is measured. -The counters appear in diagnostics beside the existing per-reason drop counters. One release carrying only these measurements runs on the Pi before the budget values are fixed. +The five gauges are in `/metrics` beside the existing per-reason drop counters. One release carrying only these measurements runs on the Pi before the budget values are fixed. + +First reading, static x86_64 musl build, one burst of 300 UDP queries: the in-flight peak reached 64 (the UDP slot pool), threads went from 14 to 205, and of 201 worker stacks 175 touched 32 kB or less while 26 touched 2 MiB each — the same 2 MiB with only forward-zone traffic and no DoT exchange, so the TLS client is not the owner. Item 3 identifies that owner before it lowers `stack_size`: a stack smaller than that path needs is a crash, not a saving. ### 1. Store each blocklist name once in memory @@ -86,7 +88,7 @@ Within one source, as compiled entries, an exact `x` in `.list` together with `x - [ ] Matcher equivalence: a deterministic corpus (hosts, domains, ABP with `||x^`, `*.x` alone, `x` alone, `x` with `*.x`, `@@`, an exact entry for a subdomain of a wildcard, nested wildcards `*.x` and `*.y.x`, overlapping sources) gives the same `Decision` for every probe name before and after item 1. - [ ] Builder: a reload holds at most one source's bodies (tested through an allocator that records peak bytes). - [ ] Boot: one snapshot build when no source refreshes; listeners never start without the initial load having run. -- [ ] Measurements: each item-0 high-water rises under a scripted load and is visible in diagnostics. +- [ ] Measurements: each of the five item-0 gauges rises under a scripted load and is visible in `/metrics`; the thread count and the worker stack residency are recorded from `/proc` beside them. - [ ] Bounds: with the query budget full, a further UDP query is dropped and counted; a stream listener at its slot cap refuses and counts; a keep-alive connection between requests holds no query permit; `ConcurrencyUnavailable` is counted at each enumerated boundary, including the logger wait and the listener races. - [ ] SSE: no ring is allocated without a subscriber; an allocation failure is counted, reported as its own error and leaves the slot reusable; overflow still ends the stream. - [ ] SQLite: each runtime handle reports its configured `cache_size`; migrations open without it. diff --git a/src/server/doh_server.zig b/src/server/doh_server.zig index a33d3d7..753ac44 100644 --- a/src/server/doh_server.zig +++ b/src/server/doh_server.zig @@ -148,6 +148,10 @@ pub const DohServer = struct { return self.core.boundAddress(); } + pub fn connectionsPeak(self: *const DohServer) u64 { + return self.core.connectionsPeak(); + } + pub fn snapshotStats(self: *const DohServer) Snapshot { const core = &self.core.stats; return .{ diff --git a/src/server/dot_server.zig b/src/server/dot_server.zig index 24c8a5e..1ad6318 100644 --- a/src/server/dot_server.zig +++ b/src/server/dot_server.zig @@ -125,6 +125,10 @@ pub const DotServer = struct { self.* = undefined; } + pub fn connectionsPeak(self: *const DotServer) u64 { + return self.core.connectionsPeak(); + } + pub fn snapshotStats(self: *const DotServer) StatsSnapshot { const core = &self.core.stats; return .{ diff --git a/src/server/handler.zig b/src/server/handler.zig index 3d6a4c8..a9ca641 100644 --- a/src/server/handler.zig +++ b/src/server/handler.zig @@ -130,6 +130,11 @@ pub const Handler = struct { pause: ?*pause.Pause = null, tracker: ?*clients.Tracker = null, stats: Stats = .{}, + /// Queries inside `handle` right now, across every listener, and the most + /// there have been at once (specs/memory-budget.md item 0). Gauges, so they + /// stay out of `Stats`, which `/metrics` exports field by field as counters. + in_flight: std.atomic.Value(u64) = .init(0), + in_flight_peak: std.atomic.Value(u64) = .init(0), pub const Stats = struct { /// Queries answered by the upstream pool. @@ -229,6 +234,21 @@ pub const Handler = struct { query: []const u8, response_buf: []u8, scratch: *Scratch, + ) Outcome { + const now_in_flight = self.in_flight.fetchAdd(1, .monotonic) + 1; + _ = self.in_flight_peak.fetchMax(now_in_flight, .monotonic); + defer _ = self.in_flight.fetchSub(1, .monotonic); + return self.answer(io, which, from, query, response_buf, scratch); + } + + fn answer( + self: *Handler, + io: std.Io, + which: Transport, + from: address.NetAddress, + query: []const u8, + response_buf: []u8, + scratch: *Scratch, ) Outcome { std.debug.assert(response_buf.len >= udp_limit_min); std.debug.assert(response_buf.len <= transport.max_message_len); @@ -1459,6 +1479,24 @@ test "a response arriving on the listener port is dropped" { try testing.expectEqual(@as(usize, 0), fake.calls); } +test "every query counts in flight for exactly its own handling" { + var t: TestIo = .init(); + defer t.deinit(); + + var fake: FakeUpstream = .{ .reply = response_bytes }; + var h_owner: upstream_owner.Borrowed = .{}; + var h = bare(h_owner.client(fake.client())); + // Two queries other listeners are still handling. + h.in_flight.store(2, .monotonic); + + var buf: [udp_limit_min]u8 = undefined; + try testing.expect(udp(&h, t.io(), query_bytes, &buf) == .reply); + try testing.expectEqual(Handler.Outcome.drop, udp(&h, t.io(), response_bytes, &buf)); + + try testing.expectEqual(@as(u64, 2), h.in_flight.load(.monotonic)); + try testing.expectEqual(@as(u64, 3), h.in_flight_peak.load(.monotonic)); +} + test "a query shorter than a header is dropped" { var t: TestIo = .init(); defer t.deinit(); diff --git a/src/server/listener.zig b/src/server/listener.zig index fb1fbcf..39ca972 100644 --- a/src/server/listener.zig +++ b/src/server/listener.zig @@ -83,6 +83,10 @@ pub const CoreStats = struct { accept_errors: std.atomic.Value(u64) = .init(0), connection_errors: std.atomic.Value(u64) = .init(0), idle_timeouts: std.atomic.Value(u64) = .init(0), + /// The most slots that have been taken at once (specs/memory-budget.md + /// item 0). A gauge, so no listener's `Snapshot` carries it: those export + /// every field as a counter. Written only under `Core.mutex`. + connections_peak: std.atomic.Value(u64) = .init(0), }; // --------------------------------------------------------------------------- @@ -255,6 +259,9 @@ pub fn Core(comptime Cfg: type) type { /// that shuts the active connections down, so a stream that arrives /// after that scan can never claim a slot the scan will not visit again. shutdown_begun: bool, + /// Guarded by `mutex`. Slots that are not `.free`: a slot is taken from + /// `claim` until `finish` frees it, closing included. + taken: u64, stats: CoreStats, run_state: std.atomic.Value(State), stopped: std.Io.Event, @@ -299,6 +306,7 @@ pub fn Core(comptime Cfg: type) type { .conns = conns, .mutex = .init, .shutdown_begun = false, + .taken = 0, .stats = .{}, .run_state = .init(.idle), .stopped = .unset, @@ -336,6 +344,10 @@ pub fn Core(comptime Cfg: type) type { return self.listener.socket.address; } + pub fn connectionsPeak(self: *const Self) u64 { + return self.stats.connections_peak.load(.monotonic); + } + /// Accept loop. Returns when the task is canceled or `deinit` stops it. pub fn serve(self: *Self, io: std.Io) void { if (self.run_state.cmpxchgStrong(.idle, .serving, .acq_rel, .acquire) != null) return; @@ -443,6 +455,10 @@ pub fn Core(comptime Cfg: type) type { self.conns[index].stream = stream; self.conns[index].peer = stream.socket.address; self.conns[index].state = .active; + self.taken += 1; + if (self.taken > self.stats.connections_peak.load(.monotonic)) { + self.stats.connections_peak.store(self.taken, .monotonic); + } }, .at_capacity, .shutting_down => {}, } @@ -464,6 +480,7 @@ pub fn Core(comptime Cfg: type) type { self.mutex.lockUncancelable(io); conn.state = .free; + self.taken -= 1; self.mutex.unlock(io); } diff --git a/src/server/tcp_server.zig b/src/server/tcp_server.zig index 8add337..f7f6fea 100644 --- a/src/server/tcp_server.zig +++ b/src/server/tcp_server.zig @@ -99,6 +99,10 @@ pub const TcpServer = struct { return self.core.boundAddress(); } + pub fn connectionsPeak(self: *const TcpServer) u64 { + return self.core.connectionsPeak(); + } + /// The counters, read one at a time. A scrape that lands mid-accept can see /// a connection counted before its outcome is; a lock would buy a /// consistency no consumer needs. diff --git a/src/server/tcp_server_integration_test.zig b/src/server/tcp_server_integration_test.zig index a7fc454..d7dc581 100644 --- a/src/server/tcp_server_integration_test.zig +++ b/src/server/tcp_server_integration_test.zig @@ -329,6 +329,52 @@ test "a canceled serve does not wait for a live connection" { try client_result; } +test "the connections peak is the most slots held at once" { + if (!build_options.integration) return error.SkipZigTest; + + const gpa = testing.allocator; + var threaded: std.Io.Threaded = .init(gpa, .{}); + defer threaded.deinit(); + const io = threaded.io(); + + var fake: FakeUpstream = .{ .reply = response_bytes }; + var h_owner: upstream_owner.Borrowed = .{}; + var h = bareHandler(h_owner.client(fake.client())); + + const listen_address: net.IpAddress = try .parse("127.0.0.1", 0); + var server = try tcp_server.TcpServer.listen(gpa, io, listen_address, &h, .{ + .max_connections = 3, + .idle_timeout = .{ .raw = .fromSeconds(600), .clock = .awake }, + }); + const server_address = server.boundAddress(); + + var serving: std.Io.Group = .init; + try serving.concurrent(io, tcp_server.TcpServer.serve, .{ &server, io }); + + var answered: [2]std.Io.Event = @splat(.unset); + var client_results: [2]anyerror!void = @splat({}); + var clients_group: std.Io.Group = .init; + for (&answered, &client_results) |*event, *result| { + try clients_group.concurrent(io, holdConnection, .{ io, server_address, event, result }); + } + for (&answered) |*event| event.waitUncancelable(io); + + // Both replies prove both slots are claimed, and the listener has room for + // a third that never arrives. + const held_open = server.connectionsPeak(); + + server.deinit(io); + serving.await(io) catch |err| switch (err) { + error.Canceled => unreachable, + }; + clients_group.await(io) catch |err| switch (err) { + error.Canceled => unreachable, + }; + + for (client_results) |result| try result; + try testing.expectEqual(@as(u64, 2), held_open); +} + /// The first slot the server would hand out, read after `serve` has returned so /// nothing can be writing it. fn firstFreeSlot(server: *const tcp_server.TcpServer) ?usize { diff --git a/src/storage/logger.zig b/src/storage/logger.zig index 84c5037..2af7e47 100644 --- a/src/storage/logger.zig +++ b/src/storage/logger.zig @@ -320,7 +320,68 @@ fn emptyAsNull(value: []const u8) ?[]const u8 { return if (value.len == 0) null else value; } -const EntryQueue = std.Io.Queue(Entry); +/// `std.Io.Queue` plus the occupancy high-water that sizes +/// `logging.query_log_buffer_max` (specs/memory-budget.md item 0). +/// +/// The queue keeps its length behind its own mutex with no accessor, so the +/// count lives beside it. Raised before an insert and lowered after a removal, +/// it never reads below the true occupancy; it overstates it by at most the +/// producers inside `tryPut` plus the entries a getter has taken and not yet +/// subtracted. An upper bound is the safe side for sizing a buffer. +const EntryQueue = struct { + inner: std.Io.Queue(Entry), + held: std.atomic.Value(u64) = .init(0), + peak: std.atomic.Value(u64) = .init(0), + + pub fn init(buffer: []Entry) EntryQueue { + return .{ .inner = .init(buffer) }; + } + + pub fn capacity(q: *const EntryQueue) usize { + return q.inner.capacity(); + } + + pub fn close(q: *EntryQueue, io: std.Io) void { + q.inner.close(io); + } + + pub fn peakOccupancy(q: *const EntryQueue) u64 { + return q.peak.load(.monotonic); + } + + /// Takes `entry` if there is room or a waiting getter, without blocking. + pub fn tryPut(q: *EntryQueue, io: std.Io, entry: Entry) (std.Io.QueueClosedError || std.Io.Cancelable)!bool { + const held = q.held.fetchAdd(1, .monotonic) + 1; + const put = q.inner.put(io, &.{entry}, 0) catch |err| { + _ = q.held.fetchSub(1, .monotonic); + return err; + }; + if (put == 0) { + _ = q.held.fetchSub(1, .monotonic); + return false; + } + _ = q.peak.fetchMax(held, .monotonic); + return true; + } + + pub fn get(q: *EntryQueue, io: std.Io, buffer: []Entry, min: usize) (std.Io.QueueClosedError || std.Io.Cancelable)!usize { + const n = try q.inner.get(io, buffer, min); + _ = q.held.fetchSub(n, .monotonic); + return n; + } + + pub fn getUncancelable(q: *EntryQueue, io: std.Io, buffer: []Entry, min: usize) std.Io.QueueClosedError!usize { + const n = try q.inner.getUncancelable(io, buffer, min); + _ = q.held.fetchSub(n, .monotonic); + return n; + } + + pub fn getOne(q: *EntryQueue, io: std.Io) (std.Io.QueueClosedError || std.Io.Cancelable)!Entry { + const entry = try q.inner.getOne(io); + _ = q.held.fetchSub(1, .monotonic); + return entry; + } +}; /// A database failure is not in here: a batch the database refuses is dropped, /// counted and reported where it happens, and the writer carries on. @@ -525,8 +586,7 @@ pub const Logger = struct { while (true) { // A closed queue or a canceled task means shutdown is underway; // both leave this entry unwritten, which is what the counter says. - const put = self.queue.put(io, &.{entry}, 0) catch break; - if (put == 1) return; + if (self.queue.tryPut(io, entry) catch break) return; // A zero-capacity queue holds nothing to drop: the put above was // this entry's one chance at a waiting getter. @@ -1442,6 +1502,34 @@ test "a zero-capacity queue drops every entry exactly once" { try testing.expectEqual(@as(u64, 5), logger.queries_dropped.load(.monotonic)); } +test "the queue peak is the most entries held at once, never the arrivals" { + var threaded: std.Io.Threaded = .init(testing.allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + + var buf: [4]Entry = undefined; + var logger: Logger = .init(.{}, &buf); + try testing.expectEqual(@as(u64, 0), logger.queue.peakOccupancy()); + + logger.log(io, sampleEntry(1, "one.example")); + logger.log(io, sampleEntry(2, "two.example")); + logger.log(io, sampleEntry(3, "three.example")); + _ = try logger.queue.getOne(io); + _ = try logger.queue.getOne(io); + logger.log(io, sampleEntry(4, "four.example")); + try testing.expectEqual(@as(u64, 3), logger.queue.peakOccupancy()); + + // A full queue evicts to make room, so the peak stops at the capacity. + for (5..10) |i| logger.log(io, sampleEntry(@intCast(i), "more.example")); + try testing.expectEqual(@as(u64, 4), logger.queue.peakOccupancy()); + try testing.expectEqual(@as(u64, 4), logger.queue.held.load(.monotonic)); + + var drained: [4]Entry = undefined; + try testing.expectEqual(@as(usize, 4), try logger.queue.get(io, &drained, 0)); + try testing.expectEqual(@as(u64, 0), logger.queue.held.load(.monotonic)); + try testing.expectEqual(@as(u64, 4), logger.queue.peakOccupancy()); +} + test "log after shutdown drops instead of blocking" { var threaded: std.Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); diff --git a/src/storage/logger_controller.zig b/src/storage/logger_controller.zig index 646d23b..1eb5879 100644 --- a/src/storage/logger_controller.zig +++ b/src/storage/logger_controller.zig @@ -63,12 +63,14 @@ const Totals = struct { rows_written: u64 = 0, batches_gated: u64 = 0, last_drop_s: i64 = 0, + queue_peak: u64 = 0, fn fold(self: *Totals, lg: *const logger.Logger) void { self.queries_dropped += lg.queries_dropped.load(.monotonic); self.rows_written += lg.rows_written.load(.monotonic); self.batches_gated += lg.batches_gated.load(.monotonic); self.last_drop_s = @max(self.last_drop_s, lg.last_drop_s.load(.monotonic)); + self.queue_peak = @max(self.queue_peak, lg.queue.peakOccupancy()); } }; @@ -80,6 +82,8 @@ pub const Sample = struct { batches_gated: u64, /// Null while nothing has been dropped, exactly as `Logger.lastDropSeconds`. last_drop_s: ?i64, + /// The most entries any generation's queue has held at once. + queue_peak: u64, /// The LIVE writer's state. A retired writer's failures are in the counters. writer_failed: bool, /// The worst episode any outstanding generation is in. A retired writer @@ -376,6 +380,7 @@ pub const Controller = struct { .rows_written = totals.rows_written, .batches_gated = totals.batches_gated, .last_drop_s = if (totals.last_drop_s == 0) null else totals.last_drop_s, + .queue_peak = totals.queue_peak, .writer_failed = self.live.logger.writer_failed.load(.acquire), .gate_episode = episode, }; diff --git a/src/web/metrics.zig b/src/web/metrics.zig index 1cebc5c..aa45529 100644 --- a/src/web/metrics.zig +++ b/src/web/metrics.zig @@ -130,6 +130,20 @@ pub const DiagnosticsSample = struct { write_failures: u64, }; +/// The most of each kind of concurrent work there has been at once since start +/// (specs/memory-budget.md item 0): what the admission budget, the listener +/// caps and the query-log queue are sized from. Gauges, so none of them rides +/// in a counter group. A null field is a collaborator the state does not have. +pub const PeakSample = struct { + dns_in_flight: ?u64 = null, + querylog_queue: ?u64 = null, + /// The larger of the two address families' peaks. Each family has its own + /// slot pool, so a cap is sized per listener, never from their sum. + tcp_connections: ?u64 = null, + dot_connections: ?u64 = null, + doh_connections: ?u64 = null, +}; + /// Everything one scrape reports. A null section is a collaborator the state /// does not have. pub const Sample = struct { @@ -172,6 +186,7 @@ pub const Sample = struct { /// label and cannot live in `UpstreamSample`. Absent while no pool is /// wired, like every other collaborator. upstream_budget_exhausted_total: ?u64 = null, + peaks: PeakSample = .{}, }; pub fn handle( @@ -193,6 +208,7 @@ pub fn collect(state: *server.WebState, io: std.Io, arena: Allocator) Allocator. if (state.handler) |handler| { sample.dns = dnsCounters(&handler.stats); + sample.peaks.dns_in_flight = handler.in_flight_peak.load(.monotonic); // The pointer is loaded under the same mutex a query loads it under, // never before it: a live `cache.size` or `dns.rate_limit` change @@ -223,6 +239,7 @@ pub fn collect(state: *server.WebState, io: std.Io, arena: Allocator) Allocator. .rows_written = reading.rows_written, .batches_gated = reading.batches_gated, }; + sample.peaks.querylog_queue = reading.queue_peak; } if (state.tracker) |tracker| sample.tracker = .{ @@ -270,9 +287,12 @@ pub fn collect(state: *server.WebState, io: std.Io, arena: Allocator) Allocator. }; } if (state.dot_listener) |listener| sample.dot_listener = listener.snapshotStats(); + if (state.doh_listener) |listener| sample.peaks.doh_connections = listener.connectionsPeak(); + if (state.dot_listener) |listener| sample.peaks.dot_connections = listener.connectionsPeak(); sample.udp_listener = sumListeners(udp_server.Snapshot, udp_server.UdpServer, state.udp_listeners); sample.tcp_listener = sumListeners(tcp_server.Snapshot, tcp_server.TcpServer, state.tcp_listeners); + sample.peaks.tcp_connections = largestConnectionsPeak(state.tcp_listeners); if (state.upstreams) |owner| { const generation = owner.acquire(io); @@ -308,6 +328,13 @@ fn sumListeners(comptime Snapshot: type, comptime Server: type, listeners: []con return total; } +/// Null for an empty slice, like `sumListeners`. +fn largestConnectionsPeak(listeners: []const *tcp_server.TcpServer) ?u64 { + var largest: ?u64 = null; + for (listeners) |listener| largest = @max(largest orelse 0, listener.connectionsPeak()); + return largest; +} + fn dnsCounters(stats: *const dns_handler.Handler.Stats) DnsCounters { var out: DnsCounters = undefined; inline for (dns_stat_fields, 0..) |field, i| { @@ -471,6 +498,8 @@ pub fn render(w: *std.Io.Writer, sample: Sample) std.Io.Writer.Error!void { try counterGroup(w, "nxdns_dot_server_", "DoT listener counter", listener); } + try renderPeaks(w, sample.peaks); + if (sample.doh_certs != null or sample.dot_certs != null) try renderCerts(w, sample); if (sample.upstreams.len != 0) try renderUpstreams(w, sample.upstreams); @@ -485,6 +514,39 @@ pub fn render(w: *std.Io.Writer, sample: Sample) std.Io.Writer.Error!void { } } +fn renderPeaks(w: *std.Io.Writer, peaks: PeakSample) std.Io.Writer.Error!void { + if (peaks.dns_in_flight) |value| try gauge( + w, + "nxdns_dns_in_flight_peak", + "Most DNS queries handled at once since start, across every listener.", + value, + ); + if (peaks.querylog_queue) |value| try gauge( + w, + "nxdns_querylog_queue_peak", + "Upper bound on the most entries the query log queue has held at once since start.", + value, + ); + if (peaks.tcp_connections) |value| try gauge( + w, + "nxdns_tcp_server_connections_peak", + "Most connections one TCP/53 listener has held at once since start.", + value, + ); + if (peaks.dot_connections) |value| try gauge( + w, + "nxdns_dot_server_connections_peak", + "Most connections the DoT listener has held at once since start.", + value, + ); + if (peaks.doh_connections) |value| try gauge( + w, + "nxdns_doh_server_connections_peak", + "Most connections the DoH listener has held at once since start.", + value, + ); +} + fn renderCerts(w: *std.Io.Writer, sample: Sample) std.Io.Writer.Error!void { try labeledHead(w, "nxdns_cert_reloads_total", "Certificate reloads that published a new context.", "counter"); if (sample.doh_certs) |stats| try endpointValue(w, "nxdns_cert_reloads_total", "doh", stats.reloads); @@ -979,6 +1041,35 @@ test "an unwired collaborator omits its family rather than reporting zeros" { try testing.expect(!std.mem.containsAtLeast(u8, text, 1, "nxdns_dot_server_")); try testing.expect(!std.mem.containsAtLeast(u8, text, 1, "nxdns_udp_server_")); try testing.expect(!std.mem.containsAtLeast(u8, text, 1, "nxdns_tcp_server_")); + try testing.expect(!std.mem.containsAtLeast(u8, text, 1, "_peak")); +} + +test "each peak is a gauge, and an unwired one is absent" { + const text = try renderToString(testing.allocator, .{ .peaks = .{ + .dns_in_flight = 7, + .querylog_queue = 120, + .tcp_connections = 3, + } }); + defer testing.allocator.free(text); + + try testing.expect(std.mem.containsAtLeast(u8, text, 1, "# TYPE nxdns_dns_in_flight_peak gauge\nnxdns_dns_in_flight_peak 7\n")); + try testing.expect(std.mem.containsAtLeast(u8, text, 1, "# TYPE nxdns_querylog_queue_peak gauge\nnxdns_querylog_queue_peak 120\n")); + try testing.expect(std.mem.containsAtLeast(u8, text, 1, "# TYPE nxdns_tcp_server_connections_peak gauge\nnxdns_tcp_server_connections_peak 3\n")); + try testing.expect(!std.mem.containsAtLeast(u8, text, 1, "nxdns_dot_server_connections_peak")); + try testing.expect(!std.mem.containsAtLeast(u8, text, 1, "nxdns_doh_server_connections_peak")); +} + +test "the TCP peak is the larger listener's, never the sum" { + var tcp6: tcp_server.TcpServer = undefined; + tcp6.core.stats = .{}; + tcp6.core.stats.connections_peak.store(4, .monotonic); + + var tcp4: tcp_server.TcpServer = undefined; + tcp4.core.stats = .{}; + tcp4.core.stats.connections_peak.store(6, .monotonic); + + try testing.expectEqual(@as(?u64, 6), largestConnectionsPeak(&.{ &tcp6, &tcp4 })); + try testing.expectEqual(@as(?u64, null), largestConnectionsPeak(&.{})); } test "the plain-DNS listener families carry every counter of both listeners" {