feat: extend main with MetricCounter and incr
This commit is contained in:
1 parent
ad58b3f05c
commit
95c1bed183
2 files changed
+200
No files matched your search
@@ -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 {
|
||||
|
||||
+162
@@ -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;
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user