wankai123 commented on code in PR #1234: URL: https://github.com/apache/skywalking-banyandb/pull/1234#discussion_r3670729777
########## 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 one is a false positive — the code compiles as written. `rules` is a *defined* type whose underlying type is `[]rule`, and `Sampler.rules` is the unnamed type `[]rule`. Go's assignability rule covers exactly this case: a value of type `V` is assignable to type `T` when `V` and `T` have identical underlying types and at least one of them is not a named type ([spec](https://go.dev/ref/spec#Assignability)). Here the underlying types are identical and `[]rule` is unnamed, so no conversion is required. Verified rather than argued: `go build ./plugins/...` succeeds, `go test ./plugins/...` passes all four packages, `golangci-lint` (pinned v1.64.8 with the repo config) reports no findings, and the resulting `.so` files load through `plugin.Open` in the end-to-end runs. Minimal repro of the rule: ```go type rule struct{ K string } type rules []rule // defined type, underlying []rule type S struct{ rules []rule } var c rules = rules{{K: "x"}} s := S{rules: c} // compiles: identical underlying types, []rule is unnamed ``` Leaving the code as-is. -- 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]
