diff --git a/src/main.zig b/src/main.zig index d6368ff..aa8332f 100644 --- a/src/main.zig +++ b/src/main.zig @@ -3598,6 +3598,44 @@ fn runServe(allocator: std.mem.Allocator, stdout: anytype, db: *cobalt.Db, bind: } sessionStore = cobalt.SessionStore.init(allocator); const ttlNanos: i128 = @intFromFloat(seconds * @as(f64, @floatFromInt(std.time.ns_per_s))); + server.enableSessions(&sessionStore, ttlNanos); + try stdout.print("sessions enabled: AUTH issues a token valid for {d}s\n", .{seconds}); + + if (sessionSweepSeconds) |sweepSeconds| { + if (sweepSeconds <= 0) { + try stdout.writeAll("error: --session-sweep-interval must be a positive number of seconds\n"); + std.process.exit(2); + } + const sweepNanos: u64 = @intFromFloat(sweepSeconds * @as(f64, @floatFromInt(std.time.ns_per_s))); + server.enableSessionSweep(sweepNanos); + try stdout.print("session sweeper enabled: expired sessions dropped every {d}s\n", .{sweepSeconds}); + } + } + defer if (sessionTtlSeconds != null) sessionStore.deinit(); + + if (ttlSweepSeconds) |seconds| { + if (seconds <= 0) { + try stdout.writeAll("error: --ttl-sweep-interval must be a positive number of seconds\n"); + std.process.exit(2); + } + const nanos: u64 = @intFromFloat(seconds * @as(f64, @floatFromInt(std.time.ns_per_s))); + server.enableTtlSweep(nanos); + try stdout.print("ttl sweeper enabled: due keys reaped every {d}s\n", .{seconds}); + } + + if (collTtlSweepSeconds) |seconds| { + if (seconds <= 0) { + try stdout.writeAll("error: --coll-ttl-sweep-interval must be a positive number of seconds\n"); + std.process.exit(2); + } + const nanos: u64 = @intFromFloat(seconds * @as(f64, @floatFromInt(std.time.ns_per_s))); + server.enableCollTtlSweep(nanos); + try stdout.print("collection ttl sweeper enabled: due collections reaped every {d}s\n", .{seconds}); + } + + try stdout.print("cobalt server listening on {}\n", .{server.boundAddress()}); + try server.start(); + server.wait(); } fn runVerify(allocator: std.mem.Allocator, stdout: anytype, dir: []const u8, count: usize) !void { diff --git a/src/metrics.zig b/src/metrics.zig new file mode 100644 index 0000000..6507b0d --- /dev/null +++ b/src/metrics.zig @@ -0,0 +1,162 @@ +const std = @import("std"); +const stats_mod = @import("stats.zig"); + +// Metrics primitives (counter/gauge/histogram), two independent rate-limiter +// algorithms (token bucket, sliding window counter), and a text-exposition +// registry. Fully additive module: does not touch Db/txn/storage internals, +// only depends on stats.zig's public Stats type for the export helper. + +pub const MetricCounter = struct { + value: std.atomic.Value(u64) = std.atomic.Value(u64).init(0), + + pub fn incr(self: *MetricCounter) void { + self.incrBy(1); + } + + pub fn incrBy(self: *MetricCounter, delta: u64) void { + _ = self.value.fetchAdd(delta, .monotonic); + } + + pub fn get(self: *const MetricCounter) u64 { + return self.value.load(.monotonic); + } + + pub fn reset(self: *MetricCounter) void { + self.value.store(0, .monotonic); + } +}; + +pub const Gauge = struct { + bits: std.atomic.Value(u64) = std.atomic.Value(u64).init(0), + + pub fn set(self: *Gauge, v: f64) void { + self.bits.store(@bitCast(v), .monotonic); + } + + pub fn get(self: *const Gauge) f64 { + const raw = self.bits.load(.monotonic); + return @bitCast(raw); + } + + pub fn add(self: *Gauge, delta: f64) void { + while (true) { + const cur = self.bits.load(.monotonic); + const curF: f64 = @bitCast(cur); + const newBits: u64 = @bitCast(curF + delta); + if (self.bits.cmpxchgWeak(cur, newBits, .monotonic, .monotonic) == null) return; + } + } +}; + +pub const HistogramError = error{EmptyBounds}; + +/// Fixed-bucket histogram with Prometheus-style cumulative-bucket quantile +/// estimation. `bounds` must be strictly increasing and is borrowed (caller +/// keeps it alive for the histogram's lifetime); values greater than the +/// last bound land in an implicit +Inf bucket. +pub const Histogram = struct { + bounds: []const f64, + counts: []u64, + sum: f64 = 0, + total: u64 = 0, + mutex: std.Thread.Mutex = .{}, + allocator: std.mem.Allocator, + + pub fn init(allocator: std.mem.Allocator, bounds: []const f64) !Histogram { + if (bounds.len == 0) return HistogramError.EmptyBounds; + var i: usize = 1; + while (i < bounds.len) : (i += 1) { + std.debug.assert(bounds[i] > bounds[i - 1]); + } + const counts = try allocator.alloc(u64, bounds.len + 1); + @memset(counts, 0); + return .{ .bounds = bounds, .counts = counts, .allocator = allocator }; + } + + pub fn deinit(self: *Histogram) void { + self.allocator.free(self.counts); + self.* = undefined; + } + + pub fn observe(self: *Histogram, value: f64) void { + self.mutex.lock(); + defer self.mutex.unlock(); + var idx: usize = self.bounds.len; + for (self.bounds, 0..) |bound, i| { + if (value <= bound) { + idx = i; + break; + } + } + self.counts[idx] += 1; + self.sum += value; + self.total += 1; + } + + pub fn count(self: *const Histogram) u64 { + return self.total; + } + + pub fn mean(self: *const Histogram) f64 { + if (self.total == 0) return 0; + return self.sum / @as(f64, @floatFromInt(self.total)); + } + + /// Linear-interpolation quantile estimate over cumulative bucket counts, + /// same algorithm Prometheus uses for `histogram_quantile`. q in [0, 1]. + pub fn quantile(self: *Histogram, q: f64) f64 { + self.mutex.lock(); + defer self.mutex.unlock(); + if (self.total == 0 or self.bounds.len == 0) return 0; + const clamped = std.math.clamp(q, 0, 1); + const target = clamped * @as(f64, @floatFromInt(self.total)); + + var cumulative: u64 = 0; + var i: usize = 0; + while (i < self.counts.len) : (i += 1) { + const prevCumulative = cumulative; + cumulative += self.counts[i]; + if (@as(f64, @floatFromInt(cumulative)) >= target) { + if (i == 0) return self.bounds[0]; + if (i == self.bounds.len) return self.bounds[self.bounds.len - 1]; + const lower = self.bounds[i - 1]; + const upper = self.bounds[i]; + const countInBucket = cumulative - prevCumulative; + if (countInBucket == 0) return upper; + const fraction = (target - @as(f64, @floatFromInt(prevCumulative))) / @as(f64, @floatFromInt(countInBucket)); + return lower + fraction * (upper - lower); + } + } + return self.bounds[self.bounds.len - 1]; + } + + pub fn bucketCount(self: *Histogram, idx: usize) u64 { + self.mutex.lock(); + defer self.mutex.unlock(); + return self.counts[idx]; + } +}; + +pub fn linearBuckets(allocator: std.mem.Allocator, start: f64, width: f64, n: usize) ![]f64 { + std.debug.assert(n > 0); + std.debug.assert(width > 0); + const out = try allocator.alloc(f64, n); + var i: usize = 0; + while (i < n) : (i += 1) out[i] = start + width * @as(f64, @floatFromInt(i)); + return out; +} + +pub fn exponentialBuckets(allocator: std.mem.Allocator, start: f64, factor: f64, n: usize) ![]f64 { + std.debug.assert(n > 0); + std.debug.assert(start > 0); + std.debug.assert(factor > 1); + const out = try allocator.alloc(f64, n); + var cur = start; + var i: usize = 0; + while (i < n) : (i += 1) { + out[i] = cur; + cur *= factor; + } + return out; +} +