mokhtar/nxdns
1
0
Fork 0
Code Pull Requests Actions Packages Releases 21 Activity

milestone 30: overview as a dashboard, explicit health contract, period aggregations

Browse Source
This commit is contained in:
mokhtar
2026-08-22 16:45:15 +02:00
parent 0e83477d80
commit 623667e475
89 changed files with 7222 additions and 4239 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/upstream/pool.zig
+15 -98
View File
@@ -49,7 +49,6 @@ const std = @import("std");
const events = @import("../storage/events.zig");
const health = @import("health.zig");
const history_mod = @import("history.zig");
const safe_url = @import("../safe_url.zig");
const transport = @import("transport.zig");
@@ -90,14 +89,13 @@ pub const Entry = struct {
busy: std.Io.Mutex = .init,
};
/// A copy of one entry's health, taken under the mutex. Feeds
/// `GET /api/upstream/health`.
/// A copy of one entry's health, taken under the mutex. Feeds `/metrics` and
/// the `/api/health` upstream condition.
pub const Snapshot = struct {
/// Whole, not redacted. `GET /api/upstream/health` returns this to a session
/// that `GET /api/upstreams` already serves the same url to in full, so
/// redacting here would hide nothing from that reader and would make two
/// responses of one API disagree. A consumer reachable without a session has
/// to redact it itself.
/// Whole, not redacted. `GET /api/upstreams` already serves the same url in
/// full to a session, so redacting here would hide nothing from that reader
/// and would make two responses of one API disagree. A consumer reachable
/// without a session has to redact it itself.
url: []const u8,
enabled: bool,
available: bool,
@@ -131,14 +129,11 @@ pub const Pool = struct {
timeouts: Timeouts,
mutex: std.Io.Mutex,
rng: std.Random.DefaultPrng,
/// Where recorded outcomes also go, as per-minute aggregates for the
/// dashboard's ranged view (m26). Defaulted rather than an `init`
/// parameter: the composition root wires it after the pool exists, and the
/// pool is fully usable without it — `nxdns check` and every unit test here
/// run with no history at all.
history: ?*history_mod.Accumulator = null,
/// The diagnostics store, wired the same way and for the same reason as
/// `history`. Every emit here sits outside `mutex`; see `recordHistory`.
/// The diagnostics store. Defaulted rather than an `init` parameter: the
/// composition root wires it after the pool exists, and the pool is fully
/// usable without it — `nxdns check` and every unit test here run with no
/// store at all. Every emit here sits outside `mutex`; see
/// `recordDiagnostics`.
diagnostics: ?*events.Store = null,
pub fn init(
@@ -345,9 +340,8 @@ pub const Pool = struct {
entry.health.recordSuccess(at);
}
// The block above closes before this line, and that ordering is the
// constraint: the accumulator takes a mutex of its own, and no task may
// hold one of the two while it takes the other.
self.recordHistory(io, entry, .success);
// constraint: the store takes a mutex of its own, and no task may hold
// one of the two while it takes the other.
self.recordDiagnostics(io, entry, .success);
}
@@ -365,26 +359,13 @@ pub const Pool = struct {
}
// After the pool mutex is released, for the reason `recordSuccess`
// states.
self.recordHistory(io, entry, .{ .failure = @errorName(err) });
self.recordDiagnostics(io, entry, .{ .failure = @errorName(err) });
}
const Outcome = union(enum) { success, failure: []const u8 };
/// The wall clock, not the `.awake` timestamp the health state runs on:
/// history is aggregated into wall-clock minutes so a dashboard period
/// means the same thing here as everywhere else on the page.
fn recordHistory(self: *Pool, io: std.Io, entry: *Entry, outcome: Outcome) void {
const history = self.history orelse return;
const wall_s = std.Io.Clock.real.now(io).toSeconds();
switch (outcome) {
.success => history.recordSuccess(io, entry.endpoint.url, wall_s),
.failure => |name| history.recordFailure(io, entry.endpoint.url, wall_s, name),
}
}
/// The same placement discipline as `recordHistory`: the store takes a
/// mutex of its own, so this runs after the pool's is released.
/// The store takes a mutex of its own, so this runs after the pool's is
/// released.
///
/// A success is the steady state of the whole program, so `resolve` is
/// built to issue no SQL when nothing is open (`storage/events.zig`).
@@ -408,10 +389,7 @@ pub const Pool = struct {
}
};
const db = @import("../storage/db.zig");
const events_fixture = @import("../storage/events_fixture.zig");
const querylog_schema = @import("../storage/querylog_schema.zig");
const upstream_history_repo = @import("../storage/repositories/upstream_history_repo.zig");
const testing = std.testing;
@@ -921,67 +899,6 @@ test "every entry disabled yields ConnectFailed without waiting out the total bu
try testing.expectEqual(@as(usize, 0), two.calls);
}
test "a wired accumulator receives both outcomes the pool records" {
var threaded: std.Io.Threaded = .init(testing.allocator, .{});
defer threaded.deinit();
const io = threaded.io();
var bad: Fake = .{ .behavior = .{ .fail = error.Timeout } };
var good: Fake = .{ .behavior = .{ .reply = response_bytes } };
var entries = [_]Entry{
testEntry("https://bad.example/dns-query", &bad, 10),
testEntry("https://good.example/dns-query", &good, 20),
};
var pool: Pool = .init(&entries, test_cfg, test_timeouts, 1);
const acc = try testing.allocator.create(history_mod.Accumulator);
defer testing.allocator.destroy(acc);
acc.* = .init;
pool.history = acc;
var buf: [512]u8 = undefined;
// One exchange: the first entry fails over into the second, so this drives
// one failure and one success.
var selected: ?[]const u8 = null;
_ = try pool.exchange(io, query_bytes, &buf, &selected);
// Two cells, one per url, in whatever minute the wall clock is in.
try testing.expectEqual(@as(u32, 2), acc.snapshotStats(io).pending);
// Read back through the flush path rather than through the accumulator's
// private cells: the whole point of the hook is that these outcomes reach
// storage under the right url.
var database = try db.Db.open(":memory:", .{ .mode = .memory });
defer database.close();
try db.applyPragmas(&database, .{});
try database.exec(querylog_schema.ddl);
acc.flushOnce(io, &database, upstream_history_repo.flush);
// A window wide enough that a minute boundary crossed mid-test changes
// nothing about what it contains.
const now = std.Io.Clock.real.now(io).toSeconds();
const failing = try upstream_history_repo.windowStats(
&database,
"https://bad.example/dns-query",
now - 3600,
now + 3600,
);
try testing.expectEqual(@as(u64, 1), failing.failures);
try testing.expectEqual(@as(u64, 0), failing.successes);
try testing.expect(failing.last_failure_ts != null);
try testing.expectEqualStrings("Timeout", failing.lastFailureError());
const succeeding = try upstream_history_repo.windowStats(
&database,
"https://good.example/dns-query",
now - 3600,
now + 3600,
);
try testing.expectEqual(@as(u64, 1), succeeding.successes);
try testing.expectEqual(@as(u64, 0), succeeding.failures);
try testing.expectEqual(@as(?i64, null), succeeding.last_failure_ts);
}
test "snapshot reports the counters in pool order" {
var threaded: std.Io.Threaded = .init(testing.allocator, .{});
defer threaded.deinit();