From b7ff3165c339edba7d7d7f51d63a0c809e133ede Mon Sep 17 00:00:00 2001 From: jaronoff97 Date: Wed, 5 Aug 2026 13:49:55 -0400 Subject: [PATCH 1/4] feat: update to policy v1.7.0 --- README.md | 5 ++ proto/tero/policy/v1/policy.proto | 32 ++++++++ src/policy/policy_engine.zig | 63 ++++++++++++++++ src/policy/provider.zig | 54 ++++++++++++++ src/policy/provider_http.zig | 119 +++++++++++++++++++++++++++++- src/policy/registry.zig | 105 ++++++++++++++++++++++++++ src/proto/tero/policy/v1.pb.zig | 103 ++++++++++++++++++++++++++ 7 files changed, 477 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index 366b1f5..2790995 100644 --- a/README.md +++ b/README.md @@ -114,6 +114,11 @@ var policy_id_buf: [16][]const u8 = undefined; const result = engine.evaluate(.log, &my_log_accessor, &my_log_ctx, &policy_id_buf, .{ .scratch = arena.allocator(), .io = io, + // Optional (spec v1.7.0 volume tracking): the record's uncompressed OTLP + // protobuf size. Every evaluate call is counted regardless of match and + // reported on the next sync; passing this adds the byte totals, which only + // the caller can measure. Omit it to report record counts alone. + .record_bytes = my_log_size, }); // Transforms whose required primitive (set/delete/move) is unwired on the diff --git a/proto/tero/policy/v1/policy.proto b/proto/tero/policy/v1/policy.proto index b1ddbca..7ef2d57 100644 --- a/proto/tero/policy/v1/policy.proto +++ b/proto/tero/policy/v1/policy.proto @@ -126,6 +126,34 @@ message PolicySyncStatus { TransformStageStatus add = 13; } +// VolumeStats reports the total telemetry a client observed since the last +// sync, regardless of whether any policy matched it. Counts are of records +// entering policy evaluation, before any keep or transform stage runs, and are +// reset on each successful sync. +// +// Reporting volume is optional, and every field is individually optional: an +// implementation may report record counts without byte counts, or a subset of +// signals. Any field left at 0 means "not tracked" as much as it means "none +// seen", so consumers must not read 0 as an observation. +// +// Byte counts, when reported, are the uncompressed OTLP protobuf serialized +// size of the records as received, and are an estimate; implementations that +// cannot measure this cheaply may approximate it. A size in any other encoding +// must not be reported here — leave the field at 0 instead. +message VolumeStats { + // Log records seen, and their total size in bytes. + int64 log_records = 1; + int64 log_bytes = 2; + + // Metric data points seen, and their total size in bytes. + int64 metric_data_points = 3; + int64 metric_bytes = 4; + + // Spans seen, and their total size in bytes. + int64 spans = 5; + int64 span_bytes = 6; +} + // SyncRequest is sent by clients to request policy updates. message SyncRequest { // Client identification and capabilities @@ -142,6 +170,10 @@ message SyncRequest { // Status of individual policies within this set. repeated PolicySyncStatus policy_statuses = 5; + + // Optional. Total telemetry observed since the last sync, regardless of + // policy match. Clients that do not track volume omit this. + VolumeStats volume = 6; } enum SyncType { diff --git a/src/policy/policy_engine.zig b/src/policy/policy_engine.zig index 8e33845..4a2e90b 100644 --- a/src/policy/policy_engine.zig +++ b/src/policy/policy_engine.zig @@ -226,6 +226,12 @@ pub const EvaluateOptions = struct { /// records are handed to the sink after keep resolution, before /// transforms. Null costs a single branch. extension_sink: ?policy_types.ExtensionSink = null, + /// Uncompressed OTLP protobuf serialized size of this record (v1.7.0 + /// volume tracking). The engine reads records through accessors and has no + /// serialized form to measure, so only the caller — which already holds + /// the wire bytes on ingest — can supply this. Leave at 0 to report record + /// counts without bytes, which the spec permits. + record_bytes: usize = 0, }; // ============================================================================= @@ -336,6 +342,11 @@ pub const PolicyEngine = struct { policy_id_buf: [][]const u8, options: EvaluateOptions, ) PolicyResult { + // Volume tracking (v1.7.0): every record entering evaluation counts, + // regardless of match — including records with no policies loaded for + // their signal, so this must precede both early returns below. + self.registry.volume.record(T, options.record_bytes); + // Get current snapshot from registry (lock-free) const snapshot = self.registry.getSnapshot() orelse { const event: EvaluateEmpty = .{}; @@ -973,6 +984,58 @@ test "PolicyEngine: empty registry returns unset" { try testing.expectEqual(@as(usize, 0), result.matched_policy_ids.len); } +test "PolicyEngine: volume counts every record, including with no policies loaded" { + const allocator = testing.allocator; + + var noop_bus: NoopEventBus = undefined; + noop_bus.init(std.Options.debug_io); + var registry = PolicyRegistry.init(allocator, noop_bus.eventBus()); + defer registry.deinit(); + + const engine = PolicyEngine.init(noop_bus.eventBus(), ®istry); + + // No snapshot at all: evaluation returns early, volume still counts. + var test_log: TestLogContext = .{ .message = "hello" }; + var policy_id_buf: [16][]const u8 = undefined; + _ = engine.evaluate(.log, &TestLogContext.accessor, &test_log, &policy_id_buf, .{ + .io = std.Options.debug_io, + .record_bytes = 120, + }); + + // A dropped record counts at its pre-policy size too. + var policy: Policy = .{ + .id = try allocator.dupe(u8, "policy-1"), + .enabled = true, + .target = .{ .log = .{ .keep = try allocator.dupe(u8, "none") } }, + }; + try policy.target.?.log.match.append(allocator, .{ + .field = .{ .log_field = .LOG_FIELD_BODY }, + .match = .{ .regex = try allocator.dupe(u8, "error") }, + }); + defer policy.deinit(allocator); + try registry.updatePolicies(&.{policy}, "file-provider", .file); + + var error_log: TestLogContext = .{ .message = "an error occurred" }; + const dropped = engine.evaluate(.log, &TestLogContext.accessor, &error_log, &policy_id_buf, .{ + .io = std.Options.debug_io, + .record_bytes = 80, + }); + try testing.expectEqual(FilterDecision.drop, dropped.decision); + + // Bytes are optional per record: omitting them still counts the record. + _ = evalTestLog(&engine, &test_log, &policy_id_buf); + + const volume = registry.volume.readAndReset(); + try testing.expectEqual(@as(i64, 3), volume.log_records); + try testing.expectEqual(@as(i64, 200), volume.log_bytes); + // Other signals are untracked, not zero-observed — both look like 0 here. + try testing.expectEqual(@as(i64, 0), volume.spans); + try testing.expectEqual(@as(i64, 0), volume.metric_data_points); + + // Drained: a second read reports nothing. + try testing.expect(registry.volume.readAndReset().isZero()); +} + test "PolicyEngine: single policy drop match" { const allocator = testing.allocator; diff --git a/src/policy/provider.zig b/src/policy/provider.zig index c0552a0..82b6121 100644 --- a/src/policy/provider.zig +++ b/src/policy/provider.zig @@ -19,6 +19,42 @@ pub const PolicyStatsSnapshot = struct { errors: []const []const u8 = &.{}, }; +/// Total telemetry that entered policy evaluation since the last successful +/// sync, regardless of match (spec v1.7.0 `VolumeStats`). Counted before the +/// keep and transform stages, so dropped/sampled/redacted records are included +/// at their pre-policy size. +/// +/// Every field is independently optional: `0` means "not tracked" as much as +/// "none seen". Byte counts are only populated when the caller passes +/// `EvaluateOptions.record_bytes` — the engine reads records through accessors +/// and has no serialized form to measure. +pub const VolumeSnapshot = struct { + log_records: i64 = 0, + log_bytes: i64 = 0, + metric_data_points: i64 = 0, + metric_bytes: i64 = 0, + spans: i64 = 0, + span_bytes: i64 = 0, + + pub fn isZero(self: VolumeSnapshot) bool { + inline for (@typeInfo(VolumeSnapshot).@"struct".fields) |f| { + if (@field(self, f.name) != 0) return false; + } + return true; + } + + pub fn toProto(self: VolumeSnapshot) proto.policy.VolumeStats { + return .{ + .log_records = self.log_records, + .log_bytes = self.log_bytes, + .metric_data_points = self.metric_data_points, + .metric_bytes = self.metric_bytes, + .spans = self.spans, + .span_bytes = self.span_bytes, + }; + } +}; + /// Pull-based stats source handed to a provider by the registry at subscribe /// time. The provider invokes `collect` immediately before each sync, passing a /// per-sync arena; the registry returns one snapshot per policy — including @@ -28,10 +64,28 @@ pub const PolicyStatsSnapshot = struct { pub const StatsCollector = struct { context: *anyopaque, collect: *const fn (arena: std.mem.Allocator, context: *anyopaque) anyerror![]PolicyStatsSnapshot, + /// Drain the registry's volume counters (spec v1.7.0). Optional: a + /// collector that leaves it null reports no volume, which is conformant. + collect_volume: ?*const fn (context: *anyopaque) VolumeSnapshot = null, + /// Add a previously drained reading back, so a failed sync retains its + /// counters for the next attempt as the spec requires. Concurrent + /// evaluation keeps incrementing across the drain, hence add-back rather + /// than restore-by-assignment. + restore_volume: ?*const fn (context: *anyopaque, snapshot: VolumeSnapshot) void = null, pub fn call(self: StatsCollector, arena: std.mem.Allocator) anyerror![]PolicyStatsSnapshot { return self.collect(arena, self.context); } + + pub fn drainVolume(self: StatsCollector) VolumeSnapshot { + const f = self.collect_volume orelse return .{}; + return f(self.context); + } + + pub fn returnVolume(self: StatsCollector, snapshot: VolumeSnapshot) void { + const f = self.restore_volume orelse return; + f(self.context, snapshot); + } }; /// Extension sync plumbing (spec v1.6.0), implemented outside policy_zig by diff --git a/src/policy/provider_http.zig b/src/policy/provider_http.zig index c52366e..ff07a4a 100644 --- a/src/policy/provider_http.zig +++ b/src/policy/provider_http.zig @@ -7,6 +7,7 @@ const o11y = @import("observability"); const PolicyCallback = policy_provider.PolicyCallback; const StatsCollector = policy_provider.StatsCollector; const PolicyStatsSnapshot = policy_provider.PolicyStatsSnapshot; +const VolumeSnapshot = policy_provider.VolumeSnapshot; const SyncRequest = proto.policy.SyncRequest; const SyncResponse = proto.policy.SyncResponse; const ClientMetadata = proto.policy.ClientMetadata; @@ -39,6 +40,11 @@ const HttpSyncRequestSucceeded = struct { url: []const u8, policy_statuses_sent: // Emitted once per policy status about to be sent, so we can confirm at runtime // exactly which policies (and counts) are reported each sync. const HttpSyncStatusReported = struct { id: []const u8, match_hits: i64, match_misses: i64, errors: usize }; +// Emitted when a sync carries volume, for the same reason: confirming at +// runtime what was reported. Omitted volume (all-zero) emits nothing. +const HttpSyncVolumeReported = struct { volume: VolumeSnapshot }; +// A failed sync hands its drained volume back for the next attempt. +const HttpSyncVolumeRetained = struct { volume: VolumeSnapshot }; const HttpFetchStarted = struct {}; const HttpFetchCompleted = struct {}; @@ -181,6 +187,16 @@ pub const HttpProvider = struct { self.stats_collector = collector; } + /// Hand a drained volume reading back to the registry after a failed sync, + /// so the next attempt includes it (spec v1.7.0). + fn retainVolume(self: *HttpProvider, volume: VolumeSnapshot) void { + if (volume.isZero()) return; + const collector = self.stats_collector orelse return; + collector.returnVolume(volume); + const event: HttpSyncVolumeRetained = .{ .volume = volume }; + self.bus.warn(event); + } + pub fn subscribe(self: *HttpProvider, callback: PolicyCallback) !void { self.callback = callback; @@ -211,9 +227,11 @@ pub const HttpProvider = struct { /// Returns the final sync's error rather than swallowing it — the caller /// decides whether a lost tail report is worth logging or acting on. The /// flush is attempted at most once: `flushed` is set before the sync (the - /// stats collector resets its counters when pulled, so the tail is gone on - /// failure and a retry would only report zeros), so a second call is a - /// no-op and the loop is stopped either way. + /// stats collector resets its per-policy counters when pulled, so the tail + /// is gone on failure and a retry would only report zeros), so a second + /// call is a no-op and the loop is stopped either way. Volume counters are + /// the exception — a failed sync hands them back — but with the loop + /// stopped there is no further sync to carry them. /// /// Reuses the io bound at init (the same process-wide io is still live at /// shutdown), so this must run before that io is torn down — i.e. in the @@ -398,6 +416,15 @@ pub const HttpProvider = struct { const policy_statuses_list = try statsToSyncStatuses(temp_allocator, stats); + // Drain total observed volume (v1.7.0). Unlike per-policy stats, this + // is handed back if the sync fails: it is the denominator for every + // policy's match rate, so a dropped interval skews usage rather than + // just losing a tail. `volume_sent` gates the add-back so a sync the + // server accepted is never counted twice. + const volume: VolumeSnapshot = if (self.stats_collector) |c| c.drainVolume() else .{}; + var volume_sent = false; + errdefer if (!volume_sent) self.retainVolume(volume); + // Log exactly what we're about to report, so runtime expectations can be // verified against the live snapshot. for (policy_statuses_list.items) |status| { @@ -449,8 +476,16 @@ pub const HttpProvider = struct { .last_sync_timestamp_unix_nano = self.last_sync_timestamp, .last_successful_hash = last_hash, .policy_statuses = policy_statuses_list, + // Omitted rather than sent zero-valued when nothing was seen, per + // the spec's guidance for implementations that track no volume. + .volume = if (volume.isZero()) null else volume.toProto(), }; + if (!volume.isZero()) { + const volume_event: HttpSyncVolumeReported = .{ .volume = volume }; + self.bus.debug(volume_event); + } + // Encode SyncRequest to JSON const request_body = try sync_request.jsonEncode(.{}, .{ .emit_oneof_field_name = false }, temp_allocator); // No defer needed - arena handles cleanup @@ -504,6 +539,11 @@ pub const HttpProvider = struct { return error.HttpRequestFailed; } + // The server accepted the request, so the reported interval is closed: + // the drained counters stay drained even if decoding the response below + // fails. + volume_sent = true; + const sent_event: HttpSyncRequestSucceeded = .{ .url = self.config_url, .policy_statuses_sent = policy_statuses_list.items.len, @@ -730,16 +770,29 @@ test "HttpProvider.close: flushes final stats once, then is idempotent" { // the tail stat that must reach the control plane before teardown. const Collector = struct { var calls: usize = 0; + var restores: usize = 0; fn collect(arena: std.mem.Allocator, _: *anyopaque) anyerror![]PolicyStatsSnapshot { calls += 1; const rows = try arena.alloc(PolicyStatsSnapshot, 1); rows[0] = .{ .id = "hot", .hits = 5, .misses = 1 }; return rows; } + fn drain(_: *anyopaque) VolumeSnapshot { + return .{ .log_records = 9, .log_bytes = 512 }; + } + fn restore(_: *anyopaque, _: VolumeSnapshot) void { + restores += 1; + } }; Collector.calls = 0; + Collector.restores = 0; var ctx: u8 = 0; - provider.setStatsCollector(.{ .context = &ctx, .collect = Collector.collect }); + provider.setStatsCollector(.{ + .context = &ctx, + .collect = Collector.collect, + .collect_volume = Collector.drain, + .restore_volume = Collector.restore, + }); try provider.close(); server_future.await(io); @@ -750,6 +803,10 @@ test "HttpProvider.close: flushes final stats once, then is idempotent" { // The final sync carried the tail stats: the policy id and its hit count. try testing.expect(std.mem.indexOf(u8, stub.body, "hot") != null); try testing.expect(std.mem.indexOf(u8, stub.body, "matchHits") != null); + // …and the observed volume, which the accepted sync does not hand back. + try testing.expect(std.mem.indexOf(u8, stub.body, "\"logRecords\":\"9\"") != null); + try testing.expect(std.mem.indexOf(u8, stub.body, "\"logBytes\":\"512\"") != null); + try testing.expectEqual(@as(usize, 0), Collector.restores); // Second close is a no-op: no extra collect, no extra request (the stub // only served one and its port is about to close). @@ -785,6 +842,60 @@ test "HttpProvider.close: propagates the final sync error" { try provider.close(); } +test "HttpProvider: a failed sync hands its volume back for the next attempt" { + const allocator = testing.allocator; + var threaded: std.Io.Threaded = .init(allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + + var noop_bus: o11y.NoopEventBus = undefined; + noop_bus.init(io); + + // Stand-in for the registry's counters: drained on pull, summed on + // add-back. (The real VolumeCounters round-trip is covered in registry.zig.) + const Counters = struct { + var pending: VolumeSnapshot = .{}; + fn collect(_: std.mem.Allocator, _: *anyopaque) anyerror![]PolicyStatsSnapshot { + return &.{}; + } + fn drain(_: *anyopaque) VolumeSnapshot { + defer pending = .{}; + return pending; + } + fn restore(_: *anyopaque, snapshot: VolumeSnapshot) void { + pending.log_records += snapshot.log_records; + pending.log_bytes += snapshot.log_bytes; + pending.spans += snapshot.spans; + } + }; + Counters.pending = .{ .log_records = 1, .log_bytes = 64, .spans = 1 }; + + // Port 1 has nothing listening → the POST fails after the drain. + var provider = try HttpProvider.init( + allocator, + io, + noop_bus.eventBus(), + .{ .id = "p", .url = "http://127.0.0.1:1/sync", .poll_interval_seconds = 3600 }, + ); + defer provider.deinit(); + var ctx: u8 = 0; + provider.setStatsCollector(.{ + .context = &ctx, + .collect = Counters.collect, + .collect_volume = Counters.drain, + .restore_volume = Counters.restore, + }); + + if (provider.close()) |_| { + try testing.expect(false); // expected the unreachable endpoint to error + } else |_| {} + + // Nothing reached the control plane, so the interval is still open. + try testing.expectEqual(@as(i64, 1), Counters.pending.log_records); + try testing.expectEqual(@as(i64, 64), Counters.pending.log_bytes); + try testing.expectEqual(@as(i64, 1), Counters.pending.spans); +} + test "statsToSyncStatuses: maps hits/misses/errors and reports every policy" { var arena = std.heap.ArenaAllocator.init(testing.allocator); defer arena.deinit(); diff --git a/src/policy/registry.zig b/src/policy/registry.zig index e59d8be..9391b22 100644 --- a/src/policy/registry.zig +++ b/src/policy/registry.zig @@ -53,6 +53,58 @@ pub const PolicyAtomicStats = struct { } }; +// ============================================================================= +// Volume Counters (spec v1.7.0) +// ============================================================================= + +/// Total telemetry entering policy evaluation, regardless of match. Lives on +/// the registry rather than the snapshot: a recompile allocates fresh +/// per-policy stats, and volume must survive that to stay a valid denominator +/// for hits/misses across a sync interval. +/// +/// ponytail: one shared cache line of atomics, same shape as the per-policy +/// hit/miss counters. If per-record contention ever shows up in a profile, +/// shard per thread and sum on drain. +pub const VolumeCounters = struct { + log_records: std.atomic.Value(i64) = .init(0), + log_bytes: std.atomic.Value(i64) = .init(0), + metric_data_points: std.atomic.Value(i64) = .init(0), + metric_bytes: std.atomic.Value(i64) = .init(0), + spans: std.atomic.Value(i64) = .init(0), + span_bytes: std.atomic.Value(i64) = .init(0), + + /// Count one record of signal `T`. `bytes` is the caller's estimate of the + /// record's uncompressed OTLP protobuf size; 0 means "not tracked" and + /// leaves the byte counter alone. + pub inline fn record(self: *VolumeCounters, comptime T: policy_types.TelemetryType, bytes: usize) void { + const counters = switch (T) { + .log => .{ &self.log_records, &self.log_bytes }, + .metric => .{ &self.metric_data_points, &self.metric_bytes }, + .trace => .{ &self.spans, &self.span_bytes }, + }; + _ = counters[0].fetchAdd(1, .monotonic); + if (bytes != 0) _ = counters[1].fetchAdd(@intCast(bytes), .monotonic); + } + + /// Read and zero every counter, for reporting on a sync request. + pub fn readAndReset(self: *VolumeCounters) policy_provider.VolumeSnapshot { + var out: policy_provider.VolumeSnapshot = .{}; + inline for (@typeInfo(policy_provider.VolumeSnapshot).@"struct".fields) |f| { + @field(out, f.name) = @field(self, f.name).swap(0, .monotonic); + } + return out; + } + + /// Add a drained reading back after a failed sync, so its records are + /// included in the next attempt instead of being lost. + pub fn add(self: *VolumeCounters, snapshot: policy_provider.VolumeSnapshot) void { + inline for (@typeInfo(policy_provider.VolumeSnapshot).@"struct".fields) |f| { + const v = @field(snapshot, f.name); + if (v != 0) _ = @field(self, f.name).fetchAdd(v, .monotonic); + } + } +}; + // ============================================================================= // Observability Events // ============================================================================= @@ -258,6 +310,10 @@ pub const PolicyRegistry = struct { // subscribe for capability advertisement + broadcast-config routing. extension_sync_hooks: ?policy_provider.ExtensionSyncHooks, + // Total telemetry seen by the engine since the last successful sync + // (v1.7.0), independent of any policy or snapshot. + volume: VolumeCounters, + /// Subscription context for provider callbacks. /// Allocated with stable address so the callback pointer remains valid. const Subscription = struct { @@ -294,6 +350,7 @@ pub const PolicyRegistry = struct { .bus = bus, .extension_resolver = null, .extension_sync_hooks = null, + .volume = .{}, }; } @@ -329,6 +386,8 @@ pub const PolicyRegistry = struct { prov.setStatsCollector(.{ .context = self, .collect = collectStatsThunk, + .collect_volume = drainVolumeThunk, + .restore_volume = restoreVolumeThunk, }); if (self.extension_sync_hooks) |hooks| { @@ -346,6 +405,16 @@ pub const PolicyRegistry = struct { return self.collectStats(arena); } + fn drainVolumeThunk(context: *anyopaque) policy_provider.VolumeSnapshot { + const self: *PolicyRegistry = @ptrCast(@alignCast(context)); + return self.volume.readAndReset(); + } + + fn restoreVolumeThunk(context: *anyopaque, snapshot: policy_provider.VolumeSnapshot) void { + const self: *PolicyRegistry = @ptrCast(@alignCast(context)); + self.volume.add(snapshot); + } + /// Collect a stats row for every policy in the current snapshot, resetting /// the underlying atomic counters, and attach any compile errors recorded /// since the last recompile. Every policy is reported, including zero-hit @@ -1744,6 +1813,42 @@ test "PolicyRegistry: collectStats drains hit/miss counters and reports every po try testing.expectEqual(@as(i64, 0), hot_row2.misses); } +test "PolicyRegistry: volume counters survive recompiles and accept add-back" { + const allocator = testing.allocator; + var noop_bus: NoopEventBus = undefined; + noop_bus.init(std.Options.debug_io); + var registry = PolicyRegistry.init(allocator, noop_bus.eventBus()); + defer registry.deinit(); + + registry.volume.record(.log, 100); + registry.volume.record(.metric, 0); // bytes untracked for this record + registry.volume.record(.trace, 40); + + // A recompile allocates fresh per-policy stats; volume lives outside the + // snapshot and must not be zeroed by it. + var p = try createTestPolicy(allocator, "p"); + defer freeTestPolicy(allocator, &p); + try registry.updatePolicies(&.{p}, "http-provider", .http); + + const drained = registry.volume.readAndReset(); + try testing.expectEqual(@as(i64, 1), drained.log_records); + try testing.expectEqual(@as(i64, 100), drained.log_bytes); + try testing.expectEqual(@as(i64, 1), drained.metric_data_points); + try testing.expectEqual(@as(i64, 0), drained.metric_bytes); + try testing.expectEqual(@as(i64, 1), drained.spans); + try testing.expectEqual(@as(i64, 40), drained.span_bytes); + try testing.expect(registry.volume.readAndReset().isZero()); + + // A failed sync hands the reading back; records seen in the meantime are + // preserved, so the next attempt reports the sum. + registry.volume.record(.log, 7); + registry.volume.add(drained); + const retried = registry.volume.readAndReset(); + try testing.expectEqual(@as(i64, 2), retried.log_records); + try testing.expectEqual(@as(i64, 107), retried.log_bytes); + try testing.expectEqual(@as(i64, 1), retried.spans); +} + test "PolicyRegistry: collectStats reports exactly the current policy set across updates" { const allocator = testing.allocator; var noop_bus: NoopEventBus = undefined; diff --git a/src/proto/tero/policy/v1.pb.zig b/src/proto/tero/policy/v1.pb.zig index 981994e..84b6a89 100644 --- a/src/proto/tero/policy/v1.pb.zig +++ b/src/proto/tero/policy/v1.pb.zig @@ -2501,6 +2501,107 @@ pub const PolicySyncStatus = struct { } }; +/// VolumeStats reports the total telemetry a client observed since the last +/// sync, regardless of whether any policy matched it. Counts are of records +/// entering policy evaluation, before any keep or transform stage runs, and are +/// reset on each successful sync. +/// +/// Reporting volume is optional, and every field is individually optional: an +/// implementation may report record counts without byte counts, or a subset of +/// signals. Any field left at 0 means "not tracked" as much as it means "none +/// seen", so consumers must not read 0 as an observation. +/// +/// Byte counts, when reported, are the uncompressed OTLP protobuf serialized +/// size of the records as received, and are an estimate; implementations that +/// cannot measure this cheaply may approximate it. A size in any other encoding +/// must not be reported here — leave the field at 0 instead. +pub const VolumeStats = struct { + log_records: i64 = 0, + log_bytes: i64 = 0, + metric_data_points: i64 = 0, + metric_bytes: i64 = 0, + spans: i64 = 0, + span_bytes: i64 = 0, + + pub const _desc_table = .{ + .log_records = fd(1, .{ .scalar = .int64 }), + .log_bytes = fd(2, .{ .scalar = .int64 }), + .metric_data_points = fd(3, .{ .scalar = .int64 }), + .metric_bytes = fd(4, .{ .scalar = .int64 }), + .spans = fd(5, .{ .scalar = .int64 }), + .span_bytes = fd(6, .{ .scalar = .int64 }), + }; + + /// Encodes the message to the writer + /// The allocator is used to generate submessages internally. + /// Hence, an ArenaAllocator is a preferred choice if allocations are a bottleneck. + pub fn encode( + self: @This(), + writer: *std.Io.Writer, + allocator: std.mem.Allocator, + ) (std.Io.Writer.Error || std.mem.Allocator.Error)!void { + return protobuf.encode(writer, allocator, self); + } + + /// Decodes the message from the bytes read from the reader. + pub fn decode( + reader: *std.Io.Reader, + allocator: std.mem.Allocator, + ) (protobuf.DecodingError || std.Io.Reader.Error || std.mem.Allocator.Error)!@This() { + return protobuf.decode(@This(), reader, allocator); + } + + /// Deinitializes and frees the memory associated with the message. + pub fn deinit(self: *@This(), allocator: std.mem.Allocator) void { + return protobuf.deinit(allocator, self); + } + + /// Duplicates the message. + pub fn dupe(self: @This(), allocator: std.mem.Allocator) std.mem.Allocator.Error!@This() { + return protobuf.dupe(@This(), self, allocator); + } + + /// Decodes the message from the JSON string. + pub fn jsonDecode( + input: []const u8, + options: std.json.ParseOptions, + allocator: std.mem.Allocator, + ) !std.json.Parsed(@This()) { + return protobuf.json.decode(@This(), input, options, allocator); + } + + /// Decodes the message from the JSON string, honoring pb options + /// (e.g. hex_bytes_fields for OTLP trace_id/span_id). + pub fn jsonDecodeOpts( + input: []const u8, + options: std.json.ParseOptions, + pb_options: protobuf.json.Options, + allocator: std.mem.Allocator, + ) !std.json.Parsed(@This()) { + return protobuf.json.decodeOpts(@This(), input, options, pb_options, allocator); + } + + /// Encodes the message to a JSON string. + pub fn jsonEncode( + self: @This(), + options: std.json.Stringify.Options, + pb_options: protobuf.json.Options, + allocator: std.mem.Allocator, + ) ![]const u8 { + return protobuf.json.encode(self, options, pb_options, allocator); + } + + /// This method is used by std.json + /// internally for deserialization. DO NOT RENAME! + pub fn jsonParse( + allocator: std.mem.Allocator, + source: anytype, + options: std.json.ParseOptions, + ) !@This() { + return protobuf.json.parse(@This(), allocator, source, options); + } +}; + /// SyncRequest is sent by clients to request policy updates. pub const SyncRequest = struct { client_metadata: ?ClientMetadata = null, @@ -2508,6 +2609,7 @@ pub const SyncRequest = struct { last_sync_timestamp_unix_nano: u64 = 0, last_successful_hash: []const u8 = &.{}, policy_statuses: std.ArrayList(PolicySyncStatus) = .empty, + volume: ?VolumeStats = null, pub const _desc_table = .{ .client_metadata = fd(1, .submessage), @@ -2515,6 +2617,7 @@ pub const SyncRequest = struct { .last_sync_timestamp_unix_nano = fd(3, .{ .scalar = .fixed64 }), .last_successful_hash = fd(4, .{ .scalar = .string }), .policy_statuses = fd(5, .{ .repeated = .submessage }), + .volume = fd(6, .submessage), }; /// Encodes the message to the writer From 65865fc74bb996914c861624cedd4f1f4cc6c6e0 Mon Sep 17 00:00:00 2001 From: jaronoff97 Date: Wed, 5 Aug 2026 14:15:48 -0400 Subject: [PATCH 2/4] better impl --- README.md | 10 +- src/policy/policy_engine.zig | 76 +++++++--- src/policy/provider.zig | 64 ++++++++- src/policy/provider_http.zig | 269 +++++++++++++++++++++++++++-------- src/policy/registry.zig | 126 ++++++++++++++-- 5 files changed, 448 insertions(+), 97 deletions(-) diff --git a/README.md b/README.md index 2790995..4a295ac 100644 --- a/README.md +++ b/README.md @@ -114,13 +114,13 @@ var policy_id_buf: [16][]const u8 = undefined; const result = engine.evaluate(.log, &my_log_accessor, &my_log_ctx, &policy_id_buf, .{ .scratch = arena.allocator(), .io = io, - // Optional (spec v1.7.0 volume tracking): the record's uncompressed OTLP - // protobuf size. Every evaluate call is counted regardless of match and - // reported on the next sync; passing this adds the byte totals, which only - // the caller can measure. Omit it to report record counts alone. - .record_bytes = my_log_size, }); +// Volume (spec v1.7.0): every evaluate call is counted regardless of match and +// reported on the next sync. Byte volume is opt-in — pass the batch's +// uncompressed OTLP protobuf size, which only the caller can measure. +registry.volume.addBytes(.log, @intCast(request_size_bytes)); + // Transforms whose required primitive (set/delete/move) is unwired on the // accessor are *eliminated at compile time* for that callsite — no // runtime branch, no code emitted. Different consumers can apply diff --git a/src/policy/policy_engine.zig b/src/policy/policy_engine.zig index 4a2e90b..e8db913 100644 --- a/src/policy/policy_engine.zig +++ b/src/policy/policy_engine.zig @@ -226,12 +226,6 @@ pub const EvaluateOptions = struct { /// records are handed to the sink after keep resolution, before /// transforms. Null costs a single branch. extension_sink: ?policy_types.ExtensionSink = null, - /// Uncompressed OTLP protobuf serialized size of this record (v1.7.0 - /// volume tracking). The engine reads records through accessors and has no - /// serialized form to measure, so only the caller — which already holds - /// the wire bytes on ingest — can supply this. Leave at 0 to report record - /// counts without bytes, which the spec permits. - record_bytes: usize = 0, }; // ============================================================================= @@ -344,8 +338,9 @@ pub const PolicyEngine = struct { ) PolicyResult { // Volume tracking (v1.7.0): every record entering evaluation counts, // regardless of match — including records with no policies loaded for - // their signal, so this must precede both early returns below. - self.registry.volume.record(T, options.record_bytes); + // their signal, so this must precede both early returns below. Byte + // volume is opt-in via `registry.volume.addBytes`. + self.registry.volume.record(T); // Get current snapshot from registry (lock-free) const snapshot = self.registry.getSnapshot() orelse { @@ -997,10 +992,8 @@ test "PolicyEngine: volume counts every record, including with no policies loade // No snapshot at all: evaluation returns early, volume still counts. var test_log: TestLogContext = .{ .message = "hello" }; var policy_id_buf: [16][]const u8 = undefined; - _ = engine.evaluate(.log, &TestLogContext.accessor, &test_log, &policy_id_buf, .{ - .io = std.Options.debug_io, - .record_bytes = 120, - }); + _ = evalTestLog(&engine, &test_log, &policy_id_buf); + registry.volume.addBytes(.log, 120); // A dropped record counts at its pre-policy size too. var policy: Policy = .{ @@ -1016,13 +1009,11 @@ test "PolicyEngine: volume counts every record, including with no policies loade try registry.updatePolicies(&.{policy}, "file-provider", .file); var error_log: TestLogContext = .{ .message = "an error occurred" }; - const dropped = engine.evaluate(.log, &TestLogContext.accessor, &error_log, &policy_id_buf, .{ - .io = std.Options.debug_io, - .record_bytes = 80, - }); + const dropped = evalTestLog(&engine, &error_log, &policy_id_buf); try testing.expectEqual(FilterDecision.drop, dropped.decision); + registry.volume.addBytes(.log, 80); - // Bytes are optional per record: omitting them still counts the record. + // Bytes are opt-in and independent of record counting. _ = evalTestLog(&engine, &test_log, &policy_id_buf); const volume = registry.volume.readAndReset(); @@ -1036,6 +1027,57 @@ test "PolicyEngine: volume counts every record, including with no policies loade try testing.expect(registry.volume.readAndReset().isZero()); } +test "PolicyEngine: volume routes each signal to its own counter" { + const allocator = testing.allocator; + + var noop_bus: NoopEventBus = undefined; + noop_bus.init(std.Options.debug_io); + var registry = PolicyRegistry.init(allocator, noop_bus.eventBus()); + defer registry.deinit(); + + // A log-only policy set: the metric and trace indices are empty, so those + // evaluations take the `index.isEmpty()` early return and must still count. + var policy: Policy = .{ + .id = try allocator.dupe(u8, "policy-1"), + .enabled = true, + .target = .{ .log = .{ .keep = try allocator.dupe(u8, "all") } }, + }; + try policy.target.?.log.match.append(allocator, .{ + .field = .{ .log_field = .LOG_FIELD_BODY }, + .match = .{ .regex = try allocator.dupe(u8, "hello") }, + }); + defer policy.deinit(allocator); + try registry.updatePolicies(&.{policy}, "file-provider", .file); + + const engine = PolicyEngine.init(noop_bus.eventBus(), ®istry); + var policy_id_buf: [16][]const u8 = undefined; + + var test_log: TestLogContext = .{ .message = "hello" }; + _ = evalTestLog(&engine, &test_log, &policy_id_buf); + + var test_metric: TestMetricContext = .{ .name = "http_requests_total" }; + _ = evalMetric(&engine, &test_metric, &policy_id_buf); + _ = evalMetric(&engine, &test_metric, &policy_id_buf); + + var test_span: TestTraceContext = .{ .name = "GET /users" }; + _ = evalTrace(&engine, &test_span, &policy_id_buf); + _ = evalTrace(&engine, &test_span, &policy_id_buf); + _ = evalTrace(&engine, &test_span, &policy_id_buf); + + // One counter per signal, no cross-talk between them. + const volume = registry.volume.readAndReset(); + try testing.expectEqual(@as(i64, 1), volume.log_records); + try testing.expectEqual(@as(i64, 2), volume.metric_data_points); + try testing.expectEqual(@as(i64, 3), volume.spans); + + // Bytes stay untracked unless the consumer opts in, per signal. + registry.volume.addBytes(.metric, 42); + const bytes = registry.volume.readAndReset(); + try testing.expectEqual(@as(i64, 42), bytes.metric_bytes); + try testing.expectEqual(@as(i64, 0), bytes.log_bytes); + try testing.expectEqual(@as(i64, 0), bytes.span_bytes); +} + test "PolicyEngine: single policy drop match" { const allocator = testing.allocator; diff --git a/src/policy/provider.zig b/src/policy/provider.zig index 82b6121..2c91e22 100644 --- a/src/policy/provider.zig +++ b/src/policy/provider.zig @@ -25,9 +25,9 @@ pub const PolicyStatsSnapshot = struct { /// at their pre-policy size. /// /// Every field is independently optional: `0` means "not tracked" as much as -/// "none seen". Byte counts are only populated when the caller passes -/// `EvaluateOptions.record_bytes` — the engine reads records through accessors -/// and has no serialized form to measure. +/// "none seen". Byte counts are only populated when the consumer calls +/// `registry.volume.addBytes` — the engine reads records through accessors and +/// has no serialized form to measure. pub const VolumeSnapshot = struct { log_records: i64 = 0, log_bytes: i64 = 0, @@ -132,3 +132,61 @@ pub const PolicyCallback = struct { try self.onUpdate(self.context, update); } }; + +// ============================================================================= +// Tests +// ============================================================================= + +const testing = std.testing; + +test "VolumeSnapshot.toProto: every field maps to its own proto field" { + // Distinct values so a transposed pair (e.g. metric_bytes ↔ span_bytes) + // fails instead of silently reporting the wrong signal's volume. + const snap: VolumeSnapshot = .{ + .log_records = 11, + .log_bytes = 22, + .metric_data_points = 33, + .metric_bytes = 44, + .spans = 55, + .span_bytes = 66, + }; + + const out = snap.toProto(); + try testing.expectEqual(@as(i64, 11), out.log_records); + try testing.expectEqual(@as(i64, 22), out.log_bytes); + try testing.expectEqual(@as(i64, 33), out.metric_data_points); + try testing.expectEqual(@as(i64, 44), out.metric_bytes); + try testing.expectEqual(@as(i64, 55), out.spans); + try testing.expectEqual(@as(i64, 66), out.span_bytes); +} + +test "VolumeSnapshot.isZero: any single tracked field defeats omission" { + try testing.expect((VolumeSnapshot{}).isZero()); + + // A consumer tracking only one signal, or only bytes, must still be + // reported — omission is reserved for "nothing observed at all". + inline for (@typeInfo(VolumeSnapshot).@"struct".fields) |f| { + var snap: VolumeSnapshot = .{}; + @field(snap, f.name) = 1; + try testing.expect(!snap.isZero()); + } +} + +test "StatsCollector: volume seam is optional and no-ops when unwired" { + // A collector from a provider the registry never gave volume fns to (or an + // older consumer building one by hand) must not crash and must report no + // volume. + var ctx: u8 = 0; + const collector: StatsCollector = .{ + .context = &ctx, + .collect = struct { + fn collect(_: std.mem.Allocator, _: *anyopaque) anyerror![]PolicyStatsSnapshot { + return &.{}; + } + }.collect, + }; + + try testing.expect(collector.drainVolume().isZero()); + collector.returnVolume(.{ .log_records = 5 }); // discarded, not a crash + try testing.expect(collector.drainVolume().isZero()); +} diff --git a/src/policy/provider_http.zig b/src/policy/provider_http.zig index ff07a4a..c3bef11 100644 --- a/src/policy/provider_http.zig +++ b/src/policy/provider_http.zig @@ -419,8 +419,9 @@ pub const HttpProvider = struct { // Drain total observed volume (v1.7.0). Unlike per-policy stats, this // is handed back if the sync fails: it is the denominator for every // policy's match rate, so a dropped interval skews usage rather than - // just losing a tail. `volume_sent` gates the add-back so a sync the - // server accepted is never counted twice. + // just losing a tail. `volume_sent` gates the add-back so a completed + // sync is never counted twice. The reading lives on this call's stack + // until then, so overlapping syncs each drain a disjoint delta. const volume: VolumeSnapshot = if (self.stats_collector) |c| c.drainVolume() else .{}; var volume_sent = false; errdefer if (!volume_sent) self.retainVolume(volume); @@ -539,11 +540,6 @@ pub const HttpProvider = struct { return error.HttpRequestFailed; } - // The server accepted the request, so the reported interval is closed: - // the drained counters stay drained even if decoding the response below - // fails. - volume_sent = true; - const sent_event: HttpSyncRequestSucceeded = .{ .url = self.config_url, .policy_statuses_sent = policy_statuses_list.items.len, @@ -566,6 +562,11 @@ pub const HttpProvider = struct { return err; }; + // Only a fully decoded response closes the reported interval. A 200 with + // an unparseable body is still a failed sync from our side, so its + // volume goes back for the next attempt (matching policy-go). + volume_sent = true; + return .{ .parsed = parsed, .response_body = response_body, @@ -689,6 +690,68 @@ test "HttpProvider: refreshClientClock advances stale now but leaves first reque try testing.expect(provider.http_client.now.?.nanoseconds > stale.nanoseconds); } +/// Minimal control-plane stub: accept one POST, record its body, answer with +/// `response` (200 and an empty SyncResponse by default). Serves exactly one +/// request, so an unexpected extra sync is observable as a hang/failure. +const SyncStub = struct { + allocator: std.mem.Allocator, + listener: std.Io.net.Server, + port: u16, + body: []u8 = &.{}, + served: bool = false, + /// Response body. Set to something unparseable to exercise the decode + /// failure path. + response: []const u8 = "{}", + + fn start(a: std.mem.Allocator, sio: std.Io) !SyncStub { + var attempt: u16 = 0; + while (attempt < 32) : (attempt += 1) { + const port: u16 = 42801 + attempt; + const addr = try std.Io.net.IpAddress.parse("127.0.0.1", port); + const listener = addr.listen(sio, .{}) catch |err| switch (err) { + error.AddressInUse => continue, + else => return err, + }; + return .{ .allocator = a, .listener = listener, .port = port }; + } + return error.AddressInUse; + } + + fn deinit(self: *SyncStub, sio: std.Io) void { + defer self.* = undefined; + self.listener.deinit(sio); + if (self.body.len > 0) self.allocator.free(self.body); + } + + fn serve(self: *SyncStub, sio: std.Io) void { + var stream = self.listener.accept(sio) catch return; + defer stream.close(sio); + var recv_buf: [16 * 1024]u8 = undefined; + var send_buf: [4 * 1024]u8 = undefined; + var conn_reader = stream.reader(sio, &recv_buf); + var conn_writer = stream.writer(sio, &send_buf); + var http_server = std.http.Server.init(&conn_reader.interface, &conn_writer.interface); + var request = http_server.receiveHead() catch return; + var transfer_buf: [8 * 1024]u8 = undefined; + const body_reader = request.readerExpectContinue(&transfer_buf) catch return; + self.body = body_reader.allocRemaining(self.allocator, .limited(1 << 20)) catch return; + self.served = true; + request.respond(self.response, .{ .status = .ok }) catch return; + } + + /// Start a stub plus a provider pointed at it, in the shape every sync test + /// needs. Caller owns both (see the deinit pattern in the tests below). + fn provider(self: *const SyncStub, a: std.mem.Allocator, sio: std.Io, bus: *EventBus) !*HttpProvider { + var url_buf: [64]u8 = undefined; + const url = try std.fmt.bufPrint(&url_buf, "http://127.0.0.1:{d}/sync", .{self.port}); + return HttpProvider.init(a, sio, bus, .{ + .id = "p", + .url = url, + .poll_interval_seconds = 3600, + }); + } +}; + test "HttpProvider.close: flushes final stats once, then is idempotent" { const allocator = testing.allocator; // Own Threaded io so the stub control-plane server gets a real thread, @@ -697,58 +760,9 @@ test "HttpProvider.close: flushes final stats once, then is idempotent" { defer threaded.deinit(); const io = threaded.io(); - // Minimal control-plane stub: accept one POST, record its body, answer 200 - // with an empty SyncResponse. Refuses to serve more than one request so a - // second close() firing a redundant sync would be observable as a hang/fail. - const Stub = struct { - const Self = @This(); - - allocator: std.mem.Allocator, - listener: std.Io.net.Server, - port: u16, - body: []u8 = &.{}, - served: bool = false, - - fn start(a: std.mem.Allocator, sio: std.Io) !Self { - var attempt: u16 = 0; - while (attempt < 32) : (attempt += 1) { - const port: u16 = 42801 + attempt; - const addr = try std.Io.net.IpAddress.parse("127.0.0.1", port); - const listener = addr.listen(sio, .{}) catch |err| switch (err) { - error.AddressInUse => continue, - else => return err, - }; - return .{ .allocator = a, .listener = listener, .port = port }; - } - return error.AddressInUse; - } - - fn deinit(self: *Self, sio: std.Io) void { - defer self.* = undefined; - self.listener.deinit(sio); - if (self.body.len > 0) self.allocator.free(self.body); - } - - fn serve(self: *Self, sio: std.Io) void { - var stream = self.listener.accept(sio) catch return; - defer stream.close(sio); - var recv_buf: [16 * 1024]u8 = undefined; - var send_buf: [4 * 1024]u8 = undefined; - var conn_reader = stream.reader(sio, &recv_buf); - var conn_writer = stream.writer(sio, &send_buf); - var http_server = std.http.Server.init(&conn_reader.interface, &conn_writer.interface); - var request = http_server.receiveHead() catch return; - var transfer_buf: [8 * 1024]u8 = undefined; - const body_reader = request.readerExpectContinue(&transfer_buf) catch return; - self.body = body_reader.allocRemaining(self.allocator, .limited(1 << 20)) catch return; - self.served = true; - request.respond("{}", .{ .status = .ok }) catch return; - } - }; - - var stub = try Stub.start(allocator, io); + var stub = try SyncStub.start(allocator, io); defer stub.deinit(io); - var server_future = io.concurrent(Stub.serve, .{ &stub, io }) catch + var server_future = io.concurrent(SyncStub.serve, .{ &stub, io }) catch return error.SkipZigTest; defer server_future.cancel(io); @@ -842,6 +856,145 @@ test "HttpProvider.close: propagates the final sync error" { try provider.close(); } +test "HttpProvider: an all-zero volume is omitted from the sync request" { + const allocator = testing.allocator; + var threaded: std.Io.Threaded = .init(allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + + var stub = try SyncStub.start(allocator, io); + defer stub.deinit(io); + var server_future = io.concurrent(SyncStub.serve, .{ &stub, io }) catch + return error.SkipZigTest; + defer server_future.cancel(io); + + var noop_bus: o11y.NoopEventBus = undefined; + noop_bus.init(io); + + var provider = try stub.provider(allocator, io, noop_bus.eventBus()); + defer provider.deinit(); + + // Collector wired, but nothing observed yet: the spec says omit the message + // rather than send a zero-valued one. + var ctx: u8 = 0; + provider.setStatsCollector(.{ + .context = &ctx, + .collect = struct { + fn collect(_: std.mem.Allocator, _: *anyopaque) anyerror![]PolicyStatsSnapshot { + return &.{}; + } + }.collect, + .collect_volume = struct { + fn drain(_: *anyopaque) VolumeSnapshot { + return .{}; + } + }.drain, + }); + + try provider.close(); + server_future.await(io); + + try testing.expect(stub.served); + try testing.expect(std.mem.indexOf(u8, stub.body, "volume") == null); +} + +test "HttpProvider: a provider with no volume seam wired syncs without volume" { + const allocator = testing.allocator; + var threaded: std.Io.Threaded = .init(allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + + var stub = try SyncStub.start(allocator, io); + defer stub.deinit(io); + var server_future = io.concurrent(SyncStub.serve, .{ &stub, io }) catch + return error.SkipZigTest; + defer server_future.cancel(io); + + var noop_bus: o11y.NoopEventBus = undefined; + noop_bus.init(io); + + var provider = try stub.provider(allocator, io, noop_bus.eventBus()); + defer provider.deinit(); + + // A collector built without the volume fns (an older consumer, or a + // provider the registry never handed them to) must still sync cleanly. + var ctx: u8 = 0; + provider.setStatsCollector(.{ + .context = &ctx, + .collect = struct { + fn collect(arena: std.mem.Allocator, _: *anyopaque) anyerror![]PolicyStatsSnapshot { + const rows = try arena.alloc(PolicyStatsSnapshot, 1); + rows[0] = .{ .id = "p", .hits = 1 }; + return rows; + } + }.collect, + }); + + try provider.close(); + server_future.await(io); + + try testing.expect(stub.served); + // Per-policy stats still reported; volume simply absent. + try testing.expect(std.mem.indexOf(u8, stub.body, "matchHits") != null); + try testing.expect(std.mem.indexOf(u8, stub.body, "volume") == null); +} + +test "HttpProvider: volume is retained when a 200 response cannot be decoded" { + const allocator = testing.allocator; + var threaded: std.Io.Threaded = .init(allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + + var stub = try SyncStub.start(allocator, io); + defer stub.deinit(io); + // 200, but the body is not a SyncResponse: the sync failed from our side, + // so the interval stays open (matching policy-go). + stub.response = "not json at all"; + var server_future = io.concurrent(SyncStub.serve, .{ &stub, io }) catch + return error.SkipZigTest; + defer server_future.cancel(io); + + var noop_bus: o11y.NoopEventBus = undefined; + noop_bus.init(io); + + var provider = try stub.provider(allocator, io, noop_bus.eventBus()); + defer provider.deinit(); + + const Counters = struct { + var pending: VolumeSnapshot = .{}; + fn collect(_: std.mem.Allocator, _: *anyopaque) anyerror![]PolicyStatsSnapshot { + return &.{}; + } + fn drain(_: *anyopaque) VolumeSnapshot { + defer pending = .{}; + return pending; + } + fn restore(_: *anyopaque, snapshot: VolumeSnapshot) void { + pending.metric_data_points += snapshot.metric_data_points; + pending.metric_bytes += snapshot.metric_bytes; + } + }; + Counters.pending = .{ .metric_data_points = 4, .metric_bytes = 256 }; + + var ctx: u8 = 0; + provider.setStatsCollector(.{ + .context = &ctx, + .collect = Counters.collect, + .collect_volume = Counters.drain, + .restore_volume = Counters.restore, + }); + + if (provider.close()) |_| { + try testing.expect(false); // expected the undecodable body to error + } else |_| {} + server_future.await(io); + + // The request did carry the volume, but we can't confirm the server took it. + try testing.expect(std.mem.indexOf(u8, stub.body, "\"metricDataPoints\":\"4\"") != null); + try testing.expectEqual(@as(i64, 4), Counters.pending.metric_data_points); + try testing.expectEqual(@as(i64, 256), Counters.pending.metric_bytes); +} + test "HttpProvider: a failed sync hands its volume back for the next attempt" { const allocator = testing.allocator; var threaded: std.Io.Threaded = .init(allocator, .{}); diff --git a/src/policy/registry.zig b/src/policy/registry.zig index 9391b22..2b553c5 100644 --- a/src/policy/registry.zig +++ b/src/policy/registry.zig @@ -73,17 +73,33 @@ pub const VolumeCounters = struct { spans: std.atomic.Value(i64) = .init(0), span_bytes: std.atomic.Value(i64) = .init(0), - /// Count one record of signal `T`. `bytes` is the caller's estimate of the - /// record's uncompressed OTLP protobuf size; 0 means "not tracked" and - /// leaves the byte counter alone. - pub inline fn record(self: *VolumeCounters, comptime T: policy_types.TelemetryType, bytes: usize) void { - const counters = switch (T) { - .log => .{ &self.log_records, &self.log_bytes }, - .metric => .{ &self.metric_data_points, &self.metric_bytes }, - .trace => .{ &self.spans, &self.span_bytes }, + /// Count one record of signal `T`. Called by the engine for every record + /// entering evaluation; consumers never call this directly. + pub inline fn record(self: *VolumeCounters, comptime T: policy_types.TelemetryType) void { + const counter = switch (T) { + .log => &self.log_records, + .metric => &self.metric_data_points, + .trace => &self.spans, }; - _ = counters[0].fetchAdd(1, .monotonic); - if (bytes != 0) _ = counters[1].fetchAdd(@intCast(bytes), .monotonic); + _ = counter.fetchAdd(1, .monotonic); + } + + /// Add to the reported byte volume for signal `T`. Records are counted + /// automatically by `evaluate`; bytes are opt-in and must be the + /// uncompressed OTLP protobuf serialized size of the records as received + /// (an estimate is fine) — the engine reads records through accessors and + /// has no serialized form to measure. Leave it unreported rather than + /// reporting another encoding's size. + /// + /// Typically called once per received batch with the batch's serialized + /// size, not once per record. + pub inline fn addBytes(self: *VolumeCounters, comptime T: policy_types.TelemetryType, bytes: i64) void { + const counter = switch (T) { + .log => &self.log_bytes, + .metric => &self.metric_bytes, + .trace => &self.span_bytes, + }; + _ = counter.fetchAdd(bytes, .monotonic); } /// Read and zero every counter, for reporting on a sync request. @@ -1820,9 +1836,11 @@ test "PolicyRegistry: volume counters survive recompiles and accept add-back" { var registry = PolicyRegistry.init(allocator, noop_bus.eventBus()); defer registry.deinit(); - registry.volume.record(.log, 100); - registry.volume.record(.metric, 0); // bytes untracked for this record - registry.volume.record(.trace, 40); + registry.volume.record(.log); + registry.volume.addBytes(.log, 100); + registry.volume.record(.metric); // metric bytes left untracked + registry.volume.record(.trace); + registry.volume.addBytes(.trace, 40); // A recompile allocates fresh per-policy stats; volume lives outside the // snapshot and must not be zeroed by it. @@ -1841,7 +1859,8 @@ test "PolicyRegistry: volume counters survive recompiles and accept add-back" { // A failed sync hands the reading back; records seen in the meantime are // preserved, so the next attempt reports the sum. - registry.volume.record(.log, 7); + registry.volume.record(.log); + registry.volume.addBytes(.log, 7); registry.volume.add(drained); const retried = registry.volume.readAndReset(); try testing.expectEqual(@as(i64, 2), retried.log_records); @@ -1849,6 +1868,85 @@ test "PolicyRegistry: volume counters survive recompiles and accept add-back" { try testing.expectEqual(@as(i64, 1), retried.spans); } +test "VolumeCounters: add folds every field back into its own counter" { + var counters: VolumeCounters = .{}; + + // All six distinct and non-zero, so a field dropped or crossed over in + // `add` shows up instead of being masked by a zero. + counters.add(.{ + .log_records = 1, + .log_bytes = 2, + .metric_data_points = 3, + .metric_bytes = 4, + .spans = 5, + .span_bytes = 6, + }); + // Folding is additive, not assignment: a second add sums. + counters.add(.{ + .log_records = 10, + .log_bytes = 20, + .metric_data_points = 30, + .metric_bytes = 40, + .spans = 50, + .span_bytes = 60, + }); + + const out = counters.readAndReset(); + try testing.expectEqual(@as(i64, 11), out.log_records); + try testing.expectEqual(@as(i64, 22), out.log_bytes); + try testing.expectEqual(@as(i64, 33), out.metric_data_points); + try testing.expectEqual(@as(i64, 44), out.metric_bytes); + try testing.expectEqual(@as(i64, 55), out.spans); + try testing.expectEqual(@as(i64, 66), out.span_bytes); + try testing.expect(counters.readAndReset().isZero()); +} + +test "PolicyRegistry: subscribe wires the volume seam to this registry" { + const provider_http = @import("./provider_http.zig"); + + const allocator = testing.allocator; + var threaded: std.Io.Threaded = .init(allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + + var noop_bus: NoopEventBus = undefined; + noop_bus.init(io); + var registry = PolicyRegistry.init(allocator, noop_bus.eventBus()); + defer registry.deinit(); + + // Port 1 has nothing listening: subscribe's initial fetch fails and is + // warned about, which is fine — the wiring under test happens before it. + var provider = try provider_http.HttpProvider.init( + allocator, + io, + noop_bus.eventBus(), + .{ .id = "p", .url = "http://127.0.0.1:1/sync", .poll_interval_seconds = 3600 }, + ); + defer provider.deinit(); + try registry.subscribe(.{ .http = provider }); + // Stop the poll thread before the registry it drains goes away. + provider.shutdown(); + + // Nothing else in the suite proves subscribe passed the volume fns (the + // provider tests install collectors by hand), so assert the seam exists and + // that both directions land on *this* registry's counters. + const collector = provider.stats_collector orelse return error.NoCollector; + try testing.expect(collector.collect_volume != null); + try testing.expect(collector.restore_volume != null); + + registry.volume.record(.log); + registry.volume.addBytes(.log, 64); + const drained = collector.drainVolume(); + try testing.expectEqual(@as(i64, 1), drained.log_records); + try testing.expectEqual(@as(i64, 64), drained.log_bytes); + try testing.expect(registry.volume.readAndReset().isZero()); + + collector.returnVolume(drained); + const restored = registry.volume.readAndReset(); + try testing.expectEqual(@as(i64, 1), restored.log_records); + try testing.expectEqual(@as(i64, 64), restored.log_bytes); +} + test "PolicyRegistry: collectStats reports exactly the current policy set across updates" { const allocator = testing.allocator; var noop_bus: NoopEventBus = undefined; From c57971578b788173248e194d03c34d546135d7ac Mon Sep 17 00:00:00 2001 From: jaronoff97 Date: Wed, 5 Aug 2026 14:50:04 -0400 Subject: [PATCH 3/4] fix bug improve coverage --- src/policy/provider_http.zig | 106 ++++++++++++++++++++++++++++++++++- 1 file changed, 103 insertions(+), 3 deletions(-) diff --git a/src/policy/provider_http.zig b/src/policy/provider_http.zig index c3bef11..af08672 100644 --- a/src/policy/provider_http.zig +++ b/src/policy/provider_http.zig @@ -36,6 +36,9 @@ const HttpPoliciesUnchanged = struct { reason: []const u8 }; const HttpPolicyHashUpdated = struct { hash: []const u8 }; const HttpPoliciesLoaded = struct { count: usize, url: []const u8, sync_timestamp: u64 }; const HttpSyncRequestFailed = struct { url: []const u8, status: u16 }; +// A 200 whose SyncResponse carries `error_message`: the control plane rejected +// the sync in-band, which is a failure even though the transport succeeded. +const HttpSyncRejected = struct { url: []const u8, err: []const u8 }; const HttpSyncRequestSucceeded = struct { url: []const u8, policy_statuses_sent: usize }; // Emitted once per policy status about to be sent, so we can confirm at runtime // exactly which policies (and counts) are reported each sync. @@ -562,9 +565,26 @@ pub const HttpProvider = struct { return err; }; - // Only a fully decoded response closes the reported interval. A 200 with - // an unparseable body is still a failed sync from our side, so its - // volume goes back for the next attempt (matching policy-go). + // A 200 carrying `error_message` is an in-band rejection: the server + // did not accept this sync, so it must not advance the hash, deliver + // policies, or close the volume interval. Checked before `volume_sent` + // for the same reason policy-go and policy-rs check it before touching + // sync state. + if (parsed.value.error_message.len > 0) { + var mut = parsed; + defer mut.deinit(); + const event: HttpSyncRejected = .{ + .url = self.config_url, + .err = parsed.value.error_message, + }; + self.bus.err(event); + return error.SyncRejected; + } + + // Only a fully accepted response closes the reported interval. A 200 + // with an unparseable or rejected body is still a failed sync from our + // side, so its volume goes back for the next attempt (matching + // policy-go and policy-rs). volume_sent = true; return .{ @@ -995,6 +1015,86 @@ test "HttpProvider: volume is retained when a 200 response cannot be decoded" { try testing.expectEqual(@as(i64, 256), Counters.pending.metric_bytes); } +test "HttpProvider: a 200 carrying error_message is a failed sync and retains volume" { + const allocator = testing.allocator; + var threaded: std.Io.Threaded = .init(allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + + var stub = try SyncStub.start(allocator, io); + defer stub.deinit(io); + // Well-formed SyncResponse carrying policies, a hash and a timestamp, but + // the control plane rejected the sync in-band. None of it may be adopted. + stub.response = + \\{"policies":[{"id":"p1","enabled":true}], + \\ "hash":"should-not-be-adopted", + \\ "syncTimestampUnixNano":"1234", + \\ "errorMessage":"client not provisioned"} + ; + var server_future = io.concurrent(SyncStub.serve, .{ &stub, io }) catch + return error.SkipZigTest; + defer server_future.cancel(io); + + var noop_bus: o11y.NoopEventBus = undefined; + noop_bus.init(io); + + var provider = try stub.provider(allocator, io, noop_bus.eventBus()); + defer provider.deinit(); + + const Counters = struct { + var pending: VolumeSnapshot = .{}; + fn collect(_: std.mem.Allocator, _: *anyopaque) anyerror![]PolicyStatsSnapshot { + return &.{}; + } + fn drain(_: *anyopaque) VolumeSnapshot { + defer pending = .{}; + return pending; + } + fn restore(_: *anyopaque, snapshot: VolumeSnapshot) void { + pending.spans += snapshot.spans; + pending.span_bytes += snapshot.span_bytes; + } + }; + Counters.pending = .{ .spans = 3, .span_bytes = 96 }; + + var ctx: u8 = 0; + provider.setStatsCollector(.{ + .context = &ctx, + .collect = Counters.collect, + .collect_volume = Counters.drain, + .restore_volume = Counters.restore, + }); + + // Wired directly rather than via subscribe(), which would fire its own + // initial fetch and consume the stub's single response. + const Sink = struct { + var notified: bool = false; + fn onUpdate(_: *anyopaque, _: policy_provider.PolicyUpdate) anyerror!void { + notified = true; + return; + } + }; + Sink.notified = false; + provider.callback = .{ .context = &ctx, .onUpdate = Sink.onUpdate }; + + if (provider.close()) |_| { + try testing.expect(false); // an in-band rejection is not a success + } else |err| { + try testing.expectEqual(error.SyncRejected, err); + } + server_future.await(io); + + // Volume survives for the next attempt… + try testing.expectEqual(@as(i64, 3), Counters.pending.spans); + try testing.expectEqual(@as(i64, 96), Counters.pending.span_bytes); + // …the rejected response's hash was not adopted as last-successful… + try testing.expect(provider.last_successful_hash == null); + // …its timestamp did not advance, so the next attempt is still a full sync… + try testing.expectEqual(@as(u64, 0), provider.last_sync_timestamp); + // …and its policies were never delivered to the registry. + try testing.expect(!Sink.notified); +} + test "HttpProvider: a failed sync hands its volume back for the next attempt" { const allocator = testing.allocator; var threaded: std.Io.Threaded = .init(allocator, .{}); From 16d32bf9a7597c562a1e7278097d6ad0fd0b9543 Mon Sep 17 00:00:00 2001 From: jaronoff97 Date: Wed, 5 Aug 2026 15:12:30 -0400 Subject: [PATCH 4/4] go off of 1.7.1 --- README.md | 8 +- proto/tero/policy/v1/policy.proto | 10 +- src/policy/policy_engine.zig | 2 +- src/policy/provider.zig | 29 +++--- src/policy/provider_http.zig | 166 +++++------------------------- src/policy/registry.zig | 80 +++----------- src/proto/tero/policy/v1.pb.zig | 10 +- 7 files changed, 75 insertions(+), 230 deletions(-) diff --git a/README.md b/README.md index 4a295ac..00eb4fd 100644 --- a/README.md +++ b/README.md @@ -116,9 +116,11 @@ const result = engine.evaluate(.log, &my_log_accessor, &my_log_ctx, &policy_id_b .io = io, }); -// Volume (spec v1.7.0): every evaluate call is counted regardless of match and -// reported on the next sync. Byte volume is opt-in — pass the batch's -// uncompressed OTLP protobuf size, which only the caller can measure. +// Volume (spec v1.7.1): every evaluate call is counted regardless of match and +// reported on the next sync, which resets the counters whether or not that sync +// succeeds — reported volume is a lower bound, never replayed. Byte volume is +// opt-in: pass the batch's uncompressed OTLP protobuf size, which only the +// caller can measure. registry.volume.addBytes(.log, @intCast(request_size_bytes)); // Transforms whose required primitive (set/delete/move) is unwired on the diff --git a/proto/tero/policy/v1/policy.proto b/proto/tero/policy/v1/policy.proto index 7ef2d57..e2a3ee5 100644 --- a/proto/tero/policy/v1/policy.proto +++ b/proto/tero/policy/v1/policy.proto @@ -128,8 +128,14 @@ message PolicySyncStatus { // VolumeStats reports the total telemetry a client observed since the last // sync, regardless of whether any policy matched it. Counts are of records -// entering policy evaluation, before any keep or transform stage runs, and are -// reset on each successful sync. +// entering policy evaluation, before any keep or transform stage runs. +// +// Counters are reset when they are read into a sync request, whether or not +// that sync then succeeds — the same rule PolicySyncStatus.match_hits and +// match_misses follow. A failed sync loses its interval from the numerator and +// the denominator alike, so match rates stay meaningful; counters from a failed +// sync must never be replayed, since the server cannot tell a replay from new +// telemetry. Reported volume is a lower bound, not an exact total. // // Reporting volume is optional, and every field is individually optional: an // implementation may report record counts without byte counts, or a subset of diff --git a/src/policy/policy_engine.zig b/src/policy/policy_engine.zig index e8db913..25b2a1d 100644 --- a/src/policy/policy_engine.zig +++ b/src/policy/policy_engine.zig @@ -336,7 +336,7 @@ pub const PolicyEngine = struct { policy_id_buf: [][]const u8, options: EvaluateOptions, ) PolicyResult { - // Volume tracking (v1.7.0): every record entering evaluation counts, + // Volume tracking (v1.7.1): every record entering evaluation counts, // regardless of match — including records with no policies loaded for // their signal, so this must precede both early returns below. Byte // volume is opt-in via `registry.volume.addBytes`. diff --git a/src/policy/provider.zig b/src/policy/provider.zig index 2c91e22..72397c3 100644 --- a/src/policy/provider.zig +++ b/src/policy/provider.zig @@ -19,11 +19,18 @@ pub const PolicyStatsSnapshot = struct { errors: []const []const u8 = &.{}, }; -/// Total telemetry that entered policy evaluation since the last successful -/// sync, regardless of match (spec v1.7.0 `VolumeStats`). Counted before the +/// Total telemetry that entered policy evaluation since the counters were last +/// read, regardless of match (spec v1.7.1 `VolumeStats`). Counted before the /// keep and transform stages, so dropped/sampled/redacted records are included /// at their pre-policy size. /// +/// Reported at most once: draining resets, whether or not the sync then +/// succeeds, so a failed sync loses its interval rather than replaying it. That +/// keeps volume on the same footing as `hits`/`misses` above — retaining one +/// side of `(hits + misses) / ` without the other would skew the +/// ratio — and a replay would double count, since the server has no interval +/// identifier to dedupe on. Volume is a lower bound, not an exact total. +/// /// Every field is independently optional: `0` means "not tracked" as much as /// "none seen". Byte counts are only populated when the consumer calls /// `registry.volume.addBytes` — the engine reads records through accessors and @@ -64,14 +71,10 @@ pub const VolumeSnapshot = struct { pub const StatsCollector = struct { context: *anyopaque, collect: *const fn (arena: std.mem.Allocator, context: *anyopaque) anyerror![]PolicyStatsSnapshot, - /// Drain the registry's volume counters (spec v1.7.0). Optional: a - /// collector that leaves it null reports no volume, which is conformant. + /// Drain the registry's volume counters (spec v1.7.1), resetting them. + /// Optional: a collector that leaves it null reports no volume, which is + /// conformant. collect_volume: ?*const fn (context: *anyopaque) VolumeSnapshot = null, - /// Add a previously drained reading back, so a failed sync retains its - /// counters for the next attempt as the spec requires. Concurrent - /// evaluation keeps incrementing across the drain, hence add-back rather - /// than restore-by-assignment. - restore_volume: ?*const fn (context: *anyopaque, snapshot: VolumeSnapshot) void = null, pub fn call(self: StatsCollector, arena: std.mem.Allocator) anyerror![]PolicyStatsSnapshot { return self.collect(arena, self.context); @@ -81,11 +84,6 @@ pub const StatsCollector = struct { const f = self.collect_volume orelse return .{}; return f(self.context); } - - pub fn returnVolume(self: StatsCollector, snapshot: VolumeSnapshot) void { - const f = self.restore_volume orelse return; - f(self.context, snapshot); - } }; /// Extension sync plumbing (spec v1.6.0), implemented outside policy_zig by @@ -173,7 +171,7 @@ test "VolumeSnapshot.isZero: any single tracked field defeats omission" { } test "StatsCollector: volume seam is optional and no-ops when unwired" { - // A collector from a provider the registry never gave volume fns to (or an + // A collector from a provider the registry never gave a volume fn to (or an // older consumer building one by hand) must not crash and must report no // volume. var ctx: u8 = 0; @@ -187,6 +185,5 @@ test "StatsCollector: volume seam is optional and no-ops when unwired" { }; try testing.expect(collector.drainVolume().isZero()); - collector.returnVolume(.{ .log_records = 5 }); // discarded, not a crash try testing.expect(collector.drainVolume().isZero()); } diff --git a/src/policy/provider_http.zig b/src/policy/provider_http.zig index af08672..5b5bb1f 100644 --- a/src/policy/provider_http.zig +++ b/src/policy/provider_http.zig @@ -46,8 +46,6 @@ const HttpSyncStatusReported = struct { id: []const u8, match_hits: i64, match_m // Emitted when a sync carries volume, for the same reason: confirming at // runtime what was reported. Omitted volume (all-zero) emits nothing. const HttpSyncVolumeReported = struct { volume: VolumeSnapshot }; -// A failed sync hands its drained volume back for the next attempt. -const HttpSyncVolumeRetained = struct { volume: VolumeSnapshot }; const HttpFetchStarted = struct {}; const HttpFetchCompleted = struct {}; @@ -190,16 +188,6 @@ pub const HttpProvider = struct { self.stats_collector = collector; } - /// Hand a drained volume reading back to the registry after a failed sync, - /// so the next attempt includes it (spec v1.7.0). - fn retainVolume(self: *HttpProvider, volume: VolumeSnapshot) void { - if (volume.isZero()) return; - const collector = self.stats_collector orelse return; - collector.returnVolume(volume); - const event: HttpSyncVolumeRetained = .{ .volume = volume }; - self.bus.warn(event); - } - pub fn subscribe(self: *HttpProvider, callback: PolicyCallback) !void { self.callback = callback; @@ -230,11 +218,9 @@ pub const HttpProvider = struct { /// Returns the final sync's error rather than swallowing it — the caller /// decides whether a lost tail report is worth logging or acting on. The /// flush is attempted at most once: `flushed` is set before the sync (the - /// stats collector resets its per-policy counters when pulled, so the tail - /// is gone on failure and a retry would only report zeros), so a second - /// call is a no-op and the loop is stopped either way. Volume counters are - /// the exception — a failed sync hands them back — but with the loop - /// stopped there is no further sync to carry them. + /// stats collector resets its counters when pulled, so the tail is gone on + /// failure and a retry would only report zeros), so a second call is a + /// no-op and the loop is stopped either way. /// /// Reuses the io bound at init (the same process-wide io is still live at /// shutdown), so this must run before that io is torn down — i.e. in the @@ -419,15 +405,11 @@ pub const HttpProvider = struct { const policy_statuses_list = try statsToSyncStatuses(temp_allocator, stats); - // Drain total observed volume (v1.7.0). Unlike per-policy stats, this - // is handed back if the sync fails: it is the denominator for every - // policy's match rate, so a dropped interval skews usage rather than - // just losing a tail. `volume_sent` gates the add-back so a completed - // sync is never counted twice. The reading lives on this call's stack - // until then, so overlapping syncs each drain a disjoint delta. + // Drain total observed volume (v1.7.1). Draining is the reset, exactly + // like the per-policy counters above: if this sync fails its interval is + // gone rather than replayed, since the server cannot tell a replay from + // new telemetry. Reported volume is a lower bound, not an exact total. const volume: VolumeSnapshot = if (self.stats_collector) |c| c.drainVolume() else .{}; - var volume_sent = false; - errdefer if (!volume_sent) self.retainVolume(volume); // Log exactly what we're about to report, so runtime expectations can be // verified against the live snapshot. @@ -565,11 +547,10 @@ pub const HttpProvider = struct { return err; }; - // A 200 carrying `error_message` is an in-band rejection: the server - // did not accept this sync, so it must not advance the hash, deliver - // policies, or close the volume interval. Checked before `volume_sent` - // for the same reason policy-go and policy-rs check it before touching - // sync state. + // A 200 carrying `error_message` is an in-band rejection: the server did + // not accept this sync, so it must not advance the hash or deliver + // policies. Checked before touching sync state, as policy-go and + // policy-rs do. if (parsed.value.error_message.len > 0) { var mut = parsed; defer mut.deinit(); @@ -581,12 +562,6 @@ pub const HttpProvider = struct { return error.SyncRejected; } - // Only a fully accepted response closes the reported interval. A 200 - // with an unparseable or rejected body is still a failed sync from our - // side, so its volume goes back for the next attempt (matching - // policy-go and policy-rs). - volume_sent = true; - return .{ .parsed = parsed, .response_body = response_body, @@ -804,7 +779,6 @@ test "HttpProvider.close: flushes final stats once, then is idempotent" { // the tail stat that must reach the control plane before teardown. const Collector = struct { var calls: usize = 0; - var restores: usize = 0; fn collect(arena: std.mem.Allocator, _: *anyopaque) anyerror![]PolicyStatsSnapshot { calls += 1; const rows = try arena.alloc(PolicyStatsSnapshot, 1); @@ -814,18 +788,13 @@ test "HttpProvider.close: flushes final stats once, then is idempotent" { fn drain(_: *anyopaque) VolumeSnapshot { return .{ .log_records = 9, .log_bytes = 512 }; } - fn restore(_: *anyopaque, _: VolumeSnapshot) void { - restores += 1; - } }; Collector.calls = 0; - Collector.restores = 0; var ctx: u8 = 0; provider.setStatsCollector(.{ .context = &ctx, .collect = Collector.collect, .collect_volume = Collector.drain, - .restore_volume = Collector.restore, }); try provider.close(); @@ -837,10 +806,9 @@ test "HttpProvider.close: flushes final stats once, then is idempotent" { // The final sync carried the tail stats: the policy id and its hit count. try testing.expect(std.mem.indexOf(u8, stub.body, "hot") != null); try testing.expect(std.mem.indexOf(u8, stub.body, "matchHits") != null); - // …and the observed volume, which the accepted sync does not hand back. + // …and the observed volume. try testing.expect(std.mem.indexOf(u8, stub.body, "\"logRecords\":\"9\"") != null); try testing.expect(std.mem.indexOf(u8, stub.body, "\"logBytes\":\"512\"") != null); - try testing.expectEqual(@as(usize, 0), Collector.restores); // Second close is a no-op: no extra collect, no extra request (the stub // only served one and its port is about to close). @@ -959,63 +927,7 @@ test "HttpProvider: a provider with no volume seam wired syncs without volume" { try testing.expect(std.mem.indexOf(u8, stub.body, "volume") == null); } -test "HttpProvider: volume is retained when a 200 response cannot be decoded" { - const allocator = testing.allocator; - var threaded: std.Io.Threaded = .init(allocator, .{}); - defer threaded.deinit(); - const io = threaded.io(); - - var stub = try SyncStub.start(allocator, io); - defer stub.deinit(io); - // 200, but the body is not a SyncResponse: the sync failed from our side, - // so the interval stays open (matching policy-go). - stub.response = "not json at all"; - var server_future = io.concurrent(SyncStub.serve, .{ &stub, io }) catch - return error.SkipZigTest; - defer server_future.cancel(io); - - var noop_bus: o11y.NoopEventBus = undefined; - noop_bus.init(io); - - var provider = try stub.provider(allocator, io, noop_bus.eventBus()); - defer provider.deinit(); - - const Counters = struct { - var pending: VolumeSnapshot = .{}; - fn collect(_: std.mem.Allocator, _: *anyopaque) anyerror![]PolicyStatsSnapshot { - return &.{}; - } - fn drain(_: *anyopaque) VolumeSnapshot { - defer pending = .{}; - return pending; - } - fn restore(_: *anyopaque, snapshot: VolumeSnapshot) void { - pending.metric_data_points += snapshot.metric_data_points; - pending.metric_bytes += snapshot.metric_bytes; - } - }; - Counters.pending = .{ .metric_data_points = 4, .metric_bytes = 256 }; - - var ctx: u8 = 0; - provider.setStatsCollector(.{ - .context = &ctx, - .collect = Counters.collect, - .collect_volume = Counters.drain, - .restore_volume = Counters.restore, - }); - - if (provider.close()) |_| { - try testing.expect(false); // expected the undecodable body to error - } else |_| {} - server_future.await(io); - - // The request did carry the volume, but we can't confirm the server took it. - try testing.expect(std.mem.indexOf(u8, stub.body, "\"metricDataPoints\":\"4\"") != null); - try testing.expectEqual(@as(i64, 4), Counters.pending.metric_data_points); - try testing.expectEqual(@as(i64, 256), Counters.pending.metric_bytes); -} - -test "HttpProvider: a 200 carrying error_message is a failed sync and retains volume" { +test "HttpProvider: a 200 carrying error_message is a failed sync" { const allocator = testing.allocator; var threaded: std.Io.Threaded = .init(allocator, .{}); defer threaded.deinit(); @@ -1041,29 +953,7 @@ test "HttpProvider: a 200 carrying error_message is a failed sync and retains vo var provider = try stub.provider(allocator, io, noop_bus.eventBus()); defer provider.deinit(); - const Counters = struct { - var pending: VolumeSnapshot = .{}; - fn collect(_: std.mem.Allocator, _: *anyopaque) anyerror![]PolicyStatsSnapshot { - return &.{}; - } - fn drain(_: *anyopaque) VolumeSnapshot { - defer pending = .{}; - return pending; - } - fn restore(_: *anyopaque, snapshot: VolumeSnapshot) void { - pending.spans += snapshot.spans; - pending.span_bytes += snapshot.span_bytes; - } - }; - Counters.pending = .{ .spans = 3, .span_bytes = 96 }; - var ctx: u8 = 0; - provider.setStatsCollector(.{ - .context = &ctx, - .collect = Counters.collect, - .collect_volume = Counters.drain, - .restore_volume = Counters.restore, - }); // Wired directly rather than via subscribe(), which would fire its own // initial fetch and consume the stub's single response. @@ -1084,10 +974,7 @@ test "HttpProvider: a 200 carrying error_message is a failed sync and retains vo } server_future.await(io); - // Volume survives for the next attempt… - try testing.expectEqual(@as(i64, 3), Counters.pending.spans); - try testing.expectEqual(@as(i64, 96), Counters.pending.span_bytes); - // …the rejected response's hash was not adopted as last-successful… + // The rejected response's hash was not adopted as last-successful… try testing.expect(provider.last_successful_hash == null); // …its timestamp did not advance, so the next attempt is still a full sync… try testing.expectEqual(@as(u64, 0), provider.last_sync_timestamp); @@ -1095,7 +982,7 @@ test "HttpProvider: a 200 carrying error_message is a failed sync and retains vo try testing.expect(!Sink.notified); } -test "HttpProvider: a failed sync hands its volume back for the next attempt" { +test "HttpProvider: a failed sync drops its volume instead of replaying it" { const allocator = testing.allocator; var threaded: std.Io.Threaded = .init(allocator, .{}); defer threaded.deinit(); @@ -1104,24 +991,24 @@ test "HttpProvider: a failed sync hands its volume back for the next attempt" { var noop_bus: o11y.NoopEventBus = undefined; noop_bus.init(io); - // Stand-in for the registry's counters: drained on pull, summed on - // add-back. (The real VolumeCounters round-trip is covered in registry.zig.) + // Stand-in for the registry's counters, draining on pull exactly as + // VolumeCounters.readAndReset does. There is deliberately no way to put a + // reading back: the server cannot tell a replay from new telemetry, so a + // failed sync's interval is lost rather than double counted. const Counters = struct { var pending: VolumeSnapshot = .{}; + var drains: usize = 0; fn collect(_: std.mem.Allocator, _: *anyopaque) anyerror![]PolicyStatsSnapshot { return &.{}; } fn drain(_: *anyopaque) VolumeSnapshot { + drains += 1; defer pending = .{}; return pending; } - fn restore(_: *anyopaque, snapshot: VolumeSnapshot) void { - pending.log_records += snapshot.log_records; - pending.log_bytes += snapshot.log_bytes; - pending.spans += snapshot.spans; - } }; Counters.pending = .{ .log_records = 1, .log_bytes = 64, .spans = 1 }; + Counters.drains = 0; // Port 1 has nothing listening → the POST fails after the drain. var provider = try HttpProvider.init( @@ -1136,17 +1023,16 @@ test "HttpProvider: a failed sync hands its volume back for the next attempt" { .context = &ctx, .collect = Counters.collect, .collect_volume = Counters.drain, - .restore_volume = Counters.restore, }); if (provider.close()) |_| { try testing.expect(false); // expected the unreachable endpoint to error } else |_| {} - // Nothing reached the control plane, so the interval is still open. - try testing.expectEqual(@as(i64, 1), Counters.pending.log_records); - try testing.expectEqual(@as(i64, 64), Counters.pending.log_bytes); - try testing.expectEqual(@as(i64, 1), Counters.pending.spans); + // The interval was read into the failed request and is gone: reported volume + // is a lower bound on what was observed, never a replay. + try testing.expectEqual(@as(usize, 1), Counters.drains); + try testing.expect(Counters.pending.isZero()); } test "statsToSyncStatuses: maps hits/misses/errors and reports every policy" { diff --git a/src/policy/registry.zig b/src/policy/registry.zig index 2b553c5..2855d97 100644 --- a/src/policy/registry.zig +++ b/src/policy/registry.zig @@ -54,7 +54,7 @@ pub const PolicyAtomicStats = struct { }; // ============================================================================= -// Volume Counters (spec v1.7.0) +// Volume Counters (spec v1.7.1) // ============================================================================= /// Total telemetry entering policy evaluation, regardless of match. Lives on @@ -102,7 +102,9 @@ pub const VolumeCounters = struct { _ = counter.fetchAdd(bytes, .monotonic); } - /// Read and zero every counter, for reporting on a sync request. + /// Read and zero every counter, for reporting on a sync request. Reading is + /// the reset: a sync that then fails drops its interval rather than + /// replaying it, matching the per-policy counters above. pub fn readAndReset(self: *VolumeCounters) policy_provider.VolumeSnapshot { var out: policy_provider.VolumeSnapshot = .{}; inline for (@typeInfo(policy_provider.VolumeSnapshot).@"struct".fields) |f| { @@ -110,15 +112,6 @@ pub const VolumeCounters = struct { } return out; } - - /// Add a drained reading back after a failed sync, so its records are - /// included in the next attempt instead of being lost. - pub fn add(self: *VolumeCounters, snapshot: policy_provider.VolumeSnapshot) void { - inline for (@typeInfo(policy_provider.VolumeSnapshot).@"struct".fields) |f| { - const v = @field(snapshot, f.name); - if (v != 0) _ = @field(self, f.name).fetchAdd(v, .monotonic); - } - } }; // ============================================================================= @@ -327,7 +320,7 @@ pub const PolicyRegistry = struct { extension_sync_hooks: ?policy_provider.ExtensionSyncHooks, // Total telemetry seen by the engine since the last successful sync - // (v1.7.0), independent of any policy or snapshot. + // (v1.7.1), independent of any policy or snapshot. volume: VolumeCounters, /// Subscription context for provider callbacks. @@ -403,7 +396,6 @@ pub const PolicyRegistry = struct { .context = self, .collect = collectStatsThunk, .collect_volume = drainVolumeThunk, - .restore_volume = restoreVolumeThunk, }); if (self.extension_sync_hooks) |hooks| { @@ -426,11 +418,6 @@ pub const PolicyRegistry = struct { return self.volume.readAndReset(); } - fn restoreVolumeThunk(context: *anyopaque, snapshot: policy_provider.VolumeSnapshot) void { - const self: *PolicyRegistry = @ptrCast(@alignCast(context)); - self.volume.add(snapshot); - } - /// Collect a stats row for every policy in the current snapshot, resetting /// the underlying atomic counters, and attach any compile errors recorded /// since the last recompile. Every policy is reported, including zero-hit @@ -1857,48 +1844,14 @@ test "PolicyRegistry: volume counters survive recompiles and accept add-back" { try testing.expectEqual(@as(i64, 40), drained.span_bytes); try testing.expect(registry.volume.readAndReset().isZero()); - // A failed sync hands the reading back; records seen in the meantime are - // preserved, so the next attempt reports the sum. + // Reading is the reset: a drained interval is reported at most once, so + // only telemetry seen since the drain shows up next time. registry.volume.record(.log); registry.volume.addBytes(.log, 7); - registry.volume.add(drained); - const retried = registry.volume.readAndReset(); - try testing.expectEqual(@as(i64, 2), retried.log_records); - try testing.expectEqual(@as(i64, 107), retried.log_bytes); - try testing.expectEqual(@as(i64, 1), retried.spans); -} - -test "VolumeCounters: add folds every field back into its own counter" { - var counters: VolumeCounters = .{}; - - // All six distinct and non-zero, so a field dropped or crossed over in - // `add` shows up instead of being masked by a zero. - counters.add(.{ - .log_records = 1, - .log_bytes = 2, - .metric_data_points = 3, - .metric_bytes = 4, - .spans = 5, - .span_bytes = 6, - }); - // Folding is additive, not assignment: a second add sums. - counters.add(.{ - .log_records = 10, - .log_bytes = 20, - .metric_data_points = 30, - .metric_bytes = 40, - .spans = 50, - .span_bytes = 60, - }); - - const out = counters.readAndReset(); - try testing.expectEqual(@as(i64, 11), out.log_records); - try testing.expectEqual(@as(i64, 22), out.log_bytes); - try testing.expectEqual(@as(i64, 33), out.metric_data_points); - try testing.expectEqual(@as(i64, 44), out.metric_bytes); - try testing.expectEqual(@as(i64, 55), out.spans); - try testing.expectEqual(@as(i64, 66), out.span_bytes); - try testing.expect(counters.readAndReset().isZero()); + const next = registry.volume.readAndReset(); + try testing.expectEqual(@as(i64, 1), next.log_records); + try testing.expectEqual(@as(i64, 7), next.log_bytes); + try testing.expectEqual(@as(i64, 0), next.spans); } test "PolicyRegistry: subscribe wires the volume seam to this registry" { @@ -1927,24 +1880,19 @@ test "PolicyRegistry: subscribe wires the volume seam to this registry" { // Stop the poll thread before the registry it drains goes away. provider.shutdown(); - // Nothing else in the suite proves subscribe passed the volume fns (the + // Nothing else in the suite proves subscribe passed the volume fn (the // provider tests install collectors by hand), so assert the seam exists and - // that both directions land on *this* registry's counters. + // that draining through it lands on *this* registry's counters. const collector = provider.stats_collector orelse return error.NoCollector; try testing.expect(collector.collect_volume != null); - try testing.expect(collector.restore_volume != null); registry.volume.record(.log); registry.volume.addBytes(.log, 64); const drained = collector.drainVolume(); try testing.expectEqual(@as(i64, 1), drained.log_records); try testing.expectEqual(@as(i64, 64), drained.log_bytes); + // Draining through the seam reset the registry's counters. try testing.expect(registry.volume.readAndReset().isZero()); - - collector.returnVolume(drained); - const restored = registry.volume.readAndReset(); - try testing.expectEqual(@as(i64, 1), restored.log_records); - try testing.expectEqual(@as(i64, 64), restored.log_bytes); } test "PolicyRegistry: collectStats reports exactly the current policy set across updates" { diff --git a/src/proto/tero/policy/v1.pb.zig b/src/proto/tero/policy/v1.pb.zig index 84b6a89..fe2ae01 100644 --- a/src/proto/tero/policy/v1.pb.zig +++ b/src/proto/tero/policy/v1.pb.zig @@ -2503,8 +2503,14 @@ pub const PolicySyncStatus = struct { /// VolumeStats reports the total telemetry a client observed since the last /// sync, regardless of whether any policy matched it. Counts are of records -/// entering policy evaluation, before any keep or transform stage runs, and are -/// reset on each successful sync. +/// entering policy evaluation, before any keep or transform stage runs. +/// +/// Counters are reset when they are read into a sync request, whether or not +/// that sync then succeeds — the same rule PolicySyncStatus.match_hits and +/// match_misses follow. A failed sync loses its interval from the numerator and +/// the denominator alike, so match rates stay meaningful; counters from a failed +/// sync must never be replayed, since the server cannot tell a replay from new +/// telemetry. Reported volume is a lower bound, not an exact total. /// /// Reporting volume is optional, and every field is individually optional: an /// implementation may report record counts without byte counts, or a subset of