From 1384bff5ea32d05cf4f36b4219190fefcbffd556 Mon Sep 17 00:00:00 2001 From: jaronoff97 Date: Thu, 6 Aug 2026 14:00:43 -0400 Subject: [PATCH] feat: update to policy-zig 0.7.0 --- bench/logging/uv.lock | 6 +- build.zig.zon | 4 +- src/signals/otlp/metrics.zig | 155 ++++++++++++++++++++++++++--------- 3 files changed, 122 insertions(+), 43 deletions(-) diff --git a/bench/logging/uv.lock b/bench/logging/uv.lock index 4a908cab..4ec298f9 100644 --- a/bench/logging/uv.lock +++ b/bench/logging/uv.lock @@ -37,11 +37,11 @@ wheels = [ [[package]] name = "packaging" -version = "26.2" +version = "26.3" source = { registry = "https://pypi.org/simple" } -sdist = { url = "https://files.pythonhosted.org/packages/d7/f1/e7a6dd94a8d4a5626c03e4e99c87f241ba9e350cd9e6d75123f992427270/packaging-26.2.tar.gz", hash = "sha256:ff452ff5a3e828ce110190feff1178bb1f2ea2281fa2075aadb987c2fb221661", size = 228134, upload-time = "2026-04-24T20:15:23.917Z" } +sdist = { url = "https://files.pythonhosted.org/packages/7d/fa/3944b40b07da9ce895c0e6303a5ab7d53da063554f534556b134a54d6093/packaging-26.3.tar.gz", hash = "sha256:94edc256424af38762eb31306eed28beb9f0efc50a8837492c9d6fd6004aed79", size = 313412, upload-time = "2026-08-04T18:15:28.737Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/df/b2/87e62e8c3e2f4b32e5fe99e0b86d576da1312593b39f47d8ceef365e95ed/packaging-26.2-py3-none-any.whl", hash = "sha256:5fc45236b9446107ff2415ce77c807cee2862cb6fac22b8a73826d0693b0980e", size = 100195, upload-time = "2026-04-24T20:15:22.081Z" }, + { url = "https://files.pythonhosted.org/packages/63/34/ba1c580383c9eada3711951fef0795c80b829a078d72188184bcab9dd527/packaging-26.3-py3-none-any.whl", hash = "sha256:d7193f7c8e4e93f444fde0262bf90af30e16fa0ad0ad44cb553c87339b23cd1c", size = 129956, upload-time = "2026-08-04T18:15:27.159Z" }, ] [[package]] diff --git a/build.zig.zon b/build.zig.zon index 2e874374..9ba031bf 100644 --- a/build.zig.zon +++ b/build.zig.zon @@ -49,8 +49,8 @@ .hash = "zbench-0.13.0-YTdc7xVBAQCCMC-IdLLuotBeiNNNm8k9Pi2V4VYqrwfI", }, .policy_zig = .{ - .url = "https://github.com/usetero/policy-zig/archive/refs/tags/v0.6.2.tar.gz", - .hash = "policy_zig-0.6.2-5_dp3rwyFQAzK2eIEWZSWKctNEOuVL4P1zlX2vRhcxb9", + .url = "https://github.com/usetero/policy-zig/archive/refs/tags/v0.7.0.tar.gz", + .hash = "policy_zig-0.7.0-5_dp3qqnFQBMOLHViVc_0XhuiMadfzsdee3oGpASi1T5", }, }, .paths = .{ diff --git a/src/signals/otlp/metrics.zig b/src/signals/otlp/metrics.zig index 20d260a2..4f9cb8d8 100644 --- a/src/signals/otlp/metrics.zig +++ b/src/signals/otlp/metrics.zig @@ -57,9 +57,9 @@ const MetricsProcessingFailed = struct { err: []const u8, contentType: []const u pub const ProcessResult = struct { /// Whether any transformations were applied (not yet supported for metrics) was_transformed: bool = false, - /// Number of metrics that were dropped by filter policies + /// Number of data points that were dropped by filter policies dropped_count: usize, - /// Original number of metrics before filtering + /// Original number of data points before filtering original_count: usize, /// The processed data (caller owns this slice) data: []u8, @@ -241,18 +241,11 @@ pub fn metricValue(ctx: *const anyopaque, field: MetricFieldRef) ?[]const u8 { }; } -/// Get attributes from the first datapoint of a metric -fn getDatapointAttrs(metric: *const Metric) []const KeyValue { - const data = metric.data orelse return &.{}; +/// Number of data points carried by a metric, across every data variant. +fn dataPointCount(metric: *const Metric) usize { + const data = metric.data orelse return 0; return switch (data) { - .gauge => |g| if (g.data_points.items.len > 0) g.data_points.items[0].attributes.items else &.{}, - .sum => |s| if (s.data_points.items.len > 0) s.data_points.items[0].attributes.items else &.{}, - .histogram => |h| if (h.data_points.items.len > 0) h.data_points.items[0].attributes.items else &.{}, - .exponential_histogram => |eh| if (eh.data_points.items.len > 0) - eh.data_points.items[0].attributes.items - else - &.{}, - .summary => |s| if (s.data_points.items.len > 0) s.data_points.items[0].attributes.items else &.{}, + inline else => |variant| variant.data_points.items.len, }; } @@ -286,7 +279,7 @@ pub const metric_accessor: policy.MetricAccessor = .{ .typed_value = metricTypedValue, }; -/// Result of filtering metrics in-place +/// Result of filtering metrics in-place, counted in data points const FilterCounts = struct { original_count: usize, dropped_count: usize, @@ -311,30 +304,44 @@ fn filterMetricsInPlace( // Structure: MetricsData -> ResourceMetrics[] -> ScopeMetrics[] -> Metric[] for (metrics_data.resource_metrics.items) |*resource_metrics| { for (resource_metrics.scope_metrics.items) |*scope_metrics| { - // Count original metrics - original_count += scope_metrics.metrics.items.len; - // Filter metrics in place by shrinking the list var write_idx: usize = 0; for (scope_metrics.metrics.items) |*metric| { - var ctx: OtlpMetricContext = .{ - .metric = metric, - .resource_metrics = resource_metrics, - .scope_metrics = scope_metrics, - .datapoint_attributes = getDatapointAttrs(metric), + // The data point is the record unit for metrics (policy spec + // v1.7.1): each one is evaluated against its own attributes and + // dropped individually, and the metric itself goes only once + // every data point is gone. + if (metric.data) |*metric_data| switch (metric_data.*) { + inline else => |*variant| { + var d: usize = 0; + while (d < variant.data_points.items.len) { + original_count += 1; + var ctx: OtlpMetricContext = .{ + .metric = metric, + .resource_metrics = resource_metrics, + .scope_metrics = scope_metrics, + .datapoint_attributes = variant.data_points.items[d].attributes.items, + }; + + const result = engine.evaluate(.metric, &metric_accessor, &ctx, &policy_id_buf, .{ + .io = bus.io, + }); + + if (result.decision.shouldContinue()) { + d += 1; + } else { + _ = variant.data_points.orderedRemove(d); + dropped_count += 1; + } + } + }, }; - const result = engine.evaluate(.metric, &metric_accessor, &ctx, &policy_id_buf, .{ .io = bus.io }); - - if (result.decision.shouldContinue()) { - // Keep this metric - move to write position if needed - if (write_idx != scope_metrics.metrics.items.len - 1) { - scope_metrics.metrics.items[write_idx] = metric.*; - } - write_idx += 1; - } else { - dropped_count += 1; - } + // A metric with no data at all is left alone; one whose data + // points were all dropped is pruned. + if (metric.data != null and dataPointCount(metric) == 0) continue; + scope_metrics.metrics.items[write_idx] = metric.*; + write_idx += 1; } // Shrink the list to only kept items (zero allocation) @@ -612,7 +619,9 @@ test "processMetrics - no policies keeps all metrics" { const metrics = \\{"resourceMetrics":[{"resource":{},"scopeMetrics":[{"scope":{}, ++ - \\"metrics":[{"name":"metric1"},{"name":"metric2"}]}]}]} + \\"metrics":[{"name":"metric1","sum":{"dataPoints":[{"asInt":"1"}]}}, + ++ + \\{"name":"metric2","sum":{"dataPoints":[{"asInt":"2"}]}}]}]}]} ; var in_reader = std.Io.Reader.fixed(metrics); @@ -669,7 +678,9 @@ test "processMetrics - DROP policy filters metrics by name" { const metrics = \\{"resourceMetrics":[{"resource":{},"scopeMetrics":[{"scope":{}, ++ - \\"metrics":[{"name":"http.requests"},{"name":"debug.internal"}]}]}]} + \\"metrics":[{"name":"http.requests","sum":{"dataPoints":[{"asInt":"1"}]}}, + ++ + \\{"name":"debug.internal","sum":{"dataPoints":[{"asInt":"2"}]}}]}]}]} ; var in_reader = std.Io.Reader.fixed(metrics); @@ -731,11 +742,15 @@ test "processMetrics - DROP policy filters metrics by resource attribute" { ++ \\"value":{"stringValue":"test-service"}}]},"scopeMetrics": ++ - \\[{"scope":{},"metrics":[{"name":"from.test"}]}]},{"resource": + \\[{"scope":{},"metrics":[{"name":"from.test","sum":{"dataPoints":[{"asInt":"1"}]}}]}]}, + ++ + \\{"resource": ++ \\{"attributes":[{"key":"service.name","value":{"stringValue":"prod-service"}}]}, ++ - \\"scopeMetrics":[{"scope":{},"metrics":[{"name":"from.prod"}]}]}]} + \\"scopeMetrics":[{"scope":{},"metrics":[{"name":"from.prod", + ++ + \\"sum":{"dataPoints":[{"asInt":"2"}]}}]}]}]} ; var in_reader = std.Io.Reader.fixed(metrics); @@ -792,7 +807,9 @@ test "processMetrics - all metrics dropped returns empty structure" { const metrics = \\{"resourceMetrics":[{"resource":{},"scopeMetrics":[{"scope":{}, ++ - \\"metrics":[{"name":"metric1"},{"name":"metric2"}]}]}]} + \\"metrics":[{"name":"metric1","sum":{"dataPoints":[{"asInt":"1"}]}}, + ++ + \\{"name":"metric2","sum":{"dataPoints":[{"asInt":"2"}]}}]}]}]} ; var in_reader = std.Io.Reader.fixed(metrics); @@ -819,6 +836,68 @@ test "processMetrics - all metrics dropped returns empty structure" { try std.testing.expect(result.allDropped()); } +test "processMetrics - DROP policy filters individual data points" { + const allocator = std.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(); + + var drop_policy: proto.policy.Policy = .{ + .id = try allocator.dupe(u8, "drop-canary-datapoints"), + .name = try allocator.dupe(u8, "drop-canary-datapoints"), + .enabled = true, + .target = .{ .metric = .{ .keep = false } }, + }; + var attr_path: proto.policy.AttributePath = .{}; + try attr_path.path.append(allocator, try allocator.dupe(u8, "env")); + try drop_policy.target.?.metric.match.append(allocator, .{ + .field = .{ .datapoint_attribute = attr_path }, + .match = .{ .regex = try allocator.dupe(u8, "canary") }, + }); + defer drop_policy.deinit(allocator); + + try registry.updatePolicies(&.{drop_policy}, "file-provider", .file); + + // One metric, three data points: only the canary one matches, so the metric + // survives with the other two intact. + const metrics = + \\{"resourceMetrics":[{"resource":{},"scopeMetrics":[{"scope":{},"metrics":[ + ++ + \\{"name":"http.requests","sum":{"dataPoints":[ + ++ + \\{"asInt":"1","attributes":[{"key":"env","value":{"stringValue":"prod"}}]}, + ++ + \\{"asInt":"2","attributes":[{"key":"env","value":{"stringValue":"canary"}}]}, + ++ + \\{"asInt":"3","attributes":[{"key":"env","value":{"stringValue":"staging"}}]} + ++ + \\]}}]}]}]} + ; + + var in_reader = std.Io.Reader.fixed(metrics); + var out_writer: std.Io.Writer.Allocating = .init(allocator); + defer out_writer.deinit(); + const result = try processMetricsStream( + allocator, + ®istry, + noop_bus.eventBus(), + &in_reader, + &out_writer.writer, + "application/json", + ); + + const data = out_writer.written(); + try std.testing.expect(std.mem.indexOf(u8, data, "http.requests") != null); + try std.testing.expect(std.mem.indexOf(u8, data, "prod") != null); + try std.testing.expect(std.mem.indexOf(u8, data, "staging") != null); + try std.testing.expect(std.mem.indexOf(u8, data, "canary") == null); + try std.testing.expectEqual(@as(usize, 3), result.original_count); + try std.testing.expectEqual(@as(usize, 1), result.dropped_count); + try std.testing.expect(!result.allDropped()); +} + test "ContentFormat.fromContentType" { try std.testing.expectEqual(ContentFormat.json, ContentFormat.fromContentType("application/json")); try std.testing.expectEqual(ContentFormat.json, ContentFormat.fromContentType("application/json; charset=utf-8"));