Copilot commented on code in PR #1234:
URL:
https://github.com/apache/skywalking-banyandb/pull/1234#discussion_r3670568489
##########
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 stays a plain `[]byte`, with no shared
config struct.
- **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 mechanism. Two things are **opt-in and
default off**: the span-id column (`Projection.SpanIDs`) and the heavy
span-body column (`Projection.Spans`). A subtlety in the native layout makes
this matter: `spanIDs` and `spans` are encoded **together** in one data block
(`mustWriteSpansTo`/`mustReadSpansFrom`, `banyand/trace/block.go`), so reading
span ids is **not** free — requesting either one forces a read of the spans
stream. Only `trace_id` and `MinTS`/`MaxTS` are genuinely intrinsic (they come
from `bl
ockMetadata` with no decode). So the tiers are: intrinsic-always (`trace_id`,
`minTS`/`maxTS`), opt-in-by-name (tags), and opt-in-and-default-off (the spans
stream — span ids and/or span bodies) — which is what makes spans *more*
optional than tags. The declare-up-front handshake matches the
projection-pushdown contracts in [DuckDB's C table
API](https://duckdb.org/docs/stable/clients/c/table_functions)
(`duckdb_init_get_column_index`) and [DataFusion's
`TableProvider::scan`](https://datafusion.apache.org/library-user-guide/custom-table-providers.html)
(`projection: Option<&Vec<usize>>`). `min_duration`-style checks are free from
`minTS`/`maxTS`, and an error predicate is just a projected tag (e.g.
`is_error`), so a plugin that requests neither tags, span ids, nor span bodies
stays on the merge raw fast path (`mustReadRaw` → `mustWriteRawBlock`, §7.1)
and pays no decode at all.
+**(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 mechanism. Two things are **opt-in and
default off**: the span-id column (`Projection.SpanIDs`) and the heavy
span-body column (`Projection.Spans`). A subtlety in the native layout makes
this matter: `spanIDs` and `spans` are encoded **together** in one data block
(`mustWriteSpansTo`/`mustReadSpansFrom`, `banyand/trace/block.go`), so reading
span ids is **not** free — requesting either one forces a read of the spans
stream. Only `trace_id` and `MinTS`/`MaxTS` are genuinely intrinsic (they come
from `bl
ockMetadata` with no decode). So the tiers are: intrinsic-always (`trace_id`,
`minTS`/`maxTS`), opt-in-by-name (tags), and opt-in-and-default-off (the spans
stream — span ids and/or span bodies) — which is what makes spans *more*
optional than tags. The declare-up-front handshake matches the
projection-pushdown contracts in [DuckDB's C table
API](https://duckdb.org/docs/stable/clients/c/table_functions)
(`duckdb_init_get_column_index`) and [DataFusion's
`TableProvider::scan`](https://datafusion.apache.org/library-user-guide/custom-table-providers.html)
(`projection: Option<&Vec<usize>>`). an error predicate is just a projected
tag (e.g. `is_error`), and while `minTS`/`maxTS` are free, they are the spread
of per-row *start* timestamps — **not** a trace duration (they are 0 for a
single-row trace), so a duration predicate must project the schema's own
start/duration tags, so a plugin that requests neither tags, span ids, nor span
bodies stays on the merge raw fast path (`mustRea
dRaw` → `mustWriteRawBlock`, §7.1) and pays no decode at all.
Review Comment:
Sentence starts mid-line after a period; this looks like a typo/grammar
issue in the design doc. Capitalize "An" to keep the paragraph readable.
##########
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,
+ }
Review Comment:
This does not compile: c.KeepTagRules has the named slice type rules, but
Sampler.rules is []rule, so the composite literal assignment is not assignable.
Convert explicitly (or change the field type) so the plugin builds.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
