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

milestone 26: upstream health answers for the selected period

Browse Source
This commit is contained in:
mokhtar
2026-08-17 18:21:56 +02:00
parent 3ed9a57822
commit e0a7cd8a6b
28 changed files with 2865 additions and 182 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/app.zig
+52 -12
View File
@@ -48,6 +48,7 @@ const faults = @import("config/faults.zig");
const fetcher = @import("filter/fetcher.zig");
const forward_zones = @import("local/forward_zones.zig");
const handler = @import("server/handler.zig");
const history_mod = @import("upstream/history.zig");
const http_util = @import("web/http_util.zig");
const loader = @import("config/loader.zig");
const local_records = @import("local/records.zig");
@@ -68,6 +69,7 @@ const shutdown = @import("server/shutdown.zig");
const sse = @import("web/sse.zig");
const static = @import("web/static.zig");
const tcp_server = @import("server/tcp_server.zig");
const upstream_history_repo = @import("storage/repositories/upstream_history_repo.zig");
const transport = @import("upstream/transport.zig");
const udp_server = @import("server/udp_server.zig");
const validate = @import("config/validate.zig");
@@ -448,6 +450,13 @@ fn serve(r: cli.Runner, args: cli.RunArgs) !u8 {
@truncate(@as(u96, @bitCast(std.Io.Clock.real.now(io).nanoseconds))),
);
// On the heap, not in this frame: the accumulator carries its pending cells
// and the flush task's buffer inline, which is about a megabyte.
const history = try gpa.create(history_mod.Accumulator);
defer gpa.destroy(history);
history.* = .init;
pool.history = history;
// -----------------------------------------------------------------------
// per-query state
// -----------------------------------------------------------------------
@@ -524,6 +533,8 @@ fn serve(r: cli.Runner, args: cli.RunArgs) !u8 {
defer querylog_writer_db.close();
var querylog_retention_db = try data.reopenQuerylogDb(io);
defer querylog_retention_db.close();
var querylog_history_db = try data.reopenQuerylogDb(io);
defer querylog_history_db.close();
var tracker_db = try data.openConfigDb(io);
defer tracker_db.close();
@@ -646,6 +657,7 @@ fn serve(r: cli.Runner, args: cli.RunArgs) !u8 {
.local_tables = &tables,
.logger = &query_logger,
.retention = &retention,
.history = history,
.sessions = if (sessions) |*s| s else null,
.limiter = if (web_limiter) |*l| l else null,
.hub = hub,
@@ -735,12 +747,42 @@ fn serve(r: cli.Runner, args: cli.RunArgs) !u8 {
shutdown.install(io);
// Declared after everything it borrows, so its `cancel` — which both
// requests cancellation and joins — is the first thing that runs on the way
// out (ruling 22). Nothing below this line may be released while a task
// could still touch it.
// Declared after everything it borrows, so the teardown below — whose
// `cancel` both requests cancellation and joins — is the first thing that
// runs on the way out (ruling 22). Nothing below this line may be released
// while a task could still touch it.
var group: std.Io.Group = .init;
defer group.cancel(io);
// Ruling 4's shutdown order, on the one path every exit from here takes:
// the logger sees a closed queue and drains what it holds rather than
// losing it to cancellation (ruling 22), then every task stops, and only
// then does the final flush run — with no recording task left that could
// add a cell after it.
//
// A `defer` and not straight-line code after `shutdown.wait`, because a
// `concurrent` spawn below can fail with the DNS listeners already
// serving; an orderly error teardown owes the operator the same drain a
// signal gets. The `querylog_history_db` this flush writes through is
// declared above, so its `close` runs after it.
defer {
query_logger.shutdown(io);
group.cancel(io);
history.flushOnce(io, &querylog_history_db, upstream_history_repo.flush);
}
// The gate every non-essential write consults. Reading it before the
// monitor's own task has sampled is safe: a fresh `Monitor` publishes `.ok`
// (disk_monitor.zig:63), so nothing is refused for want of a sample.
const gate: ?*disk_monitor.Monitor = &monitor;
// The writer starts before the listeners, and that order is the deferred
// drain's precondition: a listener that is already accepting queries
// enqueues log entries, and `Logger.shutdown` only closes the queue —
// someone has to be on the other end to write what it hands over. Spawned
// after the listeners, a `concurrent` failure in between would leave those
// entries with no consumer and `group.cancel` nothing to drain, which is
// exactly the loss the teardown above exists to prevent.
try group.concurrent(io, logger_mod.Logger.runWriter, .{ &query_logger, io, &querylog_writer_db, gate });
if (udp6) |*s| try group.concurrent(io, udp_server.UdpServer.serve, .{ s, io });
if (udp4) |*s| try group.concurrent(io, udp_server.UdpServer.serve, .{ s, io });
@@ -751,9 +793,10 @@ fn serve(r: cli.Runner, args: cli.RunArgs) !u8 {
if (doh_certs) |*store| try group.concurrent(io, cert_store.CertStore.watch, .{ store, io });
if (dot_certs) |*store| try group.concurrent(io, cert_store.CertStore.watch, .{ store, io });
const gate: ?*disk_monitor.Monitor = &monitor;
try group.concurrent(io, logger_mod.Logger.runWriter, .{ &query_logger, io, &querylog_writer_db, gate });
try group.concurrent(io, retention_mod.Retention.run, .{ &retention, io, &querylog_retention_db, gate });
// Ungated: a flush writes at most one row per upstream per minute, the same
// category as the query logger's own writes, which are ungated too.
try group.concurrent(io, history_mod.Accumulator.run, .{ history, io, &querylog_history_db });
try group.concurrent(io, disk_monitor.Monitor.run, .{ &monitor, io });
try group.concurrent(io, manager_mod.Manager.runScheduler, .{ &manager, io });
try group.concurrent(io, clients.Tracker.run, .{ &tracker, io, &tracker_db, gate, &client_names_resolver });
@@ -774,14 +817,11 @@ fn serve(r: cli.Runner, args: cli.RunArgs) !u8 {
});
// A canceled wait is a shutdown request too: whoever canceled this task
// wants the process to stop, and the teardown below is how it stops.
// wants the process to stop, and returning into the teardown deferred above
// is how it stops.
shutdown.wait(io) catch {};
log.info("shutting down", .{});
// Before the group is canceled, so the writer sees a closed queue and
// drains what it holds rather than losing it to cancellation (ruling 22).
query_logger.shutdown(io);
return cli.exit_ok;
}
src/cli.zig
+4 -2
View File
@@ -341,8 +341,10 @@ pub const DataDir = struct {
}
/// An additional connection to a `querylog.db` that `openQuerylogDb` has
/// already established. A running server needs two — the log writer and the
/// retention pass each own one (`retention.zig`'s contract).
/// already established. A running server needs three background ones — the
/// log writer, the retention pass and the upstream-history flush each own
/// one (`retention.zig`'s contract) — plus a fourth for the web task when
/// the web interface is enabled.
pub fn reopenQuerylogDb(self: *const DataDir, io: std.Io) !db.Db {
_ = io;
var database = try db.Db.open(self.querylog_db_path, .{ .mode = .read_write_existing });
src/storage/querylog_schema.zig
+22 -3
View File
@@ -45,6 +45,24 @@ pub const ddl: [:0]const u8 =
\\CREATE INDEX idx_query_log_ts ON query_log(timestamp);
\\CREATE INDEX idx_query_log_client ON query_log(client_ip);
\\CREATE INDEX idx_query_log_domain ON query_log(domain_id);
\\
\\CREATE TABLE upstream_targets (
\\ id INTEGER PRIMARY KEY,
\\ url TEXT NOT NULL UNIQUE -- the historical identity: config.db ids cannot cross database files
\\);
\\
\\CREATE TABLE upstream_minute (
\\ upstream_id INTEGER NOT NULL REFERENCES upstream_targets(id),
\\ minute_ts INTEGER NOT NULL,
\\ successes INTEGER NOT NULL,
\\ failures INTEGER NOT NULL,
\\ last_failure_ts INTEGER,
\\ last_error TEXT,
\\ PRIMARY KEY (upstream_id, minute_ts),
\\ CHECK (successes >= 0),
\\ CHECK (failures >= 0)
\\) WITHOUT ROWID;
\\CREATE INDEX idx_upstream_minute_ts ON upstream_minute(minute_ts);
;
/// `PRAGMA user_version` is a signed 32-bit field. Deriving the fingerprint from
@@ -215,20 +233,21 @@ test "fingerprint matches a fresh hash of the DDL" {
try testing.expectEqual(fingerprint, @as(i32, @bitCast(std.hash.Crc32.hash(ddl))));
}
test "ddl creates domains, query_log and the three indexes" {
test "ddl creates the query-log tables, the upstream-history tables and every index" {
var database = try db.Db.open(":memory:", .{ .mode = .memory });
defer database.close();
try db.applyPragmas(&database, .{});
try database.exec(ddl);
try testing.expectEqual(
@as(i64, 2),
@as(i64, 4),
try database.queryInt("SELECT count(*) FROM sqlite_schema WHERE type='table'"),
);
const objects = [_][]const u8{
"domains", "query_log",
"idx_query_log_ts", "idx_query_log_client",
"idx_query_log_domain",
"idx_query_log_domain", "upstream_targets",
"upstream_minute", "idx_upstream_minute_ts",
};
for (objects) |name| {
var stmt = try database.prepare("SELECT count(*) FROM sqlite_schema WHERE name = ?1");
src/storage/repositories/upstream_history_repo.zig
+493
View File
@@ -0,0 +1,493 @@
//! `upstream_minute` and its `upstream_targets` dimension table in
//! `querylog.db` (milestone-26 rulings 2, 4, 5).
//!
//! One row per upstream per wall-clock UTC minute that had at least one
//! attempt. Rows are additive facts: a flush adds to whatever is already there,
//! so a restart inside a minute continues that minute's row rather than
//! replacing it, and nothing here can lower a stored count.
//!
//! Identity is the url, not the `config.db` upstream id: ids cannot be foreign
//! keys across database files and may be deleted or reused. Editing an
//! upstream's url deliberately starts a new history.
//!
//! Nothing here retries. The accumulator owns what a failed flush means
//! (`upstream/history.zig`).
const std = @import("std");
const db = @import("../db.zig");
const health = @import("../../upstream/health.zig");
/// How far back `Retention` keeps minute rows. A fixed window, not a knob
/// (m26 anti-requirements), and deliberately wider than the widest dashboard
/// period: `stats.window` derives `since = until - width * count` with
/// `until > now`, so a 30-day window never asks for anything older than
/// `now - 30d`. The extra day is slack for a retention pass that runs late.
///
/// `logging.retention_days` does not apply here. It bounds the query log, whose
/// rows are per-query; these are per-minute aggregates whose whole purpose is
/// to outlive them.
pub const retention_window_s: i64 = 31 * 86_400;
/// One minute of one upstream's outcomes, as the accumulator hands it over.
/// Every string is borrowed for the duration of the call: `Stmt.bindText` binds
/// with `SQLITE_TRANSIENT`, so SQLite copies before `flush` returns.
pub const FlushRow = struct {
url: []const u8,
minute_ts: i64,
successes: u32,
failures: u32,
last_failure_ts: ?i64,
/// Empty when the minute held no failure.
last_error: []const u8,
};
const insert_target_sql = "INSERT OR IGNORE INTO upstream_targets (url) VALUES (?1)";
const select_target_sql = "SELECT id FROM upstream_targets WHERE url = ?1";
/// Additive, and the timestamp columns are max-wins, which is what makes a
/// flush safe to repeat against a row another process already wrote.
///
/// `max()` over a NULL is NULL in SQLite, so the coalesce is what keeps an
/// existing `last_failure_ts` when the incoming row carries none. The `CASE`
/// moves `last_error` with the timestamp it belongs to: a success-only upsert
/// leaves the stored failure and its name exactly as they were.
const upsert_minute_sql =
\\INSERT INTO upstream_minute (upstream_id, minute_ts, successes, failures, last_failure_ts, last_error)
\\VALUES (?1, ?2, ?3, ?4, ?5, ?6)
\\ON CONFLICT(upstream_id, minute_ts) DO UPDATE SET
\\ successes = successes + excluded.successes,
\\ failures = failures + excluded.failures,
\\ last_failure_ts = coalesce(max(last_failure_ts, excluded.last_failure_ts), last_failure_ts, excluded.last_failure_ts),
\\ last_error = CASE
\\ WHEN excluded.last_failure_ts IS NOT NULL
\\ AND (last_failure_ts IS NULL OR excluded.last_failure_ts >= last_failure_ts)
\\ THEN excluded.last_error
\\ ELSE last_error
\\ END
;
/// One transaction for the whole batch: either every minute of the pass lands
/// or none of it does, so a failed flush leaves nothing half-written for the
/// caller's merge-back to double-count.
pub fn flush(database: *db.Db, rows: []const FlushRow) db.Error!void {
if (rows.len == 0) return;
var insert_target = try database.prepare(insert_target_sql);
defer insert_target.deinit();
var select_target = try database.prepare(select_target_sql);
defer select_target.deinit();
var upsert = try database.prepare(upsert_minute_sql);
defer upsert.deinit();
var tx = try db.Tx.begin(database);
errdefer tx.rollback();
for (rows) |row| {
const upstream_id = try internTarget(&insert_target, &select_target, row.url);
try upsert.reset();
try upsert.bindInt(1, upstream_id);
try upsert.bindInt(2, row.minute_ts);
try upsert.bindInt(3, row.successes);
try upsert.bindInt(4, row.failures);
if (row.last_failure_ts) |at| try upsert.bindInt(5, at) else try upsert.bindNull(5);
try upsert.bindText(6, row.last_error);
try upsert.exec();
}
try tx.commit();
}
fn internTarget(insert: *db.Stmt, select: *db.Stmt, url: []const u8) db.Error!i64 {
try insert.reset();
try insert.bindText(1, url);
try insert.exec();
try select.reset();
try select.bindText(1, url);
// The insert above either created the row or found it already there, so a
// miss means the table changed under this connection.
if (!try select.step()) return error.NotFound;
const id = select.columnInt(0);
// A statement stopped on a row keeps its cursor open until it is reset;
// the transaction must not carry that to the next row.
try select.reset();
return id;
}
/// What `GET /api/upstream/health` reports for one upstream over one window.
pub const WindowStats = struct {
attempts: u64,
successes: u64,
failures: u64,
last_failure_ts: ?i64,
/// The error name of the row holding the newest `last_failure_ts` in the
/// window; empty when the window holds no failure.
///
/// By value rather than by slice: the caller loops over upstreams and
/// reuses one `WindowStats`, so a borrowed slice would dangle into the
/// storage the next iteration overwrites.
last_failure_error_buf: [health.error_name_capacity]u8,
last_failure_error_len: u8,
pub fn lastFailureError(self: *const WindowStats) []const u8 {
return self.last_failure_error_buf[0..self.last_failure_error_len];
}
};
/// **One statement, deliberately.** Two statements would not share a SQLite
/// snapshot: the flush connection can commit between them, and the read would
/// then pair a `max(last_failure_ts)` taken from one state with an error text
/// taken from another.
///
/// The error lookup is a scalar subquery for the same reason it is not a bare
/// column: `SELECT max(last_failure_ts), last_error` lets SQLite return the
/// `last_error` of an arbitrary row of the group. `ORDER BY ... DESC, minute_ts
/// DESC` makes the choice deterministic when two minutes share a timestamp.
const window_stats_sql =
\\SELECT coalesce(sum(m.successes), 0), coalesce(sum(m.failures), 0), max(m.last_failure_ts),
\\ (SELECT e.last_error FROM upstream_minute e
\\ WHERE e.upstream_id = m.upstream_id AND e.minute_ts >= ?2 AND e.minute_ts < ?3
\\ AND e.last_failure_ts IS NOT NULL
\\ ORDER BY e.last_failure_ts DESC, e.minute_ts DESC LIMIT 1)
\\ FROM upstream_minute m JOIN upstream_targets t ON t.id = m.upstream_id
\\ WHERE t.url = ?1 AND m.minute_ts >= ?2 AND m.minute_ts < ?3
;
/// Aggregates `[since, until)` by `minute_ts`. An unknown url or an empty
/// window is zeros, a null timestamp and the empty error — not an error.
pub fn windowStats(database: *db.Db, url: []const u8, since: i64, until: i64) db.Error!WindowStats {
var stmt = try database.prepare(window_stats_sql);
defer stmt.deinit();
try stmt.bindText(1, url);
try stmt.bindInt(2, since);
try stmt.bindInt(3, until);
// A bare aggregate always produces exactly one row; no row means the
// statement is not the one this function prepared.
if (!try stmt.step()) return error.Misuse;
const successes = try countOf(stmt.columnInt(0));
const failures = try countOf(stmt.columnInt(1));
var out: WindowStats = .{
.attempts = successes + failures,
.successes = successes,
.failures = failures,
.last_failure_ts = if (stmt.isNull(2)) null else stmt.columnInt(2),
.last_failure_error_buf = @splat(0),
.last_failure_error_len = 0,
};
const name = stmt.columnText(3);
const copied = @min(name.len, out.last_failure_error_buf.len);
@memcpy(out.last_failure_error_buf[0..copied], name[0..copied]);
out.last_failure_error_len = @intCast(copied);
return out;
}
/// `sum` over `CHECK (… >= 0)` columns cannot go negative; a negative value
/// means the row came from something other than this schema.
fn countOf(value: i64) db.Error!u64 {
if (value < 0) return error.Mismatch;
return @intCast(value);
}
/// Deletes every `upstream_minute` row strictly older than `cutoff_ts`, then
/// the targets no surviving row references, and returns how many **minute**
/// rows went.
///
/// One transaction: a target dropped without its rows, or rows dropped while
/// the target delete failed, would leave the foreign key pointing at nothing.
/// The count is minute rows only, so the metric an operator watches counts
/// aggregates rather than dimension-table housekeeping.
pub fn pruneOlderThan(database: *db.Db, cutoff_ts: i64) db.Error!i64 {
var tx = try db.Tx.begin(database);
errdefer tx.rollback();
var minutes = try database.prepare("DELETE FROM upstream_minute WHERE minute_ts < ?1");
defer minutes.deinit();
try minutes.bindInt(1, cutoff_ts);
try minutes.exec();
const deleted = database.changes();
try database.exec(
\\DELETE FROM upstream_targets
\\ WHERE id NOT IN (SELECT upstream_id FROM upstream_minute);
);
try tx.commit();
return deleted;
}
pub fn countMinutes(database: *db.Db) db.Error!i64 {
return database.queryInt("SELECT count(*) FROM upstream_minute");
}
pub fn countTargets(database: *db.Db) db.Error!i64 {
return database.queryInt("SELECT count(*) FROM upstream_targets");
}
// ---------------------------------------------------------------------------
// tests
// ---------------------------------------------------------------------------
const querylog_schema = @import("../querylog_schema.zig");
const testing = std.testing;
fn openLog() !db.Db {
var database = try db.Db.open(":memory:", .{ .mode = .memory });
errdefer database.close();
try db.applyPragmas(&database, .{});
try database.exec(querylog_schema.ddl);
return database;
}
/// `columnText` is borrowed until the statement is finalized, so the stored
/// error name is copied out rather than returned as a slice.
const StoredMinute = struct {
successes: u32,
failures: u32,
last_failure_ts: ?i64,
error_buf: [health.error_name_capacity]u8,
error_len: u8,
fn lastError(self: *const StoredMinute) []const u8 {
return self.error_buf[0..self.error_len];
}
};
fn readMinute(database: *db.Db, url: []const u8, minute_ts: i64) !StoredMinute {
var stmt = try database.prepare(
\\SELECT m.successes, m.failures, m.last_failure_ts, m.last_error
\\ FROM upstream_minute m JOIN upstream_targets t ON t.id = m.upstream_id
\\ WHERE t.url = ?1 AND m.minute_ts = ?2
);
defer stmt.deinit();
try stmt.bindText(1, url);
try stmt.bindInt(2, minute_ts);
try testing.expect(try stmt.step());
var out: StoredMinute = .{
.successes = @intCast(stmt.columnInt(0)),
.failures = @intCast(stmt.columnInt(1)),
.last_failure_ts = if (stmt.isNull(2)) null else stmt.columnInt(2),
.error_buf = @splat(0),
.error_len = 0,
};
const name = stmt.columnText(3);
@memcpy(out.error_buf[0..name.len], name);
out.error_len = @intCast(name.len);
return out;
}
test "the upsert adds to the row already there rather than replacing it" {
var database = try openLog();
defer database.close();
try flush(&database, &.{
.{ .url = "https://a.example", .minute_ts = 60, .successes = 3, .failures = 1, .last_failure_ts = 70, .last_error = "Timeout" },
});
// The shape a restart inside one minute takes: a second process writes the
// same (url, minute) and the counts continue.
try flush(&database, &.{
.{ .url = "https://a.example", .minute_ts = 60, .successes = 2, .failures = 4, .last_failure_ts = 90, .last_error = "ConnectFailed" },
});
const row = try readMinute(&database, "https://a.example", 60);
try testing.expectEqual(@as(u32, 5), row.successes);
try testing.expectEqual(@as(u32, 5), row.failures);
try testing.expectEqual(@as(?i64, 90), row.last_failure_ts);
try testing.expectEqualStrings("ConnectFailed", row.lastError());
// One row and one target, not two of either.
try testing.expectEqual(@as(i64, 1), try countMinutes(&database));
try testing.expectEqual(@as(i64, 1), try countTargets(&database));
}
test "a success-only upsert keeps the failure timestamp and its error" {
var database = try openLog();
defer database.close();
try flush(&database, &.{
.{ .url = "https://a.example", .minute_ts = 60, .successes = 0, .failures = 1, .last_failure_ts = 70, .last_error = "Timeout" },
});
// `max()` over a NULL is NULL in SQLite, so without the coalesce this
// upsert would erase the timestamp it knows nothing about.
try flush(&database, &.{
.{ .url = "https://a.example", .minute_ts = 60, .successes = 5, .failures = 0, .last_failure_ts = null, .last_error = "" },
});
const row = try readMinute(&database, "https://a.example", 60);
try testing.expectEqual(@as(u32, 5), row.successes);
try testing.expectEqual(@as(u32, 1), row.failures);
try testing.expectEqual(@as(?i64, 70), row.last_failure_ts);
try testing.expectEqualStrings("Timeout", row.lastError());
}
test "an older failure does not overwrite the newer error already stored" {
var database = try openLog();
defer database.close();
try flush(&database, &.{
.{ .url = "https://a.example", .minute_ts = 60, .successes = 0, .failures = 1, .last_failure_ts = 90, .last_error = "Timeout" },
});
try flush(&database, &.{
.{ .url = "https://a.example", .minute_ts = 60, .successes = 0, .failures = 1, .last_failure_ts = 70, .last_error = "ConnectFailed" },
});
const row = try readMinute(&database, "https://a.example", 60);
try testing.expectEqual(@as(?i64, 90), row.last_failure_ts);
try testing.expectEqualStrings("Timeout", row.lastError());
}
test "flush interns each url once and writes every minute of the batch" {
var database = try openLog();
defer database.close();
try flush(&database, &.{
.{ .url = "https://a.example", .minute_ts = 60, .successes = 1, .failures = 0, .last_failure_ts = null, .last_error = "" },
.{ .url = "https://a.example", .minute_ts = 120, .successes = 2, .failures = 0, .last_failure_ts = null, .last_error = "" },
.{ .url = "https://b.example", .minute_ts = 60, .successes = 3, .failures = 0, .last_failure_ts = null, .last_error = "" },
});
try testing.expectEqual(@as(i64, 3), try countMinutes(&database));
try testing.expectEqual(@as(i64, 2), try countTargets(&database));
// An empty batch opens no transaction: one is already open here, so a
// `BEGIN IMMEDIATE` would fail.
var tx = try db.Tx.begin(&database);
try flush(&database, &.{});
tx.rollback();
}
test "windowStats sums only the window and pairs the newest failure with its own error" {
var database = try openLog();
defer database.close();
try flush(&database, &.{
// Before the window.
.{ .url = "https://a.example", .minute_ts = 0, .successes = 9, .failures = 9, .last_failure_ts = 30, .last_error = "Outside" },
.{ .url = "https://a.example", .minute_ts = 60, .successes = 2, .failures = 1, .last_failure_ts = 100, .last_error = "ConnectFailed" },
.{ .url = "https://a.example", .minute_ts = 120, .successes = 4, .failures = 2, .last_failure_ts = 170, .last_error = "Timeout" },
// A later minute with no failure at all: the error must still come from
// the minute holding the newest `last_failure_ts`, not from this one.
.{ .url = "https://a.example", .minute_ts = 180, .successes = 1, .failures = 0, .last_failure_ts = null, .last_error = "" },
// At the exclusive end of the window.
.{ .url = "https://a.example", .minute_ts = 240, .successes = 7, .failures = 7, .last_failure_ts = 250, .last_error = "After" },
// A different upstream in the same minutes.
.{ .url = "https://b.example", .minute_ts = 120, .successes = 5, .failures = 5, .last_failure_ts = 175, .last_error = "Other" },
});
const stats = try windowStats(&database, "https://a.example", 60, 240);
try testing.expectEqual(@as(u64, 7), stats.successes);
try testing.expectEqual(@as(u64, 3), stats.failures);
try testing.expectEqual(@as(u64, 10), stats.attempts);
try testing.expectEqual(@as(?i64, 170), stats.last_failure_ts);
try testing.expectEqualStrings("Timeout", stats.lastFailureError());
}
test "two minutes sharing the newest failure timestamp resolve to the later minute" {
// The tiebreak the subquery's ORDER BY owns. `max(last_failure_ts)` alone
// cannot choose between these two rows, so without a deterministic second
// key the answer is whichever row SQLite happened to visit.
var database = try openLog();
defer database.close();
try flush(&database, &.{
.{ .url = "https://a.example", .minute_ts = 60, .successes = 0, .failures = 1, .last_failure_ts = 100, .last_error = "Earlier" },
.{ .url = "https://a.example", .minute_ts = 120, .successes = 0, .failures = 1, .last_failure_ts = 100, .last_error = "Later" },
});
const stats = try windowStats(&database, "https://a.example", 0, 1000);
try testing.expectEqual(@as(?i64, 100), stats.last_failure_ts);
try testing.expectEqualStrings("Later", stats.lastFailureError());
}
test "windowStats over an unknown url or an empty window is zeros and no error" {
var database = try openLog();
defer database.close();
try flush(&database, &.{
.{ .url = "https://a.example", .minute_ts = 60, .successes = 1, .failures = 1, .last_failure_ts = 70, .last_error = "Timeout" },
});
for ([_][3]i64{ .{ 0, 60, 0 }, .{ 120, 180, 0 }, .{ 60, 60, 0 } }) |window| {
const stats = try windowStats(&database, "https://a.example", window[0], window[1]);
try testing.expectEqual(@as(u64, 0), stats.attempts);
try testing.expectEqual(@as(u64, 0), stats.successes);
try testing.expectEqual(@as(u64, 0), stats.failures);
try testing.expectEqual(@as(?i64, null), stats.last_failure_ts);
try testing.expectEqualStrings("", stats.lastFailureError());
}
const unknown = try windowStats(&database, "https://never.example", 0, 1000);
try testing.expectEqual(@as(u64, 0), unknown.attempts);
try testing.expectEqual(@as(?i64, null), unknown.last_failure_ts);
try testing.expectEqualStrings("", unknown.lastFailureError());
}
test "a window whose only failures are outside it reports no failure at all" {
var database = try openLog();
defer database.close();
try flush(&database, &.{
.{ .url = "https://a.example", .minute_ts = 0, .successes = 0, .failures = 1, .last_failure_ts = 30, .last_error = "Timeout" },
.{ .url = "https://a.example", .minute_ts = 60, .successes = 4, .failures = 0, .last_failure_ts = null, .last_error = "" },
});
const stats = try windowStats(&database, "https://a.example", 60, 120);
try testing.expectEqual(@as(u64, 4), stats.attempts);
try testing.expectEqual(@as(?i64, null), stats.last_failure_ts);
try testing.expectEqualStrings("", stats.lastFailureError());
}
test "pruneOlderThan counts minute rows only and drops the orphaned target" {
var database = try openLog();
defer database.close();
try flush(&database, &.{
.{ .url = "https://old.example", .minute_ts = 60, .successes = 1, .failures = 0, .last_failure_ts = null, .last_error = "" },
.{ .url = "https://old.example", .minute_ts = 120, .successes = 1, .failures = 0, .last_failure_ts = null, .last_error = "" },
.{ .url = "https://kept.example", .minute_ts = 120, .successes = 1, .failures = 0, .last_failure_ts = null, .last_error = "" },
.{ .url = "https://kept.example", .minute_ts = 300, .successes = 1, .failures = 0, .last_failure_ts = null, .last_error = "" },
});
// Three minute rows go; the two target deletes must not join the count.
try testing.expectEqual(@as(i64, 3), try pruneOlderThan(&database, 300));
try testing.expectEqual(@as(i64, 1), try countMinutes(&database));
// "old.example" has nothing left, "kept.example" still does.
try testing.expectEqual(@as(i64, 1), try countTargets(&database));
// The row exactly at the cutoff stays, and a second pass finds nothing.
try testing.expectEqual(@as(i64, 0), try pruneOlderThan(&database, 300));
try testing.expectEqual(@as(i64, 1), try countMinutes(&database));
}
test "a failed prune leaves both tables as they were" {
var database = try openLog();
defer database.close();
try flush(&database, &.{
.{ .url = "https://a.example", .minute_ts = 60, .successes = 1, .failures = 0, .last_failure_ts = null, .last_error = "" },
});
try database.exec(
\\CREATE TRIGGER refuse_target_delete BEFORE DELETE ON upstream_targets
\\BEGIN SELECT RAISE(ABORT, 'refused'); END;
);
// The minute delete succeeds and the target delete does not; one
// transaction means neither survives.
try testing.expectError(error.Constraint, pruneOlderThan(&database, 300));
try testing.expectEqual(@as(i64, 1), try countMinutes(&database));
try testing.expectEqual(@as(i64, 1), try countTargets(&database));
}
test "an error name longer than the buffer is truncated, not overflowed" {
var database = try openLog();
defer database.close();
const long = "A" ** 200;
try flush(&database, &.{
.{ .url = "https://a.example", .minute_ts = 60, .successes = 0, .failures = 1, .last_failure_ts = 70, .last_error = long },
});
const stats = try windowStats(&database, "https://a.example", 60, 120);
try testing.expectEqual(@as(usize, health.error_name_capacity), stats.lastFailureError().len);
try testing.expectEqualStrings(long[0..health.error_name_capacity], stats.lastFailureError());
}
src/storage/retention.zig
+100 -1
View File
@@ -15,6 +15,7 @@ const db = @import("db.zig");
const disk_monitor = @import("disk_monitor.zig");
const model = @import("../config/model.zig");
const queries_repo = @import("repositories/queries_repo.zig");
const upstream_history_repo = @import("repositories/upstream_history_repo.zig");
const log = std.log.scoped(.retention);
@@ -31,6 +32,10 @@ pub const pass_interval_s = 86_400;
pub const Stats = struct {
passes: u64 = 0,
rows_pruned: u64 = 0,
/// Upstream-history minute rows, counted apart from `rows_pruned`: that
/// counter is the query log's, and an operator watching it must not see it
/// move because a different table was tidied.
upstream_rows_pruned: u64 = 0,
checkpoints: u64 = 0,
vacuums: u64 = 0,
/// Vacuums the disk monitor refused. The pass still pruned and
@@ -44,6 +49,7 @@ pub const Stats = struct {
const Counters = struct {
passes: std.atomic.Value(u64) = .init(0),
rows_pruned: std.atomic.Value(u64) = .init(0),
upstream_rows_pruned: std.atomic.Value(u64) = .init(0),
checkpoints: std.atomic.Value(u64) = .init(0),
vacuums: std.atomic.Value(u64) = .init(0),
vacuums_gated: std.atomic.Value(u64) = .init(0),
@@ -67,6 +73,7 @@ pub const Retention = struct {
return .{
.passes = self.counters.passes.load(.monotonic),
.rows_pruned = self.counters.rows_pruned.load(.monotonic),
.upstream_rows_pruned = self.counters.upstream_rows_pruned.load(.monotonic),
.checkpoints = self.counters.checkpoints.load(.monotonic),
.vacuums = self.counters.vacuums.load(.monotonic),
.vacuums_gated = self.counters.vacuums_gated.load(.monotonic),
@@ -97,7 +104,8 @@ pub const Retention = struct {
monitor: ?*disk_monitor.Monitor,
) void {
add(&self.counters.passes, 1);
const cutoff = std.Io.Clock.real.now(io).toSeconds() - model.retentionSeconds(self.cfg);
const now = std.Io.Clock.real.now(io).toSeconds();
const cutoff = now - model.retentionSeconds(self.cfg);
if (queries_repo.pruneOlderThan(database, cutoff)) |deleted| {
add(&self.counters.rows_pruned, @intCast(deleted));
@@ -105,6 +113,20 @@ pub const Retention = struct {
log.warn("retention prune before {d} failed: {s}", .{ cutoff, @errorName(err) });
}
// Before the vacuum-cadence logic below, which returns early on six
// passes out of seven and again whenever the monitor refuses the
// vacuum. A step placed after it would almost never run.
//
// `logging.retention_days` is not the window here: these are per-minute
// aggregates whose whole purpose is to outlive the per-query rows, so
// the window is the repository's own constant.
const history_cutoff = now - upstream_history_repo.retention_window_s;
if (upstream_history_repo.pruneOlderThan(database, history_cutoff)) |deleted| {
add(&self.counters.upstream_rows_pruned, @intCast(deleted));
} else |err| {
log.warn("upstream history prune before {d} failed: {s}", .{ history_cutoff, @errorName(err) });
}
if (queries_repo.checkpointTruncate(database)) {
add(&self.counters.checkpoints, 1);
} else |err| {
@@ -372,6 +394,83 @@ test "a failing prune counts the pass and leaves the rows alone" {
try testing.expectEqual(@as(u64, 1), retention.snapshotStats().checkpoints);
}
test "the upstream-history window is its own, and a one-day query log does not shrink it" {
var threaded: std.Io.Threaded = .init(testing.allocator, .{});
defer threaded.deinit();
const io = threaded.io();
var database = try openLog();
defer database.close();
const now = std.Io.Clock.real.now(io).toSeconds();
const day = 86_400;
try writeRows(&database, &.{ now - 2 * day, now - 60 });
try upstream_history_repo.flush(&database, &.{
.{ .url = "https://gone.example", .minute_ts = now - 32 * day, .successes = 1, .failures = 0, .last_failure_ts = null, .last_error = "" },
.{ .url = "https://kept.example", .minute_ts = now - 29 * day, .successes = 1, .failures = 0, .last_failure_ts = null, .last_error = "" },
});
// A query log kept for one day, and 30 days of upstream minutes beside it.
var retention: Retention = .init(.{ .retention_days = 1 });
retention.runOnce(io, &database, null);
const stats = retention.snapshotStats();
// One query-log row is older than one day; one minute row is older than the
// fixed 31-day upstream window. Each counter moved by its own amount.
try testing.expectEqual(@as(u64, 1), stats.rows_pruned);
try testing.expectEqual(@as(u64, 1), stats.upstream_rows_pruned);
try testing.expectEqual(@as(i64, 1), try upstream_history_repo.countMinutes(&database));
// The day-29 row is exactly what a 30-day dashboard window asks for.
const kept = try upstream_history_repo.windowStats(
&database,
"https://kept.example",
now - 30 * day,
now,
);
try testing.expectEqual(@as(u64, 1), kept.attempts);
// And the emptied target went with its rows.
try testing.expectEqual(@as(i64, 1), try upstream_history_repo.countTargets(&database));
}
test "the history prune runs on the passes where the vacuum logic returns early" {
var threaded: std.Io.Threaded = .init(testing.allocator, .{});
defer threaded.deinit();
const io = threaded.io();
var database = try openLog();
defer database.close();
const now = std.Io.Clock.real.now(io).toSeconds();
const old_minute = now - 40 * 86_400;
// The first pass: `passes_since_vacuum` is 1, so the vacuum block returns
// before it does anything. A prune placed after that block would never run
// on six passes out of seven.
try upstream_history_repo.flush(&database, &.{
.{ .url = "https://a.example", .minute_ts = old_minute, .successes = 1, .failures = 0, .last_failure_ts = null, .last_error = "" },
});
var early: Retention = .init(.{});
early.runOnce(io, &database, null);
try testing.expectEqual(@as(u64, 1), early.snapshotStats().upstream_rows_pruned);
try testing.expectEqual(@as(i64, 0), try upstream_history_repo.countMinutes(&database));
// The gated pass: the disk monitor refuses the vacuum and that branch
// returns too, and the prune still has to have happened before it.
var monitor: disk_monitor.Monitor = .init(.{}, std.Io.Dir.cwd(), ".", null);
monitor.state_raw.store(@intFromEnum(disk_monitor.State.critical), .monotonic);
var gated: Retention = .init(.{});
for (0..vacuum_every_passes - 1) |_| gated.runOnce(io, &database, &monitor);
try upstream_history_repo.flush(&database, &.{
.{ .url = "https://b.example", .minute_ts = old_minute, .successes = 1, .failures = 0, .last_failure_ts = null, .last_error = "" },
});
gated.runOnce(io, &database, &monitor);
try testing.expectEqual(@as(u64, 1), gated.snapshotStats().vacuums_gated);
try testing.expectEqual(@as(u64, 1), gated.snapshotStats().upstream_rows_pruned);
try testing.expectEqual(@as(i64, 0), try upstream_history_repo.countMinutes(&database));
}
test "the next pass retries what the failed one could not do" {
var threaded: std.Io.Threaded = .init(testing.allocator, .{});
defer threaded.deinit();
src/tests.zig
+2
View File
@@ -75,6 +75,8 @@ comptime {
_ = @import("cache/dns_cache.zig");
_ = @import("server/rate_limiter.zig");
_ = @import("storage/repositories/queries_repo.zig");
_ = @import("storage/repositories/upstream_history_repo.zig");
_ = @import("upstream/history.zig");
_ = @import("storage/logger.zig");
_ = @import("platform/statfs.zig");
_ = @import("storage/disk_monitor.zig");
src/upstream/health.zig
+7 -1
View File
@@ -45,6 +45,12 @@ pub const Config = struct {
/// `State.window`.
pub const window_len = 32;
/// Bytes kept of an `@errorName`, truncated to fit. Shared rather than repeated:
/// `history.Accumulator.Cell` and `upstream_history_repo.WindowStats` carry the
/// same name through the minute aggregates, and three buffers of three different
/// sizes would truncate one error name three ways.
pub const error_name_capacity = 48;
/// The shift is capped so `base_backoff_ms << shift` cannot run away; by then
/// `max_backoff_ms` has clamped the result many doublings ago.
const max_shift = 20;
@@ -55,7 +61,7 @@ pub const State = struct {
total_failures: u64,
last_success_at: ?std.Io.Timestamp,
last_error_at: ?std.Io.Timestamp,
last_error_buf: [48]u8,
last_error_buf: [error_name_capacity]u8,
/// Length of the `@errorName` held in `last_error_buf`, truncated to fit.
last_error_len: u8,
backoff_until: ?std.Io.Timestamp,
src/upstream/history.zig
+634
View File
@@ -0,0 +1,634 @@
//! Per-minute upstream outcome history (milestone-26 rulings 1, 3, 4).
//!
//! Outcomes are aggregated into their wall-clock UTC minute at the moment the
//! pool records them, and the aggregates are flushed to `querylog.db` once a
//! minute. **Nothing samples a lifetime counter and subtracts.** That is what
//! makes the stored numbers additive facts: a restart inside a minute adds to
//! the same row, a crash loses at most the cells that had not flushed yet, and
//! no path anywhere can produce a negative delta.
//!
//! The query path may not touch SQLite, so recording is memory-only under this
//! module's own mutex and the writing happens on a task of its own.
//!
//! Two failure modes, deliberately kept apart:
//!
//! * **Overflow.** More live `(url, minute)` pairs than `max_pending`. The
//! oldest minute is dropped, `rows_dropped` counts it and
//! `last_drop_minute` remembers how new the newest lost minute was, so a
//! window that starts after it can still be reported as complete.
//! * **A failed flush.** Nothing is dropped: the rows go back into the
//! accumulator and the next pass writes them again.
//!
//! The wall clock, not `.awake`: history participates in wall-clock periods, so
//! a minute here is the same minute the dashboard's period picker means.
//! Routing state (`health.zig`) stays on `.awake` and is untouched by this
//! file.
const std = @import("std");
const db = @import("../storage/db.zig");
const health = @import("health.zig");
const upstream_history_repo = @import("../storage/repositories/upstream_history_repo.zig");
const log = std.log.scoped(.upstream_history);
/// Live `(url, minute)` cells. Not a knob (m26 anti-requirements). A household
/// pool is a handful of upstreams, so the bound is only reachable when flushing
/// has been failing for hours.
pub const max_pending = 4096;
/// One flush pass per minute, so an unflushed cell is at most about a minute
/// old. `.boot` rather than `.awake`, for `retention.zig`'s reason: a box that
/// suspends must still see its interval elapse.
pub const flush_interval_s = 60;
/// The UTC minute `wall_s` falls in, as its start in seconds. `@divFloor`, not
/// `@divTrunc`: a negative second belongs to the minute before it.
pub fn minuteOf(wall_s: i64) i64 {
return @divFloor(wall_s, 60) * 60;
}
/// The write seam. Production passes `upstream_history_repo.flush`; a test
/// passes a stub that fails, or one that records what it was handed.
pub const WriteFn = *const fn (*db.Db, []const upstream_history_repo.FlushRow) db.Error!void;
pub const Accumulator = struct {
pub const Cell = struct {
/// Borrowed from the pool entry's endpoint, which lives as long as the
/// process. Nothing here copies it, and nothing here may outlive it.
url: []const u8,
minute_ts: i64,
/// Both counters saturate instead of wrapping: every write site uses
/// `+|=` — `recordSuccess`, `recordFailure`, and `mergeBack`, which
/// sums a failed flush's copy back into the live cell. The saturation
/// is deliberate and unreachable: a cell counts one upstream's
/// outcomes inside a single wall-clock minute, so filling a `u32`
/// would take about 72 million exchanges per second with that one
/// upstream. Nothing reports it, by design — `last_drop_minute` and
/// the `complete` flag it feeds describe capacity drops, and a
/// saturated counter is not a drop.
successes: u32,
failures: u32,
last_failure_ts: ?i64,
last_error_buf: [health.error_name_capacity]u8,
last_error_len: u8,
fn lastError(self: *const Cell) []const u8 {
return self.last_error_buf[0..self.last_error_len];
}
};
/// A consistent copy for `/metrics`, `/api/health` and the API layer.
pub const Stats = struct {
flushes: u64 = 0,
flush_failures: u64 = 0,
rows_dropped: u64 = 0,
pending: u32 = 0,
/// The newest minute capacity has ever cost this process, or null when
/// nothing was ever dropped. A window that starts after it is complete
/// again, so one historical overflow does not mark every later answer.
last_drop_minute: ?i64 = null,
/// Current state, not a count: set by a failed flush and cleared by the
/// next successful one. Feeds the `/api/health` rollup.
last_flush_failed: bool = false,
};
/// Atomic for the reason `retention.zig`'s are: the flush task writes them
/// and the web task reads them, on different threads. They are bumped
/// outside the mutex, so the lock is not what orders them.
const Counters = struct {
flushes: std.atomic.Value(u64) = .init(0),
flush_failures: std.atomic.Value(u64) = .init(0),
rows_dropped: std.atomic.Value(u64) = .init(0),
};
mutex: std.Io.Mutex = .init,
cells: [max_pending]Cell,
count: u32,
last_drop_minute: ?i64,
last_flush_failed: bool,
counters: Counters,
/// Owned by whichever task runs `flushOnce`, which is one task. It is a
/// field rather than a local so that the megabyte it costs lives wherever
/// the accumulator was placed instead of on a task's stack.
flush_cells: [max_pending]Cell,
flush_rows: [max_pending]upstream_history_repo.FlushRow,
flush_count: u32,
/// `cells` and `flush_cells` are `undefined`: a cell is always written
/// before it is read, and `count` is what says which ones exist.
pub const init: Accumulator = .{
.mutex = .init,
.cells = undefined,
.count = 0,
.last_drop_minute = null,
.last_flush_failed = false,
.counters = .{},
.flush_cells = undefined,
.flush_rows = undefined,
.flush_count = 0,
};
pub fn recordSuccess(self: *Accumulator, io: std.Io, url: []const u8, wall_s: i64) void {
// Uncancelable for the reason the pool's health sections are: this
// takes no Io and never blocks on a peer, and losing the record of a
// completed exchange to a cancellation would undercount for good.
self.mutex.lockUncancelable(io);
defer self.mutex.unlock(io);
const cell = self.cellFor(url, minuteOf(wall_s));
cell.successes +|= 1;
}
pub fn recordFailure(
self: *Accumulator,
io: std.Io,
url: []const u8,
wall_s: i64,
error_name: []const u8,
) void {
self.mutex.lockUncancelable(io);
defer self.mutex.unlock(io);
const cell = self.cellFor(url, minuteOf(wall_s));
cell.failures +|= 1;
noteFailure(cell, wall_s, error_name);
}
/// The only read surface. Every field above is private to this module, so
/// no consumer can read one of them without the mutex.
pub fn snapshotStats(self: *Accumulator, io: std.Io) Stats {
self.mutex.lockUncancelable(io);
defer self.mutex.unlock(io);
return .{
.flushes = self.counters.flushes.load(.monotonic),
.flush_failures = self.counters.flush_failures.load(.monotonic),
.rows_dropped = self.counters.rows_dropped.load(.monotonic),
.pending = self.count,
.last_drop_minute = self.last_drop_minute,
.last_flush_failed = self.last_flush_failed,
};
}
/// The cell for `(url, minute_ts)`, created if it is not there yet. The
/// caller holds the mutex.
///
/// Linear: the live set is one cell per upstream per unflushed minute,
/// which at household scale is a handful. A miss at capacity evicts the
/// oldest minute — see `evictOldest`.
fn cellFor(self: *Accumulator, url: []const u8, minute_ts: i64) *Cell {
for (self.cells[0..self.count]) |*cell| {
if (cell.minute_ts != minute_ts) continue;
// Pointer equality first: every call from the pool passes the same
// `Entry.endpoint.url`, so the byte compare is the cold path.
if (cell.url.ptr == url.ptr and cell.url.len == url.len) return cell;
if (std.mem.eql(u8, cell.url, url)) return cell;
}
const slot = if (self.count < max_pending) fresh: {
const index = self.count;
self.count += 1;
break :fresh &self.cells[index];
} else self.evictOldest();
slot.* = .{
.url = url,
.minute_ts = minute_ts,
.successes = 0,
.failures = 0,
.last_failure_ts = null,
.last_error_buf = @splat(0),
.last_error_len = 0,
};
return slot;
}
/// Frees the cell holding the oldest minute and accounts for what it cost.
///
/// `last_drop_minute` moves through `@max` and never through assignment: a
/// merge-back after a failed flush can evict a cell older than one already
/// dropped, and a watermark that moved backwards would report a window as
/// complete when outcomes inside it are gone.
fn evictOldest(self: *Accumulator) *Cell {
var oldest: usize = 0;
for (self.cells[1..self.count], 1..) |*cell, i| {
if (cell.minute_ts < self.cells[oldest].minute_ts) oldest = i;
}
const evicted = self.cells[oldest].minute_ts;
self.last_drop_minute = @max(self.last_drop_minute orelse evicted, evicted);
_ = self.counters.rows_dropped.fetchAdd(1, .monotonic);
return &self.cells[oldest];
}
/// One flush pass: swap the dirty cells out, write them, and on failure put
/// them back.
///
/// **A swap, never a subtraction.** SQLite runs outside the mutex, so while
/// it does, a full accumulator can evict a cell that was copied out and
/// then recreate the same `(url, minute_ts)`. A post-flush subtract would
/// then destroy outcomes recorded during the write. Moving the cells out
/// makes the flush own them: what is recorded beside it is new data, and a
/// failure merges the two additively.
///
/// Every failure is counted and warned about once; nothing here returns an
/// error, because there is no caller that could do anything the next pass
/// will not do anyway.
pub fn flushOnce(self: *Accumulator, io: std.Io, database: *db.Db, write: WriteFn) void {
{
self.mutex.lockUncancelable(io);
defer self.mutex.unlock(io);
if (self.count == 0) return;
@memcpy(self.flush_cells[0..self.count], self.cells[0..self.count]);
self.flush_count = self.count;
self.count = 0;
}
const rows = self.flush_rows[0..self.flush_count];
for (self.flush_cells[0..self.flush_count], rows) |*cell, *row| {
row.* = .{
.url = cell.url,
.minute_ts = cell.minute_ts,
.successes = cell.successes,
.failures = cell.failures,
.last_failure_ts = cell.last_failure_ts,
.last_error = cell.lastError(),
};
}
if (write(database, rows)) {
_ = self.counters.flushes.fetchAdd(1, .monotonic);
self.mutex.lockUncancelable(io);
defer self.mutex.unlock(io);
self.last_flush_failed = false;
return;
} else |err| {
_ = self.counters.flush_failures.fetchAdd(1, .monotonic);
log.warn("flushing {d} upstream history rows failed: {s}", .{ rows.len, @errorName(err) });
self.mergeBack(io);
}
}
/// Puts a failed pass's cells back through the rules recording uses: a cell
/// recorded during the write keeps its outcomes and the merge sums into it,
/// and a merge that overflows follows the ordinary drop policy.
fn mergeBack(self: *Accumulator, io: std.Io) void {
self.mutex.lockUncancelable(io);
defer self.mutex.unlock(io);
self.last_flush_failed = true;
for (self.flush_cells[0..self.flush_count]) |*saved| {
const cell = self.cellFor(saved.url, saved.minute_ts);
cell.successes +|= saved.successes;
cell.failures +|= saved.failures;
if (saved.last_failure_ts) |at| noteFailure(cell, at, saved.lastError());
}
}
/// Daily-loop shape (`retention.zig`): flush first, then sleep, so a
/// process that is about to be canceled has already written once.
pub fn run(self: *Accumulator, io: std.Io, database: *db.Db) std.Io.Cancelable!void {
const interval: std.Io.Clock.Duration = .{
.raw = .fromSeconds(flush_interval_s),
.clock = .boot,
};
while (true) {
self.flushOnce(io, database, upstream_history_repo.flush);
try interval.sleep(io);
}
}
};
/// Max-wins, matching the SQL upsert exactly: the newest failure in the minute
/// is the one whose name the cell keeps.
fn noteFailure(cell: *Accumulator.Cell, at: i64, error_name: []const u8) void {
if (cell.last_failure_ts) |existing| {
if (at < existing) return;
}
cell.last_failure_ts = at;
const copied = @min(error_name.len, cell.last_error_buf.len);
@memcpy(cell.last_error_buf[0..copied], error_name[0..copied]);
cell.last_error_len = @intCast(copied);
}
// ---------------------------------------------------------------------------
// tests
// ---------------------------------------------------------------------------
const querylog_schema = @import("../storage/querylog_schema.zig");
const testing = std.testing;
fn openLog() !db.Db {
var database = try db.Db.open(":memory:", .{ .mode = .memory });
errdefer database.close();
try db.applyPragmas(&database, .{});
try database.exec(querylog_schema.ddl);
return database;
}
/// The accumulator is about a megabyte, which is more than a test frame should
/// carry.
fn newAccumulator() !*Accumulator {
const acc = try testing.allocator.create(Accumulator);
acc.* = .init;
return acc;
}
fn failingWrite(_: *db.Db, _: []const upstream_history_repo.FlushRow) db.Error!void {
return error.Busy;
}
/// What the last stubbed flush was handed, copied out so an assertion can read
/// it after the pass returned.
var recorded: [8]upstream_history_repo.FlushRow = undefined;
var recorded_len: usize = 0;
fn recordingWrite(_: *db.Db, rows: []const upstream_history_repo.FlushRow) db.Error!void {
recorded_len = @min(rows.len, recorded.len);
@memcpy(recorded[0..recorded_len], rows[0..recorded_len]);
}
/// The interleaving of ruling 4: a flush is in flight, and the recording side
/// recreates a swapped-out cell and then fills the accumulator to overflow.
var interleaved: ?*Accumulator = null;
var interleave_io: ?std.Io = null;
var interleave_urls: [max_pending][8]u8 = undefined;
fn interleavingWrite(_: *db.Db, _: []const upstream_history_repo.FlushRow) db.Error!void {
const acc = interleaved.?;
const io = interleave_io.?;
// The very `(url, minute_ts)` the flush is holding, recorded again while
// the write runs.
acc.recordSuccess(io, "https://a.example", 60);
// And then enough distinct minutes to fill the accumulator and evict.
for (&interleave_urls, 0..) |*name, i| {
const url = std.fmt.bufPrint(name, "u{d:0>6}", .{i}) catch unreachable;
acc.recordSuccess(io, url, @as(i64, @intCast(i)) * 600 + 6000);
}
return error.Busy;
}
test "minuteOf floors to the minute, including before the epoch" {
try testing.expectEqual(@as(i64, 0), minuteOf(0));
try testing.expectEqual(@as(i64, 0), minuteOf(59));
try testing.expectEqual(@as(i64, 60), minuteOf(60));
try testing.expectEqual(@as(i64, 120), minuteOf(179));
try testing.expectEqual(@as(i64, -60), minuteOf(-1));
}
test "outcomes land in the cell of their own minute and upstream" {
var threaded: std.Io.Threaded = .init(testing.allocator, .{});
defer threaded.deinit();
const io = threaded.io();
const acc = try newAccumulator();
defer testing.allocator.destroy(acc);
acc.recordSuccess(io, "https://a.example", 65);
acc.recordSuccess(io, "https://a.example", 119);
acc.recordFailure(io, "https://a.example", 100, "Timeout");
acc.recordSuccess(io, "https://a.example", 130);
acc.recordSuccess(io, "https://b.example", 70);
// Three cells: a/60, a/120 and b/60.
try testing.expectEqual(@as(u32, 3), acc.snapshotStats(io).pending);
const first = acc.cellFor("https://a.example", 60);
try testing.expectEqual(@as(u32, 2), first.successes);
try testing.expectEqual(@as(u32, 1), first.failures);
try testing.expectEqual(@as(?i64, 100), first.last_failure_ts);
try testing.expectEqualStrings("Timeout", first.lastError());
const second = acc.cellFor("https://a.example", 120);
try testing.expectEqual(@as(u32, 1), second.successes);
try testing.expectEqual(@as(u32, 0), second.failures);
try testing.expectEqual(@as(?i64, null), second.last_failure_ts);
}
test "a cell keeps the newest failure's error and ignores an older one" {
var threaded: std.Io.Threaded = .init(testing.allocator, .{});
defer threaded.deinit();
const io = threaded.io();
const acc = try newAccumulator();
defer testing.allocator.destroy(acc);
acc.recordFailure(io, "https://a.example", 100, "Timeout");
acc.recordFailure(io, "https://a.example", 80, "ConnectFailed");
const cell = acc.cellFor("https://a.example", 60);
try testing.expectEqual(@as(u32, 2), cell.failures);
try testing.expectEqual(@as(?i64, 100), cell.last_failure_ts);
try testing.expectEqualStrings("Timeout", cell.lastError());
acc.recordFailure(io, "https://a.example", 110, "BadResponse");
try testing.expectEqual(@as(?i64, 110), cell.last_failure_ts);
try testing.expectEqualStrings("BadResponse", cell.lastError());
const long = "A" ** 200;
acc.recordFailure(io, "https://a.example", 115, long);
try testing.expectEqual(@as(usize, health.error_name_capacity), cell.lastError().len);
}
test "at capacity the oldest minute is dropped, counted, and the watermark only moves forward" {
var threaded: std.Io.Threaded = .init(testing.allocator, .{});
defer threaded.deinit();
const io = threaded.io();
const acc = try newAccumulator();
defer testing.allocator.destroy(acc);
var names: [max_pending][8]u8 = undefined;
for (&names, 0..) |*name, i| {
const url = std.fmt.bufPrint(name, "u{d:0>6}", .{i}) catch unreachable;
// Minute 600 is the oldest; every later cell is newer.
acc.recordSuccess(io, url, 600 + @as(i64, @intCast(i)) * 60);
}
try testing.expectEqual(@as(u32, max_pending), acc.snapshotStats(io).pending);
try testing.expectEqual(@as(u64, 0), acc.snapshotStats(io).rows_dropped);
try testing.expectEqual(@as(?i64, null), acc.snapshotStats(io).last_drop_minute);
// One more cell evicts the oldest minute and nothing else.
acc.recordSuccess(io, "https://new.example", 10_000_000);
const after = acc.snapshotStats(io);
try testing.expectEqual(@as(u32, max_pending), after.pending);
try testing.expectEqual(@as(u64, 1), after.rows_dropped);
try testing.expectEqual(@as(?i64, 600), after.last_drop_minute);
// A later eviction of an *older* minute must not move the watermark back.
acc.recordSuccess(io, "https://older.example", 120);
const back = acc.snapshotStats(io);
try testing.expectEqual(@as(u64, 2), back.rows_dropped);
try testing.expectEqual(@as(?i64, 660), back.last_drop_minute);
acc.recordSuccess(io, "https://newer.example", 20_000_000);
const forward = acc.snapshotStats(io);
try testing.expectEqual(@as(u64, 3), forward.rows_dropped);
// The minute just evicted is 120, older than the 660 already recorded.
try testing.expectEqual(@as(?i64, 660), forward.last_drop_minute);
}
test "a successful flush hands over every cell, empties the accumulator and counts once" {
var threaded: std.Io.Threaded = .init(testing.allocator, .{});
defer threaded.deinit();
const io = threaded.io();
var database = try openLog();
defer database.close();
const acc = try newAccumulator();
defer testing.allocator.destroy(acc);
acc.recordSuccess(io, "https://a.example", 65);
acc.recordFailure(io, "https://a.example", 100, "Timeout");
acc.flushOnce(io, &database, recordingWrite);
try testing.expectEqual(@as(usize, 1), recorded_len);
try testing.expectEqualStrings("https://a.example", recorded[0].url);
try testing.expectEqual(@as(i64, 60), recorded[0].minute_ts);
try testing.expectEqual(@as(u32, 1), recorded[0].successes);
try testing.expectEqual(@as(u32, 1), recorded[0].failures);
try testing.expectEqual(@as(?i64, 100), recorded[0].last_failure_ts);
try testing.expectEqualStrings("Timeout", recorded[0].last_error);
const stats = acc.snapshotStats(io);
try testing.expectEqual(@as(u32, 0), stats.pending);
try testing.expectEqual(@as(u64, 1), stats.flushes);
try testing.expectEqual(@as(u64, 0), stats.flush_failures);
try testing.expect(!stats.last_flush_failed);
}
test "a pass with nothing pending counts neither a flush nor a failure" {
var threaded: std.Io.Threaded = .init(testing.allocator, .{});
defer threaded.deinit();
const io = threaded.io();
var database = try openLog();
defer database.close();
const acc = try newAccumulator();
defer testing.allocator.destroy(acc);
acc.flushOnce(io, &database, failingWrite);
acc.flushOnce(io, &database, recordingWrite);
const stats = acc.snapshotStats(io);
try testing.expectEqual(@as(u64, 0), stats.flushes);
try testing.expectEqual(@as(u64, 0), stats.flush_failures);
try testing.expect(!stats.last_flush_failed);
}
test "a failed flush keeps the rows, sets the flag, and the next success clears it" {
var threaded: std.Io.Threaded = .init(testing.allocator, .{});
defer threaded.deinit();
const io = threaded.io();
var database = try openLog();
defer database.close();
const acc = try newAccumulator();
defer testing.allocator.destroy(acc);
acc.recordSuccess(io, "https://a.example", 65);
acc.recordFailure(io, "https://a.example", 100, "Timeout");
acc.flushOnce(io, &database, failingWrite);
const failed = acc.snapshotStats(io);
try testing.expectEqual(@as(u64, 0), failed.flushes);
try testing.expectEqual(@as(u64, 1), failed.flush_failures);
try testing.expectEqual(@as(u64, 0), failed.rows_dropped);
try testing.expect(failed.last_flush_failed);
// Nothing was lost: the cell is back, whole.
try testing.expectEqual(@as(u32, 1), failed.pending);
const cell = acc.cellFor("https://a.example", 60);
try testing.expectEqual(@as(u32, 1), cell.successes);
try testing.expectEqual(@as(u32, 1), cell.failures);
try testing.expectEqualStrings("Timeout", cell.lastError());
// The retry writes what the failed pass could not.
acc.recordSuccess(io, "https://a.example", 70);
acc.flushOnce(io, &database, recordingWrite);
const cleared = acc.snapshotStats(io);
try testing.expectEqual(@as(u64, 1), cleared.flushes);
try testing.expectEqual(@as(u64, 1), cleared.flush_failures);
try testing.expect(!cleared.last_flush_failed);
try testing.expectEqual(@as(usize, 1), recorded_len);
try testing.expectEqual(@as(u32, 2), recorded[0].successes);
try testing.expectEqual(@as(u32, 1), recorded[0].failures);
}
test "a merge-back sums into what was recorded beside the flush" {
var threaded: std.Io.Threaded = .init(testing.allocator, .{});
defer threaded.deinit();
const io = threaded.io();
var database = try openLog();
defer database.close();
const acc = try newAccumulator();
defer testing.allocator.destroy(acc);
// The write recreates the swapped-out cell before it fails, which is the
// interleaving the swap makes possible.
interleaved = acc;
interleave_io = io;
defer {
interleaved = null;
interleave_io = null;
}
acc.recordSuccess(io, "https://a.example", 65);
acc.recordFailure(io, "https://a.example", 100, "Timeout");
acc.flushOnce(io, &database, interleavingWrite);
const stats = acc.snapshotStats(io);
try testing.expect(stats.last_flush_failed);
try testing.expectEqual(@as(u64, 1), stats.flush_failures);
// Two drops, and they are different drops: filling to capacity during the
// write evicted the recreated minute 60, and the merge-back then found no
// room either and evicted the oldest of what the write had left, 6000.
// Overflow loss is the specced policy; what matters is that it is counted.
try testing.expectEqual(@as(u32, max_pending), stats.pending);
try testing.expectEqual(@as(u64, 2), stats.rows_dropped);
// Forward only: the second eviction was the newer minute of the two.
try testing.expectEqual(@as(?i64, 6000), stats.last_drop_minute);
// The merged cell is back, carrying what the failed flush was holding.
const merged = acc.cellFor("https://a.example", 60);
try testing.expectEqual(@as(u32, 1), merged.successes);
try testing.expectEqual(@as(u32, 1), merged.failures);
try testing.expectEqualStrings("Timeout", merged.lastError());
// The flush-owned buffer was not touched by any of the recording that
// happened beside it: it still holds exactly what was swapped out.
try testing.expectEqual(@as(u32, 1), acc.flush_count);
try testing.expectEqual(@as(u32, 1), acc.flush_cells[0].successes);
try testing.expectEqual(@as(u32, 1), acc.flush_cells[0].failures);
try testing.expectEqual(@as(i64, 60), acc.flush_cells[0].minute_ts);
}
test "a flush against the real repository writes the minute rows" {
var threaded: std.Io.Threaded = .init(testing.allocator, .{});
defer threaded.deinit();
const io = threaded.io();
var database = try openLog();
defer database.close();
const acc = try newAccumulator();
defer testing.allocator.destroy(acc);
acc.recordSuccess(io, "https://a.example", 65);
acc.recordFailure(io, "https://a.example", 100, "Timeout");
acc.recordSuccess(io, "https://b.example", 130);
acc.flushOnce(io, &database, upstream_history_repo.flush);
// The second pass is the restart shape: the same minute, added to.
acc.recordSuccess(io, "https://a.example", 90);
acc.flushOnce(io, &database, upstream_history_repo.flush);
const stats = try upstream_history_repo.windowStats(&database, "https://a.example", 0, 200);
try testing.expectEqual(@as(u64, 2), stats.successes);
try testing.expectEqual(@as(u64, 1), stats.failures);
try testing.expectEqual(@as(u64, 3), stats.attempts);
try testing.expectEqual(@as(?i64, 100), stats.last_failure_ts);
try testing.expectEqualStrings("Timeout", stats.lastFailureError());
try testing.expectEqual(@as(i64, 2), try upstream_history_repo.countMinutes(&database));
try testing.expectEqual(@as(u64, 2), acc.snapshotStats(io).flushes);
}
src/upstream/pool.zig
+106 -9
View File
@@ -48,6 +48,7 @@
const std = @import("std");
const health = @import("health.zig");
const history_mod = @import("history.zig");
const safe_url = @import("../safe_url.zig");
const transport = @import("transport.zig");
@@ -129,6 +130,12 @@ 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,
pub fn init(
entries: []Entry,
@@ -307,12 +314,19 @@ pub const Pool = struct {
}
fn recordSuccess(self: *Pool, io: std.Io, entry: *Entry, at: std.Io.Timestamp) void {
// Uncancelable: this section takes no Io and never blocks on a peer.
// Losing the bookkeeping for a completed exchange to a cancellation
// that arrives one instruction later would corrupt health for good.
self.mutex.lockUncancelable(io);
defer self.mutex.unlock(io);
entry.health.recordSuccess(at);
{
// Uncancelable: this section takes no Io and never blocks on a
// peer. Losing the bookkeeping for a completed exchange to a
// cancellation that arrives one instruction later would corrupt
// health for good.
self.mutex.lockUncancelable(io);
defer self.mutex.unlock(io);
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);
}
fn recordFailure(
@@ -322,12 +336,35 @@ pub const Pool = struct {
at: std.Io.Timestamp,
err: transport.ExchangeError,
) void {
self.mutex.lockUncancelable(io);
defer self.mutex.unlock(io);
entry.health.recordFailure(at, @errorName(err), self.cfg, self.rng.random().int(u32));
{
self.mutex.lockUncancelable(io);
defer self.mutex.unlock(io);
entry.health.recordFailure(at, @errorName(err), self.cfg, self.rng.random().int(u32));
}
// After the pool mutex is released, for the reason `recordSuccess`
// states.
self.recordHistory(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),
}
}
};
const db = @import("../storage/db.zig");
const querylog_schema = @import("../storage/querylog_schema.zig");
const upstream_history_repo = @import("../storage/repositories/upstream_history_repo.zig");
const testing = std.testing;
/// A query for example.com A: id 0x1234, RD set, one question.
@@ -710,6 +747,66 @@ 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.
_ = try pool.exchange(io, query_bytes, &buf);
// 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();
src/web/handlers/health.zig
+70 -4
View File
@@ -52,6 +52,15 @@ pub const Input = struct {
upstreams_total: u32 = 0,
queries_dropped: u64 = 0,
writer_failed: bool = false,
/// Current state, not a count: the upstream-history flush is failing right
/// now. Cleared by the next flush that succeeds (m26 ruling 7).
///
/// `rows_dropped` deliberately does not appear here. It is cumulative, and
/// a rollup that is computed statelessly cannot ask whether a counter grew
/// — so feeding it in would latch `/api/health` to degraded forever after
/// one overflow. Drops surface through the metric and through the API's
/// per-window `complete` instead.
history_flush_failing: bool = false,
refreshes_gated: u64 = 0,
snapshot_generation: ?u64 = null,
};
@@ -59,11 +68,15 @@ pub const Input = struct {
pub const status_ok = "ok";
pub const status_degraded = "degraded";
/// Ruling 22's three conditions. Each one is something an operator must act on:
/// a disk that is filling stops the query log, a pool with nothing available
/// stops resolution, and a failed writer means rows are being lost right now.
/// Conditions an operator must act on, and every one of them is a fact about
/// now rather than a count of the past: a disk that is filling stops the query
/// log, a pool with nothing available stops resolution, a failed writer means
/// rows are being lost right now, and a failing history flush means the
/// dashboard's upstream numbers are not being recorded. Each clears itself when
/// the underlying condition does.
pub fn degraded(input: Input) bool {
return input.disk_state != .ok or input.upstreams_available == 0 or input.writer_failed;
return input.disk_state != .ok or input.upstreams_available == 0 or
input.writer_failed or input.history_flush_failing;
}
pub fn rollup(input: Input) Body {
@@ -115,6 +128,10 @@ pub fn collect(state: *server.WebState, io: std.Io) Input {
input.writer_failed = logger.writer_failed.load(.monotonic);
}
if (state.history) |history| {
input.history_flush_failing = history.snapshotStats(io).last_flush_failed;
}
if (state.manager) |manager| {
input.refreshes_gated = manager.refreshesGated();
if (manager.acquire(io)) |acquired| {
@@ -130,7 +147,10 @@ pub fn collect(state: *server.WebState, io: std.Io) Input {
// tests
// ---------------------------------------------------------------------------
const db = @import("../../storage/db.zig");
const history_mod = @import("../../upstream/history.zig");
const logger_mod = @import("../../storage/logger.zig");
const upstream_history_repo = @import("../../storage/repositories/upstream_history_repo.zig");
const testing = std.testing;
/// A box with nothing wrong with it: one upstream up, disk ok, writer alive.
@@ -148,6 +168,10 @@ test "the degraded matrix covers disk state, availability and the writer" {
.{ .input = withDisk(healthy, .critical), .degraded = true },
.{ .input = withAvailable(healthy, 0), .degraded = true },
.{ .input = withWriterFailed(healthy), .degraded = true },
// A failing upstream-history flush is losing the dashboard's numbers
// right now, and it recovers on its own the moment a flush succeeds.
.{ .input = withHistoryFailing(healthy, true), .degraded = true },
.{ .input = withHistoryFailing(healthy, false), .degraded = false },
// Two faults at once still report one status.
.{ .input = withWriterFailed(withDisk(healthy, .critical)), .degraded = true },
// Some upstreams down is not degraded while one still answers.
@@ -182,6 +206,48 @@ fn withWriterFailed(input: Input) Input {
return out;
}
fn withHistoryFailing(input: Input, failing: bool) Input {
var out = input;
out.history_flush_failing = failing;
return out;
}
test "a history overflow that already happened does not degrade the rollup" {
// `rows_dropped` is cumulative and the rollup is stateless, so the only
// thing it could do with a drop count is latch on it. The accumulator's
// drops reach an operator through `/metrics` and through the per-window
// `complete` flag, and never through this.
const dropped: Input = .{
.upstreams_available = 1,
.upstreams_total = 1,
.history_flush_failing = false,
};
try testing.expect(!degraded(dropped));
try testing.expectEqualStrings(status_ok, rollup(dropped).status);
}
test "collect reads the accumulator's current flush state" {
var threaded: std.Io.Threaded = .init(testing.allocator, .{});
defer threaded.deinit();
const io = threaded.io();
const acc = try testing.allocator.create(history_mod.Accumulator);
defer testing.allocator.destroy(acc);
acc.* = .init;
var state: server.WebState = .{ .gpa = testing.allocator, .history = acc };
try testing.expect(!collect(&state, io).history_flush_failing);
var database = try db.Db.open(":memory:", .{ .mode = .memory });
defer database.close();
acc.recordSuccess(io, "https://a.example", 60);
// No schema in this database, so the real write fails and the flag is set
// by the production path rather than by a test poking a field.
acc.flushOnce(io, &database, upstream_history_repo.flush);
try testing.expect(collect(&state, io).history_flush_failing);
try testing.expectEqualStrings("degraded", rollup(collect(&state, io)).status);
}
test "the body reports every input verbatim" {
const body = rollup(.{
.disk_state = .warn,
src/web/handlers/upstream_health.zig
+479 -70
View File
@@ -1,52 +1,116 @@
//! `GET /api/upstream/health` — the pool's own view of its upstreams.
//! `GET /api/upstream/health?period=` — the pool's upstreams over the window
//! the dashboard's period picker selected (milestone-26 ruling 6).
//!
//! The rows are `Pool.Snapshot` with the borrowed strings copied. `last_error`
//! points into the entry that produced it and is rewritten by that entry's next
//! failure, so it is duplicated into the request arena before the pool's mutex
//! is out of sight.
//! Two kinds of fact, kept apart on the wire because they answer different
//! questions. `enabled`/`available` are live routing state, read from the pool
//! under its mutex: what the resolver would do with this upstream right now.
//! Everything under `period` is history, aggregated out of `upstream_minute`
//! over `[since, until)` — the same window `/api/stats` reports, so a page
//! cannot show a rate that disagrees with the chart beside it.
//!
//! No timestamps: the health fields are stamped on the `awake` clock, which
//! stops while the box is suspended and means nothing to a client reading wall
//! time. What an operator needs — is it up, how often does it fail, what did it
//! say last — is here without them.
//! Nothing here reads a process-lifetime counter. The lifetime totals, the
//! last-32-exchange window and the consecutive-failure count still live in
//! `health.State` for routing and in `/metrics`; they are not this response's
//! business, because a number that starts at process start cannot be scoped to
//! a period and a dashboard that shows one beside a picker lies about it.
//!
//! The aggregation runs on the web task's own query-log connection (m7 ruling
//! 21) and this file owns no SQL: `upstream_history_repo` does.
const std = @import("std");
const Allocator = std.mem.Allocator;
const db = @import("../../storage/db.zig");
const history_mod = @import("../../upstream/history.zig");
const http_util = @import("../http_util.zig");
const metrics = @import("../metrics.zig");
const pool_mod = @import("../../upstream/pool.zig");
const server = @import("../server.zig");
const stats = @import("stats.zig");
const upstream_history_repo = @import("../../storage/repositories/upstream_history_repo.zig");
const log = std.log.scoped(.web_upstream_health);
/// One upstream's outcomes inside the selected window.
pub const PeriodStats = struct {
attempts: u64,
successes: u64,
failures: u64,
/// Null when `attempts == 0`. No observations is not perfect reliability,
/// and a `100.0%` from an idle upstream is the exact misreading this
/// milestone exists to remove.
success_rate: ?f32,
/// The newest failure inside the window, on the wall clock the minute rows
/// are stamped with. Null when the window holds no failure, even if the
/// upstream failed before it.
last_failure_at: ?i64,
/// The error name belonging to `last_failure_at`; null exactly when it is.
last_failure_error: ?[]const u8,
};
pub const Upstream = struct {
url: []const u8,
/// Live: configuration, not history.
enabled: bool,
/// Live: false while the upstream is backing off.
available: bool,
consecutive_failures: u32,
total_successes: u64,
total_failures: u64,
success_rate: f32,
/// "" when the upstream has never failed.
last_error: []const u8,
period: PeriodStats,
};
pub const Body = struct {
upstreams: []const Upstream,
period: []const u8,
since: i64,
until: i64,
available: u32,
total: u32,
/// See `isComplete`.
complete: bool,
upstreams: []const Upstream,
};
/// The same text `/api/stats` sends (`stats.zig`'s `badPeriod`). One period
/// grammar serves the whole dashboard, so the two routes must not disagree
/// about what a typo means.
pub const bad_period_message = "period must be one of 1h, 24h, 7d, 30d";
pub fn handle(
state: *server.WebState,
io: std.Io,
request: *http_util.Request,
) http_util.HandlerError!void {
const period = stats.periodParam(request.query) catch
return http_util.respondError(request, .bad_request, bad_period_message);
const pool = state.pool orelse
return http_util.respondError(request, .service_unavailable, "no upstream pool");
return http_util.respondJson(request, .ok, try collect(pool, io, request.arena), &.{});
const database = state.querylog_db orelse
return http_util.respondError(request, .service_unavailable, "query log unavailable");
const now = std.Io.Clock.real.now(io).toSeconds();
const body = collect(request.arena, io, pool, state.history, database, period, now) catch |err| {
if (err == error.OutOfMemory) return error.OutOfMemory;
// A failed aggregate is a fault in the box, not a property of the
// request (ruling 8, PLAN §19).
log.warn("upstream health window failed: {s}", .{@errorName(err)});
return http_util.respondError(request, .internal_server_error, "internal error");
};
return http_util.respondJson(request, .ok, body, &.{});
}
pub fn collect(pool: *pool_mod.Pool, io: std.Io, arena: Allocator) Allocator.Error!Body {
/// `db.Error` already carries `OutOfMemory`, so the arena's failures and
/// SQLite's share one set.
pub const Error = Allocator.Error || db.Error;
pub fn collect(
arena: Allocator,
io: std.Io,
pool: *pool_mod.Pool,
history: ?*history_mod.Accumulator,
database: *db.Db,
period: stats.Period,
now_unix: i64,
) Error!Body {
const span = stats.window(period, now_unix);
var raw: [metrics.max_upstreams]pool_mod.Snapshot = undefined;
const count = metrics.poolSnapshot(pool, io, &raw);
@@ -54,25 +118,76 @@ pub fn collect(pool: *pool_mod.Pool, io: std.Io, arena: Allocator) Allocator.Err
var available: u32 = 0;
for (raw[0..count], out) |entry, *slot| {
if (entry.available) available += 1;
// Rows come from the current pool only: an upstream deleted from the
// configuration keeps its history in storage until retention takes it,
// and nothing joins it back into this response.
const window_stats = try upstream_history_repo.windowStats(
database,
entry.url,
span.since,
span.until,
);
slot.* = .{
.url = try arena.dupe(u8, entry.url),
.enabled = entry.enabled,
.available = entry.available,
.consecutive_failures = entry.consecutive_failures,
.total_successes = entry.total_successes,
.total_failures = entry.total_failures,
.success_rate = entry.success_rate,
.last_error = try arena.dupe(u8, entry.last_error),
.period = .{
.attempts = window_stats.attempts,
.successes = window_stats.successes,
.failures = window_stats.failures,
.success_rate = successRate(window_stats),
.last_failure_at = window_stats.last_failure_ts,
// `WindowStats` carries its error name by value, in storage this
// loop is done with as soon as the iteration ends. The copy into
// the arena is what keeps the response from pointing at bytes
// the next upstream's row overwrites.
.last_failure_error = if (window_stats.last_failure_ts == null)
null
else
try arena.dupe(u8, window_stats.lastFailureError()),
},
};
}
return .{ .upstreams = out, .available = available, .total = @intCast(count) };
return .{
.period = period.label(),
.since = span.since,
.until = span.until,
.available = available,
.total = @intCast(count),
.complete = isComplete(history, io, span.since),
.upstreams = out,
};
}
fn successRate(window_stats: upstream_history_repo.WindowStats) ?f32 {
if (window_stats.attempts == 0) return null;
const successes: f32 = @floatFromInt(window_stats.successes);
const attempts: f32 = @floatFromInt(window_stats.attempts);
return successes / attempts;
}
/// Per-window and stateless (ruling 6): false iff capacity has cost this
/// process a minute that falls inside the window. A window that starts after
/// the newest such minute is complete again, so one historical overflow does
/// not mark every later response.
///
/// It says nothing about the newest outcomes, which may not have flushed yet,
/// and nothing about an unclean shutdown, which is not detectable here — the
/// openapi description spells both out.
fn isComplete(history: ?*history_mod.Accumulator, io: std.Io, since: i64) bool {
const accumulator = history orelse return true;
const dropped = accumulator.snapshotStats(io).last_drop_minute orelse return true;
return dropped < since;
}
// ---------------------------------------------------------------------------
// tests
// ---------------------------------------------------------------------------
const querylog_schema = @import("../../storage/querylog_schema.zig");
const transport = @import("../../upstream/transport.zig");
const testing = std.testing;
@@ -94,83 +209,377 @@ fn testPool(entries: []pool_mod.Entry) pool_mod.Pool {
}, 1);
}
test "every upstream is copied, counted and owned by the arena" {
fn openLog() !db.Db {
var database = try db.Db.open(":memory:", .{ .mode = .memory });
errdefer database.close();
try db.applyPragmas(&database, .{});
try database.exec(querylog_schema.ddl);
return database;
}
const url_a = "https://a.test/dns-query";
const url_b = "https://b.test/dns-query";
/// A minute-aligned instant, so a window derived from it lands on round
/// numbers the assertions below can name.
const aligned_now: i64 = 1_699_999_980;
test "the window sums the minutes inside it and nothing outside" {
var threaded: std.Io.Threaded = .init(testing.allocator, .{});
defer threaded.deinit();
const io = threaded.io();
var entries = [_]pool_mod.Entry{
testEntry("https://a.test/dns-query", true),
testEntry("https://b.test/dns-query", false),
};
var database = try openLog();
defer database.close();
const span = stats.window(.@"1h", aligned_now);
try upstream_history_repo.flush(&database, &.{
// One minute before the window.
.{ .url = url_a, .minute_ts = span.since - 60, .successes = 100, .failures = 100, .last_failure_ts = span.since - 30, .last_error = "Outside" },
.{ .url = url_a, .minute_ts = span.since, .successes = 3, .failures = 1, .last_failure_ts = span.since + 10, .last_error = "Timeout" },
.{ .url = url_a, .minute_ts = span.until - 60, .successes = 5, .failures = 0, .last_failure_ts = null, .last_error = "" },
// The window's exclusive end.
.{ .url = url_a, .minute_ts = span.until, .successes = 200, .failures = 200, .last_failure_ts = span.until + 5, .last_error = "After" },
});
var entries = [_]pool_mod.Entry{testEntry(url_a, true)};
var pool = testPool(&entries);
var arena: std.heap.ArenaAllocator = .init(testing.allocator);
defer arena.deinit();
const body = try collect(&pool, io, arena.allocator());
try testing.expectEqual(@as(u32, 2), body.total);
const body = try collect(arena.allocator(), io, &pool, null, &database, .@"1h", aligned_now);
try testing.expectEqualStrings("1h", body.period);
try testing.expectEqual(span.since, body.since);
try testing.expectEqual(span.until, body.until);
try testing.expectEqual(@as(u32, 1), body.total);
try testing.expectEqual(@as(u32, 1), body.available);
const period = body.upstreams[0].period;
try testing.expectEqual(@as(u64, 8), period.successes);
try testing.expectEqual(@as(u64, 1), period.failures);
try testing.expectEqual(@as(u64, 9), period.attempts);
try testing.expectEqual(@as(?i64, span.since + 10), period.last_failure_at);
try testing.expectEqualStrings("Timeout", period.last_failure_error.?);
}
test "an upstream with no attempts in the window reports null, never a perfect rate" {
var threaded: std.Io.Threaded = .init(testing.allocator, .{});
defer threaded.deinit();
const io = threaded.io();
var database = try openLog();
defer database.close();
const span = stats.window(.@"1h", aligned_now);
// Only `b` has history, and only outside the window.
try upstream_history_repo.flush(&database, &.{
.{ .url = url_b, .minute_ts = span.since - 600, .successes = 4, .failures = 0, .last_failure_ts = null, .last_error = "" },
});
var entries = [_]pool_mod.Entry{ testEntry(url_a, true), testEntry(url_b, true) };
var pool = testPool(&entries);
var arena: std.heap.ArenaAllocator = .init(testing.allocator);
defer arena.deinit();
const body = try collect(arena.allocator(), io, &pool, null, &database, .@"1h", aligned_now);
for (body.upstreams) |upstream| {
try testing.expectEqual(@as(u64, 0), upstream.period.attempts);
try testing.expectEqual(@as(u64, 0), upstream.period.successes);
try testing.expectEqual(@as(u64, 0), upstream.period.failures);
try testing.expectEqual(@as(?f32, null), upstream.period.success_rate);
try testing.expectEqual(@as(?i64, null), upstream.period.last_failure_at);
try testing.expectEqual(@as(?[]const u8, null), upstream.period.last_failure_error);
}
}
test "the success rate is the window's own, not a lifetime one" {
var threaded: std.Io.Threaded = .init(testing.allocator, .{});
defer threaded.deinit();
const io = threaded.io();
var database = try openLog();
defer database.close();
const span = stats.window(.@"1h", aligned_now);
try upstream_history_repo.flush(&database, &.{
// A clean past that a lifetime rate would average into the window.
.{ .url = url_a, .minute_ts = span.since - 600, .successes = 1000, .failures = 0, .last_failure_ts = null, .last_error = "" },
.{ .url = url_a, .minute_ts = span.since, .successes = 1, .failures = 3, .last_failure_ts = span.since + 1, .last_error = "Timeout" },
});
var entries = [_]pool_mod.Entry{testEntry(url_a, true)};
var pool = testPool(&entries);
var arena: std.heap.ArenaAllocator = .init(testing.allocator);
defer arena.deinit();
const body = try collect(arena.allocator(), io, &pool, null, &database, .@"1h", aligned_now);
try testing.expectEqual(@as(?f32, 0.25), body.upstreams[0].period.success_rate);
}
test "the newest failure inside the window wins over an older one outside it" {
var threaded: std.Io.Threaded = .init(testing.allocator, .{});
defer threaded.deinit();
const io = threaded.io();
var database = try openLog();
defer database.close();
const span = stats.window(.@"1h", aligned_now);
try upstream_history_repo.flush(&database, &.{
.{ .url = url_a, .minute_ts = span.since - 120, .successes = 0, .failures = 1, .last_failure_ts = span.since - 100, .last_error = "Older" },
.{ .url = url_a, .minute_ts = span.since, .successes = 0, .failures = 1, .last_failure_ts = span.since + 5, .last_error = "Newer" },
// A later minute with no failure at all must not blank the error.
.{ .url = url_a, .minute_ts = span.since + 60, .successes = 2, .failures = 0, .last_failure_ts = null, .last_error = "" },
});
var entries = [_]pool_mod.Entry{testEntry(url_a, true)};
var pool = testPool(&entries);
var arena: std.heap.ArenaAllocator = .init(testing.allocator);
defer arena.deinit();
const body = try collect(arena.allocator(), io, &pool, null, &database, .@"1h", aligned_now);
try testing.expectEqual(@as(?i64, span.since + 5), body.upstreams[0].period.last_failure_at);
try testing.expectEqualStrings("Newer", body.upstreams[0].period.last_failure_error.?);
}
test "two upstreams keep their own last-failure errors" {
// The by-value `WindowStats` buffer is reused per iteration, so a response
// that borrowed it would show the second upstream's error on the first, or
// point at stack storage that is gone by the time it is serialized.
var threaded: std.Io.Threaded = .init(testing.allocator, .{});
defer threaded.deinit();
const io = threaded.io();
var database = try openLog();
defer database.close();
const span = stats.window(.@"1h", aligned_now);
try upstream_history_repo.flush(&database, &.{
.{ .url = url_a, .minute_ts = span.since, .successes = 1, .failures = 1, .last_failure_ts = span.since + 1, .last_error = "ConnectFailed" },
.{ .url = url_b, .minute_ts = span.since, .successes = 0, .failures = 2, .last_failure_ts = span.since + 2, .last_error = "TlsHandshakeFailed" },
});
var entries = [_]pool_mod.Entry{ testEntry(url_a, true), testEntry(url_b, true) };
var pool = testPool(&entries);
var arena: std.heap.ArenaAllocator = .init(testing.allocator);
defer arena.deinit();
const body = try collect(arena.allocator(), io, &pool, null, &database, .@"1h", aligned_now);
try testing.expectEqual(@as(usize, 2), body.upstreams.len);
try testing.expectEqualStrings("https://a.test/dns-query", body.upstreams[0].url);
try testing.expect(body.upstreams[0].enabled);
try testing.expect(body.upstreams[0].available);
try testing.expectEqualStrings(url_a, body.upstreams[0].url);
try testing.expectEqualStrings("ConnectFailed", body.upstreams[0].period.last_failure_error.?);
try testing.expectEqualStrings(url_b, body.upstreams[1].url);
try testing.expectEqualStrings("TlsHandshakeFailed", body.upstreams[1].period.last_failure_error.?);
// Serializing after every row is read is what a real response does; the
// texts must still be the ones their own rows carried.
var allocating: std.Io.Writer.Allocating = .init(testing.allocator);
defer allocating.deinit();
try std.json.Stringify.value(body, .{}, &allocating.writer);
const text = allocating.written();
try testing.expect(std.mem.containsAtLeast(u8, text, 1, "\"last_failure_error\":\"ConnectFailed\""));
try testing.expect(std.mem.containsAtLeast(u8, text, 1, "\"last_failure_error\":\"TlsHandshakeFailed\""));
}
test "a disabled upstream is not counted available and still gets its window" {
var threaded: std.Io.Threaded = .init(testing.allocator, .{});
defer threaded.deinit();
const io = threaded.io();
var database = try openLog();
defer database.close();
const span = stats.window(.@"24h", aligned_now);
try upstream_history_repo.flush(&database, &.{
.{ .url = url_b, .minute_ts = span.since, .successes = 2, .failures = 0, .last_failure_ts = null, .last_error = "" },
});
var entries = [_]pool_mod.Entry{ testEntry(url_a, true), testEntry(url_b, false) };
var pool = testPool(&entries);
var arena: std.heap.ArenaAllocator = .init(testing.allocator);
defer arena.deinit();
const body = try collect(arena.allocator(), io, &pool, null, &database, .@"24h", aligned_now);
try testing.expectEqual(@as(u32, 2), body.total);
try testing.expectEqual(@as(u32, 1), body.available);
try testing.expect(!body.upstreams[1].enabled);
try testing.expect(!body.upstreams[1].available);
// A disabled upstream is not available, so it is not counted.
try testing.expectEqual(@as(u32, 1), body.available);
try testing.expectEqualStrings("", body.upstreams[0].last_error);
try testing.expectEqual(@as(u64, 2), body.upstreams[1].period.attempts);
}
test "the copied strings survive the entry they came from" {
/// Fills the accumulator and then overflows it, so `last_drop_minute` is
/// `minute` — the only way to set it, because the accumulator's fields are
/// private to its module and `snapshotStats` is the read surface.
fn accumulatorDroppingAt(
io: std.Io,
minute: i64,
names: *[history_mod.max_pending][8]u8,
) !*history_mod.Accumulator {
const accumulator = try testing.allocator.create(history_mod.Accumulator);
accumulator.* = .init;
for (names, 0..) |*name, i| {
const url = std.fmt.bufPrint(name, "u{d:0>6}", .{i}) catch unreachable;
accumulator.recordSuccess(io, url, minute);
}
// One more cell than capacity: the oldest minute goes, and every cell above
// holds `minute`.
accumulator.recordSuccess(io, "https://overflow.test", minute + 60);
return accumulator;
}
test "complete is false only while a dropped minute falls inside the window" {
var threaded: std.Io.Threaded = .init(testing.allocator, .{});
defer threaded.deinit();
const io = threaded.io();
var entries = [_]pool_mod.Entry{testEntry("https://a.test/dns-query", true)};
var pool = testPool(&entries);
var database = try openLog();
defer database.close();
const at = std.Io.Clock.awake.now(io);
entries[0].health.recordFailure(at, "ConnectFailed", .{}, 0);
var entries = [_]pool_mod.Entry{testEntry(url_a, true)};
var pool = testPool(&entries);
var arena: std.heap.ArenaAllocator = .init(testing.allocator);
defer arena.deinit();
const body = try collect(&pool, io, arena.allocator());
try testing.expectEqualStrings("ConnectFailed", body.upstreams[0].last_error);
// The entry rewrites its buffer; the copy must not change with it.
entries[0].health.recordFailure(at, "Timeout", .{}, 0);
try testing.expectEqualStrings("ConnectFailed", body.upstreams[0].last_error);
// Two `now`s one minute apart, so the same drop is inside the first
// window and one minute before the second.
const inside_now = aligned_now;
const inside = stats.window(.@"1h", inside_now);
const after = stats.window(.@"1h", inside_now + 60);
try testing.expectEqual(inside.since + 60, after.since);
var names: [history_mod.max_pending][8]u8 = undefined;
const accumulator = try accumulatorDroppingAt(io, inside.since, &names);
defer testing.allocator.destroy(accumulator);
try testing.expectEqual(@as(?i64, inside.since), accumulator.snapshotStats(io).last_drop_minute);
const flagged = try collect(arena.allocator(), io, &pool, accumulator, &database, .@"1h", inside_now);
try testing.expect(!flagged.complete);
const recovered = try collect(arena.allocator(), io, &pool, accumulator, &database, .@"1h", inside_now + 60);
try testing.expect(recovered.complete);
// No accumulator at all is no known drop, not an incomplete window.
const unwired = try collect(arena.allocator(), io, &pool, null, &database, .@"1h", inside_now);
try testing.expect(unwired.complete);
}
test "the body serializes with snake_case field names" {
const upstreams = [_]Upstream{.{
.url = "https://a.test/dns-query",
test "a drop with no overflow leaves every window complete" {
var threaded: std.Io.Threaded = .init(testing.allocator, .{});
defer threaded.deinit();
const io = threaded.io();
var database = try openLog();
defer database.close();
var entries = [_]pool_mod.Entry{testEntry(url_a, true)};
var pool = testPool(&entries);
const accumulator = try testing.allocator.create(history_mod.Accumulator);
defer testing.allocator.destroy(accumulator);
accumulator.* = .init;
accumulator.recordFailure(io, url_a, aligned_now, "Timeout");
var arena: std.heap.ArenaAllocator = .init(testing.allocator);
defer arena.deinit();
const body = try collect(arena.allocator(), io, &pool, accumulator, &database, .@"1h", aligned_now);
try testing.expect(body.complete);
}
test "the route's period grammar and its 400 text are the ones /api/stats serves" {
// The picker scopes the whole page, so one bad spelling must mean the same
// thing on every route it drives.
try testing.expectEqual(stats.default_period, try stats.periodParam(""));
try testing.expectEqual(stats.Period.@"24h", try stats.periodParam(""));
try testing.expectEqual(stats.Period.@"7d", try stats.periodParam("period=7d"));
try testing.expectError(error.BadPeriod, stats.periodParam("period=12h"));
try testing.expectError(error.BadPeriod, stats.periodParam("period=1hhhhhhhhhh"));
try testing.expectEqualStrings("period must be one of 1h, 24h, 7d, 30d", bad_period_message);
}
test "every period the grammar accepts produces the window that period names" {
var threaded: std.Io.Threaded = .init(testing.allocator, .{});
defer threaded.deinit();
const io = threaded.io();
var database = try openLog();
defer database.close();
var entries = [_]pool_mod.Entry{testEntry(url_a, true)};
var pool = testPool(&entries);
var arena: std.heap.ArenaAllocator = .init(testing.allocator);
defer arena.deinit();
for (std.enums.values(stats.Period)) |period| {
const span = stats.window(period, aligned_now);
const body = try collect(arena.allocator(), io, &pool, null, &database, period, aligned_now);
try testing.expectEqualStrings(period.label(), body.period);
try testing.expectEqual(span.since, body.since);
try testing.expectEqual(span.until, body.until);
}
}
test "the body serializes exactly the ranged field set" {
const upstreams = [_]Upstream{ .{
.url = url_a,
.enabled = true,
.available = false,
.consecutive_failures = 3,
.total_successes = 10,
.total_failures = 4,
.success_rate = 0.5,
.last_error = "ConnectFailed",
}};
.period = .{
.attempts = 8,
.successes = 6,
.failures = 2,
.success_rate = 0.75,
.last_failure_at = 1_700_000_000,
.last_failure_error = "ConnectFailed",
},
}, .{
.url = url_b,
.enabled = false,
.available = false,
.period = .{
.attempts = 0,
.successes = 0,
.failures = 0,
.success_rate = null,
.last_failure_at = null,
.last_failure_error = null,
},
} };
var allocating: std.Io.Writer.Allocating = .init(testing.allocator);
defer allocating.deinit();
try std.json.Stringify.value(
Body{ .upstreams = &upstreams, .available = 0, .total = 1 },
.{},
&allocating.writer,
);
const text = allocating.written();
try std.json.Stringify.value(Body{
.period = "1h",
.since = 1_699_996_400,
.until = 1_700_000_000,
.available = 0,
.total = 2,
.complete = true,
.upstreams = &upstreams,
}, .{}, &allocating.writer);
try testing.expectEqualStrings(
\\{"period":"1h","since":1699996400,"until":1700000000,"available":0,"total":2,"complete":true,"upstreams":[{"url":"https://a.test/dns-query","enabled":true,"available":false,"period":{"attempts":8,"successes":6,"failures":2,"success_rate":0.75,"last_failure_at":1700000000,"last_failure_error":"ConnectFailed"}},{"url":"https://b.test/dns-query","enabled":false,"available":false,"period":{"attempts":0,"successes":0,"failures":0,"success_rate":null,"last_failure_at":null,"last_failure_error":null}}]}
, allocating.written());
// The lifetime fields m26 removed. They still exist in `health.State` and in
// `/metrics`; a client of this route must not find them here and start
// reading them as if they were scoped to the period.
for ([_][]const u8{
"\"consecutive_failures\":3",
"\"total_successes\":10",
"\"total_failures\":4",
"\"last_error\":\"ConnectFailed\"",
"\"available\":0",
"\"total\":1",
}) |fragment| {
try testing.expect(std.mem.containsAtLeast(u8, text, 1, fragment));
"consecutive_failures",
"total_successes",
"total_failures",
"last_error_age_s",
"\"last_error\"",
}) |gone| {
try testing.expect(!std.mem.containsAtLeast(u8, allocating.written(), 1, gone));
}
}
src/web/metrics.zig
+61
View File
@@ -30,6 +30,7 @@ const dns_cache = @import("../cache/dns_cache.zig");
const dns_handler = @import("../server/handler.zig");
const disk_monitor = @import("../storage/disk_monitor.zig");
const dot_server = @import("../server/dot_server.zig");
const history_mod = @import("../upstream/history.zig");
const http_util = @import("http_util.zig");
const logging = @import("../platform/logging.zig");
const pool_mod = @import("../upstream/pool.zig");
@@ -125,6 +126,9 @@ pub const Sample = struct {
tracker: ?TrackerSample = null,
client_names: ?client_names.Resolver.Stats = null,
retention: ?retention_mod.Stats = null,
/// The upstream-history flush loop's counters (m26 ruling 7). Absent while
/// no accumulator is wired, like every other collaborator.
history: ?history_mod.Accumulator.Stats = null,
blocklist: ?BlocklistSample = null,
disk: ?DiskSample = null,
/// One entry per enabled TLS endpoint (milestone-10 ruling 10). Rendered
@@ -199,6 +203,8 @@ pub fn collect(state: *server.WebState, io: std.Io, arena: Allocator) Allocator.
if (state.retention) |retention| sample.retention = retention.snapshotStats();
if (state.history) |history| sample.history = history.snapshotStats(io);
if (state.manager) |manager| {
const generation: ?u64 = if (manager.acquire(io)) |acquired| gen: {
defer acquired.release(io);
@@ -352,6 +358,36 @@ pub fn render(w: *std.Io.Writer, sample: Sample) std.Io.Writer.Error!void {
try counterGroup(w, "nxdns_retention_", "Query log retention counter", retention);
}
if (sample.history) |history| {
// Written out rather than reflected over `Accumulator.Stats`: three of
// its fields are counters, one is a gauge, and two — the drop watermark
// and the current flush state — are not exposition numbers at all.
try counter(
w,
"nxdns_upstream_history_flushes_total",
"Upstream history flush transactions that committed.",
history.flushes,
);
try counter(
w,
"nxdns_upstream_history_flush_failures_total",
"Upstream history flush attempts that failed; the rows are retried on the next pass.",
history.flush_failures,
);
try counter(
w,
"nxdns_upstream_history_rows_dropped_total",
"Upstream history minutes dropped because the accumulator was full.",
history.rows_dropped,
);
try gauge(
w,
"nxdns_upstream_history_pending",
"Upstream history minutes recorded but not yet flushed.",
history.pending,
);
}
if (sample.blocklist) |blocklist| {
try counter(
w,
@@ -720,6 +756,31 @@ test "a full sample renders the whole exposition, byte for byte" {
));
}
test "the upstream-history family renders three counters and one gauge" {
const text = try renderToString(testing.allocator, .{
.history = .{
.flushes = 12,
.flush_failures = 2,
.rows_dropped = 5,
.pending = 3,
.last_drop_minute = 1_700_000_040,
.last_flush_failed = true,
},
});
defer testing.allocator.free(text);
try testing.expect(std.mem.containsAtLeast(u8, text, 1, "nxdns_upstream_history_flushes_total 12\n"));
try testing.expect(std.mem.containsAtLeast(u8, text, 1, "nxdns_upstream_history_flush_failures_total 2\n"));
try testing.expect(std.mem.containsAtLeast(u8, text, 1, "nxdns_upstream_history_rows_dropped_total 5\n"));
try testing.expect(std.mem.containsAtLeast(u8, text, 1, "# TYPE nxdns_upstream_history_pending gauge\n"));
try testing.expect(std.mem.containsAtLeast(u8, text, 1, "nxdns_upstream_history_pending 3\n"));
// No accumulator is an absent family, not a family of zeros.
const bare = try renderToString(testing.allocator, .{});
defer testing.allocator.free(bare);
try testing.expect(!std.mem.containsAtLeast(u8, bare, 1, "nxdns_upstream_history_"));
}
test "every HELP line has a TYPE line and a sample, and every sample a name" {
const text = try renderToString(testing.allocator, .{});
defer testing.allocator.free(text);
src/web/openapi.yaml
+69 -14
View File
@@ -342,7 +342,15 @@ paths:
/api/upstream/health:
get:
summary: Upstream pool health
summary: Upstream pool health for a period
description: |
Each upstream's live routing state beside its recorded outcomes over
the period's window, which is the same UTC-aligned window `/api/stats`
reports for that period. The outcome counts come from per-minute
history in the query log, not from process-lifetime counters, so they
scope to the period and survive a restart.
parameters:
- $ref: "#/components/parameters/Period"
responses:
"200":
description: Per-upstream state and the availability rollup.
@@ -350,10 +358,14 @@ paths:
application/json:
schema:
$ref: "#/components/schemas/UpstreamHealth"
"400":
$ref: "#/components/responses/BadRequest"
"401":
$ref: "#/components/responses/Unauthorized"
"429":
$ref: "#/components/responses/RateLimited"
"500":
$ref: "#/components/responses/Internal"
"503":
$ref: "#/components/responses/Unavailable"
@@ -1826,28 +1838,71 @@ components:
type: string
nullable: true
UpstreamPeriodStats:
type: object
required: [attempts, successes, failures, success_rate, last_failure_at, last_failure_error]
properties:
attempts:
type: integer
description: Exchanges recorded against this upstream inside the window.
successes: { type: integer }
failures: { type: integer }
success_rate:
type: number
nullable: true
description: >
`successes / attempts`, from 0 to 1. Null when `attempts` is 0: no
observations is not perfect reliability.
last_failure_at:
type: integer
nullable: true
description: >
The newest failure inside the window, unix seconds. Null when the
window holds no failure, even if the upstream failed before it.
last_failure_error:
type: string
nullable: true
description: The error name belonging to `last_failure_at`; null exactly when it is.
UpstreamHealth:
type: object
required: [upstreams, available, total]
required: [period, since, until, available, total, complete, upstreams]
properties:
period:
type: string
enum: [1h, 24h, 7d, 30d]
since:
type: integer
description: Window start, unix seconds, inclusive.
until:
type: integer
description: Window end, unix seconds, exclusive.
available:
type: integer
description: How many upstreams the pool would route to right now.
total: { type: integer }
complete:
type: boolean
description: >
No capacity drops known in this process within the selected window;
up to about a minute of the newest outcomes may not have flushed
yet, and outcomes lost in an unclean shutdown are not detectable.
upstreams:
type: array
description: The upstreams configured now; a deleted upstream's history is not returned.
items:
type: object
required: [url, enabled, available, consecutive_failures, total_successes, total_failures, success_rate, last_error]
required: [url, enabled, available, period]
properties:
url: { type: string }
enabled: { type: boolean }
available: { type: boolean }
consecutive_failures: { type: integer }
total_successes: { type: integer }
total_failures: { type: integer }
success_rate: { type: number }
last_error:
type: string
description: Empty when the upstream never failed.
available: { type: integer }
total: { type: integer }
enabled:
type: boolean
description: Live configuration, not history.
available:
type: boolean
description: Live state, not history; false while the upstream is backing off.
period:
$ref: "#/components/schemas/UpstreamPeriodStats"
Group:
type: object
src/web/server.zig
+5
View File
@@ -39,6 +39,7 @@ const logger_mod = @import("../storage/logger.zig");
const manager_mod = @import("../filter/manager.zig");
const model = @import("../config/model.zig");
const pause_mod = @import("../server/pause.zig");
const history_mod = @import("../upstream/history.zig");
const pool_mod = @import("../upstream/pool.zig");
const query_sink = @import("../server/query_sink.zig");
const retention_mod = @import("../storage/retention.zig");
@@ -135,6 +136,10 @@ pub const WebState = struct {
client_names: ?*client_names.Resolver = null,
manager: ?*manager_mod.Manager = null,
pool: ?*pool_mod.Pool = null,
/// The upstream-outcome accumulator, for `metrics.collect` and the
/// `/api/health` rollup (m26 ruling 7). The ranged endpoint reads the
/// flushed rows through `querylog_db`, not through this.
history: ?*history_mod.Accumulator = null,
monitor: ?*disk_monitor.Monitor = null,
/// The local records and forward zones the DNS path reads. The
/// local-records and forward-zones handlers rebuild and swap them