This is an automated email from the ASF dual-hosted git repository. wankai123 pushed a commit to branch feat/first-party-trace-sampler-plugins in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit 96955876036f325ad320a6a8a1723a3618070d29 Author: wankai123 <[email protected]> AuthorDate: Wed Jul 29 10:29:51 2026 +0800 feat(trace-pipeline): add first-party sw/zipkin trace-retention sampler plugins Add sw-trace-sampler.so and zipkin-trace-sampler.so under plugins/skywalking/, implementing the Scenario 6.1/6.2 keep logic against SkyWalking's two real trace schemas. Both compile the shared internal/tracesampler engine and differ only in a Schema value naming where each schema keeps the rule inputs, so the two mains stay a few lines each and cannot drift apart. The existing _example/segment-tail-sampler projects tag names the real schemas do not have (duration/status), so it cannot filter live data. These plugins read the columns OAP actually writes: searchable tags are not first-class columns but "key=value" entries flattened into one string array (tags for segments, query for Zipkin), so every keepTagRules entry resolves there. Keep rules: durationThresholdMs (end-to-end envelope max(start + duration) - min(start), which catches traces slow only through sequential spans, and is not the intrinsic MaxTS-MinTS -- that is the spread of per-row start timestamps and 0 for a single-row trace), keepErrors with an errorTag override, keepTagRules, and a deterministic FNV-1a healthySampleRate that is stable across the merge and finalize passes so a trace is never half-kept. Config handling is strict by design, because every option is a keep rule and a key that silently missed would leave a sampler that drops the entire group: unknown and empty configs are rejected, as is a rule naming a first-class column (which could never match) -- with the array-column workaround named in the error. keepTagRules also accepts a compact "key=value,key=~regex,key" string so a rule set fits in an environment variable. Any predicate that cannot be evaluated keeps the trace. Absent duration columns mean "can't tell", not "not slow": the columns are schema-declared, so their absence implies the block was written under a different schema (typically the wrong plugin attached to the group), and dropping there would silently discard exactly what the operator configured the sampler to keep. An absent tag array is deliberately not treated that way, since a trace carrying no searchable tags is ordinary data. The flattened array is decoded once per trace, from a copy. The SDK's string-array decode rewrites its source in place and the engine hands the same TraceBatch to every link of a chain, so decoding the shared buffer corrupted the value for later links -- making a rule's verdict depend on its position in the chain, and only for values containing "|" or "\". Also sync docs/design/post-trace-pipeline.md: the 6.1/6.2 scenarios now reference these plugins and their real config keys instead of samplers that do not exist, and no longer claim the merge raw fast path is preserved -- projecting any tag sets forceSlow, so the block is decoded in full. --- CHANGES.md | 1 + docs/design/post-trace-pipeline.md | 87 ++- plugins/README.md | 74 ++ .../skywalking/internal/tracesampler/sampler.go | 763 +++++++++++++++++++++ .../internal/tracesampler/sampler_test.go | 631 +++++++++++++++++ plugins/skywalking/sw-trace-sampler/main.go | 71 ++ plugins/skywalking/sw-trace-sampler/main_test.go | 93 +++ plugins/skywalking/zipkin-trace-sampler/main.go | 77 +++ .../skywalking/zipkin-trace-sampler/main_test.go | 118 ++++ 9 files changed, 1871 insertions(+), 44 deletions(-) diff --git a/CHANGES.md b/CHANGES.md index 95dbdbf6d..bdc83104a 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -76,6 +76,7 @@ Release Notes. - Register/update/remove in-merge trace retention samplers at runtime without restarting the data node, replacing the static `--trace-pipeline-config` boot loader (now removed). The sampler pipeline config is carried on `common.v1.Group` (new optional `pipeline` field, with `TracePipelineConfig`/`SamplerPlugin` relocated from `pipeline/v1` into the leaf `common/v1` to avoid a `common`→`pipeline` import cycle) and delivered through the existing `KindGroup` schema watch; the data node rebu [...] - Add FODC memory-pressure pprof capture: agent grabs heap/goroutine profiles when RSS nears the cgroup limit, served via the proxy. - Add bounded plugin telemetry (meter + logger) to the trace-pipeline sampler SDK. A sampler plugin opts in by implementing `sdk.HostAware`; the engine calls `UseHost(h)` once per group before the first `Decide`, delivering a scoped `sdk.Host` the plugin may cache and use in `Decide`. The host enforces all resource limits: metric names are forced under the `banyandb_trace_pipeline_plugin_` prefix; `{group, plugin_name}` labels are always present; per-plugin cardinality is capped at 100 s [...] +- Add the first-party trace-retention sampler plugins `sw-trace-sampler.so` and `zipkin-trace-sampler.so` (`plugins/skywalking/`), implementing the Scenario 6.1/6.2 keep logic on SkyWalking's two trace schemas. Both compile the shared `plugins/skywalking/internal/tracesampler` engine and differ only in a `Schema` value naming where each schema keeps the rule inputs: the flattened searchable-tag array (`tags` for segments, `query` for Zipkin), the error signal (a real `is_error` column vs [...] - Add a Claude/Codex plugin packaging the BanyanDB MCP server with a parse-only `validate_bydbql` tool (backed by the `mcp/tools/bydbql-parse` Go validator) and a BydbQL skill for read-only natural-language-to-BydbQL generation over STREAM/MEASURE/TRACE/PROPERTY resources. - Introduce positional parameter binding (`?` placeholders) into BydbQL to eliminate QL injection. - Add reusable BydbQL binding: Prepare a query once, then Bind it many times without re-parsing or mutating the template. The liaison caches prepared statements on the gRPC query path (LRU bounded by entry count and bytes, on by default; `--bydbql-prepared-cache-size`/`--bydbql-prepared-cache-max-bytes`, `bydbql_prepared_cache_*` metrics) so repeated templates skip parsing. To pinpoint un-cacheable and slow queries without high-cardinality labels, the query access log tags each entry by [...] diff --git a/docs/design/post-trace-pipeline.md b/docs/design/post-trace-pipeline.md index 7b134ef3d..604b5bb13 100644 --- a/docs/design/post-trace-pipeline.md +++ b/docs/design/post-trace-pipeline.md @@ -124,7 +124,7 @@ The contract is designed around three hard requirements. - The plugin re-exports an `ABIVersion` constant; the engine refuses to load on mismatch with its own compiled `sdk.ABIVersion`, turning a silent miscompile into a clear, fail-fast error. Configuration is a structured `google.protobuf.Struct` (`SamplerPlugin.config`) set directly in the pipeline config; the engine serializes it to canonical JSON and the plugin unmarshals those `[]byte` into its own typed config — so the wire form is structured and inspectable while the `.so` boundary sta [...] - **Distribution:** operators build plugins against the released, version-tagged `pkg/pipeline/sdk` using the **same CI image / Go version / `-trimpath` / CGO flags** as the data node. (Background constraints, well-documented but outside this design's verified scope: Go plugins are **Linux/macOS only**, **cannot be unloaded** — so changing a plugin requires a node restart, there is no hot-reload — and a plugin **panic crashes the host** unless contained; see fail-open below.) -**(3) Projection / column selection — spans optional, more than tags.** The plugin declares the columns it needs up front via `Project()`, which returns a `Projection{ Tags []string; SpanIDs bool; Spans bool }`. The engine turns `Tags` into the **same `model.TagProjection`** the block reader already honors (`blockMetadata.tagProjection`), so only those tag columns are decoded into `block.tags` — literally the query engine's tag-projection path (`trace/v1/query.proto`), not a new mechanis [...] +**(3) Projection / column selection — spans optional, more than tags.** The plugin declares the columns it needs up front via `Project()`, which returns a `Projection{ Tags []string; SpanIDs bool; Spans bool }`. The engine turns `Tags` into the **same `model.TagProjection`** the block reader already honors (`blockMetadata.tagProjection`), so only those tag columns are decoded into `block.tags` — literally the query engine's tag-projection path (`trace/v1/query.proto`), not a new mechanis [...] **Go SDK (`pkg/pipeline/sdk`).** The batch types mirror the native trace `block`/`tag` (`banyand/trace/block.go`); the engine fills them with the block's own slices, shared **read-only** (not copied) — see the read-only contract below: @@ -232,7 +232,7 @@ type Verdict struct { } ``` -These types are real and live in [`pkg/pipeline/sdk`](../../pkg/pipeline/sdk); the block above is the conceptual layout, and the canonical definitions plus the value-decode helpers (`TagColumn.At`, `DecodeTagValue`) are in that package. A complete reference plugin implementing the Scenario 6.1 sampler — config parsing, `Project()`, and tag/span extraction — lives at [`pkg/pipeline/sdk/_example/segment-tail-sampler`](../../pkg/pipeline/sdk/_example/segment-tail-sampler). +These types are real and live in [`pkg/pipeline/sdk`](../../pkg/pipeline/sdk); the block above is the conceptual layout, and the canonical definitions plus the value-decode helpers (`TagColumn.At`, `DecodeTagValue`) are in that package. The shipped first-party samplers live at [`plugins/skywalking/sw-trace-sampler`](../../plugins/skywalking/sw-trace-sampler) and [`plugins/skywalking/zipkin-trace-sampler`](../../plugins/skywalking/zipkin-trace-sampler); a standalone teaching example of th [...] **Verdict shape — boolean keep-mask.** `Decide` returns a `[]bool` aligned to `batch.Traces`: `Keep[i]` retains trace `i`. This is the simplest fully-vectorized contract and makes the alignment invariant trivial to check (the engine rejects a verdict whose length ≠ `len(batch.Traces)`). The keep/drop is per `trace_id`, matching the merger's per-`trace_id` write granularity (§7.1). @@ -370,7 +370,7 @@ Both showcase trace groups (`sw_trace`, `sw_zipkinTrace`) declare a Hot → Warm - The migration engine runs the source stage's retention `plugins` chain against every trace in the partition. -- **No dynamic splitting is performed:** all retained data is written to the next stage's node group. Traces the source stage's plugin drops are omitted from the target write stream, reducing the physical size of the migrated partition. With Scenario 6.1's hot retention plugin (config `min_duration: 100ms`) a healthy `/homepage` trace (2802 ms) is kept and migrates to Warm; a PostgreSQL-touching trace is kept by the config's tag rule and migrates too; a healthy fast trace (6 ms) matches [...] +- **No dynamic splitting is performed:** all retained data is written to the next stage's node group. Traces the source stage's plugin drops are omitted from the target write stream, reducing the physical size of the migrated partition. With Scenario 6.1's hot retention plugin (config `durationThresholdMs: 100`) a healthy `/homepage` trace (2802 ms) is kept and migrates to Warm; a PostgreSQL-touching trace is kept by the config's tag rule and migrates too; a healthy fast trace (6 ms) mat [...] - When the partition matures past the Warm 7-day ttl, the Warm stage's retention plugin runs at this Warm→Cold boundary — the gate that decides what enters Cold (typically the strictest config; in Scenario 6.1 it keeps errors only). Only the traces it keeps are written into the Cold parts and then retained for the full Cold TTL; everything else is dropped here. Cold itself does no further sampling — Cold compactions are lossless. @@ -394,7 +394,7 @@ Below are two operational scenarios represented as complete `TracePipelineConfig ```Plain Text { - "metadata": { "group": "sw_trace", "name": "segment-tail-sampler" }, + "metadata": { "group": "sw_trace", "name": "sw-trace-sampler" }, "enabled": true, "stages": [ { @@ -403,14 +403,14 @@ Below are two operational scenarios represented as complete `TracePipelineConfig { "name": "hot-retention", "sampler": { - "path": "segment-stage-retention.so", + "path": "sw-trace-sampler.so", "abi_version": 1, "config": { - "min_duration": "0.100s", - "keep_errors": true, - "keep_tag_rules": [ - { "tag_key": "db.type", "equals": "PostgreSQL" }, - { "tag_key": "mq.queue", "equals": "queue-songs-ping" } + "durationThresholdMs": 100, + "keepErrors": true, + "keepTagRules": [ + { "tagKey": "db.type", "equals": "PostgreSQL" }, + { "tagKey": "mq.queue", "equals": "queue-songs-ping" } ] } } @@ -423,10 +423,10 @@ Below are two operational scenarios represented as complete `TracePipelineConfig { "name": "warm-retention", "sampler": { - "path": "segment-stage-retention.so", + "path": "sw-trace-sampler.so", "abi_version": 1, "config": { - "keep_errors": true + "keepErrors": true } } } @@ -436,17 +436,17 @@ Below are two operational scenarios represented as complete `TracePipelineConfig "schema_names": ["segment"], "plugins": [ { - "name": "segment-tail-sampler", + "name": "sw-trace-sampler", "sampler": { - "path": "segment-tail-sampler.so", + "path": "sw-trace-sampler.so", "abi_version": 1, "config": { - "duration_threshold": "0.500s", - "keep_errors": true, - "healthy_sample_rate": 0.1, - "keep_tag_rules": [ - { "tag_key": "db.type", "equals": "PostgreSQL" }, - { "tag_key": "mq.queue", "equals": "queue-songs-ping" } + "durationThresholdMs": 500, + "keepErrors": true, + "healthySampleRate": 0.1, + "keepTagRules": [ + { "tagKey": "db.type", "equals": "PostgreSQL" }, + { "tagKey": "mq.queue", "equals": "queue-songs-ping" } ] } } @@ -458,27 +458,27 @@ Below are two operational scenarios represented as complete `TracePipelineConfig } ``` -> Each plugin link's `config` is a structured `google.protobuf.Struct` set directly in the pipeline config (not an opaque blob); the engine does not interpret its keys — it serializes the object to JSON and hands the bytes to the plugin's constructor, which unmarshals them into its own typed config. The gating chain here is a single `sampler` link whose `Project()` returns `Projection{ Tags: ["is_error", "db.type", "mq.queue"], SpanIDs: false, Spans: false }`, so only those three tag col [...] +> Each plugin link's `config` is a structured `google.protobuf.Struct` set directly in the pipeline config (not an opaque blob); the engine does not interpret its keys — it serializes the object to JSON and hands the bytes to the plugin's constructor, which unmarshals them into its own typed config. The gating chain here is a single `sampler` link whose `Project()` returns `Projection{ Tags: ["start_time", "latency", "is_error", "tags"], SpanIDs: false, Spans: false }`, so the verdict it [...] - **Retention Dynamics** (real `sw_trace` traces; gating runs in Hot, owned by the gating chain; per-stage retention owned by each stage's `plugins` chain): - - The error trace `5fcdb353-…` (`POST /test`, `agent::app`, `is_error=1`, 4 ms) is a sure-keep at gating: the gating plugin keeps it because its config sets `keep_errors`. At Hot→Warm, the hot retention plugin keeps it (config `keep_errors`) → migrated. At Warm→Cold, the warm retention plugin (errors-only) keeps it → migrated into Cold, where it is retained for the full 30-day Cold TTL (Cold does no further sampling). + - The error trace `5fcdb353-…` (`POST /test`, `agent::app`, `is_error=1`, 4 ms) is a sure-keep at gating: the gating plugin keeps it because its config sets `keepErrors`. At Hot→Warm, the hot retention plugin keeps it (config `keepErrors`) → migrated. At Warm→Cold, the warm retention plugin (errors-only) keeps it → migrated into Cold, where it is retained for the full 30-day Cold TTL (Cold does no further sampling). - - The slow healthy trace `b03bb932-…` (`/homepage`, `agent::ui` → `agent::frontend`, 2802 ms) is a sure-keep at gating: the gating plugin keeps it because 2802 ms > 500 ms `duration_threshold` in its config (from `MaxTS - MinTS`, no span decode). At Hot→Warm: the hot plugin keeps it (2802 ms ≥ its 100 ms `min_duration`) → kept through Warm. At Warm→Cold: the warm plugin (errors-only) drops it (no error) → **dropped at Warm→Cold; never enters Cold**. + - The slow healthy trace `b03bb932-…` (`/homepage`, `agent::ui` → `agent::frontend`, 2802 ms) is a sure-keep at gating: the gating plugin keeps it because its 2802 ms envelope > the 500 ms `durationThresholdMs` in its config (from the `start_time`/`latency` tags, no span decode). At Hot→Warm: the hot plugin keeps it (2802 ms ≥ its 100 ms `durationThresholdMs`) → kept through Warm. At Warm→Cold: the warm plugin (errors-only) drops it (no error) → **dropped at Warm→Cold; never enters Cold**. - A PostgreSQL-touching trace (e.g. `b31e4be8-…`, `agent::songs` `UndertowDispatch`, 3 ms, `db.type=PostgreSQL`) is sure-kept at gating via the gating plugin's `db.type` keep-tag-rule. At Hot→Warm: the hot plugin keeps it (its config's `db.type` rule) → kept through Warm. At Warm→Cold: the warm plugin (errors-only) drops it (no error) → **dropped at Warm→Cold; never enters Cold**. - - A healthy fast trace such as `GET:/songs` at 6 ms (`agent::songs`, `http.status_code=200`) is only kept at gating if the gating plugin's `healthy_sample_rate` (`0.1`) hash retains it (deterministic `hash(trace_id) < 0.1`, since it matches no sure-keep rule). If kept, at Hot→Warm the hot plugin drops it (6 ms < 100 ms, no error, no tag match) → **dropped at Hot→Warm migration**. + - A healthy fast trace such as `GET:/songs` at 6 ms (`agent::songs`, `http.status_code=200`) is only kept at gating if the gating plugin's `healthySampleRate` (`0.1`) hash retains it (deterministic `hash(trace_id) < 0.1`, since it matches no sure-keep rule). If kept, at Hot→Warm the hot plugin drops it (6 ms < 100 ms, no error, no tag match) → **dropped at Hot→Warm migration**. ### 6.2 Scenario 2: Istio / Zipkin Mesh Edge Sampling (`sw_zipkinTrace`) -- **Objective**: On the showcase `sw_zipkinTrace` group (schema `zipkin_span`), apply a lower-cost edge sampler to the Istio service-mesh spans. The Zipkin schema has no first-class `is_error` column, so server errors are caught with a tag rule on the flattened `query` attributes rather than `keep_errors`; mesh gateway spans are kept by tag. Targets the group's Warm and Cold stages. As in §6.1 the gating verdict is owned by a native Go plugin chain (a single sampler link, §2.5); per-stag [...] +- **Objective**: On the showcase `sw_zipkinTrace` group (schema `zipkin_span`), apply a lower-cost edge sampler to the Istio service-mesh spans. The Zipkin schema has no first-class `is_error` column: `keepErrors` detects Zipkin's conventional `error` span tag inside the flattened `query` attributes, and 5xx-only failures are caught with an explicit `query` tag rule. Targets the group's Warm and Cold stages. As in §6.1 the gating verdict is owned by a native Go plugin chain (a single sam [...] - **Configuration JSON**: ```Plain Text { - "metadata": { "group": "sw_zipkinTrace", "name": "zipkin-edge-sampler" }, + "metadata": { "group": "sw_zipkinTrace", "name": "zipkin-trace-sampler" }, "enabled": true, "stages": [ { @@ -487,13 +487,12 @@ Below are two operational scenarios represented as complete `TracePipelineConfig { "name": "warm-retention", "sampler": { - "path": "zipkin-stage-retention.so", + "path": "zipkin-trace-sampler.so", "abi_version": 1, "config": { - "min_duration": "1s", - "keep_tag_rules": [ - { "tag_key": "query", "regex": "http\\.status_code=5\\d\\d" }, - { "tag_key": "local_endpoint_service_name", "equals": "gateway.sample-services" } + "durationThresholdMs": 1000, + "keepTagRules": [ + { "tagKey": "query", "regex": "http\\.status_code=5\\d\\d" } ] } } @@ -504,16 +503,16 @@ Below are two operational scenarios represented as complete `TracePipelineConfig "schema_names": ["zipkin_span"], "plugins": [ { - "name": "zipkin-edge-sampler", + "name": "zipkin-trace-sampler", "sampler": { - "path": "zipkin-edge-sampler.so", + "path": "zipkin-trace-sampler.so", "abi_version": 1, "config": { - "duration_threshold": "1.000s", - "keep_errors": false, - "healthy_sample_rate": 0.05, - "keep_tag_rules": [ - { "tag_key": "query", "regex": "http\\.status_code=5\\d\\d" } + "durationThresholdMs": 1000, + "keepErrors": true, + "healthySampleRate": 0.05, + "keepTagRules": [ + { "tagKey": "query", "regex": "http\\.status_code=5\\d\\d" } ] } } @@ -523,13 +522,13 @@ Below are two operational scenarios represented as complete `TracePipelineConfig } ``` -> Each link's `config` is a structured `google.protobuf.Struct` set directly in the pipeline config (not an opaque blob); the engine does not interpret its keys — it serializes the object to JSON and hands the bytes to the plugin's constructor, which unmarshals them into its own typed config. The gating chain's single `sampler` link `Project()` returns `Projection{ Tags: ["query"], SpanIDs: false, Spans: false }`, so only the `query` tag column (plus the intrinsic `trace_id` / `MinTS` / [...] +> Each link's `config` is a structured `google.protobuf.Struct` set directly in the pipeline config (not an opaque blob); the engine does not interpret its keys — it serializes the object to JSON and hands the bytes to the plugin's constructor, which unmarshals them into its own typed config. The gating chain's single `sampler` link `Project()` returns `Projection{ Tags: ["timestamp_millis", "duration", "query"], SpanIDs: false, Spans: false }`, so the verdict itself reads only those tag [...] - **Retention Dynamics** (real `sw_zipkinTrace` spans; gating runs in Hot, owned by the gating chain; per-stage retention owned by the Warm `plugins` chain): - - The slowest mesh call observed — `trace_id 0961e077…`, a 30.7 s `istio.skywalking-showcase` client span to Grafana's live-WS endpoint (`http.status_code=101`) — is a sure-keep at gating: the gating plugin keeps it because 30.7 s > 1 s `duration_threshold` in its config (from `MaxTS - MinTS`, no span decode). At Warm→Cold: the warm retention plugin keeps it (30.7 s ≥ its 1 s `min_duration`) → migrated into Cold, where it is retained for the full Cold TTL (Cold does no further sampling). + - The slowest mesh call observed — `trace_id 0961e077…`, a 30.7 s `istio.skywalking-showcase` client span to Grafana's live-WS endpoint (`http.status_code=101`) — is a sure-keep at gating: the gating plugin keeps it because its 30.7 s envelope > the 1 s `durationThresholdMs` in its config (from the `timestamp_millis`/`duration` tags, no span decode). At Warm→Cold: the warm retention plugin keeps it (30.7 s ≥ its 1 s `durationThresholdMs`) → migrated into Cold, where it is retained fo [...] - - A gateway span on `gateway.sample-services` at the mesh p90 (~19 ms): the gating plugin passes it only via the `0.05` `healthy_sample_rate` hash (19 ms < 1 s, no 5xx, so it matches no sure-keep rule). If kept, at Warm→Cold the warm plugin keeps it (19 ms < 1 s, but its `local_endpoint_service_name = gateway.sample-services` rule matches) → migrated into Cold, where it is retained for the full Cold TTL. + - A gateway span on `gateway.sample-services` at the mesh p90 (~19 ms): the gating plugin passes it only via the `0.05` `healthySampleRate` hash (19 ms < 1 s, no 5xx, so it matches no sure-keep rule). If kept, at Warm→Cold the warm plugin drops it (19 ms < 1 s and no 5xx, so it matches no sure-keep rule) → **dropped at Warm→Cold**. - A typical p50 mesh span (~2 ms) is kept at gating only via the gating plugin's `0.05` sample. At Warm→Cold the warm plugin drops it (2 ms < 1 s, not a gateway span, no 5xx) → **dropped at Warm→Cold**. @@ -642,7 +641,7 @@ flowchart TD subgraph A["A. In-merge filter at HOT-PHASE LSM COMPACTION (PIPELINE_EVENT_MERGE enabled)"] A1["1. mergeBlocks streams blocks ordered by trace_id (banyand/trace/merger.go).<br/>Maturity check: traceMaxTs (= timestampsMetadata.max) older than now − merge_grace (30s)?"] A1 -->|"NO"| A2["Pass blocks through unchanged; defer the verdict"] - A1 -->|"YES (stopped growing)"| A3["Engine builds a projected TraceBatch; plugin Decide returns a keep-mask.<br/>The §6.1 segment-tail-sampler keeps this trace because is_error is set (config keeps errors);<br/>a healthy trace below the 0.500s threshold with no matching tag would fall to the 0.1 sample.<br/>Verdict for 5fcdb353-…: KEEP (drops here reclaim space during routine compaction)"] + A1 -->|"YES (stopped growing)"| A3["Engine builds a projected TraceBatch; plugin Decide returns a keep-mask.<br/>The §6.1 sw-trace-sampler keeps this trace because is_error is set (config keeps errors);<br/>a healthy trace below the 500ms threshold with no matching tag would fall to the 0.1 sample.<br/>Verdict for 5fcdb353-…: KEEP (drops here reclaim space during routine compaction)"] end subgraph B["B. Plugin gating pass at HOT FINALIZATION (PIPELINE_EVENT_FINALIZE enabled, once per settled Hot segment)"] B2["2. Post-trace scheduler tick observes a segment with watermark past segment.End + finalize_grace (300s), not yet finalized"] @@ -650,8 +649,8 @@ flowchart TD B3 --> B4["4. For each surviving trace_id (not already dropped at A), evaluate the plugin gating policy — same as step 1.<br/>Surviving traces continue into the stage lifecycle; failures are dropped from the segment."] end subgraph C["C. Per-stage retention at MIGRATION-OUT (once per stage boundary, when a StageRule has a plugin)"] - C5["5. Source stage migrates to the next stage (e.g. Hot → Warm after the 1-day TTL). Pre-migration rewrite reads each source part with blockReader and runs the source stage's retention plugin, returning a keep-mask per trace_id:<br/>Hot → Warm (hot StageRule.plugins, config min_duration 0.100s): duration ≥ 0.100s? (no, only 4ms); keep_errors AND is_error? YES → Keep, migrate<br/>Warm → Cold (warm StageRule.plugins, config keep_errors only): keep_errors AND is_error? YES → Keep, [...] - C5 --> C6["6. RETENTION OUTCOME: the error trace survives the gating plugin at A and B, then the stage retention plugin keeps it (keep_errors) at every migration boundary in C.<br/>Retained Hot → Warm → Cold; written into the Cold (type=cold) parts. Cold stage TTL = 30 days (per the group's ResourceOpts.stages)."] + C5["5. Source stage migrates to the next stage (e.g. Hot → Warm after the 1-day TTL). Pre-migration rewrite reads each source part with blockReader and runs the source stage's retention plugin, returning a keep-mask per trace_id:<br/>Hot → Warm (hot StageRule.plugins, config durationThresholdMs 100): duration ≥ 0.100s? (no, only 4ms); keepErrors AND is_error? YES → Keep, migrate<br/>Warm → Cold (warm StageRule.plugins, config keepErrors only): keepErrors AND is_error? YES → Keep, [...] + C5 --> C6["6. RETENTION OUTCOME: the error trace survives the gating plugin at A and B, then the stage retention plugin keeps it (keepErrors) at every migration boundary in C.<br/>Retained Hot → Warm → Cold; written into the Cold (type=cold) parts. Cold stage TTL = 30 days (per the group's ResourceOpts.stages)."] end A3 --> B2 B4 --> C5 diff --git a/plugins/README.md b/plugins/README.md index 8ba0483a2..2ed393cf4 100644 --- a/plugins/README.md +++ b/plugins/README.md @@ -10,9 +10,25 @@ the `banyand` server image (see `docs/operation/plugins.md`). plugins/ README.md # this file skywalking/ + internal/tracesampler/ # shared §6.1/§6.2 sampler engine (not a plugin) latencystatussampler/main.go # seed: package main; NewSampler; ABIVersion + sw-trace-sampler/main.go # segment schema (sw_trace): duration/is_error/tags rules + zipkin-trace-sampler/main.go # zipkin_span schema (sw_zipkinTrace): duration/query rules ``` +`internal/tracesampler` is a shared library, not a plugin: it has no `main.go`, +so `make build-plugins` (which only builds `plugins/*/*/` dirs containing a +`main.go`) skips it, and the `sw-trace-sampler` / `zipkin-trace-sampler` mains +each compile it into their own `.so`. The two mains differ only in a `Schema` +value naming the columns each trace schema keeps the rule inputs in — see +[First-party sampler config](#first-party-sampler-config) for that mapping and +the keys both plugins accept. + +Because every `keepTagRules` entry resolves to the flattened array, a rule naming +a first-class column could never match; the engine rejects the ones a `Schema` +names (the error, duration, and start-time columns) at construction instead of +letting the rule silently never fire. + The `skywalking/` namespace holds plugins maintained by the SkyWalking project itself. A future third-party vendor would get its own sibling namespace (e.g. `plugins/acme/`) — but see the "Third-party plugins" section @@ -50,6 +66,64 @@ func main() {} // required by -buildmode=plugin vectorized `sdk.TraceBatch`/`sdk.TraceBlock` types are the full contract — see the package doc on `pkg/pipeline/sdk` for the authoritative reference. +## First-party sampler config + +`sw-trace-sampler` and `zipkin-trace-sampler` share one engine +(`internal/tracesampler`) and accept the same keys. This is the behavioral +reference; for `bydb.yml` wiring, the `SW_STORAGE_BANYANDB_*` environment +overrides, and tuning guidance, see SkyWalking's +[BanyanDB storage guide](https://skywalking.apache.org/docs/main/next/en/setup/backend/storages/banyandb/). + +| Key | Type | Meaning | +|---|---|---| +| `durationThresholdMs` | int | Keep a trace whose end-to-end envelope reaches this many **milliseconds**. `0` disables. | +| `keepErrors` | bool | Keep a trace carrying an error. Rejected on a schema that expresses error neither as a column nor as an array key. | +| `errorTag` | string | Override the tag `keepErrors` reads. Ignored unless `keepErrors` is set. | +| `keepTagRules` | list \| string | Sure-keep rules over the searchable tags: `{tagKey, exists\|equals\|in\|regex}`, or the compact `key=value,key=~regex,key` string form. | +| `healthySampleRate` | float | Fraction in `[0,1]` of the remaining traces to keep, by FNV-1a hash of the trace ID. Accepts a quoted number, since SkyWalking's config loader stringifies float placeholders. | + +**Unknown keys are rejected.** Every option is a keep rule, so a key that +silently missed would leave a sampler with no rules that drops the whole group — +the failure mode is data loss, not a stale setting. A wholly empty config (`{}`, +which `pipeline_loader` also substitutes for an absent `SamplerPlugin.config`) is +rejected for the same reason. Note Go matches field names case-insensitively, so +this catches wrong words, not wrong capitalization; the reference +`_example/segment-tail-sampler` uses `snake_case`, so a config copied from it is +refused outright. + +**Per-schema inputs.** The two mains differ only in this mapping: + +| Input | `sw-trace-sampler` | `zipkin-trace-sampler` | +|---|---|---| +| Searchable tags | `tags` | `query` | +| Error signal | `is_error` column | `error` key inside `query` | +| Per-row start | `start_time` | `timestamp_millis` | +| Per-row duration | `latency` (ms) | `duration` (µs) | + +Both start columns are BanyanDB timestamp columns, so both arrive in +**nanoseconds** whatever the tag is called; only the duration unit differs, and +`Schema.DurationTagNanosPerUnit` normalizes it. `durationThresholdMs` is +milliseconds on both. + +**Duration is an envelope, not a per-span test:** + +``` +envelope = max(start + duration) − min(start) over every row of the trace +keep if envelope ≥ durationThresholdMs +``` + +so a trace slow only through sequential spans is caught. It is deliberately not +the intrinsic `MaxTS − MinTS`, which is the spread of per-row *start* timestamps +(and `0` for a single-row trace). + +**Order and fail-open.** Rules are OR-ed in the order duration, errors, tag +rules, healthy sample; the first match wins, so order affects cost only. A +rule that cannot be evaluated **keeps** the trace: absent duration columns +mean "can't tell", not "not slow", since the columns are schema-declared and +their absence implies the block was written under a different schema. An +absent tag array is *not* treated that way — a trace with no searchable tags +is ordinary data, so tag rules simply do not match. + ## ABI / toolchain lock (read this before building for production) A Go plugin is loaded via `plugin.Open`, which requires the `.so` and the diff --git a/plugins/skywalking/internal/tracesampler/sampler.go b/plugins/skywalking/internal/tracesampler/sampler.go new file mode 100644 index 000000000..f9476414c --- /dev/null +++ b/plugins/skywalking/internal/tracesampler/sampler.go @@ -0,0 +1,763 @@ +// Licensed to Apache Software Foundation (ASF) under one or more contributor +// license agreements. See the NOTICE file distributed with +// this work for additional information regarding copyright +// ownership. Apache Software Foundation (ASF) licenses this file to you under +// the Apache License, Version 2.0 (the "License"); you may +// not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +// Package tracesampler is the shared post-trace sampler engine behind the +// first-party sw-trace-sampler and zipkin-trace-sampler plugins. Both plugins +// implement the same keep logic from docs/design/post-trace-pipeline.md +// (Scenario 6.1 for SkyWalking segments, 6.2 for Zipkin) — a duration +// threshold, sure-keep error and tag rules, and a deterministic healthy sample +// — and differ only in how each schema physically stores the columns those +// rules read. That per-schema knowledge is a Schema value passed to New; +// everything else lives here so the two plugins stay a few lines each and +// cannot drift apart. +// +// Tag matching accounts for the real BanyanDB trace layout SkyWalking writes: +// searchable tags are not first-class columns but "key=value" entries flattened +// into one string-array column ("tags" for segments, "query" for Zipkin), so +// every keepTagRules entry is matched against that array. +package tracesampler + +import ( + "bytes" + "encoding/json" + "errors" + "fmt" + "hash/fnv" + "regexp" + "sort" + "strconv" + "strings" + + "github.com/apache/skywalking-banyandb/pkg/pb/v1/valuetype" + "github.com/apache/skywalking-banyandb/pkg/pipeline/sdk" +) + +// Schema captures the per-plugin storage facts the shared engine needs: where a +// trace's searchable tags live, which columns carry the duration envelope, and +// how (or whether) "error" is expressed. The two first-party plugins differ only +// in the Schema they pass to New. +// +// Only the columns named below are ever read as columns. Every keepTagRules entry +// resolves to ArrayTagColumn, so a rule can only match a searchable tag — never a +// first-class column such as local_endpoint_service_name. New rejects a tagKey +// naming one of the columns it does know to be first-class, rather than letting +// the rule silently never fire. +type Schema struct { + // ArrayTagColumn is the flattened searchable-tag column: a string array of + // "key=value" entries — "tags" for the SkyWalking segment schema, "query" + // for the Zipkin schema. Every keepTagRules entry resolves here: a rule whose + // tagKey is this column's own name matches raw entries, and any other tagKey + // matches the value of the "tagKey=" entries. + ArrayTagColumn string + // ErrorTag is what keepErrors reads. An empty ErrorTag means the schema has no + // error signal at all and keepErrors is rejected at construction. + ErrorTag string + // DurationTag and StartTimeTag drive the durationThresholdMs rule, which + // keeps a trace whose end-to-end envelope reaches the threshold. The envelope + // is max(start + duration) - min(start) over the trace's rows, computed from + // these two per-row tags. This is the true trace duration (it catches traces + // that are slow only through sequential segments), not the spread of the + // intrinsic MinTS/MaxTS (which is per-row start timestamps and 0 for a + // single-row trace). + // + // DurationTag is the per-row duration column: "latency" (segment duration, ms) + // for the segment schema, "duration" (span duration, µs) for Zipkin. + DurationTag string + // StartTimeTag is the per-row start timestamp column: "start_time" for the + // segment schema, "timestamp_millis" for Zipkin. Both are stored as timestamp + // tags (unix nanoseconds), so the plugin reads them as int64 ns. + StartTimeTag string + // DurationTagNanosPerUnit converts one DurationTag unit to nanoseconds so the + // envelope math is ns-consistent with StartTimeTag: 1_000_000 for a millisecond + // tag (segment latency), 1_000 for a microsecond tag (Zipkin duration). + DurationTagNanosPerUnit int64 + // ErrorTagInArray says the error signal is a KEY INSIDE ArrayTagColumn rather + // than a column of its own. The segment schema has a real is_error column + // (false); Zipkin has none, but OAP flattens every span tag into "query" as both + // a bare key and "key=value", so a span carrying Zipkin's conventional "error" + // tag is detectable there (true). + // + // Note this is a tag CONVENTION, not an authoritative field: instrumentations + // that signal failure only through http.status_code 5xx or otel.status_code are + // not covered, and need an explicit keepTagRules entry. + ErrorTagInArray bool +} + +// firstClassColumn reports the config option covering tagKey when the schema +// stores it as a real column rather than an entry of ArrayTagColumn, or "" when +// a rule on tagKey is legitimate. +// +// errorColumn is the error column read at RUNTIME — schema.ErrorTag or the +// operator's errorTag override — and "" when keepErrors is off or the signal is an +// array entry. It is a separate parameter because checking schema.ErrorTag alone +// would miss an override: keepErrors would read the override as a column while a +// rule on the same key matched array entries, the exact silent no-op this guard +// exists to prevent. +// +// Only the columns a Schema names can be checked; a rule on some other first-class +// column (local_endpoint_service_name, say) is still a silent no-op, since the +// engine has no column inventory. Callers reject an empty tagKey before calling +// this, which also stops an unset Schema field from aliasing every rule. +func (s Schema) firstClassColumn(tagKey, errorColumn string) string { + switch tagKey { + case s.DurationTag, s.StartTimeTag: + return "durationThresholdMs" + case errorColumn: + return "keepErrors" + } + // The schema's own error column stays first-class even when keepErrors is off or + // overridden — a rule still cannot reach it. + if tagKey == s.ErrorTag && !s.ErrorTagInArray { + return "keepErrors" + } + return "" +} + +// rule is one sure-keep tag predicate. Exactly one matcher is honored, checked +// in the order exists, equals, in, regex. +type rule struct { + re *regexp.Regexp + Regex string `json:"regex"` + TagKey string `json:"tagKey"` + Equals string `json:"equals"` + In []string `json:"in"` + Exists bool `json:"exists"` +} + +// rules is a keepTagRules list that accepts either the explicit array form or a +// single compact string. The compact form exists because the array-of-objects +// form is unwieldy in an environment variable and, written inline in bydb.yml, +// has to be quoted (its ": " would otherwise start a nested mapping): +// +// keepTagRules: ${...:http.method=GET,http.status_code=~5\d\d} +// +// Grammar — rules separated by commas, each one of: +// +// key=value equals (split on the FIRST "=", so values may contain "=") +// key=~regex regex +// key exists +// +// Commas inside (), [] or {} do not separate rules, so a quantifier such as +// 5\d{2,3} survives. A value containing a top-level comma needs the array form. +type rules []rule + +// UnmarshalJSON accepts the array form verbatim, or a string in the compact form. +func (r *rules) UnmarshalJSON(data []byte) error { + trimmed := strings.TrimSpace(string(data)) + if trimmed == "null" { + return nil + } + if !strings.HasPrefix(trimmed, `"`) { + // Array form: decode through a plain alias so this method is not re-entered. + // Strict for the same reason as the top-level config — a misspelled matcher key + // would otherwise leave a rule that silently matches nothing. + var explicit []rule + dec := json.NewDecoder(bytes.NewReader(data)) + dec.DisallowUnknownFields() + if err := dec.Decode(&explicit); err != nil { + return err + } + *r = explicit + return nil + } + var compact string + if err := json.Unmarshal(data, &compact); err != nil { + return err + } + parsed, err := parseCompactRules(compact) + if err != nil { + return err + } + *r = parsed + return nil +} + +// parseCompactRules parses the compact "key=value,key=~regex,key" grammar. +func parseCompactRules(s string) (rules, error) { + var out rules + for _, part := range splitTopLevel(s, ',') { + part = strings.TrimSpace(part) + if part == "" { + continue + } + eq := strings.Index(part, "=") + if eq == 0 { + return nil, fmt.Errorf("rule %q has an empty tagKey", part) + } + if eq < 0 { + out = append(out, rule{TagKey: part, Exists: true}) + continue + } + key, value := part[:eq], part[eq+1:] + // Report an empty operand here, where the operator's actual mistake is still + // visible. Left to the generic validation below it would surface as the + // misleading "has no matcher", since an empty Equals/Regex is indistinguishable + // from an unset one. + if strings.HasPrefix(value, "~") { + if value == "~" { + return nil, fmt.Errorf("rule %q has an empty regex after %q", part, "=~") + } + out = append(out, rule{TagKey: key, Regex: value[1:]}) + continue + } + if value == "" { + return nil, fmt.Errorf("rule %q has an empty value; write %q to keep on the tag's "+ + "presence regardless of value", part, key) + } + out = append(out, rule{TagKey: key, Equals: value}) + } + return out, nil +} + +// splitTopLevel splits on sep, ignoring separators nested in (), [] or {} — so a +// regex quantifier like {2,3} is not mistaken for a rule boundary. +func splitTopLevel(s string, sep rune) []string { + var parts []string + depth, start := 0, 0 + for i, c := range s { + switch c { + case '(', '[', '{': + depth++ + case ')', ']', '}': + if depth > 0 { + depth-- + } + case sep: + if depth == 0 { + parts = append(parts, s[start:i]) + start = i + len(string(sep)) + } + } + } + return append(parts, s[start:]) +} + +// flexFloat is a float64 that also accepts a JSON string ("0.1"). SkyWalking's +// bydb.yml resolves a ${ENV:default} placeholder through a converter that keeps +// only String/Integer/Long/Boolean types, so a float written as a placeholder +// arrives as a quoted string in the config Struct. Accepting both keeps float +// options env-overridable instead of failing the whole config at admission. +type flexFloat float64 + +// UnmarshalJSON accepts a JSON number or a numeric JSON string. A JSON null is a +// no-op leaving the zero value, per the encoding/json convention — unlike the +// default decoder, a custom Unmarshaler is handed null rather than skipped, and a +// blank value in bydb.yml reaches the plugin as null. +func (f *flexFloat) UnmarshalJSON(data []byte) error { + if string(data) == "null" { + return nil + } + v, err := strconv.ParseFloat(strings.Trim(string(data), `"`), 64) + if err != nil { + return fmt.Errorf("expected a number, got %s", string(data)) + } + *f = flexFloat(v) + return nil +} + +// config is the JSON shape the operator sets in SamplerPlugin.config. +// Field order here (and in the other structs in this file) is chosen to satisfy +// govet's fieldalignment: pointer-bearing fields first, the slice last among them, +// then plain scalars. +type config struct { + ErrorTag string `json:"errorTag"` + KeepTagRules rules `json:"keepTagRules"` + // DurationThresholdMs keeps a trace whose end-to-end duration (the envelope of + // its rows' start+duration, see Schema.DurationTag/StartTimeTag) reaches this + // many milliseconds. Milliseconds match SkyWalking's own latency unit. 0 (or + // omitted) disables it. + DurationThresholdMs int64 `json:"durationThresholdMs"` + HealthySampleRate flexFloat `json:"healthySampleRate"` + KeepErrors bool `json:"keepErrors"` +} + +// Sampler keeps a trace when any sure-keep rule matches, and otherwise admits a +// deterministic fraction of the healthy remainder. It implements sdk.Sampler. +type Sampler struct { + arrayColumn string + errorTag string + durationTag string + startTimeTag string + rules []rule + requiredTags []string + errorRule rule + durationThresholdMs int64 + durationTagNanosPerUnit int64 + healthySampleRate float64 + keepErrors bool + errorTagInArray bool +} + +// New parses and validates the operator config against the given Schema, +// compiles any regex matchers, and computes the projection. A returned error +// rejects the plugin at admission. +func New(configJSON []byte, schema Schema) (sdk.Sampler, error) { + var c config + if len(configJSON) > 0 { + // Strict: an unrecognized key is an error, not a silent no-op. Ignoring one is + // catastrophic here rather than merely untidy — every option this plugin has is + // a KEEP rule, so a config whose keys all miss (a snake_case config copied from + // the _example plugin, say) yields a sampler with no rules at all, which drops + // every trace in the group. Note Go matches field names case-insensitively, so + // this catches wrong words, not wrong capitalization. + dec := json.NewDecoder(bytes.NewReader(configJSON)) + dec.DisallowUnknownFields() + if err := dec.Decode(&c); err != nil { + // Not necessarily malformed JSON: a compact keepTagRules string that fails + // its own grammar also surfaces here, via rules.UnmarshalJSON. + return nil, fmt.Errorf("tracesampler: invalid config: %w", err) + } + } + // A config carrying no keys at all leaves every option at its zero value, which for + // this plugin means no keep rule of any kind: it would drop the entire group. That + // state is reachable by OMISSION, not just by writing {} — pipeline_loader.go + // substitutes []byte("{}") when SamplerPlugin.config is unset — so a plugin + // registered without a config would silently delete the group's traces. + // + // This deliberately tests for an absent KEY, not for absent keep criteria: rate 0 + // with no rules is a supported setting, so {"healthySampleRate": 0} stays valid. + var present map[string]json.RawMessage + if len(configJSON) > 0 { + // Already known well-formed: the strict decode above would have rejected it. + _ = json.Unmarshal(configJSON, &present) + } + if len(present) == 0 { + return nil, errors.New("tracesampler: config is empty, so no keep rule is set and every " + + "trace in the group would be dropped; set at least one option (or leave the pipeline " + + "disabled if retaining nothing is the intent)") + } + if c.HealthySampleRate < 0 || c.HealthySampleRate > 1 { + return nil, fmt.Errorf("tracesampler: healthySampleRate %v out of [0,1]", c.HealthySampleRate) + } + s := &Sampler{ + arrayColumn: schema.ArrayTagColumn, + rules: c.KeepTagRules, + healthySampleRate: float64(c.HealthySampleRate), + keepErrors: c.KeepErrors, + } + if c.KeepErrors { + s.errorTag = schema.ErrorTag + if c.ErrorTag != "" { + s.errorTag = c.ErrorTag + } + if s.errorTag == "" { + return nil, fmt.Errorf("tracesampler: keepErrors is set but this schema has no error signal; "+ + "catch errors with a keepTagRules entry on %q instead", schema.ArrayTagColumn) + } + s.errorTagInArray = schema.ErrorTagInArray + // The in-array error signal is just an exists rule on the error key, evaluated + // by the same matcher as keepTagRules so the two can never diverge. + s.errorRule = rule{TagKey: s.errorTag, Exists: true} + } + if c.DurationThresholdMs < 0 { + return nil, fmt.Errorf("tracesampler: durationThresholdMs must be >= 0, got %d", c.DurationThresholdMs) + } + s.durationThresholdMs = c.DurationThresholdMs + if s.durationThresholdMs > 0 { + if schema.DurationTag == "" || schema.StartTimeTag == "" || schema.DurationTagNanosPerUnit <= 0 { + return nil, fmt.Errorf("tracesampler: durationThresholdMs is set but this schema has no duration/start-time tag configured") + } + s.durationTag = schema.DurationTag + s.startTimeTag = schema.StartTimeTag + s.durationTagNanosPerUnit = schema.DurationTagNanosPerUnit + } + + // Build the projection: the error column (when keepErrors is set), the duration + // envelope columns, and — for any rule at all — the flattened array column, the + // only place a rule can match. Compile regex matchers once, here, not per batch. + tagSet := make(map[string]struct{}) + // errorColumn is set only when the error signal is read as a real column, so it + // carries the errorTag override that schema.ErrorTag would not. + errorColumn := "" + if s.keepErrors { + if s.errorTagInArray { + tagSet[s.arrayColumn] = struct{}{} + } else { + tagSet[s.errorTag] = struct{}{} + errorColumn = s.errorTag + } + } + if s.durationTag != "" { + tagSet[s.durationTag] = struct{}{} + tagSet[s.startTimeTag] = struct{}{} + } + if err := validateRules(s.rules, schema, errorColumn); err != nil { + return nil, err + } + if len(s.rules) > 0 { + // Every rule resolves to the flattened array column, whatever its tagKey. + tagSet[s.arrayColumn] = struct{}{} + } + s.requiredTags = make([]string, 0, len(tagSet)) + for k := range tagSet { + s.requiredTags = append(s.requiredTags, k) + } + // Stable order keeps Project() reproducible across runs (Go map iteration is + // randomized); the engine treats Tags as a set, but logs, tests, and caches + // benefit from determinism. + sort.Strings(s.requiredTags) + return s, nil +} + +// validateRules rejects any rule that could never match and compiles each regex in +// place, so the cost is paid once at admission rather than per batch. errorColumn is +// the error column read at runtime; see Schema.firstClassColumn. +func validateRules(rs []rule, schema Schema, errorColumn string) error { + for i := range rs { + r := &rs[i] + if r.TagKey == "" { + return fmt.Errorf("tracesampler: keepTagRules[%d] has empty tagKey", i) + } + if !r.Exists && r.Equals == "" && len(r.In) == 0 && r.Regex == "" { + return fmt.Errorf("tracesampler: keepTagRules[%d] (tagKey %q) has no matcher; "+ + "set one of exists/equals/in/regex", i, r.TagKey) + } + if firstClass := schema.firstClassColumn(r.TagKey, errorColumn); firstClass != "" { + // A hard error, not a warning: plugins have no log channel, and silently + // accepting the rule is the failure mode this check exists to remove. The + // array-column escape hatch matters when a searchable tag legitimately shares + // a name with a column (a Zipkin span tag literally called "duration"). + return fmt.Errorf("tracesampler: keepTagRules[%d] targets %q, which this schema stores as a "+ + "first-class column, not as an entry of %q; such a rule could never match. Use the %s option "+ + "instead, or — to match a searchable tag that happens to share the name — write the rule "+ + "against %q itself, e.g. {tagKey: %q, regex: \"^%s=\"}", + i, r.TagKey, schema.ArrayTagColumn, firstClass, + schema.ArrayTagColumn, schema.ArrayTagColumn, r.TagKey) + } + if r.Regex != "" { + re, err := regexp.Compile(r.Regex) + if err != nil { + return fmt.Errorf("tracesampler: keepTagRules[%d] bad regex %q: %w", i, r.Regex, err) + } + r.re = re + } + } + return nil +} + +// Kind reports the sampler kind, satisfying the generic sdk.Plugin interface +// that sdk.Sampler embeds. +func (s *Sampler) Kind() sdk.Kind { return sdk.KindSampler } + +// Project declares the columns the verdict reads: the duration and start-time +// columns (when a duration threshold is set), the error column (when keepErrors +// is set), the flattened array column (when any tag rule is configured, or the +// error signal lives there), plus the span-id column only when a span-count rule +// is configured. +// +// Spans stays false, so the verdict never reads span bodies. That is a bound on +// what this plugin inspects, NOT on what the merge decodes: the engine sets +// forceSlow whenever a projection names any tag (merger.go, finalizer.go), which +// disables the raw-copy fast path, and the resulting decode reads the whole block +// — span bodies and every tag, not just the projected ones. Enabling a sampler +// therefore costs a full block decode per merge; keeping the projection small +// bounds only this plugin's own work. +func (s *Sampler) Project() sdk.Projection { + return sdk.Projection{Tags: s.requiredTags} +} + +// Close releases resources; this sampler holds none. +func (s *Sampler) Close() error { return nil } + +// Decide returns a keep-mask aligned to batch.Traces. The batch is read-only. +// It never returns an error: a per-row decode failure fails open for that trace +// (kept), so one malformed value can never make the sampler drop data. +func (s *Sampler) Decide(batch *sdk.TraceBatch) (sdk.Verdict, error) { + keep := make([]bool, len(batch.Traces)) + for i := range batch.Traces { + keep[i] = s.keepTrace(&batch.Traces[i]) + } + return sdk.Verdict{Keep: keep}, nil +} + +// keepTrace applies the sure-keep rules, then the deterministic healthy sample. +// Any decode error encountered while evaluating a sure-keep predicate keeps the +// trace (fail open), never drops it. +func (s *Sampler) keepTrace(b *sdk.TraceBlock) bool { + // Duration keep: the trace's end-to-end envelope reaches the threshold. + if s.durationThresholdMs > 0 { + hit, err := s.hasSlowTrace(b) + if err != nil || hit { + return true + } + } + // Decode the flattened tag array ONCE. DecodeTagValue rewrites its source slice + // in place for string arrays (vararray.UnmarshalVarArray shifts bytes left past + // every escape), so decoding the same column again — once per rule, as an earlier + // version did — reads corrupted entries whenever a value contains "|" or "\\". + var entries []string + if s.needsArrayColumn() { + decoded, err := arrayEntries(b.Tag(s.arrayColumn)) + if err != nil { + return true // fail open + } + entries = decoded + } + // Error keep. + if s.keepErrors { + if s.errorTagInArray { + if matchEntries(entries, &s.errorRule, s.arrayColumn) { + return true + } + } else { + hit, err := s.hasErrorColumn(b) + if err != nil || hit { + return true + } + } + } + // Sure-keep tag rules, all evaluated against the already-decoded entries. + for i := range s.rules { + if matchEntries(entries, &s.rules[i], s.arrayColumn) { + return true + } + } + // Healthy remainder: deterministic hash(trace_id) < rate, stable across + // re-evaluation at merge and finalization. + if s.healthySampleRate > 0 && sampleFraction(b.TraceID) < s.healthySampleRate { + return true + } + return false +} + +// nanosPerMillis converts the millisecond threshold to nanoseconds for the +// envelope comparison, matching the nanosecond start-time tags. +const nanosPerMillis = int64(1_000_000) + +// errNoDurationEnvelope reports that no row yielded a usable start/duration pair, +// so the envelope could not be computed at all. keepTrace treats any error as a +// keep, which is the intended reading: "can't tell", not "not slow". The columns +// are declared by the Schema, so their absence means the block was written under a +// different schema — typically the wrong plugin for the group — and answering +// "not slow" there would silently drop every trace the operator asked to keep. +// +// This is deliberately the opposite of an absent ArrayTagColumn, which stays +// fail-closed: a trace carrying no searchable tags is ordinary data, and keeping +// it would make every tagless trace survive and the tag rules pointless. +var errNoDurationEnvelope = errors.New("no row carries both a start-time and a duration value") + +// hasSlowTrace reports whether the trace's end-to-end envelope reaches the +// threshold. The envelope is max(start + duration) - min(start) over the rows, +// where start comes from StartTimeTag (a timestamp tag, unix ns) and duration +// from DurationTag (scaled to ns by DurationTagNanosPerUnit). This is the true +// trace duration — it catches traces slow only through sequential segments — +// and reads two cheap tag columns, never the span bodies. +func (s *Sampler) hasSlowTrace(b *sdk.TraceBlock) (bool, error) { + startCol := b.Tag(s.startTimeTag) + durCol := b.Tag(s.durationTag) + if startCol == nil || durCol == nil { + return false, errNoDurationEnvelope + } + rows := len(durCol.Values) + if len(startCol.Values) < rows { + rows = len(startCol.Values) + } + var minStart, maxEnd int64 + seen := false + for row := 0; row < rows; row++ { + sv, sErr := startCol.At(row) + if sErr != nil { + return false, sErr + } + dv, dErr := durCol.At(row) + if dErr != nil { + return false, dErr + } + if sv.IsNull() || dv.IsNull() { + continue + } + // StartTimeTag must be a timestamp column, which BanyanDB stores as unix ns + // (write_standalone.go: GetTimestamp().AsTime().UnixNano()) — that is what makes + // reading it as ns correct. A plain Int64 start time is deliberately NOT accepted: + // it carries no unit, and unlike DurationTag there is no nanos-per-unit to scale + // it by, so treating it as ns would silently mis-measure a millisecond column. + // Skipping the row instead makes it a can't-tell, which fails open. + // Duration is an int in the tag's own unit, scaled below. + if sv.ValueType() != valuetype.ValueTypeTimestamp || dv.ValueType() != valuetype.ValueTypeInt64 { + continue + } + start := sv.Int64() + end := start + dv.Int64()*s.durationTagNanosPerUnit + if !seen { + minStart, maxEnd, seen = start, end, true + continue + } + if start < minStart { + minStart = start + } + if end > maxEnd { + maxEnd = end + } + } + if !seen { + return false, errNoDurationEnvelope + } + return maxEnd-minStart >= s.durationThresholdMs*nanosPerMillis, nil +} + +// needsArrayColumn reports whether any predicate reads the flattened tag array. +func (s *Sampler) needsArrayColumn() bool { + return len(s.rules) > 0 || (s.keepErrors && s.errorTagInArray) +} + +// arrayEntries decodes every entry of the flattened tag array, flattened across +// rows. All tag predicates are existential over rows and entries, so collapsing +// the rows loses nothing — and decoding once is what keeps the in-place string +// array decode from corrupting later reads (see keepTrace). +func arrayEntries(col *sdk.TagColumn) ([]string, error) { + if col == nil { + return nil, nil + } + var ( + out []string + scratch = sdk.TagColumn{Name: col.Name, ValueType: col.ValueType, Values: make([][]byte, 1)} + buf []byte + ) + for row := range col.Values { + if col.Values[row] == nil { + continue + } + // Decode a COPY of the row, never the engine's buffer. The SDK's string-array + // decode rewrites its source in place (vararray.UnmarshalVarArray shifts bytes + // left past every escape), TraceBlock slices are documented read-only, and the + // engine hands the SAME TraceBatch to every link of a chain (sdk.applyChainLink). + // Decoding in place would therefore corrupt the value for every later link, making + // a rule's verdict depend on its position in the chain. Only values containing "|" + // or "\" carry an escape, so the damage is silent and data-dependent. + // + // Reusing buf across rows is safe because the decoder builds each entry with a + // string(...) conversion, which copies rather than aliasing the buffer. + buf = append(buf[:0], col.Values[row]...) + scratch.Values[0] = buf + v, err := scratch.At(0) + if err != nil { + return nil, err + } + if v.IsNull() { + continue + } + out = append(out, entriesOf(v)...) + } + return out, nil +} + +// hasErrorColumn reports whether a dedicated error column is truthy on any row. +func (s *Sampler) hasErrorColumn(b *sdk.TraceBlock) (bool, error) { + col := b.Tag(s.errorTag) + if col == nil { + return false, nil + } + for row := range col.Values { + v, err := col.At(row) + if err != nil { + return false, err + } + if v.IsNull() { + continue + } + switch v.ValueType() { + case valuetype.ValueTypeInt64: + if v.Int64() != 0 { + return true, nil + } + case valuetype.ValueTypeStr: + if str := v.Str(); str == "true" || str == "1" { + return true, nil + } + default: + // No other type carries an error signal this schema understands. + } + } + return false, nil +} + +// matchEntries reports whether the rule matches any decoded entry. A tagKey naming +// the array column itself matches raw entries; any other tagKey matches the value +// part of its "tagKey=" entries. +// +// An exists rule additionally accepts the BARE key, because the array holds a tag +// as both "key" and "key=value" (and Zipkin annotations appear as raw values with +// no "="). The comparison is exact, so a longer key such as "error_rate=0" never +// satisfies an exists rule on "error". keepErrors routes through here too, so the +// two cannot drift apart. +func matchEntries(entries []string, r *rule, arrayColumn string) bool { + prefix := "" + if r.TagKey != arrayColumn { + prefix = r.TagKey + "=" + } + for _, entry := range entries { + if r.Exists && entry == r.TagKey { + return true + } + candidate := entry + if prefix != "" { + if !strings.HasPrefix(entry, prefix) { + continue + } + candidate = entry[len(prefix):] + } + if r.Exists || matchValue(r, candidate) { + return true + } + } + return false +} + +// matchValue applies the rule's active string matcher (equals, then in, then +// regex) to a candidate. The exists matcher is handled by the caller, which +// knows whether the value or entry is present. +func matchValue(r *rule, candidate string) bool { + switch { + case r.Equals != "": + return candidate == r.Equals + case len(r.In) > 0: + for _, want := range r.In { + if candidate == want { + return true + } + } + return false + case r.re != nil: + return r.re.MatchString(candidate) + default: + return false + } +} + +// entriesOf returns the string entries of a value: the array elements for a +// string array, or the single string for a plain string tag. Other types have +// no string entries. +func entriesOf(v sdk.Value) []string { + switch v.ValueType() { + case valuetype.ValueTypeStrArr: + return v.StrArr() + case valuetype.ValueTypeStr: + return []string{v.Str()} + default: + return nil + } +} + +// sampleFraction maps a trace_id to a stable fraction in [0,1) via FNV-1a, so +// the keep decision is deterministic and reproducible across passes. The top 53 +// bits fill a float64 mantissa exactly (the technique math/rand uses), so the +// result is strictly below 1 and a healthySampleRate of 1.0 keeps every trace. +func sampleFraction(traceID string) float64 { + h := fnv.New64a() + _, _ = h.Write([]byte(traceID)) + return float64(h.Sum64()>>11) / (1 << 53) +} diff --git a/plugins/skywalking/internal/tracesampler/sampler_test.go b/plugins/skywalking/internal/tracesampler/sampler_test.go new file mode 100644 index 000000000..7f081f15f --- /dev/null +++ b/plugins/skywalking/internal/tracesampler/sampler_test.go @@ -0,0 +1,631 @@ +// Licensed to Apache Software Foundation (ASF) under one or more contributor +// license agreements. See the NOTICE file distributed with +// this work for additional information regarding copyright +// ownership. Apache Software Foundation (ASF) licenses this file to you under +// the Apache License, Version 2.0 (the "License"); you may +// not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +// White-box tests for the shared sampler engine, verified offline against the +// sdktest fixture kit (no .so build, no cluster) per plugins/README.md. +package tracesampler + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/apache/skywalking-banyandb/pkg/pb/v1/valuetype" + "github.com/apache/skywalking-banyandb/pkg/pipeline/sdk" + "github.com/apache/skywalking-banyandb/pkg/pipeline/sdk/sdktest" +) + +// segmentSchema and zipkinSchema mirror the two first-party plugins' Schema +// values so the engine is exercised exactly as they configure it. +var ( + segmentSchema = Schema{ + ArrayTagColumn: "tags", ErrorTag: "is_error", + DurationTag: "latency", StartTimeTag: "start_time", DurationTagNanosPerUnit: 1_000_000, + } + zipkinSchema = Schema{ + ArrayTagColumn: "query", ErrorTag: "error", ErrorTagInArray: true, + DurationTag: "duration", StartTimeTag: "timestamp_millis", DurationTagNanosPerUnit: 1_000, + } + // noErrorSchema has no error signal at all, so keepErrors must be rejected. + noErrorSchema = Schema{ArrayTagColumn: "tags"} +) + +func TestNew_ConfigValidation(t *testing.T) { + cases := []struct { + name string + config string + schema Schema + }{ + {"invalid JSON", `{`, segmentSchema}, + {"rate below 0", `{"healthySampleRate":-0.1}`, segmentSchema}, + {"rate above 1", `{"healthySampleRate":1.1}`, segmentSchema}, + {"empty tagKey", `{"keepTagRules":[{"equals":"x"}]}`, segmentSchema}, + {"no matcher", `{"keepTagRules":[{"tagKey":"db.type"}]}`, segmentSchema}, + {"bad regex", `{"keepTagRules":[{"tagKey":"db.type","regex":"("}]}`, segmentSchema}, + {"bad duration type", `{"durationThresholdMs":"nope"}`, segmentSchema}, + {"negative duration", `{"durationThresholdMs":-5}`, segmentSchema}, + {"keepErrors without error signal", `{"keepErrors":true}`, noErrorSchema}, + // A rule on a first-class column can never match, since every rule resolves to + // the flattened array column. Rejecting beats silently never firing. + {"rule on is_error column", `{"keepTagRules":[{"tagKey":"is_error","equals":"true"}]}`, segmentSchema}, + {"rule on latency column", `{"keepTagRules":[{"tagKey":"latency","exists":true}]}`, segmentSchema}, + {"rule on start_time column", `{"keepTagRules":[{"tagKey":"start_time","exists":true}]}`, segmentSchema}, + {"rule on zipkin duration column", `{"keepTagRules":[{"tagKey":"duration","exists":true}]}`, zipkinSchema}, + // The errorTag override is read as a COLUMN, so a rule on the same key is the same + // silent no-op. Checking schema.ErrorTag alone would have accepted this. + { + "rule on overridden error column", + `{"keepErrors":true,"errorTag":"custom","keepTagRules":[{"tagKey":"custom","exists":true}]}`, + segmentSchema, + }, + // is_error stays a column even when keepErrors is off or points elsewhere. + {"rule on is_error with keepErrors off", `{"keepTagRules":[{"tagKey":"is_error","exists":true}]}`, segmentSchema}, + { + "rule on is_error when errorTag overridden", + `{"keepErrors":true,"errorTag":"custom","keepTagRules":[{"tagKey":"is_error","exists":true}]}`, + segmentSchema, + }, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + _, err := New([]byte(c.config), c.schema) + require.Error(t, err) + }) + } +} + +func TestNew_Projection(t *testing.T) { + // Segment: keepErrors projects is_error; both flattened-tag rules collapse + // onto the single "tags" array column. + s, err := New([]byte(`{ + "keepErrors": true, + "keepTagRules": [ + {"tagKey":"db.type","equals":"PostgreSQL"}, + {"tagKey":"mq.queue","equals":"q"} + ] + }`), segmentSchema) + require.NoError(t, err) + assert.ElementsMatch(t, []string{"is_error", "tags"}, s.Project().Tags) + assert.False(t, s.Project().SpanIDs) + + // Zipkin declares no first-class columns, so every rule — whether it names the + // array column itself or a logical key inside it — collapses onto "query". + z, err := New([]byte(`{ + "keepTagRules": [ + {"tagKey":"query","regex":"http\\.status_code=5\\d\\d"}, + {"tagKey":"http.method","equals":"GET"} + ] + }`), zipkinSchema) + require.NoError(t, err) + assert.ElementsMatch(t, []string{"query"}, z.Project().Tags) + + // A duration threshold opts into the schema's duration + start-time columns. + d, err := New([]byte(`{"durationThresholdMs":500}`), segmentSchema) + require.NoError(t, err) + assert.Contains(t, d.Project().Tags, "latency") + assert.Contains(t, d.Project().Tags, "start_time") +} + +// TestNew_CompactKeepTagRules covers the compact string grammar, which is what an +// operator realistically types into an environment variable. +func TestNew_CompactKeepTagRules(t *testing.T) { + s, err := New([]byte(`{"keepTagRules":"db.type=PostgreSQL, http.status_code=~5\\d{2,3}, mq.queue"}`), segmentSchema) + require.NoError(t, err) + parsed := s.(*Sampler).rules + require.Len(t, parsed, 3, "the comma inside {2,3} must NOT split the regex rule") + + assert.Equal(t, "db.type", parsed[0].TagKey) + assert.Equal(t, "PostgreSQL", parsed[0].Equals) + + assert.Equal(t, "http.status_code", parsed[1].TagKey) + assert.Equal(t, `5\d{2,3}`, parsed[1].Regex) + + assert.Equal(t, "mq.queue", parsed[2].TagKey) + assert.True(t, parsed[2].Exists, "a bare key is an exists rule") + + // A value may contain "=" — only the first one separates key from value. + u, err := New([]byte(`{"keepTagRules":"url=http://x?a=b"}`), segmentSchema) + require.NoError(t, err) + assert.Equal(t, "http://x?a=b";, u.(*Sampler).rules[0].Equals) + + // The array form still works, and an empty string yields no rules. + a, err := New([]byte(`{"keepTagRules":[{"tagKey":"db.type","in":["A","B"]}]}`), segmentSchema) + require.NoError(t, err) + assert.Equal(t, []string{"A", "B"}, a.(*Sampler).rules[0].In) + e, err := New([]byte(`{"keepTagRules":""}`), segmentSchema) + require.NoError(t, err) + assert.Empty(t, e.(*Sampler).rules) + + // Malformed rules are rejected at admission rather than silently ignored, and the + // message must name the actual mistake — an empty operand used to surface as the + // misleading "has no matcher", which describes a rule the operator did not write. + for _, tc := range []struct { + name, config, wantMsg string + }{ + {"empty tagKey", `{"keepTagRules":"=novalue"}`, "empty tagKey"}, + {"empty value", `{"keepTagRules":"db.type="}`, "empty value"}, + {"empty regex", `{"keepTagRules":"db.type=~"}`, "empty regex"}, + } { + t.Run(tc.name, func(t *testing.T) { + _, cErr := New([]byte(tc.config), segmentSchema) + require.Error(t, cErr) + assert.Contains(t, cErr.Error(), tc.wantMsg) + assert.NotContains(t, cErr.Error(), "has no matcher", + "the generic matcher error hides which operand was empty") + }) + } + + // The empty-value error points at the exists form, so the suggested fix must work. + fix, err := New([]byte(`{"keepTagRules":"db.type"}`), segmentSchema) + require.NoError(t, err) + assert.True(t, fix.(*Sampler).rules[0].Exists) +} + +// TestDecide_CompactRulesMatch proves the compact form produces the same verdicts +// as the array form it desugars to. +func TestDecide_CompactRulesMatch(t *testing.T) { + s, err := New([]byte(`{"keepTagRules":"db.type=PostgreSQL,http.status_code=~5\\d{2,3}"}`), segmentSchema) + require.NoError(t, err) + + pg, e := sdktest.NewTrace("pg").Tag("tags", []string{"db.type=PostgreSQL"}).Build() + require.NoError(t, e) + fiveXX, e := sdktest.NewTrace("5xx").Tag("tags", []string{"http.status_code=503"}).Build() + require.NoError(t, e) + miss, e := sdktest.NewTrace("miss").Tag("tags", []string{"db.type=MySQL", "http.status_code=200"}).Build() + require.NoError(t, e) + + verdict, report := sdktest.Run(s, sdktest.Batch(pg, fiveXX, miss)) + require.NoError(t, report.Err) + assert.Empty(t, report.ProjectionDivergedIDs) + assert.Equal(t, []bool{true, true, false}, verdict.Keep) +} + +// TestDecide_EscapedEntriesSurviveMultipleRules is a regression test for in-place +// decoding. vararray.UnmarshalVarArray rewrites its source slice, shifting bytes left +// past every escape, so a column decoded a SECOND time yields corrupted entries. The +// engine used to decode the flattened array once per rule, which silently broke every +// rule after the first whenever a tag value contained "|" or "\" — a retention filter +// dropping data it was told to keep. Values here need escaping and the rule that must +// match is deliberately LAST. +func TestDecide_EscapedEntriesSurviveMultipleRules(t *testing.T) { + s, err := New([]byte(`{ + "keepTagRules": [ + {"tagKey":"never.matches","equals":"nothing"}, + {"tagKey":"db.type","equals":"MySQL"}, + {"tagKey":"db.statement","equals":"SELECT a|b"} + ] + }`), segmentSchema) + require.NoError(t, err) + + // "SELECT a|b" contains the entity delimiter and "C:\\tmp" the escape character, + // so both are stored escaped and are rewritten on decode. + hit, e := sdktest.NewTrace("escaped"). + Tag("tags", []string{`db.type=PostgreSQL`, `url=C:\tmp`, `db.statement=SELECT a|b`}).Build() + require.NoError(t, e) + miss, e := sdktest.NewTrace("miss"). + Tag("tags", []string{`db.type=PostgreSQL`, `db.statement=SELECT c|d`}).Build() + require.NoError(t, e) + + verdict, report := sdktest.Run(s, sdktest.Batch(hit, miss)) + require.NoError(t, report.Err) + assert.Empty(t, report.ProjectionDivergedIDs) + assert.Equal(t, []bool{true, false}, verdict.Keep, + "the third rule must still see intact entries after the first two were evaluated") +} + +// TestDecide_ExistsMatchesBareKey pins the exists semantics after unifying it with +// keepErrors: the key counts whether the array holds it as a bare entry or as +// "key=value". The comparison is exact, so a longer key must NOT satisfy it — +// that guard is what stops "error_rate=0" from looking like an error. +func TestDecide_ExistsMatchesBareKey(t *testing.T) { + s, err := New([]byte(`{"keepTagRules":[{"tagKey":"error","exists":true}]}`), zipkinSchema) + require.NoError(t, err) + + bare, e := sdktest.NewTrace("bare").Tag("query", []string{"error", "http.method=GET"}).Build() + require.NoError(t, e) + withValue, e := sdktest.NewTrace("with-value").Tag("query", []string{"error=boom"}).Build() + require.NoError(t, e) + emptyValue, e := sdktest.NewTrace("empty-value").Tag("query", []string{"error="}).Build() + require.NoError(t, e) + lookalike, e := sdktest.NewTrace("lookalike"). + Tag("query", []string{"error_rate=0", "errors=3", "http.method=GET"}).Build() + require.NoError(t, e) + + verdict, report := sdktest.Run(s, sdktest.Batch(bare, withValue, emptyValue, lookalike)) + require.NoError(t, report.Err) + assert.Empty(t, report.ProjectionDivergedIDs) + assert.Equal(t, []bool{true, true, true, false}, verdict.Keep, + "exists must accept the bare key and key=value, but never a longer key") +} + +// TestDecide_KeepErrorsEqualsExistsRule is the point of the unification: on an +// in-array schema, keepErrors and an explicit exists rule on the same key must +// produce identical verdicts, since the docs present them as equivalent. +func TestDecide_KeepErrorsEqualsExistsRule(t *testing.T) { + viaFlag, err := New([]byte(`{"keepErrors":true,"healthySampleRate":0}`), zipkinSchema) + require.NoError(t, err) + viaRule, err := New([]byte(`{"keepTagRules":[{"tagKey":"error","exists":true}],"healthySampleRate":0}`), zipkinSchema) + require.NoError(t, err) + + for _, tc := range []struct { + name string + query []string + }{ + {"bare", []string{"error"}}, + {"with-value", []string{"error=boom"}}, + {"empty-value", []string{"error="}}, + {"both-forms", []string{"error", "error=boom"}}, + {"lookalike", []string{"error_rate=0"}}, + {"healthy", []string{"http.status_code=200"}}, + } { + t.Run(tc.name, func(t *testing.T) { + block, e := sdktest.NewTrace(tc.name).Tag("query", tc.query).Build() + require.NoError(t, e) + flagVerdict, _ := sdktest.Run(viaFlag, sdktest.Batch(block)) + ruleVerdict, _ := sdktest.Run(viaRule, sdktest.Batch(block)) + assert.Equal(t, flagVerdict.Keep, ruleVerdict.Keep, + "keepErrors and the equivalent exists rule must agree") + }) + } +} + +func TestDecide_FlattenedTagMatchers(t *testing.T) { + s, err := New([]byte(`{ + "keepTagRules": [ + {"tagKey":"db.type","equals":"PostgreSQL"}, + {"tagKey":"http.method","in":["GET","POST"]}, + {"tagKey":"mq.queue","exists":true} + ] + }`), segmentSchema) + require.NoError(t, err) + + equalsHit, e := sdktest.NewTrace("equals").Tag("tags", []string{"db.type=PostgreSQL"}).Build() + require.NoError(t, e) + inHit, e := sdktest.NewTrace("in").Tag("tags", []string{"http.method=POST"}).Build() + require.NoError(t, e) + existsHit, e := sdktest.NewTrace("exists").Tag("tags", []string{"mq.queue=whatever"}).Build() + require.NoError(t, e) + miss, e := sdktest.NewTrace("miss").Tag("tags", []string{"db.type=MySQL", "http.method=DELETE"}).Build() + require.NoError(t, e) + + verdict, report := sdktest.Run(s, sdktest.Batch(equalsHit, inHit, existsHit, miss)) + require.NoError(t, report.Err) + assert.Empty(t, report.ProjectionDivergedIDs) + assert.Equal(t, []bool{true, true, true, false}, verdict.Keep) +} + +// TestDecide_RawArrayVsPrefixedMatch covers the two ways a rule resolves against +// the flattened array: a tagKey naming the array itself ("query") matches whole +// raw entries, while any other tagKey matches only the value part of its +// "tagKey=" entries — so a rule on "http.status_code" must not be satisfied by +// some other key that happens to contain the same text. +func TestDecide_RawArrayVsPrefixedMatch(t *testing.T) { + z, err := New([]byte(`{ + "keepTagRules": [ + {"tagKey":"query","regex":"http\\.status_code=5\\d\\d"}, + {"tagKey":"http.method","equals":"POST"} + ] + }`), zipkinSchema) + require.NoError(t, err) + + // Raw-entry regex hit on the array column's own name. + fiveXX, e := sdktest.NewTrace("5xx"). + Tag("query", []string{"http.status_code", "http.status_code=503"}).Build() + require.NoError(t, e) + // Prefixed hit: the "http.method=" entry's value equals POST. + post, e := sdktest.NewTrace("post"). + Tag("query", []string{"http.method", "http.method=POST"}).Build() + require.NoError(t, e) + // Near miss: another key's VALUE is "POST", but the http.method rule is prefix + // scoped, so it must not match. + otherKey, e := sdktest.NewTrace("other-key"). + Tag("query", []string{"rpc.verb=POST", "http.status_code=200"}).Build() + require.NoError(t, e) + + verdict, report := sdktest.Run(z, sdktest.Batch(fiveXX, post, otherKey)) + require.NoError(t, report.Err) + assert.Empty(t, report.ProjectionDivergedIDs) + assert.Equal(t, []bool{true, true, false}, verdict.Keep) +} + +// TestDecide_Duration exercises the end-to-end trace envelope: the decision is +// max(start_time + latency) - min(start_time) over the trace's segments. Segment +// start_time is a timestamp tag in unix ns; latency is ms (units-per-ns = 1e6). +// The sequential case is the one the old per-segment "max latency" rule missed. +func TestDecide_Duration(t *testing.T) { + s, err := New([]byte(`{"durationThresholdMs":500}`), segmentSchema) + require.NoError(t, err) + + const ms = int64(1_000_000) // ns per ms, for start_time timestamps + + // Single segment, latency 600ms ≥ 500ms → kept. + single, e := sdktest.NewTrace("slow-single"). + TagAs("start_time", valuetype.ValueTypeTimestamp, int64(0)). + Tag("latency", int64(600)).Build() + require.NoError(t, e) + + // Two sequential segments, each 300ms, staggered by 300ms: envelope + // = (300ms + 300ms) - 0 = 600ms ≥ 500ms → kept. Max single latency is 300ms, + // so the old per-segment rule would have WRONGLY dropped this slow trace. + sequential, e := sdktest.NewTrace("slow-sequential"). + TagAs("start_time", valuetype.ValueTypeTimestamp, int64(0)). + TagAs("start_time", valuetype.ValueTypeTimestamp, 300*ms). + Tag("latency", int64(300)). + Tag("latency", int64(300)).Build() + require.NoError(t, e) + + // Two segments, envelope = (100ms + 100ms) - 0 = 200ms < 500ms → dropped. + fast, e := sdktest.NewTrace("fast"). + TagAs("start_time", valuetype.ValueTypeTimestamp, int64(0)). + TagAs("start_time", valuetype.ValueTypeTimestamp, 100*ms). + Tag("latency", int64(100)). + Tag("latency", int64(100)).Build() + require.NoError(t, e) + + verdict, report := sdktest.Run(s, sdktest.Batch(single, sequential, fast)) + require.NoError(t, report.Err) + assert.Empty(t, report.ProjectionDivergedIDs) + assert.Equal(t, []bool{true, true, false}, verdict.Keep) +} + +// TestDecide_DurationZipkin: the Zipkin duration tag is microseconds +// (nanos-per-unit = 1000) and timestamp_millis is a ns timestamp; the envelope +// is computed in ns and compared to the ms threshold. +func TestDecide_DurationZipkin(t *testing.T) { + z, err := New([]byte(`{"durationThresholdMs":1000}`), zipkinSchema) + require.NoError(t, err) + + slow, e := sdktest.NewTrace("slow"). + TagAs("timestamp_millis", valuetype.ValueTypeTimestamp, int64(0)). + Tag("duration", int64(1_500_000)).Build() // 1.5s (µs) ≥ 1s → kept + require.NoError(t, e) + fast, e := sdktest.NewTrace("fast"). + TagAs("timestamp_millis", valuetype.ValueTypeTimestamp, int64(0)). + Tag("duration", int64(2_000)).Build() // 2ms → dropped + require.NoError(t, e) + + verdict, report := sdktest.Run(z, sdktest.Batch(slow, fast)) + require.NoError(t, report.Err) + assert.Empty(t, report.ProjectionDivergedIDs) + assert.Equal(t, []bool{true, false}, verdict.Keep) +} + +// The Zipkin "error" signal is an array entry rather than a column, so a rule on it +// stays legal — the first-class-column rejection must not over-reach. +func TestNew_ZipkinErrorTagRuleAllowed(t *testing.T) { + _, err := New([]byte(`{"keepTagRules":[{"tagKey":"error","exists":true}]}`), zipkinSchema) + require.NoError(t, err) +} + +// A rule's verdict must not depend on the link's POSITION in a chain. The engine hands +// the same *TraceBatch to every link, and the SDK's string-array decode rewrites its +// source in place, so a link ahead that also reads the array column used to corrupt the +// bytes this one decodes. Only values containing "|" or "\" carry an escape, so the +// corruption is data-dependent — TestDecide_EscapedEntriesSurviveMultipleRules covers the +// same hazard between rules inside one sampler; this covers it between links. +func TestDecide_EscapedEntriesSurviveChainedLinks(t *testing.T) { + // Both links keep, so a false verdict can only come from corruption, not from the + // chain ANDing in an unrelated drop. + target := func() sdk.Sampler { + s, err := New([]byte(`{"keepTagRules":[{"tagKey":"db.statement","equals":"SELECT a|b"}]}`), segmentSchema) + require.NoError(t, err) + return s + } + ahead := func() sdk.Sampler { + s, err := New([]byte(`{"keepTagRules":[{"tagKey":"http.method","equals":"GET"}]}`), segmentSchema) + require.NoError(t, err) + return s + } + block := func() sdk.TraceBlock { + b, e := sdktest.NewTrace("escaped"). + Tag("tags", []string{`db.statement=SELECT a|b`, "http.method=GET"}).Build() + require.NoError(t, e) + return b + } + + alone, report := sdktest.Run(target(), sdktest.Batch(block())) + require.NoError(t, report.Err) + require.Equal(t, []bool{true}, alone.Keep, "precondition: the rule matches when it runs alone") + + behind, chainReport := sdktest.RunChain([]sdk.Sampler{ahead(), target()}, sdktest.Batch(block())) + assert.Empty(t, chainReport.Bypassed) + assert.Equal(t, []bool{true}, behind.Keep, + "a link ahead that decodes the same array column must not corrupt it for this one") + + first, chainReport := sdktest.RunChain([]sdk.Sampler{target(), ahead()}, sdktest.Batch(block())) + assert.Empty(t, chainReport.Bypassed) + assert.Equal(t, []bool{true}, first.Keep, "and the verdict must not depend on chain position") +} + +// An empty config sets no keep rule at all, so it would drop the whole group. It is +// reachable by omission: pipeline_loader substitutes {} for an unset SamplerPlugin.config. +func TestNew_RejectsEmptyConfig(t *testing.T) { + for _, cfg := range []string{`{}`, ` {} `, ``} { + _, err := New([]byte(cfg), segmentSchema) + require.Error(t, err, "empty config %q must be rejected", cfg) + } + // A config with a key stays valid even when it sets no keep criteria — rate 0 with + // no rules is a supported setting, not a mistake. + _, err := New([]byte(`{"healthySampleRate":0}`), segmentSchema) + require.NoError(t, err) +} + +// Every option this plugin has is a KEEP rule, so a key that silently misses yields a +// sampler with no rules that drops the whole group. Unknown keys must therefore be a +// decode error — most importantly for a snake_case config copied from the _example +// plugin, which shares none of these spellings. +func TestNew_RejectsUnknownConfigKeys(t *testing.T) { + for _, cfg := range []string{ + `{"duration_threshold_ms":500,"keep_errors":true}`, + `{"keepErrorz":true}`, + `{"durationThresholdMs":500,"typo":1}`, + `{"keepTagRules":[{"tagKey":"db.type","equalz":"PostgreSQL"}]}`, + } { + _, err := New([]byte(cfg), segmentSchema) + require.Error(t, err, "config %s must be rejected, not silently ignored", cfg) + } + + // Go matches field names case-insensitively, so this is accepted rather than + // rejected — the guard catches wrong words, not wrong capitalization. + _, err := New([]byte(`{"durationthresholdms":500}`), segmentSchema) + require.NoError(t, err) +} + +// A Zipkin errorTag override still names an ARRAY entry, so a rule on it stays legal — +// only an override read as a column is rejected. +func TestNew_ZipkinErrorTagOverrideRuleAllowed(t *testing.T) { + _, err := New([]byte(`{"keepErrors":true,"errorTag":"otel.status_code",`+ + `"keepTagRules":[{"tagKey":"otel.status_code","exists":true}]}`), zipkinSchema) + require.NoError(t, err) +} + +// The escape hatch the rejection message points at has to actually work: a searchable tag +// whose key collides with a column name stays reachable via a rule on the array column. +func TestDecide_ArrayColumnEscapeHatch(t *testing.T) { + z, err := New([]byte(`{"keepTagRules":[{"tagKey":"query","regex":"^duration=slow$"}]}`), zipkinSchema) + require.NoError(t, err) + + hit, e := sdktest.NewTrace("collide"). + Tag("query", []string{"http.method=GET", "duration=slow"}).Build() + require.NoError(t, e) + miss, e := sdktest.NewTrace("no-collide"). + Tag("query", []string{"http.method=GET"}).Build() + require.NoError(t, e) + + verdict, report := sdktest.Run(z, sdktest.Batch(hit, miss)) + require.NoError(t, report.Err) + assert.Equal(t, []bool{true, false}, verdict.Keep) +} + +// A plain Int64 start time carries no unit, so it is skipped rather than assumed to be +// nanoseconds. The trace then has no computable envelope and fails open — a 4ms latency +// under a 500ms threshold would otherwise have been dropped. +func TestDecide_Int64StartTimeFailsOpen(t *testing.T) { + s, err := New([]byte(`{"durationThresholdMs":500}`), segmentSchema) + require.NoError(t, err) + + tr, e := sdktest.NewTrace("int64-start"). + Tag("start_time", int64(0)). + Tag("latency", int64(4)).Build() + require.NoError(t, e) + + verdict, report := sdktest.Run(s, sdktest.Batch(tr)) + require.NoError(t, report.Err) + assert.Equal(t, []bool{true}, verdict.Keep) +} + +// A trace whose duration columns are absent cannot be evaluated at all, so the +// threshold rule fails open. Answering "not slow" instead would make a plugin +// pointed at the wrong group's schema silently drop every trace it was configured +// to keep. +func TestDecide_DurationMissingColumnsFailOpen(t *testing.T) { + s, err := New([]byte(`{"durationThresholdMs":500}`), segmentSchema) + require.NoError(t, err) + + // Zipkin-shaped rows under the segment plugin: no start_time, no latency. + wrongSchema, e := sdktest.NewTrace("wrong-schema"). + TagAs("timestamp_millis", valuetype.ValueTypeTimestamp, int64(0)). + Tag("duration", int64(1)).Build() + require.NoError(t, e) + + verdict, report := sdktest.Run(s, sdktest.Batch(wrongSchema)) + require.NoError(t, report.Err) + assert.Equal(t, []bool{true}, verdict.Keep) +} + +func TestDecide_KeepErrors(t *testing.T) { + s, err := New([]byte(`{"keepErrors":true}`), segmentSchema) + require.NoError(t, err) + + intErr, e := sdktest.NewTrace("int-error").Tag("is_error", int64(1)).Build() + require.NoError(t, e) + strErr, e := sdktest.NewTrace("str-error").Tag("is_error", "true").Build() + require.NoError(t, e) + ok, e := sdktest.NewTrace("ok").Tag("is_error", int64(0)).Build() + require.NoError(t, e) + + verdict, report := sdktest.Run(s, sdktest.Batch(intErr, strErr, ok)) + require.NoError(t, report.Err) + assert.Empty(t, report.ProjectionDivergedIDs) + assert.Equal(t, []bool{true, true, false}, verdict.Keep) +} + +// TestNew_HealthySampleRateAcceptsString covers the ${ENV:default} path: OAP's +// config loader stringifies float placeholders, so the rate can arrive quoted. +func TestNew_HealthySampleRateAcceptsString(t *testing.T) { + quoted, err := New([]byte(`{"healthySampleRate":"0"}`), segmentSchema) + require.NoError(t, err) + block, e := sdktest.NewTrace("x").Build() + require.NoError(t, e) + verdict, _ := sdktest.Run(quoted, sdktest.Batch(block)) + assert.Equal(t, []bool{false}, verdict.Keep, `"0" must disable healthy sampling, not fail the config`) + + // A quoted out-of-range rate is still range-checked; a non-numeric string is rejected. + _, err = New([]byte(`{"healthySampleRate":"1.5"}`), segmentSchema) + require.Error(t, err) + _, err = New([]byte(`{"healthySampleRate":"abc"}`), segmentSchema) + require.Error(t, err) + + // A blank value in bydb.yml reaches the plugin as JSON null; treat it as "not set" + // (zero value) rather than failing the whole config. + nulled, err := New([]byte(`{"healthySampleRate":null}`), segmentSchema) + require.NoError(t, err) + verdict, _ = sdktest.Run(nulled, sdktest.Batch(block)) + assert.Equal(t, []bool{false}, verdict.Keep, "null must behave as an unset rate, not fail the config") +} + +func TestDecide_HealthySampleRate(t *testing.T) { + // Rate 1.0 keeps every trace (sampleFraction is strictly < 1). + all, err := New([]byte(`{"healthySampleRate":1.0}`), segmentSchema) + require.NoError(t, err) + for _, id := range []string{"a", "b", "c", "trace-42", ""} { + block, e := sdktest.NewTrace(id).Build() + require.NoError(t, e) + verdict, _ := sdktest.Run(all, sdktest.Batch(block)) + assert.Equal(t, []bool{true}, verdict.Keep, "rate 1.0 must keep %q", id) + } + + // Rate 0 disables healthy sampling: a trace matching no sure-keep rule drops. + none, err := New([]byte(`{"healthySampleRate":0}`), segmentSchema) + require.NoError(t, err) + block, e := sdktest.NewTrace("x").Build() + require.NoError(t, e) + verdict, _ := sdktest.Run(none, sdktest.Batch(block)) + assert.Equal(t, []bool{false}, verdict.Keep) +} + +// TestDecide_FailOpenOnDecodeError proves a malformed tag value keeps the trace +// (fail open) rather than dropping it or erroring the whole batch: an is_error +// column whose raw int64 is not 8 bytes fails to decode, and keepErrors keeps +// the trace anyway. +func TestDecide_FailOpenOnDecodeError(t *testing.T) { + s, err := New([]byte(`{"keepErrors":true}`), segmentSchema) + require.NoError(t, err) + + batch := &sdk.TraceBatch{Traces: []sdk.TraceBlock{{ + TraceID: "malformed", + Tags: []sdk.TagColumn{{ + Name: "is_error", + ValueType: valuetype.ValueTypeInt64, + Values: [][]byte{{0x01, 0x02, 0x03}}, // not 8 bytes → decode error + }}, + }}} + verdict, err := s.Decide(batch) + require.NoError(t, err) + assert.Equal(t, []bool{true}, verdict.Keep, "a decode error must fail open (keep), never drop") +} diff --git a/plugins/skywalking/sw-trace-sampler/main.go b/plugins/skywalking/sw-trace-sampler/main.go new file mode 100644 index 000000000..a69b9e238 --- /dev/null +++ b/plugins/skywalking/sw-trace-sampler/main.go @@ -0,0 +1,71 @@ +// Licensed to Apache Software Foundation (ASF) under one or more contributor +// license agreements. See the NOTICE file distributed with +// this work for additional information regarding copyright +// ownership. Apache Software Foundation (ASF) licenses this file to you under +// the Apache License, Version 2.0 (the "License"); you may +// not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +// Command sw-trace-sampler is the first-party post-trace sampler for the +// SkyWalking-native segment schema (group sw_trace, BanyanDB trace "segment"). +// It implements the Scenario 6.1 keep logic from +// docs/design/post-trace-pipeline.md on the real segment columns: +// +// - trace duration from the envelope of the segment "start_time"/"latency" tags (ms), +// - keepErrors from the first-class is_error tag, +// - keepTagRules matched against the flattened searchable-tag array "tags" +// (entries "key=value", e.g. "db.type=PostgreSQL"), +// - and a deterministic healthySampleRate hash of the trace id. +// +// Config JSON (from SamplerPlugin.config): +// +// { +// "durationThresholdMs": 500, +// "keepErrors": true, +// "healthySampleRate": 0.1, +// "keepTagRules": [ +// { "tagKey": "db.type", "equals": "PostgreSQL" }, +// { "tagKey": "mq.queue", "equals": "queue-songs-ping" } +// ] +// } +// +// Build it as a Go plugin with `make build-plugins`; it must use the same Go +// toolchain and pinned pkg/pipeline/sdk as the data node (see plugins/README.md). +package main + +import ( + "github.com/apache/skywalking-banyandb/pkg/pipeline/sdk" + "github.com/apache/skywalking-banyandb/plugins/skywalking/internal/tracesampler" +) + +// ABIVersion re-exports the SDK ABI version. The engine refuses to load the +// plugin unless this equals its own compiled sdk.ABIVersion. +var ABIVersion = sdk.ABIVersion + +// segmentSchema describes how the SkyWalking segment schema stores the columns +// this sampler reads: trace duration is the envelope of the per-segment +// "start_time" (unix-ns timestamp) and "latency" (ms) tags, errors are the +// first-class is_error tag, and searchable tags are flattened into the "tags" array. +var segmentSchema = tracesampler.Schema{ + ArrayTagColumn: "tags", + ErrorTag: "is_error", + DurationTag: "latency", + StartTimeTag: "start_time", + DurationTagNanosPerUnit: 1_000_000, +} + +// NewSampler is the constructor symbol the engine looks up. +func NewSampler(configJSON []byte) (sdk.Sampler, error) { + return tracesampler.New(configJSON, segmentSchema) +} + +func main() {} diff --git a/plugins/skywalking/sw-trace-sampler/main_test.go b/plugins/skywalking/sw-trace-sampler/main_test.go new file mode 100644 index 000000000..a77213b97 --- /dev/null +++ b/plugins/skywalking/sw-trace-sampler/main_test.go @@ -0,0 +1,93 @@ +// Licensed to Apache Software Foundation (ASF) under one or more contributor +// license agreements. See the NOTICE file distributed with +// this work for additional information regarding copyright +// ownership. Apache Software Foundation (ASF) licenses this file to you under +// the Apache License, Version 2.0 (the "License"); you may +// not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +// Offline verify-before-you-build-a-.so test for sw-trace-sampler, exercising +// the Scenario 6.1 config against the sdktest fixture kit (no .so, no cluster). +package main + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/apache/skywalking-banyandb/pkg/pb/v1/valuetype" + "github.com/apache/skywalking-banyandb/pkg/pipeline/sdk/sdktest" +) + +// config61 is the Scenario 6.1 gating config, with healthySampleRate pinned to +// 0 so every keep here is attributable to a sure-keep rule (deterministic). +const config61 = `{ + "durationThresholdMs": 500, + "keepErrors": true, + "healthySampleRate": 0, + "keepTagRules": [ + { "tagKey": "db.type", "equals": "PostgreSQL" }, + { "tagKey": "mq.queue", "equals": "queue-songs-ping" } + ] +}` + +func TestSWTraceSampler_Scenario61(t *testing.T) { + sampler, err := NewSampler([]byte(config61)) + require.NoError(t, err) + + // Slow healthy trace: envelope 2802ms ≥ 500ms → sure-keep on duration. + slow, e := sdktest.NewTrace("slow-healthy"). + TagAs("start_time", valuetype.ValueTypeTimestamp, int64(0)). + Tag("latency", int64(2802)).Build() + require.NoError(t, e) + // Every trace carries start_time: SegmentRecord stores it as a primitive column, so a + // real segment always has one, and a trace missing it is a can't-tell the duration rule + // deliberately fails open on. Omitting it here would keep these traces by that fallback + // and stop them exercising the rules they are named for. + // + // Fast error trace: sure-keep on keepErrors. + errTrace, e := sdktest.NewTrace("error"). + TagAs("start_time", valuetype.ValueTypeTimestamp, int64(0)). + Tag("latency", int64(4)).Tag("is_error", int64(1)).Build() + require.NoError(t, e) + // Fast PostgreSQL trace: sure-keep on the db.type tag rule. + pg, e := sdktest.NewTrace("postgres"). + TagAs("start_time", valuetype.ValueTypeTimestamp, int64(0)). + Tag("latency", int64(4)). + Tag("is_error", int64(0)). + Tag("tags", []string{"http.method=GET", "db.type=PostgreSQL"}).Build() + require.NoError(t, e) + // Fast healthy trace, matches nothing, rate 0 → dropped. + fast, e := sdktest.NewTrace("fast-healthy"). + TagAs("start_time", valuetype.ValueTypeTimestamp, int64(0)). + Tag("latency", int64(6)). + Tag("is_error", int64(0)). + Tag("tags", []string{"http.method=GET"}).Build() + require.NoError(t, e) + + verdict, report := sdktest.Run(sampler, sdktest.Batch(slow, errTrace, pg, fast)) + require.NoError(t, report.Err) + require.NoError(t, report.ProjectionErr) + assert.Empty(t, report.ProjectionDivergedIDs, + "sw-trace-sampler must only read the columns it projects (start_time, latency, is_error, tags)") + assert.Equal(t, []bool{true, true, true, false}, verdict.Keep) +} + +func TestSWTraceSampler_Project(t *testing.T) { + sampler, err := NewSampler([]byte(config61)) + require.NoError(t, err) + proj := sampler.Project() + assert.ElementsMatch(t, []string{"latency", "start_time", "is_error", "tags"}, proj.Tags) + assert.False(t, proj.SpanIDs) + assert.False(t, proj.Spans) +} diff --git a/plugins/skywalking/zipkin-trace-sampler/main.go b/plugins/skywalking/zipkin-trace-sampler/main.go new file mode 100644 index 000000000..9a15a8bb4 --- /dev/null +++ b/plugins/skywalking/zipkin-trace-sampler/main.go @@ -0,0 +1,77 @@ +// Licensed to Apache Software Foundation (ASF) under one or more contributor +// license agreements. See the NOTICE file distributed with +// this work for additional information regarding copyright +// ownership. Apache Software Foundation (ASF) licenses this file to you under +// the Apache License, Version 2.0 (the "License"); you may +// not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +// Command zipkin-trace-sampler is the first-party post-trace sampler for the +// Zipkin schema (group sw_zipkinTrace, BanyanDB trace "zipkin_span"). It +// implements the Scenario 6.2 keep logic from +// docs/design/post-trace-pipeline.md on the real Zipkin columns: +// +// - trace duration from the envelope of the "timestamp_millis"/"duration" tags (duration is µs), +// - keepTagRules matched against the flattened "query" array, whose entries +// include both bare keys and "key=value" such as "http.status_code=500", +// - and a deterministic healthySampleRate hash of the trace id. +// +// The Zipkin schema has no is_error column. keepErrors instead detects Zipkin's +// conventional "error" span tag, which OAP flattens into "query" as both a bare +// key and "error=<message>". That is a tag convention, not an authoritative field: +// instrumentations that only signal failure through http.status_code 5xx or +// otel.status_code need an explicit rule, e.g. 'http.status_code=~5\d\d'. +// +// Config JSON (from SamplerPlugin.config): +// +// { +// "durationThresholdMs": 1000, +// "keepErrors": true, +// "healthySampleRate": 0.05, +// "keepTagRules": [ +// { "tagKey": "query", "regex": "http\\.status_code=5\\d\\d" } +// ] +// } +// +// Build it as a Go plugin with `make build-plugins`; it must use the same Go +// toolchain and pinned pkg/pipeline/sdk as the data node (see plugins/README.md). +package main + +import ( + "github.com/apache/skywalking-banyandb/pkg/pipeline/sdk" + "github.com/apache/skywalking-banyandb/plugins/skywalking/internal/tracesampler" +) + +// ABIVersion re-exports the SDK ABI version. The engine refuses to load the +// plugin unless this equals its own compiled sdk.ABIVersion. +var ABIVersion = sdk.ABIVersion + +// zipkinSchema describes how the Zipkin schema stores the columns this sampler +// reads: trace duration is the envelope of the per-span "timestamp_millis" +// (unix-ns timestamp) and "duration" (µs) tags, searchable tags are flattened +// into the "query" array, and keepErrors looks for Zipkin's "error" tag inside +// that same array (ErrorTagInArray) since the schema has no error column. +var zipkinSchema = tracesampler.Schema{ + ArrayTagColumn: "query", + ErrorTag: "error", + ErrorTagInArray: true, + DurationTag: "duration", + StartTimeTag: "timestamp_millis", + DurationTagNanosPerUnit: 1_000, +} + +// NewSampler is the constructor symbol the engine looks up. +func NewSampler(configJSON []byte) (sdk.Sampler, error) { + return tracesampler.New(configJSON, zipkinSchema) +} + +func main() {} diff --git a/plugins/skywalking/zipkin-trace-sampler/main_test.go b/plugins/skywalking/zipkin-trace-sampler/main_test.go new file mode 100644 index 000000000..3eee68bad --- /dev/null +++ b/plugins/skywalking/zipkin-trace-sampler/main_test.go @@ -0,0 +1,118 @@ +// Licensed to Apache Software Foundation (ASF) under one or more contributor +// license agreements. See the NOTICE file distributed with +// this work for additional information regarding copyright +// ownership. Apache Software Foundation (ASF) licenses this file to you under +// the Apache License, Version 2.0 (the "License"); you may +// not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +// Offline verify-before-you-build-a-.so test for zipkin-trace-sampler, +// exercising the Scenario 6.2 config against the sdktest fixture kit. +package main + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/apache/skywalking-banyandb/pkg/pb/v1/valuetype" + "github.com/apache/skywalking-banyandb/pkg/pipeline/sdk/sdktest" +) + +// config62 is the Scenario 6.2 gating config, with healthySampleRate pinned to +// 0 so every keep here is attributable to a sure-keep rule (deterministic). +const config62 = `{ + "durationThresholdMs": 1000, + "healthySampleRate": 0, + "keepTagRules": [ + { "tagKey": "query", "regex": "http\\.status_code=5\\d\\d" } + ] +}` + +func TestZipkinTraceSampler_Scenario62(t *testing.T) { + sampler, err := NewSampler([]byte(config62)) + require.NoError(t, err) + + // Slow mesh call: envelope 30.7s (duration µs) ≥ 1s → sure-keep on duration. + slow, e := sdktest.NewTrace("slow-mesh"). + TagAs("timestamp_millis", valuetype.ValueTypeTimestamp, int64(0)). + Tag("duration", int64(30_700_000)). + Tag("query", []string{"http.status_code", "http.status_code=101"}).Build() + require.NoError(t, e) + // Every trace carries timestamp_millis: ZipkinSpanRecord stores it as a primitive + // column, so a real span always has one, and a trace missing it is a can't-tell the + // duration rule deliberately fails open on. Omitting it here would keep these traces by + // that fallback and stop them exercising the rules they are named for. + // + // Fast 5xx span: sure-keep on the query regex rule. + serverError, e := sdktest.NewTrace("server-error"). + TagAs("timestamp_millis", valuetype.ValueTypeTimestamp, int64(0)). + Tag("duration", int64(3_000)). + Tag("query", []string{"http.status_code", "http.status_code=500"}).Build() + require.NoError(t, e) + // Fast p50 span, matches nothing, rate 0 → dropped. + p50, e := sdktest.NewTrace("p50"). + TagAs("timestamp_millis", valuetype.ValueTypeTimestamp, int64(0)). + Tag("duration", int64(2_000)). + Tag("query", []string{"http.status_code", "http.status_code=200"}).Build() + require.NoError(t, e) + + verdict, report := sdktest.Run(sampler, sdktest.Batch(slow, serverError, p50)) + require.NoError(t, report.Err) + require.NoError(t, report.ProjectionErr) + assert.Empty(t, report.ProjectionDivergedIDs, + "zipkin-trace-sampler must only read the columns it projects (timestamp_millis, duration, query)") + assert.Equal(t, []bool{true, true, false}, verdict.Keep) +} + +// TestZipkinTraceSampler_KeepErrors covers keepErrors on a schema with no error +// column: it detects Zipkin's conventional "error" span tag, which OAP writes into +// "query" as both a bare key and "error=<message>". healthySampleRate is 0 so a +// keep can only come from the error signal. +func TestZipkinTraceSampler_KeepErrors(t *testing.T) { + sampler, err := NewSampler([]byte(`{"keepErrors":true,"healthySampleRate":0}`)) + require.NoError(t, err) + + // OAP's normal output: the bare key plus "error=<message>". + both, e := sdktest.NewTrace("err-both"). + Tag("query", []string{"error", "error=Connection refused", "http.method=GET"}).Build() + require.NoError(t, e) + // An empty error message still marks the span as failed. + empty, e := sdktest.NewTrace("err-empty").Tag("query", []string{"error="}).Build() + require.NoError(t, e) + // A tag whose key merely starts with "error" must NOT count. + lookalike, e := sdktest.NewTrace("lookalike"). + Tag("query", []string{"error_rate=0", "http.method=GET"}).Build() + require.NoError(t, e) + // A healthy span has no error entry at all. + healthy, e := sdktest.NewTrace("healthy").Tag("query", []string{"http.status_code=200"}).Build() + require.NoError(t, e) + + verdict, report := sdktest.Run(sampler, sdktest.Batch(both, empty, lookalike, healthy)) + require.NoError(t, report.Err) + assert.Empty(t, report.ProjectionDivergedIDs, + "keepErrors must read only the projected query column") + assert.Equal(t, []bool{true, true, false, false}, verdict.Keep) + + // keepErrors projects the array column, not a separate error column. + assert.ElementsMatch(t, []string{"query"}, sampler.Project().Tags) +} + +func TestZipkinTraceSampler_Project(t *testing.T) { + sampler, err := NewSampler([]byte(config62)) + require.NoError(t, err) + proj := sampler.Project() + assert.ElementsMatch(t, []string{"duration", "timestamp_millis", "query"}, proj.Tags) + assert.False(t, proj.SpanIDs) + assert.False(t, proj.Spans) +}
