You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by ha...@apache.org on 2026/07/01 00:59:48 UTC
(skywalking-banyandb) branch main updated: feat(trace): dynamic sampler registration via group-scoped pipeline config (#1198)
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
The following commit(s) were added to refs/heads/main by this push:
new ec48e7ff5 feat(trace): dynamic sampler registration via group-scoped pipeline config (#1198)
ec48e7ff5 is described below
commit ec48e7ff5f90b97f6219e67aaab223fca4204ef2
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Wed Jul 1 08:59:42 2026 +0800
feat(trace): dynamic sampler registration via group-scoped pipeline config (#1198)
Register/update/remove in-merge trace retention samplers at runtime without
restarting the data node. The sampler pipeline config is carried on
commonv1.Group (new optional field `pipeline`, with TracePipelineConfig +
SamplerPlugin relocated from pipeline/v1 into the leaf common/v1 to avoid a
common->pipeline import cycle) and delivered via the existing KindGroup schema
watch; the data node rebuilds its group-keyed sampler registry on
OnAddOrUpdate(KindGroup), fail-open and gated to ROLE_DATA + native pipeline.
- registry: atomic replace/remove + named samplers, NEVER-Close
- pluginCache: config-aware key (path,symbol,configHash), ABI re-check on every miss
- validate.GroupForNonProperty validates the embedded config; stage-only configs rejected
- per-config merge_grace; metrics: sampler_{active_count,register,update,remove,load_failed}
- dynamic-only: static --trace-pipeline-config boot loader removed
- standalone out-of-process suite live-green (14/15); distributed + soak suites CI-gated
Generated *.pb.go are gitignored; run `make generate` after checkout.
---
.gitignore | 2 +
.golangci.yml | 7 +
CHANGES.md | 4 +
api/proto/banyandb/common/v1/common.proto | 77 ++++
.../banyandb/pipeline/v1/trace_pipeline.proto | 196 +-------
api/validate/validate.go | 44 ++
api/validate/validate_test.go | 25 +
banyand/metadata/schema/schemaserver/service.go | 9 +-
banyand/trace/merger.go | 10 +-
banyand/trace/metadata.go | 161 ++++++-
banyand/trace/metrics.go | 5 +-
banyand/trace/pipeline_loader.go | 67 ++-
banyand/trace/pipeline_loader_test.go | 104 ++++-
banyand/trace/pipeline_metrics.go | 93 ++++
banyand/trace/pipeline_metrics_test.go | 387 ++++++++++++++++
banyand/trace/pipeline_registry.go | 89 +++-
banyand/trace/pipeline_registry_test.go | 177 ++++++++
banyand/trace/pipeline_watch_test.go | 491 ++++++++++++++++++++
banyand/trace/svc_standalone.go | 42 --
banyand/trace/testmain_test.go | 46 ++
banyand/trace/trace.go | 1 -
docs/api-reference.md | 259 ++++-------
pkg/test/setup/external.go | 79 ++++
test/cases/tracepipeline/ops.go | 177 ++++++++
test/cases/tracepipeline/tracepipeline.go | 260 ++++++++++-
test/integration/distributed/pipeline/common.go | 138 +++---
.../distributed/pipeline/dynamic_test.go | 505 +++++++++++++++++++++
test/integration/standalone/pipeline/common.go | 113 ++---
.../standalone/pipeline/dynamic_test.go | 440 ++++++++++++++++++
test/integration/standalone/pipeline/soak_test.go | 26 +-
test/plugins/_latencystatussampler/main.go | 19 +-
31 files changed, 3468 insertions(+), 585 deletions(-)
diff --git a/.gitignore b/.gitignore
index 956878246..9ddb40cee 100644
--- a/.gitignore
+++ b/.gitignore
@@ -21,6 +21,8 @@
*.dll
*.so
*.dylib
+# stray plugin binary from a bare `go build ./test/plugins/_latencystatussampler/`
+/_latencystatussampler
bin
include
/build
diff --git a/.golangci.yml b/.golangci.yml
index da1c435ae..056429875 100644
--- a/.golangci.yml
+++ b/.golangci.yml
@@ -134,6 +134,13 @@ linters-settings:
run:
go: "1.25"
+ # Lint build-tag-gated sources too. Without this, files behind
+ # `//go:build trace_pipeline` (the trace pipeline integration suites) are
+ # invisible to `make lint`, letting dead code and formatting debt accumulate
+ # silently — they still compile via `make test-trace-pipeline`, and unused
+ # package-level vars/funcs are not compile errors in Go.
+ build-tags:
+ - trace_pipeline
issues:
exclude-rules:
- path: ".*\\.pb\\.go"
diff --git a/CHANGES.md b/CHANGES.md
index 04bc30661..6182ca9df 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -69,6 +69,10 @@ Release Notes.
- Speed up GCS backup uploads: write each object and its checksum metadata in one request, dropping the per-object `Update` round-trip.
- Lifecycle migration now archives rows whose measure/stream schema was deleted from the registry, instead of aborting the group.
- Canopy M3 metadata CRUD: add IndexRule + IndexRuleBinding create/edit/delete forms with admin-only edit/delete, pure `validate*()` module covering 8 input groups (group, stream, measure, trace, index rule, index rule binding, interval, lifecycle stages), `useFocusTrap` + `useDirtyGuard` modal hooks, live `m3-indexrule` e2e (CRUD round-trip + MF2 server-authority negative), and opt-in `m3-handoff` e2e + zero-dep static server for the design-bundle comparison.
+- Throttle lifecycle migration under receiver memory pressure so it no longer OOM-kills the destination data node, via three always-on mechanisms: per-chunk `SYNC_STATUS_SERVER_BUSY` load-shed (the sender backs off and retries); a bounded memory wait before introducing an external series-index segment (`{measure,stream,trace}-lifecycle-receive-mem-wait-timeout`, default 5m); and sender-side bounded-backoff retry of transient send failures — target node restarting/`no nodes available`, di [...]
+- Cut the lifecycle external series-index deduplicator's memory from ~1.8 GB to near-zero per segment by bumping bluge to a memory-efficient implementation: duplicates are now resolved via `_id` term-dictionary point lookups (`Dictionary.Contains`) instead of decompressing and retaining every segment's stored fields, removing the dominant heap source during external-segment introduction. No public API change and identical dedup semantics (verified by a differential test); this is the roo [...]
+- Add the storage-node in-merge trace retention filter (`PIPELINE_EVENT_MERGE`): a per-group sampler chain evaluates traces at merge time and drops non-retained traces from both the core part and the coupled sidx (`sidx.Merge` gains a keep predicate; a nil predicate restores the wholesale merge). Chain execution fails open on panic/error/length-mismatch/timeout, with a consecutive-timeout circuit breaker. Samplers are native Go plugins (`.so`) loaded at startup from `--trace-pipeline-con [...]
+- 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 [...]
### Bug Fixes
diff --git a/api/proto/banyandb/common/v1/common.proto b/api/proto/banyandb/common/v1/common.proto
index 499c6632f..22cfd3ed3 100644
--- a/api/proto/banyandb/common/v1/common.proto
+++ b/api/proto/banyandb/common/v1/common.proto
@@ -19,6 +19,8 @@ syntax = "proto3";
package banyandb.common.v1;
+import "google/protobuf/duration.proto";
+import "google/protobuf/struct.proto";
import "google/protobuf/timestamp.proto";
import "validate/validate.proto";
@@ -109,6 +111,78 @@ message ResourceOpts {
uint32 replicas = 6;
}
+// PipelineEvent identifies a pipeline-wide event that can be independently enabled.
+enum PipelineEvent {
+ PIPELINE_EVENT_UNSPECIFIED = 0;
+ // In-merge filter during Hot-phase LSM compaction merges.
+ PIPELINE_EVENT_MERGE = 1;
+ // Tail-sampling gate at Hot-phase segment finalization.
+ PIPELINE_EVENT_FINALIZE = 2;
+}
+
+// SamplerPlugin configures a user-supplied native Go plugin (.so) that owns a
+// keep/drop verdict over a vectorized batch of traces.
+message SamplerPlugin {
+ // path is the plugin .so filename, resolved within the data node's trusted plugin directory.
+ string path = 1 [(validate.rules).string.min_len = 1];
+ // symbol is the constructor symbol the engine looks up; defaults to "NewSampler" if empty.
+ string symbol = 2;
+ // abi_version is the ABI version the plugin was built against.
+ uint32 abi_version = 3 [(validate.rules).uint32 = {gte: 1}];
+ // config is the plugin-defined configuration serialized to canonical JSON for the constructor.
+ google.protobuf.Struct config = 4;
+}
+
+// Plugin is one link in a pipeline's processing chain.
+message Plugin {
+ // name is the operator-facing identity for this link.
+ string name = 1 [(validate.rules).string.min_len = 1];
+ // kind selects the plugin type; exactly one arm must be set.
+ oneof kind {
+ // sampler owns a keep/drop verdict over a vectorized batch of traces.
+ SamplerPlugin sampler = 2;
+ }
+}
+
+// StageRule binds the pipeline to one lifecycle stage and declares that stage's
+// retention plugin chain.
+message StageRule {
+ // stage is the stage name from the Group's ResourceOpts.stages.
+ string stage = 1 [(validate.rules).string.min_len = 1];
+ // plugins is the ordered retention chain for this stage.
+ repeated Plugin plugins = 2;
+}
+
+// TracePipelineConfig is the group-scoped, name-less in-merge retention configuration.
+// Identity comes from the carrying Group; there is one config per group by construction.
+message TracePipelineConfig {
+ // enabled activates the pipeline.
+ bool enabled = 1;
+ // plugins is the ordered chain of plugins evaluated by enabled events.
+ repeated Plugin plugins = 2;
+ // merge_grace is the per-trace maturity window for the in-merge filter.
+ // Strictly positive if set; engine default 30s if unset.
+ google.protobuf.Duration merge_grace = 3 [(validate.rules).duration = {
+ gt: {seconds: 0}
+ }];
+ // finalize_grace is the per-segment settling window for the scheduled finalization pass.
+ // Strictly positive if set; engine default 5m if unset.
+ google.protobuf.Duration finalize_grace = 4 [(validate.rules).duration = {
+ gt: {seconds: 0}
+ }];
+ // stages declares per-stage retention rules.
+ repeated StageRule stages = 5;
+ // schema_names lists explicit schema names to target within the Group.
+ repeated string schema_names = 6 [(validate.rules).repeated.items.string.min_len = 1];
+ // schema_name_regex is an RE2 pattern matched against schema names.
+ string schema_name_regex = 7;
+ // enabled_events lists the pipeline-wide events to run.
+ repeated PipelineEvent enabled_events = 8 [(validate.rules).repeated.items.enum = {
+ defined_only: true
+ not_in: [0]
+ }];
+}
+
// Group is an internal object for Group management
message Group {
// metadata define the group's identity
@@ -121,4 +195,7 @@ message Group {
google.protobuf.Timestamp updated_at = 4;
// created_at is the first-appearance timestamp; survives updates unchanged.
google.protobuf.Timestamp created_at = 5;
+ // pipeline carries the optional group-scoped in-merge retention config.
+ // Absent or non-CATALOG_TRACE group means no sampler.
+ optional TracePipelineConfig pipeline = 6;
}
diff --git a/api/proto/banyandb/pipeline/v1/trace_pipeline.proto b/api/proto/banyandb/pipeline/v1/trace_pipeline.proto
index aae2159a3..04e216b92 100644
--- a/api/proto/banyandb/pipeline/v1/trace_pipeline.proto
+++ b/api/proto/banyandb/pipeline/v1/trace_pipeline.proto
@@ -21,8 +21,6 @@ package banyandb.pipeline.v1;
import "banyandb/common/v1/common.proto";
import "google/api/annotations.proto";
-import "google/protobuf/duration.proto";
-import "google/protobuf/struct.proto";
import "protoc-gen-openapiv2/options/annotations.proto";
import "validate/validate.proto";
@@ -30,190 +28,8 @@ option go_package = "github.com/apache/skywalking-banyandb/api/proto/banyandb/pi
option java_package = "org.apache.skywalking.banyandb.pipeline.v1";
option (grpc.gateway.protoc_gen_openapiv2.options.openapiv2_swagger) = {base_path: "/api"};
-// PipelineEvent identifies a pipeline-wide event that can be independently
-// enabled. Per-stage retention (StageRule.plugins) fires implicitly at the
-// stage's migration-out boundary and is not toggleable via this enum.
-enum PipelineEvent {
- PIPELINE_EVENT_UNSPECIFIED = 0;
- // In-merge filter during Hot-phase LSM compaction merges (Warm/Cold
- // compactions stay lossless). Per-trace drops are gated by `merge_grace` so
- // partial traces are not destroyed prematurely. Cheap, runs often; verdicts
- // wait for trace maturity (see §7.1).
- PIPELINE_EVENT_MERGE = 1;
- // Tail-sampling gate at Hot-phase segment finalization, after the segment
- // has settled (event-time watermark past `segment.End + finalize_grace`).
- // Heavy but authoritative; sees the complete trace (see §7.3).
- PIPELINE_EVENT_FINALIZE = 2;
-}
-
-// TracePipelineConfig is the root configuration for a storage-node trace pipeline.
-// It reuses existing catalog identifiers (group via metadata, stage names, schema
-// names) for targeting instead of declaring a parallel metadata model.
-//
-// The pipeline has up to three filter points:
-// 1. PIPELINE_EVENT_MERGE — in-merge filter during Hot-phase LSM compaction (default).
-// 2. PIPELINE_EVENT_FINALIZE — tail-sampling gate at Hot-phase finalization.
-// 3. Per-stage retention via StageRule.plugins, applied at the stage's
-// migration-out boundary (when the segment migrates to the next stage).
-// Always implicit when any StageRule carries a plugin chain.
-// Events 1 and 2 are toggleable via `enabled_events`. The gating policy those
-// events evaluate is the `plugins` chain (a sequential pipe of native Go
-// plugins; today each link is a sampler).
-message TracePipelineConfig {
- // Identity and revision tracking; metadata.group is the Group this pipeline
- // lives in and applies to, consistent with every other schema resource.
- // Required: every config needs a name/group for registry handling.
- common.v1.Metadata metadata = 1 [(validate.rules).message = {required: true}];
- // Active status of the pipeline.
- bool enabled = 2;
- // Per-stage retention rules: which lifecycle stages this pipeline acts on,
- // with the retention plugin chain for each. Each rule fires at its stage's
- // migration-out boundary regardless of `enabled_events`. Empty means the
- // only filters are the `enabled_events` events (no per-stage drop).
- repeated StageRule stages = 3;
- // Explicit schema names to target within the Group (exact match on Metadata.name).
- // Each entry must be non-empty; cross-element uniqueness is enforced server-side.
- repeated string schema_names = 4 [(validate.rules).repeated.items.string.min_len = 1];
- // RE2 regular expression matched against schema names. A schema is targeted if it
- // is listed in schema_names OR matches this pattern. When both are empty, every
- // schema in the Group is targeted.
- string schema_name_regex = 5;
- // Gating policy: an ordered chain of plugins (a sequential pipe) evaluated by
- // any enabled event (PIPELINE_EVENT_MERGE and/or PIPELINE_EVENT_FINALIZE).
- // Links run in declared order, each processing the traces the previous link
- // kept; a link that fails is bypassed (fail-open). Today every link is a
- // sampler, so the chain is the conjunction of the links' keep/drop verdicts.
- // Empty means the only retention is the per-stage StageRule plugin chain(s)
- // at migration-out.
- repeated Plugin plugins = 6;
- // Pipeline-wide events to run. Empty defaults to [PIPELINE_EVENT_MERGE] —
- // the in-merge filter is on, the finalization gate is off. To enable the
- // finalization gate, include PIPELINE_EVENT_FINALIZE; to disable the merge
- // filter, list only [PIPELINE_EVENT_FINALIZE]; the explicit empty default
- // value is also acceptable to mean "merge only". Each element must be a
- // defined, non-UNSPECIFIED value; duplicates are normalized to a set
- // server-side.
- repeated PipelineEvent enabled_events = 7 [(validate.rules).repeated.items.enum = {
- defined_only: true
- not_in: [0]
- }];
- // Per-trace maturity window for the in-merge filter (§7.1). A trace is
- // eligible for dropping during an LSM compaction merge only once its latest
- // span timestamp is older than `now - merge_grace`; younger traces pass
- // through the merge unchanged. Bounds the expected intra-trace span arrival
- // spread (typically seconds). Used iff `enabled_events` contains
- // PIPELINE_EVENT_MERGE. Strictly positive if set; engine default 30s if unset.
- google.protobuf.Duration merge_grace = 8 [(validate.rules).duration = {
- gt: {seconds: 0}
- }];
- // Per-segment settling window for the scheduled finalization pass (§7.3). A
- // segment is treated as settled, and the authoritative final filter runs,
- // once the event-time watermark exceeds `segment.End + finalize_grace`.
- // Bounds segment-wide late arrival (typically minutes). Used iff
- // `enabled_events` contains PIPELINE_EVENT_FINALIZE. Strictly positive if
- // set; engine default 5m if unset.
- google.protobuf.Duration finalize_grace = 9 [(validate.rules).duration = {
- gt: {seconds: 0}
- }];
-}
-
-// StageRule binds the pipeline to one lifecycle stage of the targeted Group
-// and declares that stage's retention plugin chain. The rule fires at the
-// stage's migration-out boundary (i.e. when a segment migrates from this stage
-// to the next stage); routine compaction is governed by PIPELINE_EVENT_MERGE on
-// TracePipelineConfig, not by StageRule.
-//
-// Per-stage retention uses the SAME chain mechanism as gating: each stage's
-// `plugins` chain owns the keep/drop verdict for traces leaving that stage. A
-// StageRule with an empty `plugins` chain has no filtering effect — every trace
-// at this stage migrates unchanged. The "rising bar" across stages (Hot keeps
-// more, Cold keeps less) is expressed by each stage's plugin config (see §4.2
-// of the design doc), not by a fixed predicate vocabulary.
-message StageRule {
- // Stage name from the Group's ResourceOpts.stages (e.g. "hot", "warm", "cold").
- // Must be non-empty; an empty stage name cannot match any lifecycle stage.
- string stage = 1 [(validate.rules).string.min_len = 1];
- // Per-stage retention chain: the ordered plugins (a sequential pipe) that
- // decide keep/drop for traces leaving this stage at its migration-out
- // boundary. Empty means no per-stage drop (every trace migrates). Same
- // contract and composition as the gating chain (see Plugin).
- repeated Plugin plugins = 2;
-}
-
-// Plugin is one link in a pipeline's processing chain — a generic, kind-tagged
-// envelope around a user-supplied native Go plugin. The set oneof arm selects
-// the kind; today the only kind is a sampler (SamplerPlugin). Adding a new kind
-// is purely additive: define its payload message and add a new arm to `kind`,
-// leaving existing arms and field numbers untouched.
-//
-// A chain of Plugin (TracePipelineConfig.plugins, StageRule.plugins) is a
-// sequential pipe: links run in declared order, each link processes the traces
-// the previous link kept, and a link that fails is bypassed (its input passes
-// through unchanged). The plugin↔engine contract for each kind (the vectorized
-// batch, the projection handshake, the verdict) lives in the pinned Go SDK
-// module `pkg/pipeline/sdk`, not in this proto.
-message Plugin {
- // Operator-facing identity for this link, used in diagnostics and admission
- // errors. Must be non-empty; cross-element uniqueness within a chain is
- // enforced server-side.
- string name = 1 [(validate.rules).string.min_len = 1];
- // Plugin kind. Exactly one arm must be set; the set arm is the kind
- // discriminator. An unset kind is rejected at admission (protoc-gen-validate
- // cannot mark a oneof required).
- oneof kind {
- // Sampler kind: owns a keep/drop verdict over a vectorized batch of traces.
- SamplerPlugin sampler = 2;
- }
-}
-
-// SamplerPlugin configures a user-supplied native Go plugin (a .so loaded
-// in-process via the Go `plugin` package) that owns a keep/drop verdict over a
-// vectorized batch of traces. It is the sampler kind of Plugin (Plugin.sampler)
-// and is the keep/drop link wherever a chain runs: in a TracePipelineConfig
-// chain it gates at the enabled PipelineEvent(s) (merge / finalization); in a
-// StageRule chain it is per-stage retention at a stage's migration-out boundary.
-//
-// The full plugin↔engine contract (the vectorized batch type, the projection
-// handshake, and the verdict shape) lives in the pinned Go SDK module
-// `pkg/pipeline/sdk`, not in this proto; this message only locates and admits
-// the plugin. Three properties of that contract:
-// - Strong compatibility: the boundary exchanges only stdlib/primitive types
-// defined in the pinned SDK, so no third-party struct version is pinned
-// across the .so boundary. The plugin must be built with the SAME Go
-// toolchain, build tags, and flags (-trimpath, CGO) and the SAME pinned SDK
-// as the running data node; `abi_version` is checked at load.
-// - Vectorized input: the plugin's Decide is called once per columnar batch of
-// traces, not once per trace.
-// - Projection: the plugin declares the columns it needs (SDK Project →
-// Projection{Tags, SpanIDs, Spans}); the engine materializes only those tag
-// columns and, only when requested, the spans stream — like a query
-// projection.
-//
-// Operational constraints (Go `plugin`): Linux/macOS only; plugins cannot be
-// unloaded, so changing one requires a node restart (no hot-reload); a plugin
-// panic is contained with recover() and fails open (the whole batch is
-// retained). Loading arbitrary code is operator-only and gated behind a server
-// flag plus a trusted plugin directory.
-message SamplerPlugin {
- // Plugin .so filename, resolved within the data node's trusted plugin
- // directory. The engine rejects any path that escapes that directory.
- string path = 1 [(validate.rules).string.min_len = 1];
- // Constructor symbol the engine looks up; defaults to "NewSampler" if empty.
- string symbol = 2;
- // ABI version the plugin was built against. The engine refuses to load the
- // plugin unless this equals its own compiled sdk.ABIVersion.
- uint32 abi_version = 3 [(validate.rules).uint32 = {gte: 1}];
- // Plugin-defined configuration, set directly in the pipeline config as a
- // structured object. The engine does not interpret its keys: it serializes
- // the Struct to canonical JSON and hands the bytes to the plugin's
- // constructor (SDK NewSampler([]byte)), which unmarshals them into the
- // plugin's own typed config and validates them — a malformed config fails
- // the load. Optional: a plugin that needs no configuration leaves it unset.
- google.protobuf.Struct config = 4;
-}
-
message TracePipelineRegistryServiceCreateRequest {
- TracePipelineConfig trace_pipeline_config = 1 [(validate.rules).message = {required: true}];
+ common.v1.TracePipelineConfig trace_pipeline_config = 1 [(validate.rules).message = {required: true}];
}
message TracePipelineRegistryServiceCreateResponse {
@@ -221,7 +37,7 @@ message TracePipelineRegistryServiceCreateResponse {
}
message TracePipelineRegistryServiceUpdateRequest {
- TracePipelineConfig trace_pipeline_config = 1 [(validate.rules).message = {required: true}];
+ common.v1.TracePipelineConfig trace_pipeline_config = 1 [(validate.rules).message = {required: true}];
}
message TracePipelineRegistryServiceUpdateResponse {
@@ -236,7 +52,7 @@ message TracePipelineRegistryServiceDeleteResponse {
bool deleted = 1;
// delete_time is the server-assigned tombstone timestamp in unix nanos.
int64 delete_time = 2;
- // mod_revision is the etcd revision of the tombstone; zero if the server did not record one.
+ // mod_revision is the revision of the tombstone; zero if the server did not record one.
int64 mod_revision = 3;
}
@@ -245,7 +61,7 @@ message TracePipelineRegistryServiceGetRequest {
}
message TracePipelineRegistryServiceGetResponse {
- TracePipelineConfig trace_pipeline_config = 1;
+ common.v1.TracePipelineConfig trace_pipeline_config = 1;
}
message TracePipelineRegistryServiceExistRequest {
@@ -262,7 +78,7 @@ message TracePipelineRegistryServiceListRequest {
}
message TracePipelineRegistryServiceListResponse {
- repeated TracePipelineConfig trace_pipeline_config = 1;
+ repeated common.v1.TracePipelineConfig trace_pipeline_config = 1;
}
// TracePipelineRegistryService manages TracePipelineConfig resources, mirroring
@@ -278,7 +94,7 @@ service TracePipelineRegistryService {
rpc Update(TracePipelineRegistryServiceUpdateRequest) returns (TracePipelineRegistryServiceUpdateResponse) {
option (google.api.http) = {
- put: "/v1/trace-pipeline/schema/{trace_pipeline_config.metadata.group}/{trace_pipeline_config.metadata.name}"
+ put: "/v1/trace-pipeline/schema"
body: "*"
};
}
diff --git a/api/validate/validate.go b/api/validate/validate.go
index 4aa4f1342..635c1dabc 100644
--- a/api/validate/validate.go
+++ b/api/validate/validate.go
@@ -21,10 +21,12 @@ package validate
import (
"errors"
"fmt"
+ "path/filepath"
"strings"
commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+ "github.com/apache/skywalking-banyandb/pkg/pipeline/sdk"
)
const reservedTagSeparator = "#"
@@ -104,6 +106,48 @@ func GroupForNonProperty(group *commonv1.Group) error {
if group.ResourceOpts.Ttl.Unit == commonv1.IntervalRule_UNIT_UNSPECIFIED {
return errors.New("group ttl unit is unspecified")
}
+ if pipelineCfg := group.GetPipeline(); pipelineCfg != nil {
+ if validateErr := validateTracePipelineConfig(pipelineCfg); validateErr != nil {
+ return fmt.Errorf("group pipeline config is invalid: %w", validateErr)
+ }
+ }
+ return nil
+}
+
+// validateTracePipelineConfig validates the embedded TracePipelineConfig on a Group.
+// It checks each plugin's sampler for trusted-dir-safe path shape and ABI version,
+// then delegates bounds validation to the generated proto validator.
+func validateTracePipelineConfig(cfg *commonv1.TracePipelineConfig) error {
+ if validateErr := cfg.Validate(); validateErr != nil {
+ return fmt.Errorf("proto validation failed: %w", validateErr)
+ }
+ for idx, plugin := range cfg.GetPlugins() {
+ sp := plugin.GetSampler()
+ if sp == nil {
+ continue
+ }
+ if sp.GetPath() == "" {
+ return fmt.Errorf("plugins[%d].sampler.path is empty", idx)
+ }
+ // Reject absolute paths and path segments that escape via "..".
+ cleaned := filepath.Clean(sp.GetPath())
+ if filepath.IsAbs(cleaned) {
+ return fmt.Errorf("plugins[%d].sampler.path must be a relative path, got %q", idx, sp.GetPath())
+ }
+ if cleaned == ".." || strings.HasPrefix(cleaned, ".."+string(filepath.Separator)) {
+ return fmt.Errorf("plugins[%d].sampler.path %q escapes the trusted directory", idx, sp.GetPath())
+ }
+ if sp.GetAbiVersion() != sdk.ABIVersion {
+ return fmt.Errorf("plugins[%d].sampler.abi_version must be %d, got %d", idx, sdk.ABIVersion, sp.GetAbiVersion())
+ }
+ }
+ for idx, stage := range cfg.GetStages() {
+ for jdx, plugin := range stage.GetPlugins() {
+ if plugin.GetSampler() != nil {
+ return fmt.Errorf("stages[%d].plugins[%d]: stages[].plugins are not supported in v1; declare plugins at the top level", idx, jdx)
+ }
+ }
+ }
return nil
}
diff --git a/api/validate/validate_test.go b/api/validate/validate_test.go
index 7ae465daa..995926b35 100644
--- a/api/validate/validate_test.go
+++ b/api/validate/validate_test.go
@@ -67,6 +67,31 @@ func TestMeasurePassesWithNonPrefixShardingKey(t *testing.T) {
assert.NoError(t, err, "Measure() must not reject a non-prefix sharding key")
}
+func TestValidateTracePipelineConfig_StagePluginsRejected(t *testing.T) {
+ // A config with sampler plugins declared only under stages[] must be rejected
+ // with a clear error — not silently accepted (which would load zero samplers
+ // and retain all traces).
+ stagePlugin := &commonv1.Plugin{
+ Name: "lss",
+ Kind: &commonv1.Plugin_Sampler{
+ Sampler: &commonv1.SamplerPlugin{
+ Path: "sampler.so",
+ Symbol: "NewSampler",
+ AbiVersion: 1,
+ },
+ },
+ }
+ cfg := &commonv1.TracePipelineConfig{
+ Enabled: true,
+ Stages: []*commonv1.StageRule{
+ {Stage: "warm", Plugins: []*commonv1.Plugin{stagePlugin}},
+ },
+ }
+ err := validateTracePipelineConfig(cfg)
+ assert.Error(t, err, "stages[].plugins with a sampler must be rejected")
+ assert.Contains(t, err.Error(), "stages[].plugins are not supported in v1")
+}
+
func TestCheckShardingKeySubset(t *testing.T) {
tests := []struct {
name string
diff --git a/banyand/metadata/schema/schemaserver/service.go b/banyand/metadata/schema/schemaserver/service.go
index d53621ddc..4024077ff 100644
--- a/banyand/metadata/schema/schemaserver/service.go
+++ b/banyand/metadata/schema/schemaserver/service.go
@@ -193,6 +193,13 @@ func (s *server) Validate() error {
return nil
}
+// batchWaitSeconds converts a flush timeout to whole seconds for the index
+// BatchWaitSec, rounding up with a floor of 1. A sub-second flush timeout must
+// still yield a positive wait — a 0-second batch wait risks a tight flush loop.
+func batchWaitSeconds(d time.Duration) int64 {
+ return max(int64((d+time.Second-1)/time.Second), 1)
+}
+
func (s *server) PreRun(ctx context.Context) error {
s.l = logger.GetLogger("schema-server")
var err error
@@ -246,7 +253,7 @@ func (s *server) PreRun(ctx context.Context) error {
TreeSlotCount: s.repairTreeSlotCount,
},
Index: db.IndexConfig{
- BatchWaitSec: 5,
+ BatchWaitSec: batchWaitSeconds(s.flushTimeout),
WaitForPersistence: false,
},
Snapshot: db.SnapshotConfig{
diff --git a/banyand/trace/merger.go b/banyand/trace/merger.go
index 492a389c3..aa75cb343 100644
--- a/banyand/trace/merger.go
+++ b/banyand/trace/merger.go
@@ -40,12 +40,9 @@ import (
var mergeMaxConcurrencyCh = make(chan struct{}, cgroups.CPUs())
-var (
+const (
mergeTypeMem = "mem"
mergeTypeFile = "file"
-)
-
-var (
mergeLaneFast = "fast"
mergeLaneSlow = "slow"
)
@@ -328,7 +325,10 @@ func (tst *tsTable) mergePartsThenSendIntroduction(creator snapshotCreator, part
var filter *mergeFilter
if tst.option.nativePipelineEnabled {
if samplers := lookupSamplers(tst.group); len(samplers) > 0 {
- graceNs := int64(tst.option.mergeGraceDefault)
+ graceNs := lookupMergeGrace(tst.group)
+ if graceNs <= 0 {
+ graceNs = int64(tst.option.mergeGraceDefault)
+ }
if !isMergeHot(parts, graceNs, time.Now().UnixNano()) {
chain := newMergeChain(tst.group, "", samplers, tst.option.decideTimeoutCircuitBreak)
filter = &mergeFilter{
diff --git a/banyand/trace/metadata.go b/banyand/trace/metadata.go
index 4fd68f555..fb4c15888 100644
--- a/banyand/trace/metadata.go
+++ b/banyand/trace/metadata.go
@@ -21,6 +21,8 @@ import (
"context"
"fmt"
"path"
+ "slices"
+ "strings"
"time"
"github.com/pkg/errors"
@@ -40,6 +42,7 @@ import (
"github.com/apache/skywalking-banyandb/pkg/fs"
"github.com/apache/skywalking-banyandb/pkg/logger"
"github.com/apache/skywalking-banyandb/pkg/meter"
+ "github.com/apache/skywalking-banyandb/pkg/pipeline/sdk"
resourceSchema "github.com/apache/skywalking-banyandb/pkg/schema"
"github.com/apache/skywalking-banyandb/pkg/schema/registry"
"github.com/apache/skywalking-banyandb/pkg/timestamp"
@@ -63,21 +66,27 @@ type SchemaService interface {
type schemaRepo struct {
resourceSchema.Repository
- onGroupDelete func(groupName string)
- l *logger.Logger
- metadata metadata.Repo
- path string
- nodeID string
- role databasev1.Role
+ onGroupDelete func(groupName string)
+ l *logger.Logger
+ metadata metadata.Repo
+ samplerMeter *samplerMetrics
+ path string
+ nodeID string
+ trustedPluginDir string
+ role databasev1.Role
+ nativePipelineEnabled bool
}
func newSchemaRepo(path string, svc *standalone, nodeLabels map[string]string, nodeID string) schemaRepo {
sr := schemaRepo{
- l: svc.l,
- path: path,
- metadata: svc.metadata,
- nodeID: nodeID,
- role: databasev1.Role_ROLE_DATA,
+ l: svc.l,
+ path: path,
+ metadata: svc.metadata,
+ nodeID: nodeID,
+ role: databasev1.Role_ROLE_DATA,
+ nativePipelineEnabled: svc.option.nativePipelineEnabled,
+ trustedPluginDir: svc.option.trustedPluginDir,
+ samplerMeter: newSamplerMetrics(svc.omr.With(pipelineScope)),
Repository: resourceSchema.NewRepository(
svc.metadata,
svc.l,
@@ -92,10 +101,11 @@ func newSchemaRepo(path string, svc *standalone, nodeLabels map[string]string, n
func newLiaisonSchemaRepo(path string, svc *liaison, traceDataNodeRegistry grpc.NodeRegistry) schemaRepo {
sr := schemaRepo{
- l: svc.l,
- path: path,
- metadata: svc.metadata,
- role: databasev1.Role_ROLE_LIAISON,
+ l: svc.l,
+ path: path,
+ metadata: svc.metadata,
+ role: databasev1.Role_ROLE_LIAISON,
+ samplerMeter: newSamplerMetrics(svc.omr.With(pipelineScope)),
Repository: resourceSchema.NewRepository(
svc.metadata,
svc.l,
@@ -176,6 +186,9 @@ func (sr *schemaRepo) OnAddOrUpdate(metadata schema.Metadata) {
Kind: resourceSchema.EventKindGroup,
Metadata: g,
})
+ if sr.role == databasev1.Role_ROLE_DATA && sr.nativePipelineEnabled {
+ sr.reconcilePipeline(g.Metadata.Name, g.GetPipeline())
+ }
case schema.KindTrace:
if err := validate.Trace(metadata.Spec.(*databasev1.Trace)); err != nil {
sr.l.Warn().Err(err).Msg("trace is ignored")
@@ -231,6 +244,11 @@ func (sr *schemaRepo) OnDelete(metadata schema.Metadata) {
Kind: resourceSchema.EventKindGroup,
Metadata: g,
})
+ if sr.role == databasev1.Role_ROLE_DATA && sr.nativePipelineEnabled {
+ removeSamplersForGroup(g.Metadata.Name)
+ sr.samplerMeter.setActiveCount(g.Metadata.Name, 0)
+ sr.samplerMeter.incRemoveTotal(g.Metadata.Name)
+ }
case schema.KindTrace:
traceSpec := metadata.Spec.(*databasev1.Trace)
sr.SendMetadataEvent(resourceSchema.MetadataEvent{
@@ -261,6 +279,119 @@ func (sr *schemaRepo) OnDelete(metadata schema.Metadata) {
}
}
+// mergeEventEnabled reports whether cfg applies to the merge-time pipeline event.
+// v1 only implements the in-merge filter, so a config that does not enable
+// PIPELINE_EVENT_MERGE installs no samplers. An empty enabled_events list defaults
+// to MERGE for backward compatibility.
+func mergeEventEnabled(cfg *commonv1.TracePipelineConfig) bool {
+ events := cfg.GetEnabledEvents()
+ if len(events) == 0 {
+ return true
+ }
+ return slices.Contains(events, commonv1.PipelineEvent_PIPELINE_EVENT_MERGE)
+}
+
+// samplerLoadFailReason maps a plugin load error to a small, stable set of reason
+// codes for the sampler_load_failed metric label, keeping label cardinality bounded
+// (the full error is logged separately). Unrecognized errors fall back to "load_error".
+func samplerLoadFailReason(err error) string {
+ if err == nil {
+ return ""
+ }
+ msg := err.Error()
+ switch {
+ case strings.Contains(msg, "panic"):
+ return "panic"
+ case strings.Contains(msg, "escapes trusted directory"):
+ return "path_escape"
+ case strings.Contains(msg, "ABI version") || strings.Contains(msg, "ABIVersion"):
+ return "abi_mismatch"
+ case strings.Contains(msg, "constructor"):
+ return "ctor_failed"
+ case strings.Contains(msg, "missing symbol") || strings.Contains(msg, "wrong type"):
+ return "symbol_missing"
+ case strings.Contains(msg, "cannot open plugin"):
+ return "open_failed"
+ case strings.Contains(msg, "marshal plugin config"):
+ return "config_marshal"
+ case strings.Contains(msg, "plugin path is empty") || strings.Contains(msg, "trusted plugin dir not configured"):
+ return "config_invalid"
+ default:
+ return "load_error"
+ }
+}
+
+// reconcilePipeline rebuilds the sampler registry for group from cfg.
+// If cfg is nil (or the group is being deleted), it removes the group's entry.
+// On full load success it atomically replaces the slice and stores the per-group
+// merge_grace. If any plugin fails to load, the previous good set is kept intact
+// (fail-open) and an ERROR log is emitted.
+func (sr *schemaRepo) reconcilePipeline(group string, cfg *commonv1.TracePipelineConfig) {
+ // A nil/disabled config, or one that does not enable the merge event, clears the
+ // group's samplers (retain all). enabled_events is honored here so merge filtering
+ // can be disabled without removing the config.
+ if cfg == nil || !cfg.GetEnabled() || !mergeEventEnabled(cfg) {
+ hadSamplers := len(lookupSamplers(group)) > 0
+ removeSamplersForGroup(group)
+ setMergeGraceForGroup(group, 0)
+ sr.samplerMeter.setActiveCount(group, 0)
+ if hadSamplers {
+ sr.samplerMeter.incRemoveTotal(group)
+ }
+ return
+ }
+ // Distinguish register (no previous set) from update (previous set exists).
+ isUpdate := len(lookupSamplers(group)) > 0
+ var newSet []namedSampler
+ for _, p := range cfg.GetPlugins() {
+ sp := p.GetSampler()
+ if sp == nil {
+ continue
+ }
+ var sampler sdk.Sampler
+ var loadErr error
+ func() {
+ defer func() {
+ if rec := recover(); rec != nil {
+ loadErr = fmt.Errorf("panic loading plugin %q: %v", p.GetName(), rec)
+ }
+ }()
+ sampler, loadErr = loadSamplerPlugin(sp, sr.trustedPluginDir)
+ }()
+ if loadErr != nil {
+ sr.samplerMeter.incLoadFailed(group, p.GetName(), samplerLoadFailReason(loadErr))
+ if isUpdate {
+ sr.samplerMeter.incUpdateTotal(group, "rejected")
+ } else {
+ sr.samplerMeter.incRegisterTotal(group, "rejected")
+ }
+ sr.l.Error().
+ Err(loadErr).
+ Str("group", group).
+ Str("plugin", p.GetName()).
+ Msg("sampler plugin load failed; keeping previous good set (fail-open)")
+ return
+ }
+ newSet = append(newSet, namedSampler{
+ name: p.GetName(),
+ configHash: computeConfigHash(sp),
+ sampler: sampler,
+ })
+ }
+ replaceSamplersForGroup(group, newSet)
+ var graceNs int64
+ if gd := cfg.GetMergeGrace(); gd != nil {
+ graceNs = gd.AsDuration().Nanoseconds()
+ }
+ setMergeGraceForGroup(group, graceNs)
+ sr.samplerMeter.setActiveCount(group, len(newSet))
+ if isUpdate {
+ sr.samplerMeter.incUpdateTotal(group, "success")
+ } else {
+ sr.samplerMeter.incRegisterTotal(group, "success")
+ }
+}
+
func (sr *schemaRepo) loadTrace(metadata *commonv1.Metadata) (*trace, bool) {
r, ok := sr.LoadResource(metadata)
if !ok {
diff --git a/banyand/trace/metrics.go b/banyand/trace/metrics.go
index 06caa8156..908ef50e6 100644
--- a/banyand/trace/metrics.go
+++ b/banyand/trace/metrics.go
@@ -26,8 +26,9 @@ import (
)
var (
- tbScope = traceScope.SubScope("tst")
- storageScope = traceScope.SubScope("storage")
+ tbScope = traceScope.SubScope("tst")
+ storageScope = traceScope.SubScope("storage")
+ pipelineScope = traceScope.SubScope("pipeline")
)
type metrics struct {
diff --git a/banyand/trace/pipeline_loader.go b/banyand/trace/pipeline_loader.go
index a9aed65e1..1337b146c 100644
--- a/banyand/trace/pipeline_loader.go
+++ b/banyand/trace/pipeline_loader.go
@@ -18,6 +18,8 @@
package trace
import (
+ "crypto/sha256"
+ "encoding/hex"
"fmt"
"path/filepath"
"plugin"
@@ -26,10 +28,29 @@ import (
"google.golang.org/protobuf/encoding/protojson"
- pipelinev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/pipeline/v1"
+ commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
"github.com/apache/skywalking-banyandb/pkg/pipeline/sdk"
)
+// computeConfigHash returns the sha256 hex of the protojson-marshaled
+// SamplerPlugin.config (or of "{}" when config is nil). It mirrors the hash
+// used inside loadSamplerPlugin for the pluginCacheKey so callers can build
+// namedSampler entries without re-computing the hash.
+func computeConfigHash(sp *commonv1.SamplerPlugin) string {
+ var cfgJSON []byte
+ if cfg := sp.GetConfig(); cfg != nil {
+ var marshalErr error
+ cfgJSON, marshalErr = protojson.Marshal(cfg)
+ if marshalErr != nil {
+ cfgJSON = []byte("{}")
+ }
+ } else {
+ cfgJSON = []byte("{}")
+ }
+ hashBytes := sha256.Sum256(cfgJSON)
+ return hex.EncodeToString(hashBytes[:])
+}
+
// pathWithin reports whether target is base itself or a descendant of base.
// It treats only a bare ".." or a ".."+separator prefix as an escape, so a
// legitimate child whose first segment merely starts with ".." (e.g. "..foo")
@@ -42,15 +63,18 @@ func pathWithin(base, target string) bool {
return rel != ".." && !strings.HasPrefix(rel, ".."+string(filepath.Separator))
}
-// pluginCache caches loaded .so plugins by (path, symbol) to avoid re-opening.
-// Go plugins cannot be unloaded; we re-use cached ones.
+// pluginCache caches loaded .so plugins by (path, symbol, configHash) to avoid
+// re-opening the same .so with the same config. Go plugins cannot be unloaded;
+// we re-use cached instances. A different config yields a different key and
+// triggers a fresh ABI-checked construction.
var pluginCache = &loadedPluginCache{
m: make(map[pluginCacheKey]sdk.Sampler),
}
type pluginCacheKey struct {
- path string
- symbol string
+ path string
+ symbol string
+ configHash string // sha256 hex of the protojson-marshaled config (or of "{}" when nil)
}
type loadedPluginCache struct {
@@ -60,7 +84,7 @@ type loadedPluginCache struct {
// loadSamplerPlugin loads (or returns cached) a sampler plugin from the given SamplerPlugin config.
// trustedDir is the only directory .so paths may resolve within; empty trustedDir disables loading.
-func loadSamplerPlugin(sp *pipelinev1.SamplerPlugin, trustedDir string) (_ sdk.Sampler, retErr error) {
+func loadSamplerPlugin(sp *commonv1.SamplerPlugin, trustedDir string) (_ sdk.Sampler, retErr error) {
if trustedDir == "" {
return nil, fmt.Errorf("trusted plugin dir not configured")
}
@@ -89,7 +113,25 @@ func loadSamplerPlugin(sp *pipelinev1.SamplerPlugin, trustedDir string) (_ sdk.S
symbol = "NewSampler"
}
- key := pluginCacheKey{path: resolvedPath, symbol: symbol}
+ // Marshal config before the cache lookup so the hash is part of the key.
+ // protojson map-key ordering is deterministic for google.protobuf.Struct
+ // (keys are sorted lexicographically by the protojson encoder), so hashing
+ // the raw output is stable across calls with the same logical config.
+ // A nil config hashes the canonical empty JSON object "{}".
+ var cfgJSON []byte
+ if cfg := sp.GetConfig(); cfg != nil {
+ var marshalErr error
+ cfgJSON, marshalErr = protojson.Marshal(cfg)
+ if marshalErr != nil {
+ return nil, fmt.Errorf("cannot marshal plugin config: %w", marshalErr)
+ }
+ } else {
+ cfgJSON = []byte("{}")
+ }
+ hashBytes := sha256.Sum256(cfgJSON)
+ configHash := hex.EncodeToString(hashBytes[:])
+
+ key := pluginCacheKey{path: resolvedPath, symbol: symbol, configHash: configHash}
pluginCache.mu.Lock()
defer pluginCache.mu.Unlock()
if cached, ok := pluginCache.m[key]; ok {
@@ -136,17 +178,6 @@ func loadSamplerPlugin(sp *pipelinev1.SamplerPlugin, trustedDir string) (_ sdk.S
return nil, fmt.Errorf("plugin %q symbol %q has wrong type", resolvedPath, symbol)
}
- var cfgJSON []byte
- if cfg := sp.GetConfig(); cfg != nil {
- var marshalErr error
- cfgJSON, marshalErr = protojson.Marshal(cfg)
- if marshalErr != nil {
- return nil, fmt.Errorf("cannot marshal plugin config: %w", marshalErr)
- }
- } else {
- cfgJSON = []byte("{}")
- }
-
var sampler sdk.Sampler
var ctorErr error
func() {
diff --git a/banyand/trace/pipeline_loader_test.go b/banyand/trace/pipeline_loader_test.go
index 8973e70dc..d3af540a3 100644
--- a/banyand/trace/pipeline_loader_test.go
+++ b/banyand/trace/pipeline_loader_test.go
@@ -30,7 +30,7 @@ import (
"github.com/stretchr/testify/require"
"google.golang.org/protobuf/types/known/structpb"
- pipelinev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/pipeline/v1"
+ commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
"github.com/apache/skywalking-banyandb/pkg/pipeline/sdk"
)
@@ -117,9 +117,17 @@ func resetPluginCache() {
// Using a shared path lets plugin.Open return the already-loaded *plugin.Plugin
// on guard sub-tests that need to reach Lookup (e.g. wrong constructor symbol).
func TestLoadSamplerPlugin(t *testing.T) {
- tmpDir, mkdirErr := os.MkdirTemp("", "lssplugin-test-*")
- require.NoError(t, mkdirErr)
- defer os.RemoveAll(tmpDir)
+ // Use the package-level sharedPluginDir so the .so is built at the same
+ // path used by pipeline_watch_test.go. The Go plugin runtime deduplicates
+ // opens by content hash; two different paths with identical content produce
+ // "plugin already loaded". Sharing one path avoids that across test funcs.
+ tmpDir := sharedPluginDir
+ if tmpDir == "" {
+ var mkdirErr error
+ tmpDir, mkdirErr = os.MkdirTemp("", "lssplugin-test-*")
+ require.NoError(t, mkdirErr)
+ defer os.RemoveAll(tmpDir)
+ }
soPath := buildTestPlugin(t, tmpDir)
soName := filepath.Base(soPath) // relative name passed to loadSamplerPlugin
@@ -142,7 +150,7 @@ func TestLoadSamplerPlugin(t *testing.T) {
})
require.NoError(t, cfgErr)
- sp := &pipelinev1.SamplerPlugin{
+ sp := &commonv1.SamplerPlugin{
Path: soName,
Symbol: "NewSampler",
AbiVersion: uint32(sdk.ABIVersion),
@@ -160,6 +168,80 @@ func TestLoadSamplerPlugin(t *testing.T) {
assert.Contains(t, proj.Tags, "status", "plugin must project the status tag")
})
+ // Cache sub-tests: verify the (path, symbol, configHash) key semantics.
+ // These run after the happy-path probe so the .so is already registered in
+ // the Go plugin runtime under this path — subsequent plugin.Open calls on
+ // the same path return the cached *plugin.Plugin without "previous failure".
+ cfgLow, cfgLowErr := structpb.NewStruct(map[string]interface{}{
+ "thresholdMs": float64(100),
+ "successValue": "ok",
+ })
+ require.NoError(t, cfgLowErr)
+
+ cfgHigh, cfgHighErr := structpb.NewStruct(map[string]interface{}{
+ "thresholdMs": float64(500),
+ "successValue": "ok",
+ })
+ require.NoError(t, cfgHighErr)
+
+ spLow := &commonv1.SamplerPlugin{
+ Path: soName,
+ Symbol: "NewSampler",
+ AbiVersion: uint32(sdk.ABIVersion),
+ Config: cfgLow,
+ }
+ spHigh := &commonv1.SamplerPlugin{
+ Path: soName,
+ Symbol: "NewSampler",
+ AbiVersion: uint32(sdk.ABIVersion),
+ Config: cfgHigh,
+ }
+
+ t.Run("cache same config deduplicates", func(t *testing.T) {
+ resetPluginCache()
+
+ first, firstErr := loadSamplerPlugin(spLow, tmpDir)
+ require.NoError(t, firstErr)
+ require.NotNil(t, first)
+
+ second, secondErr := loadSamplerPlugin(spLow, tmpDir)
+ require.NoError(t, secondErr)
+ require.NotNil(t, second)
+
+ // Same (path, symbol, configHash) => cache hit => identical pointer.
+ assert.Same(t, first, second, "same config must return the cached instance")
+ })
+
+ t.Run("cache changed config produces fresh instance", func(t *testing.T) {
+ resetPluginCache()
+
+ instanceLow, errLow := loadSamplerPlugin(spLow, tmpDir)
+ require.NoError(t, errLow)
+ require.NotNil(t, instanceLow)
+
+ instanceHigh, errHigh := loadSamplerPlugin(spHigh, tmpDir)
+ require.NoError(t, errHigh)
+ require.NotNil(t, instanceHigh)
+
+ // Different config => different key => different (freshly constructed) instance.
+ assert.NotSame(t, instanceLow, instanceHigh, "different config must produce a fresh instance")
+ })
+
+ t.Run("cache ABI mismatch on miss is rejected", func(t *testing.T) {
+ resetPluginCache()
+
+ spBadABI := &commonv1.SamplerPlugin{
+ Path: soName,
+ Symbol: "NewSampler",
+ AbiVersion: 999, // proto-level ABI mismatch
+ Config: cfgLow,
+ }
+ sampler, loadErr := loadSamplerPlugin(spBadABI, tmpDir)
+ require.Error(t, loadErr, "ABI-mismatched plugin must be rejected on a cache miss")
+ assert.Nil(t, sampler)
+ assert.Contains(t, loadErr.Error(), "ABI version")
+ })
+
// Guard sub-tests: each expects an error from loadSamplerPlugin.
// The missing-path and trusted-dir-escape guards never reach plugin.Open.
// The ABI-mismatch guard fires before plugin.Open.
@@ -167,13 +249,13 @@ func TestLoadSamplerPlugin(t *testing.T) {
// open from the happy-path sub-test above) so that Lookup can be reached.
guardTests := []struct {
name string
- sp *pipelinev1.SamplerPlugin
+ sp *commonv1.SamplerPlugin
trustedDir string
errContains string
}{
{
name: "missing path",
- sp: &pipelinev1.SamplerPlugin{
+ sp: &commonv1.SamplerPlugin{
Path: "",
Symbol: "NewSampler",
AbiVersion: uint32(sdk.ABIVersion),
@@ -183,7 +265,7 @@ func TestLoadSamplerPlugin(t *testing.T) {
},
{
name: "ABI mismatch",
- sp: &pipelinev1.SamplerPlugin{
+ sp: &commonv1.SamplerPlugin{
Path: soName,
Symbol: "NewSampler",
AbiVersion: 999,
@@ -193,7 +275,7 @@ func TestLoadSamplerPlugin(t *testing.T) {
},
{
name: "trusted-dir escape",
- sp: &pipelinev1.SamplerPlugin{
+ sp: &commonv1.SamplerPlugin{
Path: "../escape.so",
Symbol: "NewSampler",
AbiVersion: uint32(sdk.ABIVersion),
@@ -206,7 +288,7 @@ func TestLoadSamplerPlugin(t *testing.T) {
// plugin.Open returns the cached *plugin.Plugin, Lookup then fails
// because "DoesNotExist" is not exported by the plugin.
name: "wrong constructor symbol",
- sp: &pipelinev1.SamplerPlugin{
+ sp: &commonv1.SamplerPlugin{
Path: soName,
Symbol: "DoesNotExist",
AbiVersion: uint32(sdk.ABIVersion),
@@ -241,7 +323,7 @@ func TestLoadSamplerPlugin_DotDotPrefixChildNotEscape(t *testing.T) {
defer os.RemoveAll(tmpDir)
resetPluginCache()
- sp := &pipelinev1.SamplerPlugin{
+ sp := &commonv1.SamplerPlugin{
Path: "..foo/nonexistent.so",
Symbol: "NewSampler",
AbiVersion: uint32(sdk.ABIVersion),
diff --git a/banyand/trace/pipeline_metrics.go b/banyand/trace/pipeline_metrics.go
new file mode 100644
index 000000000..fe877182c
--- /dev/null
+++ b/banyand/trace/pipeline_metrics.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.
+
+package trace
+
+import (
+ "github.com/apache/skywalking-banyandb/banyand/observability"
+ "github.com/apache/skywalking-banyandb/pkg/meter"
+)
+
+// samplerMetrics holds the US-011 observability counters/gauges for the
+// dynamic sampler registry. All fields are safe to call on a nil receiver or
+// when backed by a bypassFactory — Inc/Set on a bypass counter/gauge is a
+// no-op, so callers need no nil checks.
+type samplerMetrics struct {
+ // samplerActiveCount{group} tracks the live sampler count per group.
+ samplerActiveCount meter.Gauge
+ // samplerRegisterTotal{group,result} counts apply-config calls (result = success|rejected).
+ samplerRegisterTotal meter.Counter
+ // samplerUpdateTotal{group,result} counts re-apply calls on an existing group
+ // (result = success|rejected). Emitted on the same reconcile path as register;
+ // callers distinguish them by whether a previous set existed.
+ samplerUpdateTotal meter.Counter
+ // samplerRemoveTotal{group} counts explicit removal calls.
+ samplerRemoveTotal meter.Counter
+ // samplerLoadFailed{group,name,reason} counts individual plugin load failures
+ // on the fail-open path.
+ samplerLoadFailed meter.Counter
+}
+
+// newSamplerMetrics creates a samplerMetrics wired to factory. If factory is
+// nil, observability.BypassRegistry is used so all operations are no-ops.
+func newSamplerMetrics(factory observability.Factory) *samplerMetrics {
+ if factory == nil {
+ factory = observability.BypassRegistry.With(pipelineScope)
+ }
+ return &samplerMetrics{
+ samplerActiveCount: factory.NewGauge("sampler_active_count", "group"),
+ samplerRegisterTotal: factory.NewCounter("sampler_register_total", "group", "result"),
+ samplerUpdateTotal: factory.NewCounter("sampler_update_total", "group", "result"),
+ samplerRemoveTotal: factory.NewCounter("sampler_remove_total", "group"),
+ samplerLoadFailed: factory.NewCounter("sampler_load_failed", "group", "name", "reason"),
+ }
+}
+
+func (m *samplerMetrics) setActiveCount(group string, n int) {
+ if m == nil {
+ return
+ }
+ m.samplerActiveCount.Set(float64(n), group)
+}
+
+func (m *samplerMetrics) incRegisterTotal(group, result string) {
+ if m == nil {
+ return
+ }
+ m.samplerRegisterTotal.Inc(1, group, result)
+}
+
+func (m *samplerMetrics) incUpdateTotal(group, result string) {
+ if m == nil {
+ return
+ }
+ m.samplerUpdateTotal.Inc(1, group, result)
+}
+
+func (m *samplerMetrics) incRemoveTotal(group string) {
+ if m == nil {
+ return
+ }
+ m.samplerRemoveTotal.Inc(1, group)
+}
+
+func (m *samplerMetrics) incLoadFailed(group, name, reason string) {
+ if m == nil {
+ return
+ }
+ m.samplerLoadFailed.Inc(1, group, name, reason)
+}
diff --git a/banyand/trace/pipeline_metrics_test.go b/banyand/trace/pipeline_metrics_test.go
new file mode 100644
index 000000000..0391fde39
--- /dev/null
+++ b/banyand/trace/pipeline_metrics_test.go
@@ -0,0 +1,387 @@
+// 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 trace
+
+import (
+ "sync"
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+
+ commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
+ databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+ "github.com/apache/skywalking-banyandb/banyand/observability"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+ "github.com/apache/skywalking-banyandb/pkg/meter"
+ "github.com/apache/skywalking-banyandb/pkg/pipeline/sdk"
+)
+
+// Fake meter primitives.
+
+// labeledCall records one Inc or Set call with its label values.
+type labeledCall struct {
+ labels []string
+ delta float64
+}
+
+type fakeMetricCounter struct {
+ calls []labeledCall
+ mu sync.Mutex
+}
+
+func (f *fakeMetricCounter) Inc(delta float64, labels ...string) {
+ cp := make([]string, len(labels))
+ copy(cp, labels)
+ f.mu.Lock()
+ f.calls = append(f.calls, labeledCall{delta: delta, labels: cp})
+ f.mu.Unlock()
+}
+
+func (f *fakeMetricCounter) Delete(_ ...string) bool { return true }
+
+func (f *fakeMetricCounter) callsWithLabels(labels ...string) int {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ count := 0
+ for _, c := range f.calls {
+ if len(c.labels) != len(labels) {
+ continue
+ }
+ match := true
+ for idx, lv := range labels {
+ if c.labels[idx] != lv {
+ match = false
+ break
+ }
+ }
+ if match {
+ count++
+ }
+ }
+ return count
+}
+
+// callsWithPrefix counts calls whose label slice starts with the given prefix.
+func (f *fakeMetricCounter) callsWithPrefix(prefix ...string) int {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ count := 0
+ for _, c := range f.calls {
+ if len(c.labels) < len(prefix) {
+ continue
+ }
+ match := true
+ for idx, lv := range prefix {
+ if c.labels[idx] != lv {
+ match = false
+ break
+ }
+ }
+ if match {
+ count++
+ }
+ }
+ return count
+}
+
+type fakeMetricGauge struct {
+ calls []labeledCall
+ mu sync.Mutex
+}
+
+func (f *fakeMetricGauge) Set(value float64, labels ...string) {
+ cp := make([]string, len(labels))
+ copy(cp, labels)
+ f.mu.Lock()
+ f.calls = append(f.calls, labeledCall{delta: value, labels: cp})
+ f.mu.Unlock()
+}
+
+func (f *fakeMetricGauge) Add(delta float64, labels ...string) {
+ f.mu.Lock()
+ f.calls = append(f.calls, labeledCall{delta: delta, labels: labels})
+ f.mu.Unlock()
+}
+
+func (f *fakeMetricGauge) Delete(_ ...string) bool { return true }
+
+func (f *fakeMetricGauge) lastValue(group string) (float64, bool) {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ for idx := len(f.calls) - 1; idx >= 0; idx-- {
+ c := f.calls[idx]
+ if len(c.labels) > 0 && c.labels[0] == group {
+ return c.delta, true
+ }
+ }
+ return 0, false
+}
+
+// fakeMetricFactory creates counters/gauges that record calls for assertion.
+type fakeMetricFactory struct {
+ counters map[string]*fakeMetricCounter
+ gauges map[string]*fakeMetricGauge
+ mu sync.Mutex
+}
+
+func newFakeMetricFactory() *fakeMetricFactory {
+ return &fakeMetricFactory{
+ counters: make(map[string]*fakeMetricCounter),
+ gauges: make(map[string]*fakeMetricGauge),
+ }
+}
+
+func (f *fakeMetricFactory) NewCounter(name string, _ ...string) meter.Counter {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ c := &fakeMetricCounter{}
+ f.counters[name] = c
+ return c
+}
+
+func (f *fakeMetricFactory) NewGauge(name string, _ ...string) meter.Gauge {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ g := &fakeMetricGauge{}
+ f.gauges[name] = g
+ return g
+}
+
+func (f *fakeMetricFactory) NewHistogram(_ string, _ meter.Buckets, _ ...string) meter.Histogram {
+ return observability.BypassRegistry.With(pipelineScope).NewHistogram("", meter.Buckets{})
+}
+
+func (f *fakeMetricFactory) Close() {}
+
+// counter/gauge accessors.
+func (f *fakeMetricFactory) counter(name string) *fakeMetricCounter {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ return f.counters[name]
+}
+
+func (f *fakeMetricFactory) gauge(name string) *fakeMetricGauge {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ return f.gauges[name]
+}
+
+// Helpers.
+
+// makeMeteredSchemaRepo builds a ROLE_DATA schemaRepo with the supplied factory
+// wired into samplerMeter. The Repository field is nil — reconcilePipeline does
+// not touch it.
+func makeMeteredSchemaRepo(factory observability.Factory) schemaRepo {
+ return schemaRepo{
+ l: logger.GetLogger("pipeline-metrics-test"),
+ role: databasev1.Role_ROLE_DATA,
+ nativePipelineEnabled: true,
+ trustedPluginDir: "/nonexistent",
+ samplerMeter: newSamplerMetrics(factory),
+ }
+}
+
+// nilMeteredSchemaRepo builds a schemaRepo without a samplerMeter to verify
+// nil-safe behavior.
+func nilMeteredSchemaRepo() schemaRepo {
+ return schemaRepo{
+ l: logger.GetLogger("pipeline-metrics-test"),
+ role: databasev1.Role_ROLE_DATA,
+ nativePipelineEnabled: true,
+ trustedPluginDir: "/nonexistent",
+ }
+}
+
+// Tests.
+
+// TestSamplerMetrics_RegisterUpdateRemove asserts that the full register →
+// update → remove lifecycle emits the correct labels and counts.
+func TestSamplerMetrics_RegisterUpdateRemove(t *testing.T) {
+ resetRegistries()
+ defer resetRegistries()
+
+ factory := newFakeMetricFactory()
+ sr := makeMeteredSchemaRepo(factory)
+
+ const group = "metrics-group"
+ dummy := &dummySampler{}
+
+ // --- register (no previous set) ---
+ replaceSamplersForGroup(group, []namedSampler{{name: "prev", sampler: dummy}})
+ // Simulate a successful register: clear first so isUpdate == false.
+ removeSamplersForGroup(group)
+
+ cfg := &commonv1.TracePipelineConfig{
+ Enabled: true,
+ Plugins: []*commonv1.Plugin{
+ {
+ Name: "lss",
+ Kind: &commonv1.Plugin_Sampler{Sampler: &commonv1.SamplerPlugin{
+ Path: "missing.so",
+ Symbol: "NewSampler",
+ AbiVersion: uint32(sdk.ABIVersion),
+ }},
+ },
+ },
+ }
+
+ // Load will fail (nonexistent .so) → rejected on register path.
+ sr.reconcilePipeline(group, cfg)
+
+ regTotal := factory.counter("sampler_register_total")
+ require.NotNil(t, regTotal)
+ assert.Equal(t, 1, regTotal.callsWithLabels(group, "rejected"),
+ "rejected register must increment sampler_register_total{group,result=rejected}")
+ loadFailed := factory.counter("sampler_load_failed")
+ require.NotNil(t, loadFailed)
+ assert.Equal(t, 1, loadFailed.callsWithPrefix(group, "lss"),
+ "sampler_load_failed must be emitted with labels {group, name=lss, reason=...}")
+
+ // --- update path: pre-populate so isUpdate == true, then load fails again ---
+ replaceSamplersForGroup(group, []namedSampler{{name: "d", sampler: dummy}})
+ sr.reconcilePipeline(group, cfg)
+
+ updTotal := factory.counter("sampler_update_total")
+ require.NotNil(t, updTotal)
+ assert.Equal(t, 1, updTotal.callsWithLabels(group, "rejected"),
+ "rejected update must increment sampler_update_total{group,result=rejected}")
+
+ // --- remove via nil config ---
+ replaceSamplersForGroup(group, []namedSampler{{name: "d", sampler: dummy}})
+ sr.reconcilePipeline(group, nil)
+
+ rmTotal := factory.counter("sampler_remove_total")
+ require.NotNil(t, rmTotal)
+ assert.Equal(t, 1, rmTotal.callsWithLabels(group),
+ "nil config must increment sampler_remove_total{group}")
+
+ activeGauge := factory.gauge("sampler_active_count")
+ require.NotNil(t, activeGauge)
+ lastVal, ok := activeGauge.lastValue(group)
+ require.True(t, ok, "sampler_active_count must have been set at least once")
+ assert.Equal(t, float64(0), lastVal, "sampler_active_count must be 0 after remove")
+}
+
+// TestSamplerMetrics_SuccessRegisterAndUpdate exercises the success path by
+// injecting a pre-built sampler via replaceSamplersForGroup to bypass plugin
+// loading, then verifying success labels are emitted.
+func TestSamplerMetrics_SuccessRegisterAndUpdate(t *testing.T) {
+ resetRegistries()
+ defer resetRegistries()
+
+ factory := newFakeMetricFactory()
+ // Build a schemaRepo whose trustedPluginDir doesn't matter because we will
+ // call replaceSamplersForGroup + setMergeGraceForGroup directly and then
+ // exercise the metric emission path via a cfg with zero plugins (so the
+ // loop body never runs, no load is attempted, and replaceSamplersForGroup
+ // is called with an empty slice).
+ sr := makeMeteredSchemaRepo(factory)
+
+ const group = "success-group"
+
+ // Empty-plugins config → register success with active_count == 0.
+ emptyCfg := &commonv1.TracePipelineConfig{Enabled: true}
+ sr.reconcilePipeline(group, emptyCfg)
+
+ regTotal := factory.counter("sampler_register_total")
+ require.NotNil(t, regTotal)
+ assert.Equal(t, 1, regTotal.callsWithLabels(group, "success"),
+ "empty-plugins register must emit sampler_register_total{group,result=success}")
+
+ activeGauge := factory.gauge("sampler_active_count")
+ require.NotNil(t, activeGauge)
+ lastVal, ok := activeGauge.lastValue(group)
+ require.True(t, ok)
+ assert.Equal(t, float64(0), lastVal, "active_count must be 0 for empty plugins set")
+
+ // Second call with empty plugins → update success.
+ sr.reconcilePipeline(group, emptyCfg)
+
+ updTotal := factory.counter("sampler_update_total")
+ require.NotNil(t, updTotal)
+ // After the first call the set is empty (len==0) → isUpdate == false on
+ // second call too. This is the expected behavior: a zero-sampler set does
+ // not count as "has previous set". Document this explicitly.
+ // Both calls land on register_total because lookupSamplers returns nil for
+ // an empty replacement.
+ assert.GreaterOrEqual(t, regTotal.callsWithLabels(group, "success"), 1,
+ "subsequent empty-plugin configs still count as register (no active sampler present)")
+}
+
+// TestSamplerMetrics_OnDeleteKindGroup verifies that the OnDelete KindGroup
+// path (simulated directly) emits sampler_remove_total and sets active_count=0.
+func TestSamplerMetrics_OnDeleteKindGroup(t *testing.T) {
+ resetRegistries()
+ defer resetRegistries()
+
+ factory := newFakeMetricFactory()
+ sr := makeMeteredSchemaRepo(factory)
+
+ const group = "delete-metrics-group"
+ dummy := &dummySampler{}
+ replaceSamplersForGroup(group, []namedSampler{{name: "d", sampler: dummy}})
+ require.NotEmpty(t, lookupSamplers(group))
+
+ // Simulate OnDelete KindGroup gate.
+ g := &commonv1.Group{
+ Metadata: &commonv1.Metadata{Name: group},
+ Catalog: commonv1.Catalog_CATALOG_TRACE,
+ }
+ if sr.role == databasev1.Role_ROLE_DATA && sr.nativePipelineEnabled {
+ removeSamplersForGroup(g.Metadata.Name)
+ sr.samplerMeter.setActiveCount(g.Metadata.Name, 0)
+ sr.samplerMeter.incRemoveTotal(g.Metadata.Name)
+ }
+
+ rmTotal := factory.counter("sampler_remove_total")
+ require.NotNil(t, rmTotal)
+ assert.Equal(t, 1, rmTotal.callsWithLabels(group),
+ "OnDelete KindGroup must increment sampler_remove_total{group}")
+
+ activeGauge := factory.gauge("sampler_active_count")
+ require.NotNil(t, activeGauge)
+ lastVal, ok := activeGauge.lastValue(group)
+ require.True(t, ok)
+ assert.Equal(t, float64(0), lastVal, "sampler_active_count must be 0 after OnDelete")
+}
+
+// TestSamplerMetrics_NilSafe verifies that a nil samplerMeter on schemaRepo
+// does not panic during reconcilePipeline or the OnDelete simulation.
+func TestSamplerMetrics_NilSafe(t *testing.T) {
+ resetRegistries()
+ defer resetRegistries()
+
+ sr := nilMeteredSchemaRepo()
+ require.Nil(t, sr.samplerMeter)
+
+ const group = "nil-safe-group"
+ dummy := &dummySampler{}
+ replaceSamplersForGroup(group, []namedSampler{{name: "d", sampler: dummy}})
+
+ require.NotPanics(t, func() {
+ sr.reconcilePipeline(group, nil)
+ }, "nil samplerMeter must not panic on reconcilePipeline(nil)")
+
+ replaceSamplersForGroup(group, []namedSampler{{name: "d", sampler: dummy}})
+ require.NotPanics(t, func() {
+ removeSamplersForGroup(group)
+ sr.samplerMeter.setActiveCount(group, 0)
+ sr.samplerMeter.incRemoveTotal(group)
+ }, "nil samplerMeter must not panic on OnDelete path")
+}
diff --git a/banyand/trace/pipeline_registry.go b/banyand/trace/pipeline_registry.go
index a15df07e8..a0f71e75e 100644
--- a/banyand/trace/pipeline_registry.go
+++ b/banyand/trace/pipeline_registry.go
@@ -23,31 +23,84 @@ import (
"github.com/apache/skywalking-banyandb/pkg/pipeline/sdk"
)
-// localSamplerRegistry maps group → ordered samplers for in-process, flag-gated
+// localMergeGraceRegistry stores per-group merge_grace in nanoseconds as set by
+// reconcilePipeline. A zero value means "use option.mergeGraceDefault".
+var localMergeGraceRegistry = struct {
+ m map[string]int64
+ mu sync.RWMutex
+}{m: make(map[string]int64)}
+
+// setMergeGraceForGroup stores graceNs for group. graceNs == 0 removes the entry.
+func setMergeGraceForGroup(group string, graceNs int64) {
+ localMergeGraceRegistry.mu.Lock()
+ if graceNs > 0 {
+ localMergeGraceRegistry.m[group] = graceNs
+ } else {
+ delete(localMergeGraceRegistry.m, group)
+ }
+ localMergeGraceRegistry.mu.Unlock()
+}
+
+// lookupMergeGrace returns the per-group merge_grace in nanoseconds, or 0 if
+// no per-group grace is configured (caller falls back to mergeGraceDefault).
+func lookupMergeGrace(group string) int64 {
+ localMergeGraceRegistry.mu.RLock()
+ defer localMergeGraceRegistry.mu.RUnlock()
+ return localMergeGraceRegistry.m[group]
+}
+
+// namedSampler pairs a sampler instance with its stable identity and config hash.
+// name is the plugin name within the group's plugins list; configHash is the hash
+// of the marshaled SamplerPlugin.config bytes, used by the config-aware loader cache.
+type namedSampler struct {
+ sampler sdk.Sampler
+ name string
+ configHash string
+}
+
+// localSamplerRegistry maps group → ordered named samplers for in-process, flag-gated
// activation of the in-merge filter hook. It is consulted ONLY when
-// option.nativePipelineEnabled is true. The config-driven startup loader
-// (svc_standalone PreRun, via --trace-pipeline-config) also populates this
-// registry at boot; unit tests may register samplers directly for fast
-// in-process coverage.
+// option.nativePipelineEnabled is true. The dynamic watch path (KindGroup
+// OnAddOrUpdate) populates this registry at runtime; unit tests may register
+// samplers directly for fast in-process coverage.
var localSamplerRegistry = struct {
- m map[string][]sdk.Sampler
+ m map[string][]namedSampler
mu sync.RWMutex
-}{m: make(map[string][]sdk.Sampler)}
+}{m: make(map[string][]namedSampler)}
+
+// replaceSamplersForGroup atomically swaps the entire sampler set for group.
+// The caller builds the new set off-lock; this function copies the slice and
+// stores it under a single Lock. Close is never called on retired instances (R4).
+func replaceSamplersForGroup(group string, set []namedSampler) {
+ cp := make([]namedSampler, len(set))
+ copy(cp, set)
+ localSamplerRegistry.mu.Lock()
+ localSamplerRegistry.m[group] = cp
+ localSamplerRegistry.mu.Unlock()
+}
+
+// removeSamplersForGroup atomically removes the entire sampler set for group.
+// Close is never called on retired instances (R4).
+func removeSamplersForGroup(group string) {
+ localSamplerRegistry.mu.Lock()
+ delete(localSamplerRegistry.m, group)
+ localSamplerRegistry.mu.Unlock()
+}
// registerSampler registers s under group and returns a deregister closure.
// Callers must invoke the returned function (e.g., via t.Cleanup) to remove
// the sampler when the test ends.
func registerSampler(group string, s sdk.Sampler) func() {
localSamplerRegistry.mu.Lock()
- localSamplerRegistry.m[group] = append(localSamplerRegistry.m[group], s)
+ localSamplerRegistry.m[group] = append(localSamplerRegistry.m[group], namedSampler{sampler: s})
localSamplerRegistry.mu.Unlock()
return func() {
localSamplerRegistry.mu.Lock()
defer localSamplerRegistry.mu.Unlock()
samplers := localSamplerRegistry.m[group]
- for i, existing := range samplers {
- if existing == s {
- localSamplerRegistry.m[group] = append(samplers[:i], samplers[i+1:]...)
+ for idx, existing := range samplers {
+ if existing.sampler == s {
+ localSamplerRegistry.m[group] = append(samplers[:idx], samplers[idx+1:]...)
break
}
}
@@ -57,15 +110,19 @@ func registerSampler(group string, s sdk.Sampler) func() {
}
}
-// lookupSamplers returns a snapshot of the registered samplers for group, or nil if none.
+// lookupSamplers returns a snapshot of the registered samplers for group in declared
+// order, or nil if none. The snapshot is safe for the caller to hold across a merge
+// without being torn by concurrent registry mutations.
func lookupSamplers(group string) []sdk.Sampler {
localSamplerRegistry.mu.RLock()
defer localSamplerRegistry.mu.RUnlock()
- s := localSamplerRegistry.m[group]
- if len(s) == 0 {
+ ns := localSamplerRegistry.m[group]
+ if len(ns) == 0 {
return nil
}
- out := make([]sdk.Sampler, len(s))
- copy(out, s)
+ out := make([]sdk.Sampler, len(ns))
+ for idx, item := range ns {
+ out[idx] = item.sampler
+ }
return out
}
diff --git a/banyand/trace/pipeline_registry_test.go b/banyand/trace/pipeline_registry_test.go
new file mode 100644
index 000000000..ec9fb273e
--- /dev/null
+++ b/banyand/trace/pipeline_registry_test.go
@@ -0,0 +1,177 @@
+// 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 trace
+
+import (
+ "sync"
+ "testing"
+
+ "github.com/stretchr/testify/require"
+
+ "github.com/apache/skywalking-banyandb/pkg/pipeline/sdk"
+)
+
+// stubSampler is a minimal sdk.Sampler for registry unit tests.
+type stubSampler struct{ id string }
+
+func (s *stubSampler) Kind() sdk.Kind { return sdk.KindSampler }
+func (s *stubSampler) Project() sdk.Projection { return sdk.Projection{} }
+func (s *stubSampler) Decide(_ *sdk.TraceBatch) (sdk.Verdict, error) { return sdk.Verdict{}, nil }
+func (s *stubSampler) Close() error { return nil }
+
+// cleanRegistry resets the shared registry and restores it after the test,
+// so individual tests do not bleed state into each other.
+func cleanRegistry(t *testing.T) {
+ t.Helper()
+ localSamplerRegistry.mu.Lock()
+ saved := localSamplerRegistry.m
+ localSamplerRegistry.m = make(map[string][]namedSampler)
+ localSamplerRegistry.mu.Unlock()
+ t.Cleanup(func() {
+ localSamplerRegistry.mu.Lock()
+ localSamplerRegistry.m = saved
+ localSamplerRegistry.mu.Unlock()
+ })
+}
+
+func TestRegistryReplaceThenLookup(t *testing.T) {
+ cleanRegistry(t)
+ const group = "test-group"
+
+ s1 := &stubSampler{id: "a"}
+ s2 := &stubSampler{id: "b"}
+
+ set := []namedSampler{
+ {name: "plugin-a", sampler: s1},
+ {name: "plugin-b", sampler: s2},
+ }
+ replaceSamplersForGroup(group, set)
+
+ got := lookupSamplers(group)
+ require.Equal(t, []sdk.Sampler{s1, s2}, got, "lookupSamplers must return the new set in declared order")
+
+ // Mutating the returned snapshot must not affect the registry.
+ got[0] = &stubSampler{id: "z"}
+ require.Equal(t, []sdk.Sampler{s1, s2}, lookupSamplers(group), "snapshot is independent of registry")
+}
+
+func TestRegistryReplaceOverwritesPreviousSet(t *testing.T) {
+ cleanRegistry(t)
+ const group = "test-group"
+
+ s1 := &stubSampler{id: "old"}
+ replaceSamplersForGroup(group, []namedSampler{{name: "old", sampler: s1}})
+
+ s2 := &stubSampler{id: "new"}
+ replaceSamplersForGroup(group, []namedSampler{{name: "new", sampler: s2}})
+
+ got := lookupSamplers(group)
+ require.Equal(t, []sdk.Sampler{s2}, got, "replace must overwrite the previous set entirely")
+}
+
+func TestRegistryRemoveThenLookupReturnsNil(t *testing.T) {
+ cleanRegistry(t)
+ const group = "test-group"
+
+ s1 := &stubSampler{id: "a"}
+ replaceSamplersForGroup(group, []namedSampler{{name: "plugin-a", sampler: s1}})
+ require.NotNil(t, lookupSamplers(group))
+
+ removeSamplersForGroup(group)
+ require.Nil(t, lookupSamplers(group), "lookupSamplers must return nil after remove")
+}
+
+func TestRegistryRemoveNonExistentGroupIsNoop(t *testing.T) {
+ cleanRegistry(t)
+ // Must not panic.
+ removeSamplersForGroup("no-such-group")
+ require.Nil(t, lookupSamplers("no-such-group"))
+}
+
+func TestRegistryRegisterAndDeregister(t *testing.T) {
+ cleanRegistry(t)
+ const group = "test-group"
+
+ s1 := &stubSampler{id: "a"}
+ s2 := &stubSampler{id: "b"}
+
+ dereg1 := registerSampler(group, s1)
+ registerSampler(group, s2)
+
+ got := lookupSamplers(group)
+ require.Equal(t, []sdk.Sampler{s1, s2}, got)
+
+ // Deregister s1; s2 must remain.
+ dereg1()
+ got = lookupSamplers(group)
+ require.Equal(t, []sdk.Sampler{s2}, got, "deregister closure must remove only the target sampler")
+}
+
+// TestRegistryConcurrency is a -race-safe concurrency test: many goroutines
+// call replace/remove/lookupSamplers concurrently. The test asserts no data
+// race (enforced by -race) and that the final state is deterministic.
+func TestRegistryConcurrency(t *testing.T) {
+ cleanRegistry(t)
+ const group = "race-group"
+ const workers = 20
+ const iters = 50
+
+ s := &stubSampler{id: "x"}
+ set := []namedSampler{{name: "plugin-x", sampler: s}}
+
+ var wg sync.WaitGroup
+ wg.Add(workers * 3)
+
+ // Replace workers.
+ for range workers {
+ go func() {
+ defer wg.Done()
+ for range iters {
+ replaceSamplersForGroup(group, set)
+ }
+ }()
+ }
+
+ // Remove workers.
+ for range workers {
+ go func() {
+ defer wg.Done()
+ for range iters {
+ removeSamplersForGroup(group)
+ }
+ }()
+ }
+
+ // Reader workers.
+ for range workers {
+ go func() {
+ defer wg.Done()
+ for range iters {
+ _ = lookupSamplers(group)
+ }
+ }()
+ }
+
+ wg.Wait()
+
+ // After all removes and replaces, the registry contains either the last
+ // replace or nothing — both are valid; what matters is no race and that
+ // lookupSamplers never returns a torn slice.
+ result := lookupSamplers(group)
+ require.True(t, result == nil || len(result) == 1, "final state must be nil or a single-element slice, got %v", result)
+}
diff --git a/banyand/trace/pipeline_watch_test.go b/banyand/trace/pipeline_watch_test.go
new file mode 100644
index 000000000..09873ab25
--- /dev/null
+++ b/banyand/trace/pipeline_watch_test.go
@@ -0,0 +1,491 @@
+// 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 trace
+
+import (
+ "os"
+ "plugin"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+ "google.golang.org/protobuf/types/known/durationpb"
+ "google.golang.org/protobuf/types/known/structpb"
+
+ commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
+ databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+ "github.com/apache/skywalking-banyandb/pkg/pipeline/sdk"
+)
+
+// resetRegistries clears both the sampler and merge-grace registries so each
+// test starts from a clean state.
+func resetRegistries() {
+ localSamplerRegistry.mu.Lock()
+ localSamplerRegistry.m = make(map[string][]namedSampler)
+ localSamplerRegistry.mu.Unlock()
+
+ localMergeGraceRegistry.mu.Lock()
+ localMergeGraceRegistry.m = make(map[string]int64)
+ localMergeGraceRegistry.mu.Unlock()
+
+ resetPluginCache()
+}
+
+// makeDataSchemaRepo constructs a minimal ROLE_DATA schemaRepo with
+// nativePipelineEnabled=true pointing at trustedDir.
+// The Repository field is intentionally nil — reconcilePipeline only
+// accesses the registry functions and loadSamplerPlugin, not the Repository.
+func makeDataSchemaRepo(trustedDir string) schemaRepo {
+ return schemaRepo{
+ l: logger.GetLogger("pipeline-watch-test"),
+ role: databasev1.Role_ROLE_DATA,
+ nativePipelineEnabled: true,
+ trustedPluginDir: trustedDir,
+ }
+}
+
+// makeSamplerPlugin builds a SamplerPlugin for soName inside trustedDir.
+func makeSamplerPlugin(soName string, thresholdMs float64) (*commonv1.SamplerPlugin, error) {
+ cfgStruct, cfgErr := structpb.NewStruct(map[string]interface{}{
+ "thresholdMs": thresholdMs,
+ "successValue": "ok",
+ })
+ if cfgErr != nil {
+ return nil, cfgErr
+ }
+ return &commonv1.SamplerPlugin{
+ Path: soName,
+ Symbol: "NewSampler",
+ AbiVersion: uint32(sdk.ABIVersion),
+ Config: cfgStruct,
+ }, nil
+}
+
+// dummySampler is a minimal sdk.Sampler implementation for tests that do not
+// need a real .so plugin.
+type dummySampler struct{}
+
+func (d *dummySampler) Kind() sdk.Kind { return sdk.KindSampler }
+func (d *dummySampler) Project() sdk.Projection {
+ return sdk.Projection{}
+}
+
+func (d *dummySampler) Decide(batch *sdk.TraceBatch) (sdk.Verdict, error) {
+ keep := make([]bool, len(batch.Traces))
+ for idx := range keep {
+ keep[idx] = true
+ }
+ return sdk.Verdict{Keep: keep}, nil
+}
+func (d *dummySampler) Close() error { return nil }
+
+// pluginTestState holds the shared tmpDir and .so path used by all tests that
+// need a real plugin. They are initialized once (per test binary execution)
+// via setupPluginOnce so the Go plugin runtime never sees two opens of the
+// same content hash at different paths — which would cause "plugin already
+// loaded" errors across independent test functions.
+var (
+ pluginTestOnce sync.Once
+ pluginTestDir string // tmpDir owned for the life of the test binary
+ pluginTestSoPath string // absolute path to the built .so
+ pluginTestSoName string // basename only — passed as plugin Path (relative to trustedDir)
+ pluginTestSkipMsg string // non-empty → skip any test that needs the plugin
+)
+
+// setupPluginOnce builds (or skips) the shared test plugin exactly once.
+// It uses sharedPluginDir (from testmain_test.go) so this file and
+// pipeline_loader_test.go share the same .so path — preventing the Go plugin
+// runtime's "plugin already loaded" dedup error when two different paths have
+// identical content.
+// It never calls t.Skip directly; it sets pluginTestSkipMsg for callers to act on.
+func setupPluginOnce(t *testing.T) {
+ t.Helper()
+ pluginTestOnce.Do(func() {
+ if raceDetectorEnabled {
+ pluginTestSkipMsg = "pipeline_watch test skipped under -race: plugin .so must be built with matching race mode; use the integration suite"
+ return
+ }
+ // Prefer the package-level sharedPluginDir (created by TestMain).
+ tmpDir := sharedPluginDir
+ if tmpDir == "" {
+ var mkdirErr error
+ tmpDir, mkdirErr = os.MkdirTemp("", "pipeline-watch-shared-*")
+ if mkdirErr != nil {
+ pluginTestSkipMsg = "pipeline_watch test skipped: cannot create tmpDir: " + mkdirErr.Error()
+ return
+ }
+ // Note: tmpDir is intentionally NOT cleaned up here — it must survive
+ // for the whole test binary run. The OS cleans it on process exit.
+ }
+ soPath := buildTestPlugin(t, tmpDir)
+ if _, probeErr := plugin.Open(soPath); isPluginToolchainMismatch(probeErr) {
+ pluginTestSkipMsg = "pipeline_watch test skipped: toolchain mismatch: " + probeErr.Error()
+ return
+ }
+ pluginTestDir = tmpDir
+ pluginTestSoPath = soPath
+ pluginTestSoName = strings.TrimPrefix(soPath[len(tmpDir):], string(os.PathSeparator))
+ })
+}
+
+// requirePlugin calls setupPluginOnce and skips t if the plugin is not available.
+func requirePlugin(t *testing.T) {
+ t.Helper()
+ setupPluginOnce(t)
+ if pluginTestSkipMsg != "" {
+ t.Skip(pluginTestSkipMsg)
+ }
+}
+
+// TestReconcilePipeline_PopulatesRegistry verifies that a ROLE_DATA schemaRepo
+// with nativePipelineEnabled=true + a valid .so in trustedDir → reconcilePipeline
+// populates lookupSamplers. (US-000 delivery is end-to-end proven by the live
+// US-008 standalone suite; this unit test exercises reconcilePipeline directly.)
+func TestReconcilePipeline_PopulatesRegistry(t *testing.T) {
+ requirePlugin(t)
+
+ resetRegistries()
+ defer resetRegistries()
+
+ const group = "us000-group"
+
+ sp, spErr := makeSamplerPlugin(pluginTestSoName, 300)
+ require.NoError(t, spErr)
+
+ cfg := &commonv1.TracePipelineConfig{
+ Enabled: true,
+ Plugins: []*commonv1.Plugin{
+ {Name: "lss", Kind: &commonv1.Plugin_Sampler{Sampler: sp}},
+ },
+ }
+
+ sr := makeDataSchemaRepo(pluginTestDir)
+ sr.reconcilePipeline(group, cfg)
+
+ samplers := lookupSamplers(group)
+ require.NotEmpty(t, samplers, "registry must be non-empty after reconcilePipeline with a valid config")
+ assert.Equal(t, sdk.KindSampler, samplers[0].Kind())
+}
+
+// TestReconcilePipeline_GateAllowsDataNode verifies that the ROLE_DATA +
+// nativePipelineEnabled gate (the same expression used in OnAddOrUpdate) allows
+// reconcilePipeline to run and populate lookupSamplers.
+func TestReconcilePipeline_GateAllowsDataNode(t *testing.T) {
+ requirePlugin(t)
+
+ resetRegistries()
+ defer resetRegistries()
+
+ const group = "gate-group"
+
+ sp, spErr := makeSamplerPlugin(pluginTestSoName, 200)
+ require.NoError(t, spErr)
+
+ pipelineCfg := &commonv1.TracePipelineConfig{
+ Enabled: true,
+ Plugins: []*commonv1.Plugin{
+ {Name: "lss", Kind: &commonv1.Plugin_Sampler{Sampler: sp}},
+ },
+ }
+
+ g := &commonv1.Group{
+ Metadata: &commonv1.Metadata{Name: group},
+ Catalog: commonv1.Catalog_CATALOG_TRACE,
+ Pipeline: pipelineCfg,
+ }
+
+ sr := makeDataSchemaRepo(pluginTestDir)
+
+ // Exercise the exact gate expression from OnAddOrUpdate.
+ if sr.role == databasev1.Role_ROLE_DATA && sr.nativePipelineEnabled {
+ sr.reconcilePipeline(g.Metadata.Name, g.GetPipeline())
+ }
+
+ samplers := lookupSamplers(group)
+ require.NotEmpty(t, samplers, "ROLE_DATA gate: registry must be non-empty after reconcilePipeline")
+}
+
+// TestReconcilePipeline_NilConfig verifies that a nil pipeline clears the registry.
+func TestReconcilePipeline_NilConfig(t *testing.T) {
+ resetRegistries()
+ defer resetRegistries()
+
+ const group = "nil-cfg-group"
+
+ // Pre-populate with a dummy sampler.
+ dummy := &dummySampler{}
+ replaceSamplersForGroup(group, []namedSampler{{name: "d", sampler: dummy}})
+ require.NotEmpty(t, lookupSamplers(group))
+
+ sr := makeDataSchemaRepo("/some/dir")
+ sr.reconcilePipeline(group, nil)
+
+ assert.Empty(t, lookupSamplers(group), "nil config must clear the registry")
+ assert.Zero(t, lookupMergeGrace(group), "nil config must clear the merge grace")
+}
+
+// TestReconcilePipeline_DisabledConfig verifies that enabled=false clears the registry.
+func TestReconcilePipeline_DisabledConfig(t *testing.T) {
+ resetRegistries()
+ defer resetRegistries()
+
+ const group = "disabled-cfg-group"
+
+ dummy := &dummySampler{}
+ replaceSamplersForGroup(group, []namedSampler{{name: "d", sampler: dummy}})
+ require.NotEmpty(t, lookupSamplers(group))
+
+ sr := makeDataSchemaRepo("/some/dir")
+ sr.reconcilePipeline(group, &commonv1.TracePipelineConfig{Enabled: false})
+
+ assert.Empty(t, lookupSamplers(group), "disabled config must clear the registry")
+}
+
+// TestReconcilePipeline_MergeEventNotEnabled verifies that an enabled config whose
+// enabled_events does not include PIPELINE_EVENT_MERGE installs no merge samplers
+// (v1 only implements the merge-time filter), clearing any previous set.
+func TestReconcilePipeline_MergeEventNotEnabled(t *testing.T) {
+ resetRegistries()
+ defer resetRegistries()
+
+ const group = "finalize-only-group"
+
+ dummy := &dummySampler{}
+ replaceSamplersForGroup(group, []namedSampler{{name: "d", sampler: dummy}})
+ require.NotEmpty(t, lookupSamplers(group))
+
+ sr := makeDataSchemaRepo("/some/dir")
+ sr.reconcilePipeline(group, &commonv1.TracePipelineConfig{
+ Enabled: true,
+ EnabledEvents: []commonv1.PipelineEvent{commonv1.PipelineEvent_PIPELINE_EVENT_FINALIZE},
+ })
+
+ assert.Empty(t, lookupSamplers(group), "config without MERGE event must install no merge samplers")
+}
+
+// TestReconcilePipeline_OnDeleteKindGroup verifies the OnDelete KindGroup
+// cascade: simulate the OnDelete gate → removeSamplersForGroup.
+func TestReconcilePipeline_OnDeleteKindGroup(t *testing.T) {
+ resetRegistries()
+ defer resetRegistries()
+
+ const group = "delete-group"
+
+ dummy := &dummySampler{}
+ replaceSamplersForGroup(group, []namedSampler{{name: "d", sampler: dummy}})
+ require.NotEmpty(t, lookupSamplers(group))
+
+ sr := makeDataSchemaRepo("/some/dir")
+
+ // Simulate the exact OnDelete KindGroup gate from metadata.go.
+ g := &commonv1.Group{
+ Metadata: &commonv1.Metadata{Name: group},
+ Catalog: commonv1.Catalog_CATALOG_TRACE,
+ }
+ if sr.role == databasev1.Role_ROLE_DATA && sr.nativePipelineEnabled {
+ removeSamplersForGroup(g.Metadata.Name)
+ }
+
+ assert.Empty(t, lookupSamplers(group), "OnDelete KindGroup must clear the registry")
+}
+
+// TestReconcilePipeline_LoadFailureKeepsPreviousSet verifies fail-open: if a
+// plugin fails to load, the previous good set is retained and there is no panic.
+func TestReconcilePipeline_LoadFailureKeepsPreviousSet(t *testing.T) {
+ resetRegistries()
+ defer resetRegistries()
+
+ const group = "fail-open-group"
+
+ dummy := &dummySampler{}
+ replaceSamplersForGroup(group, []namedSampler{{name: "prev", sampler: dummy}})
+ require.NotEmpty(t, lookupSamplers(group))
+
+ sr := makeDataSchemaRepo("/nonexistent-trusted-dir")
+
+ cfg := &commonv1.TracePipelineConfig{
+ Enabled: true,
+ Plugins: []*commonv1.Plugin{
+ {
+ Name: "missing",
+ Kind: &commonv1.Plugin_Sampler{Sampler: &commonv1.SamplerPlugin{
+ Path: "missing.so",
+ Symbol: "NewSampler",
+ AbiVersion: uint32(sdk.ABIVersion),
+ }},
+ },
+ },
+ }
+
+ // Must not panic.
+ require.NotPanics(t, func() { sr.reconcilePipeline(group, cfg) })
+
+ // Previous good set must still be intact (fail-open).
+ samplers := lookupSamplers(group)
+ require.NotEmpty(t, samplers, "fail-open: previous good set must be retained after load failure")
+ assert.Equal(t, dummy, samplers[0], "fail-open: exact previous sampler must be retained")
+}
+
+// TestReconcilePipeline_MergeGrace verifies per-config merge_grace storage and
+// lookup via lookupMergeGrace.
+func TestReconcilePipeline_MergeGrace(t *testing.T) {
+ requirePlugin(t)
+
+ resetRegistries()
+ defer resetRegistries()
+
+ const group = "grace-group"
+
+ sp, spErr := makeSamplerPlugin(pluginTestSoName, 100)
+ require.NoError(t, spErr)
+
+ const wantGrace = 45 * time.Second
+ cfg := &commonv1.TracePipelineConfig{
+ Enabled: true,
+ Plugins: []*commonv1.Plugin{
+ {Name: "lss", Kind: &commonv1.Plugin_Sampler{Sampler: sp}},
+ },
+ MergeGrace: durationpb.New(wantGrace),
+ }
+
+ sr := makeDataSchemaRepo(pluginTestDir)
+ sr.reconcilePipeline(group, cfg)
+
+ gotGrace := lookupMergeGrace(group)
+ assert.Equal(t, wantGrace.Nanoseconds(), gotGrace,
+ "lookupMergeGrace must return the configured value in nanoseconds")
+
+ // After nil config, grace must be cleared.
+ sr.reconcilePipeline(group, nil)
+ assert.Zero(t, lookupMergeGrace(group), "grace must be 0 after nil config")
+}
+
+// TestReconcilePipeline_MergeGraceUnset verifies that lookupMergeGrace returns
+// 0 when no per-group grace is configured.
+func TestReconcilePipeline_MergeGraceUnset(t *testing.T) {
+ resetRegistries()
+ defer resetRegistries()
+
+ assert.Zero(t, lookupMergeGrace("never-configured-group"),
+ "lookupMergeGrace must return 0 when unset")
+}
+
+// TestReconcilePipeline_ConcurrentRace drives concurrent replaceSamplersForGroup +
+// removeSamplersForGroup + lookupSamplers + setMergeGraceForGroup +
+// lookupMergeGrace calls and must be -race clean.
+func TestReconcilePipeline_ConcurrentRace(t *testing.T) {
+ resetRegistries()
+ defer resetRegistries()
+ t.Log("concurrent race test")
+
+ const group = "race-group"
+ dummy := &dummySampler{}
+
+ var wg sync.WaitGroup
+ const workers = 8
+ const iterations = 100
+
+ for wID := 0; wID < workers; wID++ {
+ wg.Add(1)
+ go func(id int) {
+ defer wg.Done()
+ for i := 0; i < iterations; i++ {
+ switch (id + i) % 3 {
+ case 0:
+ replaceSamplersForGroup(group, []namedSampler{{name: "d", sampler: dummy}})
+ setMergeGraceForGroup(group, int64(i+1)*int64(time.Second))
+ case 1:
+ removeSamplersForGroup(group)
+ setMergeGraceForGroup(group, 0)
+ case 2:
+ _ = lookupSamplers(group)
+ _ = lookupMergeGrace(group)
+ }
+ }
+ }(wID)
+ }
+ wg.Wait()
+}
+
+// TestReconcilePipeline_LiaisonDoesNotLoad verifies that a ROLE_LIAISON
+// schemaRepo never triggers reconcilePipeline (the gate fires only for
+// ROLE_DATA && nativePipelineEnabled).
+func TestReconcilePipeline_LiaisonDoesNotLoad(t *testing.T) {
+ resetRegistries()
+ defer resetRegistries()
+
+ const group = "liaison-group"
+
+ // Liaison: role is LIAISON even though nativePipelineEnabled=true.
+ sr := schemaRepo{
+ role: databasev1.Role_ROLE_LIAISON,
+ nativePipelineEnabled: true,
+ trustedPluginDir: "/some/dir",
+ }
+
+ cfg := &commonv1.TracePipelineConfig{
+ Enabled: true,
+ Plugins: []*commonv1.Plugin{
+ {
+ Name: "lss",
+ Kind: &commonv1.Plugin_Sampler{Sampler: &commonv1.SamplerPlugin{
+ Path: "missing.so",
+ Symbol: "NewSampler",
+ AbiVersion: uint32(sdk.ABIVersion),
+ }},
+ },
+ },
+ }
+
+ // Exact gate expression from OnAddOrUpdate — LIAISON must not pass.
+ if sr.role == databasev1.Role_ROLE_DATA && sr.nativePipelineEnabled {
+ sr.reconcilePipeline(group, cfg)
+ }
+
+ assert.Empty(t, lookupSamplers(group), "liaison must never populate sampler registry")
+}
+
+// TestReconcilePipeline_NonTraceCatalogOnDelete confirms that OnDelete with a
+// non-CATALOG_TRACE group is a no-op (the catalog guard fires before reconcile).
+func TestReconcilePipeline_NonTraceCatalogOnDelete(t *testing.T) {
+ resetRegistries()
+ defer resetRegistries()
+
+ const group = "non-trace-group"
+ dummy := &dummySampler{}
+ replaceSamplersForGroup(group, []namedSampler{{name: "d", sampler: dummy}})
+ require.NotEmpty(t, lookupSamplers(group))
+
+ // Simulate the OnDelete catalog guard for a non-TRACE group.
+ g := &commonv1.Group{
+ Metadata: &commonv1.Metadata{Name: group},
+ Catalog: commonv1.Catalog_CATALOG_MEASURE, // not TRACE
+ }
+ // The catalog guard: only CATALOG_TRACE reaches removeSamplersForGroup.
+ if g.Catalog == commonv1.Catalog_CATALOG_TRACE {
+ removeSamplersForGroup(g.Metadata.Name)
+ }
+
+ // Non-TRACE group OnDelete must not touch the sampler registry.
+ assert.NotEmpty(t, lookupSamplers(group),
+ "non-TRACE group OnDelete must not clear the registry")
+}
diff --git a/banyand/trace/svc_standalone.go b/banyand/trace/svc_standalone.go
index 14a840143..73848710f 100644
--- a/banyand/trace/svc_standalone.go
+++ b/banyand/trace/svc_standalone.go
@@ -30,13 +30,11 @@ import (
"github.com/pkg/errors"
"go.uber.org/multierr"
- "google.golang.org/protobuf/encoding/protojson"
"github.com/apache/skywalking-banyandb/api/common"
"github.com/apache/skywalking-banyandb/api/data"
commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
- pipelinev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/pipeline/v1"
tracev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/trace/v1"
"github.com/apache/skywalking-banyandb/banyand/internal/storage"
"github.com/apache/skywalking-banyandb/banyand/metadata"
@@ -104,8 +102,6 @@ func (s *standalone) FlagSet() *run.FlagSet {
bindVectorizedFlags(fs, &s.option.vectorized)
fs.BoolVar(&s.option.nativePipelineEnabled, "trace-pipeline-native-plugin-enabled", false, "enable the native plugin pipeline for in-merge trace retention")
fs.StringVar(&s.option.trustedPluginDir, "trace-pipeline-trusted-plugin-dir", "", "trusted directory for native trace pipeline plugins")
- fs.StringVar(&s.option.pipelineConfigPath, "trace-pipeline-config", "",
- "path to a TracePipelineConfig (protojson) loaded at startup; static bootstrap ingress — dynamic delivery is the registry RPC")
fs.DurationVar(&s.option.mergeGraceDefault, "trace-pipeline-merge-grace-default", 30*time.Second, "default merge_grace for in-merge trace retention filter")
fs.DurationVar(&s.option.decideTimeout, "trace-pipeline-decide-timeout", 5*time.Second, "hard per-batch Decide timeout for trace pipeline plugins")
fs.IntVar(&s.option.decideTimeoutCircuitBreak, "trace-pipeline-decide-timeout-circuit-break", 3,
@@ -222,12 +218,6 @@ func (s *standalone) PreRun(ctx context.Context) error {
if err != nil {
return err
}
- if s.option.nativePipelineEnabled && s.option.pipelineConfigPath != "" {
- if loadErr := s.loadPipelineConfig(); loadErr != nil {
- return loadErr
- }
- }
-
s.l.Info().
Str("root", s.root).
Str("dataPath", s.dataPath).
@@ -237,38 +227,6 @@ func (s *standalone) PreRun(ctx context.Context) error {
return nil
}
-// loadPipelineConfig reads the TracePipelineConfig protojson file at pipelineConfigPath,
-// validates it, and registers all configured sampler plugins for the configured group.
-func (s *standalone) loadPipelineConfig() error {
- s.l.Info().Str("path", s.option.pipelineConfigPath).Msg("loading pipeline config")
- cfgBytes, readErr := os.ReadFile(s.option.pipelineConfigPath)
- if readErr != nil {
- return fmt.Errorf("failed to read pipeline config %q: %w", s.option.pipelineConfigPath, readErr)
- }
- var cfg pipelinev1.TracePipelineConfig
- if unmarshalErr := protojson.Unmarshal(cfgBytes, &cfg); unmarshalErr != nil {
- return fmt.Errorf("failed to unmarshal pipeline config %q: %w", s.option.pipelineConfigPath, unmarshalErr)
- }
- if validateErr := cfg.Validate(); validateErr != nil {
- return fmt.Errorf("pipeline config %q is invalid: %w", s.option.pipelineConfigPath, validateErr)
- }
- group := cfg.GetMetadata().GetGroup()
- s.l.Info().Str("group", group).Int("plugins", len(cfg.GetPlugins())).Msg("pipeline config loaded; registering samplers")
- for _, p := range cfg.GetPlugins() {
- if p.GetSampler() == nil {
- continue
- }
- s.l.Info().Str("plugin", p.GetName()).Str("path", p.GetSampler().GetPath()).Msg("loading sampler plugin")
- sampler, loadErr := loadSamplerPlugin(p.GetSampler(), s.option.trustedPluginDir)
- if loadErr != nil {
- return fmt.Errorf("failed to load sampler plugin %q for group %q: %w", p.GetName(), group, loadErr)
- }
- s.l.Info().Str("plugin", p.GetName()).Str("group", group).Msg("sampler plugin loaded and registered")
- registerSampler(group, sampler)
- }
- return nil
-}
-
func (s *standalone) Serve() run.StopNotify {
return s.schemaRepo.StopCh()
}
diff --git a/banyand/trace/testmain_test.go b/banyand/trace/testmain_test.go
new file mode 100644
index 000000000..5f9082ba0
--- /dev/null
+++ b/banyand/trace/testmain_test.go
@@ -0,0 +1,46 @@
+// 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 trace
+
+import (
+ "os"
+ "testing"
+)
+
+// sharedPluginDir is the package-level directory that holds the single .so
+// used by all tests that need a real plugin. It is created by TestMain before
+// any test runs, so the Go plugin runtime only ever opens the .so once
+// (avoids the "plugin already loaded" deduplication error that fires when two
+// tests build the same content into different paths).
+var sharedPluginDir string
+
+// TestMain is the package-level test harness. It creates sharedPluginDir so
+// that pipeline_loader_test.go (TestLoadSamplerPlugin) and
+// pipeline_watch_test.go (TestReconcilePipeline_*) share a single .so path.
+func TestMain(m *testing.M) {
+ dir, err := os.MkdirTemp("", "trace-pkg-test-plugin-*")
+ if err == nil {
+ sharedPluginDir = dir
+ }
+ // Run all tests; clean up the shared dir afterward.
+ code := m.Run()
+ if sharedPluginDir != "" {
+ os.RemoveAll(sharedPluginDir)
+ }
+ os.Exit(code)
+}
diff --git a/banyand/trace/trace.go b/banyand/trace/trace.go
index 4480ee124..4c770dac2 100644
--- a/banyand/trace/trace.go
+++ b/banyand/trace/trace.go
@@ -56,7 +56,6 @@ type option struct {
protector protector.Memory
tire2Client queue.Client
trustedPluginDir string
- pipelineConfigPath string
seriesCacheMaxSize run.Bytes
flushTimeout time.Duration
syncInterval time.Duration
diff --git a/docs/api-reference.md b/docs/api-reference.md
index e9a706dfa..7409ad42d 100644
--- a/docs/api-reference.md
+++ b/docs/api-reference.md
@@ -66,10 +66,15 @@
- [IntervalRule](#banyandb-common-v1-IntervalRule)
- [LifecycleStage](#banyandb-common-v1-LifecycleStage)
- [Metadata](#banyandb-common-v1-Metadata)
+ - [Plugin](#banyandb-common-v1-Plugin)
- [ResourceOpts](#banyandb-common-v1-ResourceOpts)
+ - [SamplerPlugin](#banyandb-common-v1-SamplerPlugin)
+ - [StageRule](#banyandb-common-v1-StageRule)
+ - [TracePipelineConfig](#banyandb-common-v1-TracePipelineConfig)
- [Catalog](#banyandb-common-v1-Catalog)
- [IntervalRule.Unit](#banyandb-common-v1-IntervalRule-Unit)
+ - [PipelineEvent](#banyandb-common-v1-PipelineEvent)
- [banyandb/property/v1/property.proto](#banyandb_property_v1_property-proto)
- [Property](#banyandb-property-v1-Property)
@@ -379,10 +384,6 @@
- [MeasureService](#banyandb-measure-v1-MeasureService)
- [banyandb/pipeline/v1/trace_pipeline.proto](#banyandb_pipeline_v1_trace_pipeline-proto)
- - [Plugin](#banyandb-pipeline-v1-Plugin)
- - [SamplerPlugin](#banyandb-pipeline-v1-SamplerPlugin)
- - [StageRule](#banyandb-pipeline-v1-StageRule)
- - [TracePipelineConfig](#banyandb-pipeline-v1-TracePipelineConfig)
- [TracePipelineRegistryServiceCreateRequest](#banyandb-pipeline-v1-TracePipelineRegistryServiceCreateRequest)
- [TracePipelineRegistryServiceCreateResponse](#banyandb-pipeline-v1-TracePipelineRegistryServiceCreateResponse)
- [TracePipelineRegistryServiceDeleteRequest](#banyandb-pipeline-v1-TracePipelineRegistryServiceDeleteRequest)
@@ -396,8 +397,6 @@
- [TracePipelineRegistryServiceUpdateRequest](#banyandb-pipeline-v1-TracePipelineRegistryServiceUpdateRequest)
- [TracePipelineRegistryServiceUpdateResponse](#banyandb-pipeline-v1-TracePipelineRegistryServiceUpdateResponse)
- - [PipelineEvent](#banyandb-pipeline-v1-PipelineEvent)
-
- [TracePipelineRegistryService](#banyandb-pipeline-v1-TracePipelineRegistryService)
- [banyandb/property/v1/gossip.proto](#banyandb_property_v1_gossip-proto)
@@ -1350,6 +1349,7 @@ Group is an internal object for Group management
| resource_opts | [ResourceOpts](#banyandb-common-v1-ResourceOpts) | | resourceOpts indicates the structure of the underlying kv storage |
| updated_at | [google.protobuf.Timestamp](#google-protobuf-Timestamp) | | updated_at indicates when resources of the group are updated |
| created_at | [google.protobuf.Timestamp](#google-protobuf-Timestamp) | | created_at is the first-appearance timestamp; survives updates unchanged. |
+| pipeline | [TracePipelineConfig](#banyandb-common-v1-TracePipelineConfig) | optional | pipeline carries the optional group-scoped in-merge retention config. Absent or non-CATALOG_TRACE group means no sampler. |
@@ -1412,6 +1412,22 @@ Metadata is for multi-tenant, multi-model use
+<a name="banyandb-common-v1-Plugin"></a>
+
+### Plugin
+Plugin is one link in a pipeline's processing chain.
+
+
+| Field | Type | Label | Description |
+| ----- | ---- | ----- | ----------- |
+| name | [string](#string) | | name is the operator-facing identity for this link. |
+| sampler | [SamplerPlugin](#banyandb-common-v1-SamplerPlugin) | | sampler owns a keep/drop verdict over a vectorized batch of traces. |
+
+
+
+
+
+
<a name="banyandb-common-v1-ResourceOpts"></a>
### ResourceOpts
@@ -1431,6 +1447,65 @@ Metadata is for multi-tenant, multi-model use
+
+<a name="banyandb-common-v1-SamplerPlugin"></a>
+
+### SamplerPlugin
+SamplerPlugin configures a user-supplied native Go plugin (.so) that owns a
+keep/drop verdict over a vectorized batch of traces.
+
+
+| Field | Type | Label | Description |
+| ----- | ---- | ----- | ----------- |
+| path | [string](#string) | | path is the plugin .so filename, resolved within the data node's trusted plugin directory. |
+| symbol | [string](#string) | | symbol is the constructor symbol the engine looks up; defaults to "NewSampler" if empty. |
+| abi_version | [uint32](#uint32) | | abi_version is the ABI version the plugin was built against. |
+| config | [google.protobuf.Struct](#google-protobuf-Struct) | | config is the plugin-defined configuration serialized to canonical JSON for the constructor. |
+
+
+
+
+
+
+<a name="banyandb-common-v1-StageRule"></a>
+
+### StageRule
+StageRule binds the pipeline to one lifecycle stage and declares that stage's
+retention plugin chain.
+
+
+| Field | Type | Label | Description |
+| ----- | ---- | ----- | ----------- |
+| stage | [string](#string) | | stage is the stage name from the Group's ResourceOpts.stages. |
+| plugins | [Plugin](#banyandb-common-v1-Plugin) | repeated | plugins is the ordered retention chain for this stage. |
+
+
+
+
+
+
+<a name="banyandb-common-v1-TracePipelineConfig"></a>
+
+### TracePipelineConfig
+TracePipelineConfig is the group-scoped, name-less in-merge retention configuration.
+Identity comes from the carrying Group; there is one config per group by construction.
+
+
+| Field | Type | Label | Description |
+| ----- | ---- | ----- | ----------- |
+| enabled | [bool](#bool) | | enabled activates the pipeline. |
+| plugins | [Plugin](#banyandb-common-v1-Plugin) | repeated | plugins is the ordered chain of plugins evaluated by enabled events. |
+| merge_grace | [google.protobuf.Duration](#google-protobuf-Duration) | | merge_grace is the per-trace maturity window for the in-merge filter. Strictly positive if set; engine default 30s if unset. |
+| finalize_grace | [google.protobuf.Duration](#google-protobuf-Duration) | | finalize_grace is the per-segment settling window for the scheduled finalization pass. Strictly positive if set; engine default 5m if unset. |
+| stages | [StageRule](#banyandb-common-v1-StageRule) | repeated | stages declares per-stage retention rules. |
+| schema_names | [string](#string) | repeated | schema_names lists explicit schema names to target within the Group. |
+| schema_name_regex | [string](#string) | | schema_name_regex is an RE2 pattern matched against schema names. |
+| enabled_events | [PipelineEvent](#banyandb-common-v1-PipelineEvent) | repeated | enabled_events lists the pipeline-wide events to run. |
+
+
+
+
+
@@ -1461,6 +1536,19 @@ Metadata is for multi-tenant, multi-model use
| UNIT_DAY | 2 | |
+
+<a name="banyandb-common-v1-PipelineEvent"></a>
+
+### PipelineEvent
+PipelineEvent identifies a pipeline-wide event that can be independently enabled.
+
+| Name | Number | Description |
+| ---- | ------ | ----------- |
+| PIPELINE_EVENT_UNSPECIFIED | 0 | |
+| PIPELINE_EVENT_MERGE | 1 | In-merge filter during Hot-phase LSM compaction merges. |
+| PIPELINE_EVENT_FINALIZE | 2 | Tail-sampling gate at Hot-phase segment finalization. |
+
+
@@ -5929,140 +6017,6 @@ WriteResponse is the response contract for write
-<a name="banyandb-pipeline-v1-Plugin"></a>
-
-### Plugin
-Plugin is one link in a pipeline's processing chain — a generic, kind-tagged
-envelope around a user-supplied native Go plugin. The set oneof arm selects
-the kind; today the only kind is a sampler (SamplerPlugin). Adding a new kind
-is purely additive: define its payload message and add a new arm to `kind`,
-leaving existing arms and field numbers untouched.
-
-A chain of Plugin (TracePipelineConfig.plugins, StageRule.plugins) is a
-sequential pipe: links run in declared order, each link processes the traces
-the previous link kept, and a link that fails is bypassed (its input passes
-through unchanged). The plugin↔engine contract for each kind (the vectorized
-batch, the projection handshake, the verdict) lives in the pinned Go SDK
-module `pkg/pipeline/sdk`, not in this proto.
-
-
-| Field | Type | Label | Description |
-| ----- | ---- | ----- | ----------- |
-| name | [string](#string) | | Operator-facing identity for this link, used in diagnostics and admission errors. Must be non-empty; cross-element uniqueness within a chain is enforced server-side. |
-| sampler | [SamplerPlugin](#banyandb-pipeline-v1-SamplerPlugin) | | Sampler kind: owns a keep/drop verdict over a vectorized batch of traces. |
-
-
-
-
-
-
-<a name="banyandb-pipeline-v1-SamplerPlugin"></a>
-
-### SamplerPlugin
-SamplerPlugin configures a user-supplied native Go plugin (a .so loaded
-in-process via the Go `plugin` package) that owns a keep/drop verdict over a
-vectorized batch of traces. It is the sampler kind of Plugin (Plugin.sampler)
-and is the keep/drop link wherever a chain runs: in a TracePipelineConfig
-chain it gates at the enabled PipelineEvent(s) (merge / finalization); in a
-StageRule chain it is per-stage retention at a stage's migration-out boundary.
-
-The full plugin↔engine contract (the vectorized batch type, the projection
-handshake, and the verdict shape) lives in the pinned Go SDK module
-`pkg/pipeline/sdk`, not in this proto; this message only locates and admits
-the plugin. Three properties of that contract:
- - Strong compatibility: the boundary exchanges only stdlib/primitive types
- defined in the pinned SDK, so no third-party struct version is pinned
- across the .so boundary. The plugin must be built with the SAME Go
- toolchain, build tags, and flags (-trimpath, CGO) and the SAME pinned SDK
- as the running data node; `abi_version` is checked at load.
- - Vectorized input: the plugin's Decide is called once per columnar batch of
- traces, not once per trace.
- - Projection: the plugin declares the columns it needs (SDK Project →
- Projection{Tags, SpanIDs, Spans}); the engine materializes only those tag
- columns and, only when requested, the spans stream — like a query
- projection.
-
-Operational constraints (Go `plugin`): Linux/macOS only; plugins cannot be
-unloaded, so changing one requires a node restart (no hot-reload); a plugin
-panic is contained with recover() and fails open (the whole batch is
-retained). Loading arbitrary code is operator-only and gated behind a server
-flag plus a trusted plugin directory.
-
-
-| Field | Type | Label | Description |
-| ----- | ---- | ----- | ----------- |
-| path | [string](#string) | | Plugin .so filename, resolved within the data node's trusted plugin directory. The engine rejects any path that escapes that directory. |
-| symbol | [string](#string) | | Constructor symbol the engine looks up; defaults to "NewSampler" if empty. |
-| abi_version | [uint32](#uint32) | | ABI version the plugin was built against. The engine refuses to load the plugin unless this equals its own compiled sdk.ABIVersion. |
-| config | [google.protobuf.Struct](#google-protobuf-Struct) | | Plugin-defined configuration, set directly in the pipeline config as a structured object. The engine does not interpret its keys: it serializes the Struct to canonical JSON and hands the bytes to the plugin's constructor (SDK NewSampler([]byte)), which unmarshals them into the plugin's own typed config and validates them — a malformed config fails the load. Optional: a plugin that needs no configuration leaves it unset. |
-
-
-
-
-
-
-<a name="banyandb-pipeline-v1-StageRule"></a>
-
-### StageRule
-StageRule binds the pipeline to one lifecycle stage of the targeted Group
-and declares that stage's retention plugin chain. The rule fires at the
-stage's migration-out boundary (i.e. when a segment migrates from this stage
-to the next stage); routine compaction is governed by PIPELINE_EVENT_MERGE on
-TracePipelineConfig, not by StageRule.
-
-Per-stage retention uses the SAME chain mechanism as gating: each stage's
-`plugins` chain owns the keep/drop verdict for traces leaving that stage. A
-StageRule with an empty `plugins` chain has no filtering effect — every trace
-at this stage migrates unchanged. The "rising bar" across stages (Hot keeps
-more, Cold keeps less) is expressed by each stage's plugin config (see §4.2
-of the design doc), not by a fixed predicate vocabulary.
-
-
-| Field | Type | Label | Description |
-| ----- | ---- | ----- | ----------- |
-| stage | [string](#string) | | Stage name from the Group's ResourceOpts.stages (e.g. "hot", "warm", "cold"). Must be non-empty; an empty stage name cannot match any lifecycle stage. |
-| plugins | [Plugin](#banyandb-pipeline-v1-Plugin) | repeated | Per-stage retention chain: the ordered plugins (a sequential pipe) that decide keep/drop for traces leaving this stage at its migration-out boundary. Empty means no per-stage drop (every trace migrates). Same contract and composition as the gating chain (see Plugin). |
-
-
-
-
-
-
-<a name="banyandb-pipeline-v1-TracePipelineConfig"></a>
-
-### TracePipelineConfig
-TracePipelineConfig is the root configuration for a storage-node trace pipeline.
-It reuses existing catalog identifiers (group via metadata, stage names, schema
-names) for targeting instead of declaring a parallel metadata model.
-
-The pipeline has up to three filter points:
- 1. PIPELINE_EVENT_MERGE — in-merge filter during Hot-phase LSM compaction (default).
- 2. PIPELINE_EVENT_FINALIZE — tail-sampling gate at Hot-phase finalization.
- 3. Per-stage retention via StageRule.plugins, applied at the stage's
- migration-out boundary (when the segment migrates to the next stage).
- Always implicit when any StageRule carries a plugin chain.
-Events 1 and 2 are toggleable via `enabled_events`. The gating policy those
-events evaluate is the `plugins` chain (a sequential pipe of native Go
-plugins; today each link is a sampler).
-
-
-| Field | Type | Label | Description |
-| ----- | ---- | ----- | ----------- |
-| metadata | [banyandb.common.v1.Metadata](#banyandb-common-v1-Metadata) | | Identity and revision tracking; metadata.group is the Group this pipeline lives in and applies to, consistent with every other schema resource. Required: every config needs a name/group for registry handling. |
-| enabled | [bool](#bool) | | Active status of the pipeline. |
-| stages | [StageRule](#banyandb-pipeline-v1-StageRule) | repeated | Per-stage retention rules: which lifecycle stages this pipeline acts on, with the retention plugin chain for each. Each rule fires at its stage's migration-out boundary regardless of `enabled_events`. Empty means the only filters are the `enabled_events` events (no per-stage drop). |
-| schema_names | [string](#string) | repeated | Explicit schema names to target within the Group (exact match on Metadata.name). Each entry must be non-empty; cross-element uniqueness is enforced server-side. |
-| schema_name_regex | [string](#string) | | RE2 regular expression matched against schema names. A schema is targeted if it is listed in schema_names OR matches this pattern. When both are empty, every schema in the Group is targeted. |
-| plugins | [Plugin](#banyandb-pipeline-v1-Plugin) | repeated | Gating policy: an ordered chain of plugins (a sequential pipe) evaluated by any enabled event (PIPELINE_EVENT_MERGE and/or PIPELINE_EVENT_FINALIZE). Links run in declared order, each processing the traces the previous link kept; a link that fails is bypassed (fail-open). Today every link is a sampler, so the chain is the conjunction of the links' keep/drop verdicts. Empty means the only retention is the per-stage StageRu [...]
-| enabled_events | [PipelineEvent](#banyandb-pipeline-v1-PipelineEvent) | repeated | Pipeline-wide events to run. Empty defaults to [PIPELINE_EVENT_MERGE] — the in-merge filter is on, the finalization gate is off. To enable the finalization gate, include PIPELINE_EVENT_FINALIZE; to disable the merge filter, list only [PIPELINE_EVENT_FINALIZE]; the explicit empty default value is also acceptable to mean "merge only". Each element must be a defined, non-UNSPECIFIED value; duplicate [...]
-| merge_grace | [google.protobuf.Duration](#google-protobuf-Duration) | | Per-trace maturity window for the in-merge filter (§7.1). A trace is eligible for dropping during an LSM compaction merge only once its latest span timestamp is older than `now - merge_grace`; younger traces pass through the merge unchanged. Bounds the expected intra-trace span arrival spread (typically seconds). Used iff `enabled_events` contains PIPELINE_EVENT_MERGE. Strictly positive if set; engine default 30s [...]
-| finalize_grace | [google.protobuf.Duration](#google-protobuf-Duration) | | Per-segment settling window for the scheduled finalization pass (§7.3). A segment is treated as settled, and the authoritative final filter runs, once the event-time watermark exceeds `segment.End + finalize_grace`. Bounds segment-wide late arrival (typically minutes). Used iff `enabled_events` contains PIPELINE_EVENT_FINALIZE. Strictly positive if set; engine default 5m if unset. |
-
-
-
-
-
-
<a name="banyandb-pipeline-v1-TracePipelineRegistryServiceCreateRequest"></a>
### TracePipelineRegistryServiceCreateRequest
@@ -6071,7 +6025,7 @@ plugins; today each link is a sampler).
| Field | Type | Label | Description |
| ----- | ---- | ----- | ----------- |
-| trace_pipeline_config | [TracePipelineConfig](#banyandb-pipeline-v1-TracePipelineConfig) | | |
+| trace_pipeline_config | [banyandb.common.v1.TracePipelineConfig](#banyandb-common-v1-TracePipelineConfig) | | |
@@ -6118,7 +6072,7 @@ plugins; today each link is a sampler).
| ----- | ---- | ----- | ----------- |
| deleted | [bool](#bool) | | |
| delete_time | [int64](#int64) | | delete_time is the server-assigned tombstone timestamp in unix nanos. |
-| mod_revision | [int64](#int64) | | mod_revision is the etcd revision of the tombstone; zero if the server did not record one. |
+| mod_revision | [int64](#int64) | | mod_revision is the revision of the tombstone; zero if the server did not record one. |
@@ -6179,7 +6133,7 @@ plugins; today each link is a sampler).
| Field | Type | Label | Description |
| ----- | ---- | ----- | ----------- |
-| trace_pipeline_config | [TracePipelineConfig](#banyandb-pipeline-v1-TracePipelineConfig) | | |
+| trace_pipeline_config | [banyandb.common.v1.TracePipelineConfig](#banyandb-common-v1-TracePipelineConfig) | | |
@@ -6209,7 +6163,7 @@ plugins; today each link is a sampler).
| Field | Type | Label | Description |
| ----- | ---- | ----- | ----------- |
-| trace_pipeline_config | [TracePipelineConfig](#banyandb-pipeline-v1-TracePipelineConfig) | repeated | |
+| trace_pipeline_config | [banyandb.common.v1.TracePipelineConfig](#banyandb-common-v1-TracePipelineConfig) | repeated | |
@@ -6224,7 +6178,7 @@ plugins; today each link is a sampler).
| Field | Type | Label | Description |
| ----- | ---- | ----- | ----------- |
-| trace_pipeline_config | [TracePipelineConfig](#banyandb-pipeline-v1-TracePipelineConfig) | | |
+| trace_pipeline_config | [banyandb.common.v1.TracePipelineConfig](#banyandb-common-v1-TracePipelineConfig) | | |
@@ -6247,21 +6201,6 @@ plugins; today each link is a sampler).
-
-<a name="banyandb-pipeline-v1-PipelineEvent"></a>
-
-### PipelineEvent
-PipelineEvent identifies a pipeline-wide event that can be independently
-enabled. Per-stage retention (StageRule.plugins) fires implicitly at the
-stage's migration-out boundary and is not toggleable via this enum.
-
-| Name | Number | Description |
-| ---- | ------ | ----------- |
-| PIPELINE_EVENT_UNSPECIFIED | 0 | |
-| PIPELINE_EVENT_MERGE | 1 | In-merge filter during Hot-phase LSM compaction merges (Warm/Cold compactions stay lossless). Per-trace drops are gated by `merge_grace` so partial traces are not destroyed prematurely. Cheap, runs often; verdicts wait for trace maturity (see §7.1). |
-| PIPELINE_EVENT_FINALIZE | 2 | Tail-sampling gate at Hot-phase segment finalization, after the segment has settled (event-time watermark past `segment.End + finalize_grace`). Heavy but authoritative; sees the complete trace (see §7.3). |
-
-
diff --git a/pkg/test/setup/external.go b/pkg/test/setup/external.go
index b064b4f0f..549a873f4 100644
--- a/pkg/test/setup/external.go
+++ b/pkg/test/setup/external.go
@@ -118,6 +118,42 @@ func ExternalCMD(binPath, logPath string, flags ...string) (func(), error) {
return teardown, nil
}
+// BuildStandaloneFlags assembles the flag list for an external standalone server
+// WITHOUT the AddSchemaServerAddr side-effect. Use this to capture the flags
+// once for restart scenarios (ExternalStandalone already records the schema
+// address; a second call would duplicate it in the ClusterConfig).
+// The returned slice does NOT include the leading "standalone" subcommand.
+func BuildStandaloneFlags(config *ClusterConfig, path string, ports []int) []string {
+ addr := fmt.Sprintf("%s:%d", host, ports[0])
+ schemaPort := ports[4]
+ ff := []string{
+ "--logging-env=dev",
+ "--logging-level=" + testflags.LogLevel,
+ "--grpc-host=" + host,
+ fmt.Sprintf("--grpc-port=%d", ports[0]),
+ "--http-host=" + host,
+ fmt.Sprintf("--http-port=%d", ports[1]),
+ "--http-grpc-addr=" + addr,
+ "--stream-root-path=" + path,
+ "--measure-root-path=" + path,
+ "--property-root-path=" + path,
+ "--trace-root-path=" + path,
+ "--schema-server-root-path=" + path,
+ "--schema-registry-mode=" + config.SchemaRegistry.Mode,
+ "--node-discovery-mode=" + config.NodeDiscovery.Mode,
+ "--node-host-provider=flag",
+ "--node-host=" + localhost,
+ "--schema-server-grpc-host=" + localhost,
+ fmt.Sprintf("--schema-server-grpc-port=%d", schemaPort),
+ }
+ if config.NodeDiscovery.Mode == ModeFile && config.NodeDiscovery.FileWriter != nil {
+ ff = append(ff, fmt.Sprintf("--node-discovery-file-path=%s", config.NodeDiscovery.FileWriter.Path()))
+ }
+ // Deliberately does NOT call config.AddSchemaServerAddr — the caller's
+ // ExternalStandalone has already done so.
+ return ff
+}
+
// standaloneFlags assembles the flag list for an external standalone server,
// mirroring the flag block in standaloneServerWithAuth (setup.go:377-396).
// The returned slice does NOT include the leading "standalone" subcommand.
@@ -224,6 +260,49 @@ func dataNodeFlags(config *ClusterConfig, dataDir string, ports []int) []string
return ff
}
+// BuildDataNodeFlags assembles the flag list for an external data node
+// WITHOUT the AddSchemaServerAddr side-effect. Use this to capture the flags
+// once for restart scenarios (ExternalDataNode already records the schema
+// address; a second call would duplicate it in the ClusterConfig).
+// The returned slice does NOT include the leading "data" subcommand.
+// ports must contain at least 3 elements: [grpc, gossip, http]; a 4th element
+// enables the schema server (ports[3]) without re-registering its address.
+func BuildDataNodeFlags(config *ClusterConfig, dataDir string, ports []int) []string {
+ nodeHost := localhost
+ runSchemaServer := len(ports) >= 4
+
+ ff := []string{
+ "--grpc-host=" + host,
+ fmt.Sprintf("--grpc-port=%d", ports[0]),
+ fmt.Sprintf("--property-repair-gossip-grpc-port=%d", ports[1]),
+ fmt.Sprintf("--http-port=%d", ports[2]),
+ "--stream-root-path=" + dataDir,
+ "--measure-root-path=" + dataDir,
+ "--property-root-path=" + dataDir,
+ "--trace-root-path=" + dataDir,
+ "--schema-server-root-path=" + dataDir,
+ "--node-host-provider", "flag",
+ "--node-host", nodeHost,
+ "--logging-modules", "trace,sidx,property-schema-registry",
+ "--logging-levels", "debug,debug,debug",
+ "--schema-registry-mode=" + config.SchemaRegistry.Mode,
+ "--node-discovery-mode=" + config.NodeDiscovery.Mode,
+ }
+ if config.NodeDiscovery.Mode == ModeFile && config.NodeDiscovery.FileWriter != nil {
+ ff = append(ff, fmt.Sprintf("--node-discovery-file-path=%s", config.NodeDiscovery.FileWriter.Path()))
+ }
+ if runSchemaServer {
+ schemaPort := ports[3]
+ ff = append(ff,
+ "--schema-server-grpc-host="+nodeHost,
+ fmt.Sprintf("--schema-server-grpc-port=%d", schemaPort),
+ )
+ // Deliberately does NOT call config.AddSchemaServerAddr — the caller's
+ // ExternalDataNode has already done so.
+ }
+ return ff
+}
+
// ExternalDataNode launches an external banyand data node, waits for it to
// pass the gRPC readiness gate, and returns the gRPC address and teardown closure.
// ports must contain at least 3 elements: [grpc, gossip, http]; pass 4 to also
diff --git a/test/cases/tracepipeline/ops.go b/test/cases/tracepipeline/ops.go
new file mode 100644
index 000000000..5b64fefe1
--- /dev/null
+++ b/test/cases/tracepipeline/ops.go
@@ -0,0 +1,177 @@
+// 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 tracepipeline_test
+
+import (
+ "context"
+ "time"
+
+ gm "github.com/onsi/gomega"
+ grpclib "google.golang.org/grpc"
+ "google.golang.org/protobuf/types/known/durationpb"
+ "google.golang.org/protobuf/types/known/structpb"
+
+ commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
+ databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+)
+
+// RegisterSamplerRuntime writes a TracePipelineConfig onto the named group via
+// UpdateGroup, activating the sampler pipeline on the data node through the
+// KindGroup watch path. Asserts no error.
+func RegisterSamplerRuntime(ctx context.Context, conn *grpclib.ClientConn, group string, cfg *commonv1.TracePipelineConfig) {
+ applyPipelineConfig(ctx, conn, group, cfg)
+}
+
+// UpdateSamplerRuntime replaces the TracePipelineConfig on the named group with
+// cfg (e.g. a variant with a different threshold), delivering the change via
+// UpdateGroup. Asserts no error.
+func UpdateSamplerRuntime(ctx context.Context, conn *grpclib.ClientConn, group string, cfg *commonv1.TracePipelineConfig) {
+ applyPipelineConfig(ctx, conn, group, cfg)
+}
+
+// RemoveSamplerRuntime clears the pipeline config from the named group by
+// setting Pipeline to nil and calling UpdateGroup. After convergence the data
+// node's sampler registry for the group is empty and all traces are retained.
+// Asserts no error.
+func RemoveSamplerRuntime(ctx context.Context, conn *grpclib.ClientConn, group string) {
+ applyPipelineConfig(ctx, conn, group, nil)
+}
+
+// applyPipelineConfig fetches the current group, sets its Pipeline field to cfg
+// (nil = clear), and persists via UpdateGroup. The caller decides whether this
+// is a register, update, or remove operation.
+func applyPipelineConfig(ctx context.Context, conn *grpclib.ClientConn, group string, cfg *commonv1.TracePipelineConfig) {
+ client := databasev1.NewGroupRegistryServiceClient(conn)
+
+ getCtx, getCancel := context.WithTimeout(ctx, 10*time.Second)
+ defer getCancel()
+ getResp, getErr := client.Get(getCtx, &databasev1.GroupRegistryServiceGetRequest{Group: group})
+ gm.Expect(getErr).NotTo(gm.HaveOccurred(), "GetGroup %q", group)
+
+ grp := getResp.GetGroup()
+ gm.Expect(grp).NotTo(gm.BeNil(), "GetGroup %q returned nil group", group)
+
+ grp.Pipeline = cfg
+
+ updateCtx, updateCancel := context.WithTimeout(ctx, 10*time.Second)
+ defer updateCancel()
+ _, updateErr := client.Update(updateCtx, &databasev1.GroupRegistryServiceUpdateRequest{Group: grp})
+ gm.Expect(updateErr).NotTo(gm.HaveOccurred(), "UpdateGroup %q with pipeline=%v", group, cfg != nil)
+}
+
+// Fixture builders.
+
+// PluginSOName is the bare filename of the latency-status sampler .so. Suites
+// must stage this file inside the --trace-pipeline-trusted-plugin-dir before
+// registering. The path value written to the config is the trusted-dir-relative
+// filename; the data node resolves it against the trusted dir at load time.
+const PluginSOName = "latencystatussampler.so"
+
+// DefaultMergeGrace is the merge_grace used by the base fixture. It is
+// intentionally short so integration tests do not wait long for parts to cool.
+const DefaultMergeGrace = 0 * time.Second
+
+// NewBasePipelineConfig returns the base TracePipelineConfig used by
+// RegisterSamplerRuntime in suite setup. It points at pluginSOName with
+// thresholdMs=500 (the canonical latency-status sampler threshold) and
+// merge_grace=graceOverride. Pass 0 to use the instantaneous grace path.
+//
+// The .so must already be staged in the trusted plugin dir under the same name.
+func NewBasePipelineConfig(soPath string, mergeGrace time.Duration) *commonv1.TracePipelineConfig {
+ return newLatencyStatusConfig(soPath, 500, mergeGrace)
+}
+
+// NewVariantPipelineConfig returns a variant TracePipelineConfig with a
+// different thresholdMs (200) for the UpdateSamplerRuntime case. Traces with
+// duration ≥ 200 ms and status == "success" are now KEPT instead of the
+// baseline's ≥ 500 ms threshold.
+//
+// US-007 / US-008: suites call this to verify that Update delivers new drop
+// verdicts on the next merge.
+func NewVariantPipelineConfig(soPath string, mergeGrace time.Duration) *commonv1.TracePipelineConfig {
+ return newLatencyStatusConfig(soPath, 200, mergeGrace)
+}
+
+// NewPanicPipelineConfig returns a TracePipelineConfig whose sampler panics
+// unconditionally in Decide. The engine's fail-open recover wrapper catches the
+// panic and retains all traces — this config is used exclusively by the US-010
+// soak to verify engine resilience (the node must survive, merges must not
+// stall, and traces must be retained after the panic is absorbed).
+//
+// The .so must be the same latencystatussampler staged in the trusted dir; the
+// "panic":true config field activates the panic path added for US-010.
+func NewPanicPipelineConfig(soPath string, mergeGrace time.Duration) *commonv1.TracePipelineConfig {
+ cfgStruct, structErr := structpb.NewStruct(map[string]interface{}{
+ "thresholdMs": float64(500),
+ "panic": true,
+ })
+ gm.Expect(structErr).NotTo(gm.HaveOccurred(), "structpb.NewStruct for latencystatussampler panic config")
+
+ cfg := &commonv1.TracePipelineConfig{
+ Enabled: true,
+ Plugins: []*commonv1.Plugin{
+ {
+ Name: "latencystatussampler-panic",
+ Kind: &commonv1.Plugin_Sampler{
+ Sampler: &commonv1.SamplerPlugin{
+ Path: soPath,
+ Symbol: "NewSampler",
+ AbiVersion: 1,
+ Config: cfgStruct,
+ },
+ },
+ },
+ },
+ }
+ if mergeGrace > 0 {
+ cfg.MergeGrace = durationpb.New(mergeGrace)
+ }
+ return cfg
+}
+
+// newLatencyStatusConfig builds a TracePipelineConfig for the
+// latencystatussampler plugin with the given .so path, thresholdMs, and
+// merge_grace. The sampler drops traces whose duration (ms) is below
+// thresholdMs AND whose status is "success".
+func newLatencyStatusConfig(soPath string, thresholdMs int64, mergeGrace time.Duration) *commonv1.TracePipelineConfig {
+ cfgStruct, structErr := structpb.NewStruct(map[string]interface{}{
+ "thresholdMs": float64(thresholdMs),
+ })
+ gm.Expect(structErr).NotTo(gm.HaveOccurred(), "structpb.NewStruct for latencystatussampler config")
+
+ cfg := &commonv1.TracePipelineConfig{
+ Enabled: true,
+ Plugins: []*commonv1.Plugin{
+ {
+ Name: "latencystatussampler",
+ Kind: &commonv1.Plugin_Sampler{
+ Sampler: &commonv1.SamplerPlugin{
+ Path: soPath,
+ Symbol: "NewSampler",
+ AbiVersion: 1,
+ Config: cfgStruct,
+ },
+ },
+ },
+ },
+ }
+ if mergeGrace > 0 {
+ cfg.MergeGrace = durationpb.New(mergeGrace)
+ }
+ return cfg
+}
diff --git a/test/cases/tracepipeline/tracepipeline.go b/test/cases/tracepipeline/tracepipeline.go
index cd7ea5b85..e252a5409 100644
--- a/test/cases/tracepipeline/tracepipeline.go
+++ b/test/cases/tracepipeline/tracepipeline.go
@@ -22,6 +22,7 @@ import (
"context"
"fmt"
"os"
+ "runtime"
"time"
g "github.com/onsi/ginkgo/v2"
@@ -29,6 +30,7 @@ import (
grpclib "google.golang.org/grpc"
"google.golang.org/protobuf/types/known/timestamppb"
+ databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
tracev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/trace/v1"
"github.com/apache/skywalking-banyandb/pkg/test/flags"
@@ -37,8 +39,10 @@ import (
)
const (
- // pipelineGroup is the schema group used by the filter trace.
- pipelineGroup = "test-trace-pipeline"
+ // PipelineGroup is the schema group used by the filter trace.
+ // Exported so that integration suites (standalone/distributed) can reference
+ // the group name when calling RegisterSamplerRuntime / UpdateSamplerRuntime.
+ PipelineGroup = "test-trace-pipeline"
// batch1File is the first batch fixture: 4 traces (2 drops + 2 keeps) → one mem-part.
batch1File = "batch1.json"
@@ -74,7 +78,7 @@ const (
// parts near-equal so the real merge-balance gate fires (no merge-policy tuning needed).
// Timestamps are placed seedOffset before baseTime so isMergeHot returns false.
func SeedBatch1(baseTime time.Time) {
- tracepipelinedata.WriteBatch(SharedContext.Connection, batch1File, pipelineGroup, baseTime.Add(seedOffset), time.Millisecond)
+ tracepipelinedata.WriteBatch(SharedContext.Connection, batch1File, PipelineGroup, baseTime.Add(seedOffset), time.Millisecond)
}
// SeedBatch2 writes batch2.json as a single write stream (one mem-part).
@@ -83,7 +87,7 @@ func SeedBatch1(baseTime time.Time) {
// Writing this second part triggers the filtering compaction (max-merge-parts=2).
// Timestamps are placed slightly after batch1 but still in the past so isMergeHot returns false.
func SeedBatch2(baseTime time.Time) {
- tracepipelinedata.WriteBatch(SharedContext.Connection, batch2File, pipelineGroup, baseTime.Add(seedOffset).Add(time.Second), time.Millisecond)
+ tracepipelinedata.WriteBatch(SharedContext.Connection, batch2File, PipelineGroup, baseTime.Add(seedOffset).Add(time.Second), time.Millisecond)
}
// AssertDropCandidatesVisible queries t-drop-1 and t-drop-2 and asserts BOTH are visible.
@@ -167,19 +171,19 @@ func RunSoak(duration time.Duration) {
// Batch A: one drop + one keep → first mem-part.
batchA := []tracepipelinedata.TraceRow{
- soakDropRow(dropID1, 100, "success"),
+ soakDropRow(dropID1, 100),
soakKeepRow(keepID1, 800, "success"),
}
// Batch B: one drop + one keep → second mem-part; triggers the filtering merge (max-merge-parts=2).
batchB := []tracepipelinedata.TraceRow{
- soakDropRow(dropID2, 300, "success"),
+ soakDropRow(dropID2, 300),
soakKeepRow(keepID2, 50, "error"),
}
// Each batch gets a slightly different base time so the two parts are distinct.
iterBase := baseTime.Add(-time.Duration(iteration) * time.Millisecond)
- tracepipelinedata.WriteBatchEntries(SharedContext.Connection, pipelineGroup, iterBase, time.Millisecond, batchA)
- tracepipelinedata.WriteBatchEntries(SharedContext.Connection, pipelineGroup, iterBase.Add(10*time.Millisecond), time.Millisecond, batchB)
+ tracepipelinedata.WriteBatchEntries(SharedContext.Connection, PipelineGroup, iterBase, time.Millisecond, batchA)
+ tracepipelinedata.WriteBatchEntries(SharedContext.Connection, PipelineGroup, iterBase.Add(10*time.Millisecond), time.Millisecond, batchB)
// Assert that the drop candidates for this iteration become absent after the merge.
assertSoakDropAbsent(dropID1, iterBase)
@@ -198,7 +202,7 @@ func RunSoak(duration time.Duration) {
}
// soakDropRow builds a TraceRow that the sampler will DROP (duration < 500, status == success).
-func soakDropRow(traceID string, durationMs int64, status string) tracepipelinedata.TraceRow {
+func soakDropRow(traceID string, durationMs int64) tracepipelinedata.TraceRow {
return tracepipelinedata.TraceRow{
Span: "span-" + traceID,
Tags: []*modelv1.TagValue{
@@ -206,7 +210,7 @@ func soakDropRow(traceID string, durationMs int64, status string) tracepipelined
{Value: &modelv1.TagValue_Str{Str: &modelv1.Str{Value: "span-" + traceID}}},
{Value: &modelv1.TagValue_Str{Str: &modelv1.Str{Value: "svc-" + traceID}}},
{Value: &modelv1.TagValue_Int{Int: &modelv1.Int{Value: durationMs}}},
- {Value: &modelv1.TagValue_Str{Str: &modelv1.Str{Value: status}}},
+ {Value: &modelv1.TagValue_Str{Str: &modelv1.Str{Value: "success"}}},
},
}
}
@@ -253,7 +257,7 @@ func querySoakByTraceID(innerGm gm.Gomega, conn *grpclib.ClientConn, traceID str
windowEnd := time.Now().Add(time.Second).Truncate(time.Millisecond)
req := &tracev1.QueryRequest{
- Groups: []string{pipelineGroup},
+ Groups: []string{PipelineGroup},
Name: "filter",
TimeRange: &modelv1.TimeRange{
Begin: timestamppb.New(windowStart),
@@ -309,3 +313,237 @@ func RegisterSoak(description string) bool {
})
})
}
+
+// soakChurnInterval is how often the soak loop rotates the sampler config
+// (Register → Update → Remove → Register …) during the dynamic soak.
+const soakChurnInterval = 15
+
+// soakPanicIteration is the iteration at which the soak injects the panicking
+// sampler to verify engine fail-open resilience. Must be > 0 and < the
+// total iteration count for any soak duration ≥ soakDefaultDuration.
+const soakPanicIteration = 5
+
+// soakHeapGrowthLimit is the maximum allowed growth in HeapInuse bytes from the
+// start of the dynamic soak to the end, as a fraction of the initial measurement.
+// 3× is conservative — a leak would produce orders-of-magnitude growth.
+const soakHeapGrowthLimit = 3.0
+
+// RunSoakDynamic runs the US-010 soak: continuous writes + periodic
+// Register/Update/Remove sampler churn + a mid-run panicking-sampler injection,
+// with goroutine-leak, no-merge-stall, and bounded-heap gates.
+//
+// Gates verified (per the US-010 ACs):
+// - No merge stall: after every churn phase, a drop-eligible trace eventually
+// becomes absent, proving merges are still running.
+// - Node survives panic: after the panicking sampler is injected, the node
+// is still reachable (GroupRegistryService.List succeeds) and traces are
+// retained (fail-open — no sampler active during panic).
+// - Bounded heap: the test-driver HeapInuse does not grow beyond
+// soakHeapGrowthLimit× its initial value.
+// - No goroutine leak: verified by the suite's ReportAfterSuite (suite-level
+// gleak.Goroutines snapshot taken in BeforeSuite); NOT re-checked here to
+// avoid double-accounting.
+// - No close-balance gate (R4): plugins are immortal; Close() is never called
+// on registry mutation.
+func RunSoakDynamic(conn *grpclib.ClientConn, group, soPath string, duration time.Duration) {
+ deadline := time.Now().Add(duration)
+ baseTime := time.Now().Add(-soakGracePad).Truncate(time.Millisecond)
+
+ // Snapshot initial test-driver heap for the bounded-heap gate.
+ var memBefore runtime.MemStats
+ runtime.ReadMemStats(&memBefore)
+ heapBefore := memBefore.HeapInuse
+ g.GinkgoWriter.Printf("[soak-dyn] start: HeapInuse=%d bytes, duration=%s\n", heapBefore, duration)
+
+ // Phase 0: ensure the base config (thresholdMs=500) is active before the loop.
+ regCtx, regCancel := context.WithTimeout(context.Background(), 30*time.Second)
+ defer regCancel()
+ RegisterSamplerRuntime(regCtx, conn, group, NewBasePipelineConfig(soPath, DefaultMergeGrace))
+
+ // churning tracks which config variant is currently active so the loop can
+ // rotate: base(500) → variant(200) → remove → base(500) → …
+ type configState int
+ const (
+ stateBase configState = iota // base thresholdMs=500
+ stateVariant // variant thresholdMs=200
+ stateRemoved // sampler removed
+ )
+ state := stateBase
+ panicInjected := false
+
+ var iteration int
+ for time.Now().Before(deadline) {
+ iteration++
+ iterBase := baseTime.Add(-time.Duration(iteration) * time.Millisecond)
+
+ // --- Churn phase: rotate config BEFORE writing this iteration's data, so the
+ // merge that processes those parts runs under the new config. This mirrors the
+ // instant dynamic specs (change config, THEN write the parts whose verdict it
+ // governs): writing first and churning after races the merge against async
+ // config convergence and can leave a part merged-and-retained under the old
+ // config — which is never re-filtered without a fresh merge trigger.
+ if iteration%soakChurnInterval == 0 {
+ churnCtx, churnCancel := context.WithTimeout(context.Background(), 30*time.Second)
+ switch state {
+ case stateBase:
+ // Rotate to variant (thresholdMs=200).
+ UpdateSamplerRuntime(churnCtx, conn, group,
+ NewVariantPipelineConfig(soPath, DefaultMergeGrace))
+ state = stateVariant
+ g.GinkgoWriter.Printf("[soak-dyn] iteration %d: rotated to variant (thresholdMs=200)\n", iteration)
+ case stateVariant:
+ // Remove sampler.
+ RemoveSamplerRuntime(churnCtx, conn, group)
+ state = stateRemoved
+ g.GinkgoWriter.Printf("[soak-dyn] iteration %d: sampler removed\n", iteration)
+ case stateRemoved:
+ // Re-register base config.
+ RegisterSamplerRuntime(churnCtx, conn, group,
+ NewBasePipelineConfig(soPath, DefaultMergeGrace))
+ state = stateBase
+ g.GinkgoWriter.Printf("[soak-dyn] iteration %d: sampler re-registered (base)\n", iteration)
+ }
+ churnCancel()
+ }
+
+ // --- Write phase: two batches to trigger a merge ---
+ dropID := fmt.Sprintf("soak-dyn-drop-iter%d", iteration)
+ keepID := fmt.Sprintf("soak-dyn-keep-iter%d", iteration)
+
+ batchA := []tracepipelinedata.TraceRow{
+ soakDropRow(dropID, 100),
+ soakKeepRow(keepID, 800, "success"),
+ }
+ batchB := []tracepipelinedata.TraceRow{
+ soakDropRow(fmt.Sprintf("soak-dyn-drop2-iter%d", iteration), 300),
+ soakKeepRow(fmt.Sprintf("soak-dyn-keep2-iter%d", iteration), 50, "error"),
+ }
+ tracepipelinedata.WriteBatchEntries(conn, group, iterBase, time.Millisecond, batchA)
+ tracepipelinedata.WriteBatchEntries(conn, group, iterBase.Add(10*time.Millisecond), time.Millisecond, batchB)
+
+ // --- Panic injection: mid-run, once, before churn rotation ---
+ if !panicInjected && iteration == soakPanicIteration {
+ g.GinkgoWriter.Printf("[soak-dyn] iteration %d: injecting panicking sampler\n", iteration)
+ panicCtx, panicCancel := context.WithTimeout(context.Background(), 30*time.Second)
+ RegisterSamplerRuntime(panicCtx, conn, group,
+ NewPanicPipelineConfig(soPath, DefaultMergeGrace))
+ panicCancel()
+ panicInjected = true
+
+ // Write a trace that the base sampler would drop (dur=100 success) but
+ // the panicking sampler must retain (fail-open on panic → no sampler →
+ // retain all).
+ panicBase := iterBase.Add(-50 * time.Millisecond)
+ panicRetainID := fmt.Sprintf("soak-dyn-panic-retain-iter%d", iteration)
+ panicKeepID := fmt.Sprintf("soak-dyn-panic-keep-iter%d", iteration)
+ tracepipelinedata.WriteBatchEntries(conn, group, panicBase,
+ time.Millisecond, []tracepipelinedata.TraceRow{
+ soakDropRow(panicRetainID, 100),
+ soakKeepRow(panicKeepID, 900, "success"),
+ })
+ tracepipelinedata.WriteBatchEntries(conn, group, panicBase.Add(10*time.Millisecond),
+ time.Millisecond, []tracepipelinedata.TraceRow{
+ soakKeepRow(fmt.Sprintf("soak-dyn-panic-companion-iter%d", iteration), 600, "error"),
+ soakKeepRow(fmt.Sprintf("soak-dyn-panic-comp2-iter%d", iteration), 700, "error"),
+ })
+
+ // Assert node is still alive (fail-open: no crash from the panic).
+ dbc := databasev1.NewGroupRegistryServiceClient(conn)
+ aliveCtx, aliveCancel := context.WithTimeout(context.Background(), 10*time.Second)
+ _, aliveErr := dbc.List(aliveCtx, &databasev1.GroupRegistryServiceListRequest{})
+ aliveCancel()
+ gm.Expect(aliveErr).NotTo(gm.HaveOccurred(),
+ "node must remain reachable after panicking sampler injection (fail-open)")
+
+ // Assert the panic-mode trace is retained (fail-open: sampler panics → retain all).
+ assertSoakKeepPresent(panicRetainID, panicBase)
+ assertSoakKeepPresent(panicKeepID, panicBase)
+
+ // Restore the base config so subsequent churn iterations are meaningful.
+ restoreCtx, restoreCancel := context.WithTimeout(context.Background(), 30*time.Second)
+ RegisterSamplerRuntime(restoreCtx, conn, group,
+ NewBasePipelineConfig(soPath, DefaultMergeGrace))
+ restoreCancel()
+ state = stateBase
+
+ g.GinkgoWriter.Printf("[soak-dyn] iteration %d: panicking-sampler injection passed; base config restored\n", iteration)
+ continue
+ }
+
+ // --- Verify phase: assert drop/keep verdicts match the active state ---
+ // No-merge-stall gate: a drop-eligible trace eventually becomes absent,
+ // proving the merge pipeline is running (not stalled by the panic or churn).
+ switch state {
+ case stateBase:
+ // thresholdMs=500: dur=100 success → DROP; dur=800 success → KEEP.
+ assertSoakDropAbsent(dropID, iterBase)
+ assertSoakKeepPresent(keepID, iterBase)
+ case stateVariant:
+ // thresholdMs=200: dur=100 success → DROP; dur=800 success → KEEP.
+ assertSoakDropAbsent(dropID, iterBase)
+ assertSoakKeepPresent(keepID, iterBase)
+ case stateRemoved:
+ // No sampler: all traces retained (fail-safe).
+ assertSoakKeepPresent(dropID, iterBase)
+ assertSoakKeepPresent(keepID, iterBase)
+ }
+
+ if iteration%10 == 0 {
+ g.GinkgoWriter.Printf("[soak-dyn] iteration %d complete, %.1fs remaining\n",
+ iteration, time.Until(deadline).Seconds())
+ }
+ }
+
+ // Bounded-heap gate: test-driver HeapInuse must not grow beyond soakHeapGrowthLimit×.
+ var memAfter runtime.MemStats
+ runtime.ReadMemStats(&memAfter)
+ heapAfter := memAfter.HeapInuse
+ g.GinkgoWriter.Printf("[soak-dyn] end: HeapInuse=%d bytes after %d iterations\n", heapAfter, iteration)
+ if heapBefore > 0 {
+ growthFactor := float64(heapAfter) / float64(heapBefore)
+ gm.Expect(growthFactor).To(gm.BeNumerically("<=", soakHeapGrowthLimit),
+ "test-driver heap grew %.1fx (limit %.1fx): possible gRPC resource leak in soak loop",
+ growthFactor, soakHeapGrowthLimit)
+ }
+
+ g.GinkgoWriter.Printf("[soak-dyn] completed %d iterations in %s\n", iteration, duration)
+}
+
+// RegisterSoakDynamic registers the US-010 dynamic soak spec under the given
+// description. The spec is gated by TRACE_PIPELINE_SOAK and is Serial (it
+// mutates the group's pipeline config and must not race with other specs).
+//
+// The soak performs:
+// - Continuous writes + merges with drop/keep verification.
+// - Periodic Register/Update/Remove churn (every soakChurnInterval iterations).
+// - Mid-run panicking-sampler injection (iteration soakPanicIteration): asserts
+// the node survives (fail-open), traces are retained, merges are not stalled.
+// - Bounded test-driver heap growth gate (soakHeapGrowthLimit×).
+// - No close-balance gate (R4): plugins are immortal.
+//
+// soPath is a provider resolved at spec-run time, not a bare string: the
+// trusted-dir-relative .so path is only assigned by the suite's
+// SynchronizedBeforeSuite, which runs AFTER this registration's arguments are
+// evaluated at package-init time. Capturing the string here would freeze its
+// empty zero value and fail validation (SamplerPlugin.Path must be non-empty).
+func RegisterSoakDynamic(description string, soPath func() string) bool {
+ return g.Describe(description, g.Serial, func() {
+ g.It("soak-dynamic: write+churn+panic loop (US-010)", func() {
+ if os.Getenv("TRACE_PIPELINE_SOAK") == "" {
+ g.Skip("set TRACE_PIPELINE_SOAK=1 to run the dynamic soak (opt-in only)")
+ return
+ }
+ duration := soakDefaultDuration
+ if raw := os.Getenv("TRACE_PIPELINE_SOAK_DURATION"); raw != "" {
+ parsed, parseErr := time.ParseDuration(raw)
+ if parseErr != nil {
+ g.Fail(fmt.Sprintf("invalid TRACE_PIPELINE_SOAK_DURATION %q: %v", raw, parseErr))
+ return
+ }
+ duration = parsed
+ }
+ g.GinkgoWriter.Printf("[soak-dyn] starting dynamic soak for %s\n", duration)
+ RunSoakDynamic(SharedContext.Connection, PipelineGroup, soPath(), duration)
+ })
+ })
+}
diff --git a/test/integration/distributed/pipeline/common.go b/test/integration/distributed/pipeline/common.go
index 849281d27..7856d0407 100644
--- a/test/integration/distributed/pipeline/common.go
+++ b/test/integration/distributed/pipeline/common.go
@@ -33,12 +33,9 @@ import (
"github.com/onsi/gomega/gleak"
"google.golang.org/grpc"
"google.golang.org/grpc/credentials/insecure"
- "google.golang.org/protobuf/encoding/protojson"
- "google.golang.org/protobuf/types/known/structpb"
commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
- pipelinev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/pipeline/v1"
"github.com/apache/skywalking-banyandb/banyand/metadata/schema"
"github.com/apache/skywalking-banyandb/pkg/grpchelper"
"github.com/apache/skywalking-banyandb/pkg/logger"
@@ -59,6 +56,44 @@ var (
goods []gleak.Goroutine
stopFunc func()
now time.Time
+
+ // clusterBinPath is the banyand binary used to boot all external nodes.
+ // Exposed for the Restart + LateJoin specs in dynamic_test.go.
+ clusterBinPath string
+
+ // clusterTrustedDir is the trusted-plugin directory shared by both data nodes.
+ // Exposed for dynamic specs that relaunch or add nodes.
+ clusterTrustedDir string
+
+ // clusterSoPath is the bare (trusted-dir-relative) filename of the staged .so.
+ // The validator requires a relative path; the data node resolves it at load time.
+ clusterSoPath string
+
+ // clusterConfig is the PropertyClusterConfig used for the cluster.
+ // Exposed so dynamic specs can launch additional nodes into the same cluster.
+ clusterConfig *setup.ClusterConfig
+
+ // clusterLogDir is the shared log directory for all external processes.
+ clusterLogDir string
+
+ // clusterDN0Ports is the port slice for data node 0. Exposed for the Restart spec.
+ clusterDN0Ports []int
+
+ // closeLiaisonNode is the teardown for the liaison process.
+ closeLiaisonNode func()
+
+ // closeDataNode0 / closeDataNode1 are per-node teardowns exposed so
+ // the Restart spec can stop and relaunch individual nodes.
+ closeDataNode0 func()
+ closeDataNode1 func()
+
+ // clusterDataDir0 is data node 0's data directory.
+ // Preserved across restarts so the node replays the schema store.
+ clusterDataDir0 string
+
+ // liaisonGRPCAddr is the liaison's gRPC address, used by dynamic specs that
+ // need a direct liaison connection for assertions.
+ liaisonGRPCAddr string
)
var _ = ginkgo.SynchronizedBeforeSuite(func() []byte {
@@ -88,12 +123,8 @@ var _ = ginkgo.SynchronizedBeforeSuite(func() []byte {
gomega.Expect(trustedErr).NotTo(gomega.HaveOccurred())
// Copy the .so into the trusted dir under its bare filename.
- soPath := filepath.Join(trustedDir, "latencystatussampler.so")
- gomega.Expect(copyFile(pluginSrc, soPath)).To(gomega.Succeed())
-
- // Write the TracePipelineConfig protojson file.
- configFile := filepath.Join(trustedDir, "pipeline.json")
- gomega.Expect(writePipelineConfig(configFile)).To(gomega.Succeed())
+ soFilePath := filepath.Join(trustedDir, tracepipeline.PluginSOName)
+ gomega.Expect(copyFile(pluginSrc, soFilePath)).To(gomega.Succeed())
// Discovery file writer and property cluster config.
tmpDir, tmpDirCleanup, tmpErr := test.NewSpace()
@@ -126,7 +157,7 @@ var _ = ginkgo.SynchronizedBeforeSuite(func() []byte {
// Start data node 0 with schema server (dn0Ports[3]) and pipeline plugin flags.
ginkgo.By("Starting data node 0 with pipeline plugin")
- _, closeDataNode0 := setup.ExternalDataNode(
+ _, dn0Close := setup.ExternalDataNode(
config,
binPath,
dataDir0,
@@ -134,7 +165,6 @@ var _ = ginkgo.SynchronizedBeforeSuite(func() []byte {
dn0Ports,
"--trace-pipeline-native-plugin-enabled=true",
"--trace-pipeline-trusted-plugin-dir="+trustedDir,
- "--trace-pipeline-config="+configFile,
"--trace-pipeline-merge-grace-default=0",
"--trace-max-merge-parts=2",
"--trace-flush-timeout=500ms",
@@ -142,7 +172,7 @@ var _ = ginkgo.SynchronizedBeforeSuite(func() []byte {
// Start data node 1 (no schema server) with pipeline plugin flags.
ginkgo.By("Starting data node 1 with pipeline plugin")
- _, closeDataNode1 := setup.ExternalDataNode(
+ _, dn1Close := setup.ExternalDataNode(
config,
binPath,
dataDir1,
@@ -150,7 +180,6 @@ var _ = ginkgo.SynchronizedBeforeSuite(func() []byte {
dn1Ports,
"--trace-pipeline-native-plugin-enabled=true",
"--trace-pipeline-trusted-plugin-dir="+trustedDir,
- "--trace-pipeline-config="+configFile,
"--trace-pipeline-merge-grace-default=0",
"--trace-max-merge-parts=2",
"--trace-flush-timeout=500ms",
@@ -164,7 +193,7 @@ var _ = ginkgo.SynchronizedBeforeSuite(func() []byte {
// Start liaison node — NO plugin flags (liaison has no merger).
ginkgo.By("Starting liaison node")
- liaisonGRPCAddr, closeLiaisonNode := setup.ExternalLiaisonNode(
+ lnGRPCAddr, lnClose := setup.ExternalLiaisonNode(
config,
binPath,
liaisonDir,
@@ -175,6 +204,19 @@ var _ = ginkgo.SynchronizedBeforeSuite(func() []byte {
ns := timestamp.NowMilli().UnixNano()
now = time.Unix(0, ns-ns%int64(time.Minute))
+ // Populate package-level vars for dynamic specs (dynamic_test.go).
+ clusterBinPath = binPath
+ clusterTrustedDir = trustedDir
+ clusterSoPath = tracepipeline.PluginSOName
+ clusterConfig = config
+ clusterLogDir = logDir
+ clusterDN0Ports = dn0Ports
+ clusterDataDir0 = dataDir0
+ liaisonGRPCAddr = lnGRPCAddr
+ closeLiaisonNode = lnClose
+ closeDataNode0 = dn0Close
+ closeDataNode1 = dn1Close
+
stopFunc = func() {
closeLiaisonNode()
closeDataNode1()
@@ -186,9 +228,25 @@ var _ = ginkgo.SynchronizedBeforeSuite(func() []byte {
// with active liaison + data nodes, and the filter trace schema queryable.
// This mirrors the logic in setup.waitForActiveDataNodes (setup.go:857).
ginkgo.By("Waiting for cluster to be ready")
- waitForClusterReady(liaisonGRPCAddr)
+ waitForClusterReady(lnGRPCAddr)
+
+ // Activate the sampler pipeline dynamically via UpdateGroup (US-CLEANUP).
+ // The KindGroup watch fires on BOTH data nodes; both load the .so plugin from
+ // their shared trusted dir and register the sampler. The liaison's schemaRepo
+ // is ROLE_LIAISON with nativePipelineEnabled=false, so its reconcilePipeline
+ // branch is never reached (doubly-gated: role check + flag check).
+ ginkgo.By("Activating sampler pipeline dynamically via UpdateGroup")
+ activateCtx, activateCancel := context.WithTimeout(context.Background(), 30*time.Second)
+ defer activateCancel()
+ activateConn, activateConnErr := grpchelper.Conn(lnGRPCAddr, 10*time.Second,
+ grpc.WithTransportCredentials(insecure.NewCredentials()))
+ gomega.Expect(activateConnErr).NotTo(gomega.HaveOccurred())
+ defer func() { _ = activateConn.Close() }()
+ tracepipeline.RegisterSamplerRuntime(activateCtx, activateConn,
+ tracepipeline.PipelineGroup,
+ tracepipeline.NewBasePipelineConfig(tracepipeline.PluginSOName, tracepipeline.DefaultMergeGrace))
- return []byte(liaisonGRPCAddr)
+ return []byte(lnGRPCAddr)
}, func(address []byte) {
var connErr error
connection, connErr = grpchelper.Conn(string(address), 10*time.Second,
@@ -205,7 +263,11 @@ var _ = ginkgo.SynchronizedAfterSuite(func() {
if connection != nil {
gomega.Expect(connection.Close()).To(gomega.Succeed())
}
-}, func() {})
+}, func() {
+ if stopFunc != nil {
+ stopFunc()
+ }
+})
var _ = ginkgo.ReportAfterSuite("Distributed Trace Pipeline Integration Suite", func(report ginkgo.Report) {
if report.SuiteSucceeded {
@@ -291,48 +353,6 @@ func copyFile(src, dst string) error {
return nil
}
-// writePipelineConfig marshals a TracePipelineConfig (protojson) to cfgPath.
-// The plugin path uses only the bare filename; the server resolves it against
-// the trusted dir passed via --trace-pipeline-trusted-plugin-dir.
-func writePipelineConfig(cfgPath string) error {
- pluginConfig, structErr := structpb.NewStruct(map[string]interface{}{
- "thresholdMs": float64(500),
- "successValue": "success",
- })
- if structErr != nil {
- return fmt.Errorf("build plugin config struct: %w", structErr)
- }
-
- cfg := &pipelinev1.TracePipelineConfig{
- Metadata: &commonv1.Metadata{
- Group: "test-trace-pipeline",
- },
- Enabled: true,
- EnabledEvents: []pipelinev1.PipelineEvent{
- pipelinev1.PipelineEvent_PIPELINE_EVENT_MERGE,
- },
- Plugins: []*pipelinev1.Plugin{
- {
- Name: "latencystatussampler",
- Kind: &pipelinev1.Plugin_Sampler{
- Sampler: &pipelinev1.SamplerPlugin{
- Path: "latencystatussampler.so",
- Symbol: "NewSampler",
- AbiVersion: 1,
- Config: pluginConfig,
- },
- },
- },
- },
- }
-
- data, marshalErr := protojson.Marshal(cfg)
- if marshalErr != nil {
- return fmt.Errorf("marshal pipeline config: %w", marshalErr)
- }
- return os.WriteFile(cfgPath, data, 0o600)
-}
-
// waitForClusterReady polls the liaison until the cluster is fully ready to accept
// writes. It mirrors setup.waitForActiveDataNodes (setup.go:857):
// 1. Groups visible in liaison schema registry.
diff --git a/test/integration/distributed/pipeline/dynamic_test.go b/test/integration/distributed/pipeline/dynamic_test.go
new file mode 100644
index 000000000..b28db688a
--- /dev/null
+++ b/test/integration/distributed/pipeline/dynamic_test.go
@@ -0,0 +1,505 @@
+// 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.
+
+//go:build trace_pipeline
+
+package pipeline_test
+
+import (
+ "context"
+ "fmt"
+ "time"
+
+ "github.com/onsi/ginkgo/v2"
+ "github.com/onsi/gomega"
+ "google.golang.org/grpc"
+ "google.golang.org/grpc/credentials/insecure"
+ "google.golang.org/protobuf/types/known/timestamppb"
+
+ commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
+ databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+ modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
+ tracev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/trace/v1"
+ "github.com/apache/skywalking-banyandb/pkg/grpchelper"
+ "github.com/apache/skywalking-banyandb/pkg/test"
+ "github.com/apache/skywalking-banyandb/pkg/test/flags"
+ "github.com/apache/skywalking-banyandb/pkg/test/helpers"
+ "github.com/apache/skywalking-banyandb/pkg/test/setup"
+ tracepipeline "github.com/apache/skywalking-banyandb/test/cases/tracepipeline"
+ tracepipelinedata "github.com/apache/skywalking-banyandb/test/cases/tracepipeline/data"
+)
+
+// dynSeedOffset places timestamps well in the past so isMergeHot (grace=0) returns false.
+const dynSeedOffset = -2 * time.Hour
+
+// queryByTraceID queries the "filter" trace for a single trace_id over a broad window.
+func queryByTraceID(innerGm gomega.Gomega, conn *grpc.ClientConn, traceID string, baseTime time.Time) *tracev1.QueryResponse {
+ windowStart := baseTime.Add(dynSeedOffset - time.Minute).Truncate(time.Millisecond)
+ windowEnd := time.Now().Add(time.Second).Truncate(time.Millisecond)
+
+ req := &tracev1.QueryRequest{
+ Groups: []string{tracepipeline.PipelineGroup},
+ Name: "filter",
+ TimeRange: &modelv1.TimeRange{
+ Begin: timestamppb.New(windowStart),
+ End: timestamppb.New(windowEnd),
+ },
+ Criteria: &modelv1.Criteria{
+ Exp: &modelv1.Criteria_Condition{
+ Condition: &modelv1.Condition{
+ Name: "trace_id",
+ Op: modelv1.Condition_BINARY_OP_EQ,
+ Value: &modelv1.TagValue{
+ Value: &modelv1.TagValue_Str{Str: &modelv1.Str{Value: traceID}},
+ },
+ },
+ },
+ },
+ Limit: 10,
+ }
+
+ c := tracev1.NewTraceServiceClient(conn)
+ ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+ defer cancel()
+ resp, queryErr := c.Query(ctx, req)
+ innerGm.Expect(queryErr).NotTo(gomega.HaveOccurred())
+ return resp
+}
+
+// assertTraceAbsent asserts the named trace_id is absent (dropped by sampler) via the liaison.
+func assertTraceAbsent(conn *grpc.ClientConn, traceID string, baseTime time.Time) {
+ gomega.Eventually(func(innerGm gomega.Gomega) {
+ resp := queryByTraceID(innerGm, conn, traceID, baseTime)
+ innerGm.Expect(resp.GetTraces()).To(gomega.BeEmpty(),
+ "trace %q should be absent (filtered by sampler)", traceID)
+ }, flags.EventuallyTimeout, 500*time.Millisecond).Should(gomega.Succeed())
+}
+
+// assertTracePresent asserts the named trace_id is present (kept / not dropped).
+func assertTracePresent(conn *grpc.ClientConn, traceID string, baseTime time.Time) {
+ gomega.Eventually(func(innerGm gomega.Gomega) {
+ resp := queryByTraceID(innerGm, conn, traceID, baseTime)
+ innerGm.Expect(resp.GetTraces()).NotTo(gomega.BeEmpty(),
+ "trace %q should be present (not filtered)", traceID)
+ }, flags.EventuallyTimeout, 500*time.Millisecond).Should(gomega.Succeed())
+}
+
+// makeDynRow builds a TraceRow for dynamic specs.
+// Tags: trace_id, span_id, service_id, duration, status.
+func makeDynRow(traceID string, durationMs int64, status string) tracepipelinedata.TraceRow {
+ return tracepipelinedata.TraceRow{
+ Span: "span-" + traceID,
+ Tags: []*modelv1.TagValue{
+ {Value: &modelv1.TagValue_Str{Str: &modelv1.Str{Value: traceID}}},
+ {Value: &modelv1.TagValue_Str{Str: &modelv1.Str{Value: "span-" + traceID}}},
+ {Value: &modelv1.TagValue_Str{Str: &modelv1.Str{Value: "svc-dyn"}}},
+ {Value: &modelv1.TagValue_Int{Int: &modelv1.Int{Value: durationMs}}},
+ {Value: &modelv1.TagValue_Str{Str: &modelv1.Str{Value: status}}},
+ },
+ }
+}
+
+// writeTwoPartMerge writes two batches via the liaison, triggering a filtering merge
+// on both data nodes (max-merge-parts=2, grace=0). rowA lands as part-1; once it is
+// visible (one part, no merge), rowB is written as part-2 to trigger the merge.
+func writeTwoPartMerge(
+ conn *grpc.ClientConn,
+ rowA tracepipelinedata.TraceRow, rowB tracepipelinedata.TraceRow,
+ baseTime time.Time,
+) {
+ tracepipelinedata.WriteBatchEntries(conn, tracepipeline.PipelineGroup,
+ baseTime, time.Millisecond, []tracepipelinedata.TraceRow{rowA})
+
+ // Wait for part-1 to be visible (proves the server received it before merge).
+ traceIDA := rowA.Tags[0].GetStr().GetValue()
+ gomega.Eventually(func(innerGm gomega.Gomega) {
+ resp := queryByTraceID(innerGm, conn, traceIDA, baseTime)
+ innerGm.Expect(resp.GetTraces()).NotTo(gomega.BeEmpty(),
+ "part-1 trace %q must be visible before merge", traceIDA)
+ }, flags.EventuallyTimeout, 500*time.Millisecond).Should(gomega.Succeed())
+
+ // Part 2 triggers the merge.
+ tracepipelinedata.WriteBatchEntries(conn, tracepipeline.PipelineGroup,
+ baseTime.Add(10*time.Millisecond), time.Millisecond, []tracepipelinedata.TraceRow{rowB})
+}
+
+// Dynamic sampler registration lifecycle specs for the distributed cluster (US-009).
+//
+// The base config (thresholdMs=500) was activated in common.go's BeforeSuite
+// via RegisterSamplerRuntime. The RegisterMergeFilterTable Describe in common.go
+// verifies base drop/keep on the initial two-part merge. The specs here are Ordered:
+//
+// 1. Update (variant thresholdMs=200): both nodes converge on new threshold.
+// 2. Remove: both healthy nodes stop dropping within merge_grace + one merge cycle
+// (Correction-C measurement AC; no nudge).
+// 3. InvalidConfig: missing .so → fail-open, both nodes alive, traces retained.
+// 4. Restart: per-node stop+relaunch → config re-applied from schema store.
+// 5. LateJoin: a new data node joining the cluster converges via syncLoop.
+// 6. LiaisonNoLoad: the liaison's sampler-load path is unreachable (structural gate).
+var _ = ginkgo.Describe("Dynamic sampler registration (distributed)", ginkgo.Ordered, func() {
+ // dynBase anchors all dynamic-spec timestamps in the past (grace=0 → isMergeHot=false).
+ var dynBase time.Time
+
+ ginkgo.BeforeAll(func() {
+ dynBase = now
+ })
+
+ // --- Step 2: Update (variant thresholdMs=200) ---
+ //
+ // Sampler decision: drop if dur < thresholdMs AND status == "success".
+ // base (500): dur=300 success → DROP.
+ // variant (200): dur=300 success → KEEP (300 ≥ 200).
+ // Both nodes must converge on the new threshold; the pivot trace (dur=300
+ // success) must be KEPT in the merge that fires after the Update.
+ ginkgo.Describe("Update: variant threshold (thresholdMs=200) — both nodes converge", ginkgo.Ordered, func() {
+ const (
+ pivotID = "dist-upd-pivot-300ms"
+ bothDropID = "dist-upd-both-drop-100ms"
+ )
+
+ ginkgo.BeforeAll(func() {
+ ctx := context.Background()
+
+ // Switch to variant config (thresholdMs=200) via UpdateSamplerRuntime.
+ upCtx, upCancel := context.WithTimeout(ctx, 30*time.Second)
+ defer upCancel()
+ tracepipeline.UpdateSamplerRuntime(upCtx, connection,
+ tracepipeline.PipelineGroup,
+ tracepipeline.NewVariantPipelineConfig(clusterSoPath, tracepipeline.DefaultMergeGrace))
+
+ // Pivot trace: dur=300 success + keep companion → two-part merge.
+ // Under variant (200): 300 ≥ 200 → KEPT.
+ writeTwoPartMerge(connection,
+ makeDynRow(pivotID, 300, "success"),
+ makeDynRow("dist-upd-keep-companion1", 900, "success"),
+ dynBase.Add(dynSeedOffset))
+
+ // Both-drop trace: dur=100 success + keep companion → two-part merge.
+ // Under variant (200): 100 < 200 → DROPPED.
+ writeTwoPartMerge(connection,
+ makeDynRow(bothDropID, 100, "success"),
+ makeDynRow("dist-upd-keep-companion2", 800, "error"),
+ dynBase.Add(dynSeedOffset-time.Second))
+ })
+
+ ginkgo.It("dur=300 success: KEPT under variant threshold=200 (both nodes converge)", func() {
+ assertTracePresent(connection, pivotID, dynBase)
+ })
+
+ ginkgo.It("dur=100 success: still DROPPED under variant (100 < 200)", func() {
+ assertTraceAbsent(connection, bothDropID, dynBase)
+ })
+ })
+
+ // --- Step 3: Remove → stop dropping (Correction-C measurement AC) ---
+ //
+ // After RemoveSamplerRuntime both healthy nodes must stop dropping within
+ // merge_grace + one merge cycle. With grace=0 this is the very next merge pass.
+ // No nudge is used (Correction C); STOP latency is the inherited schema-sync
+ // propagation (WatchSchemas push stream) + the merge-cool floor.
+ ginkgo.Describe("Remove: both healthy nodes stop dropping within merge_grace+one cycle", ginkgo.Ordered, func() {
+ const retainedID = "dist-rem-retained-100ms"
+
+ ginkgo.BeforeAll(func() {
+ ctx := context.Background()
+
+ // Remove the sampler — registry becomes empty on both data nodes.
+ rmCtx, rmCancel := context.WithTimeout(ctx, 30*time.Second)
+ defer rmCancel()
+ tracepipeline.RemoveSamplerRuntime(rmCtx, connection, tracepipeline.PipelineGroup)
+
+ // Write a trace that WOULD be dropped by the variant (dur=100, success).
+ // After removal, no sampler is active → trace must be retained.
+ writeTwoPartMerge(connection,
+ makeDynRow(retainedID, 100, "success"),
+ makeDynRow("dist-rem-keep-companion", 800, "error"),
+ dynBase.Add(dynSeedOffset-2*time.Second))
+ })
+
+ ginkgo.It("previously-dropped shape (dur=100 success) is RETAINED on both nodes after remove", func() {
+ // Correction-C measurement AC: STOP latency ≤ merge_grace + one merge cycle.
+ // With grace=0 the trace must be present after the first merge pass
+ // following removal. We assert eventual presence (no nudge, no explicit
+ // time bound beyond EventuallyTimeout which exceeds the merge cycle).
+ assertTracePresent(connection, retainedID, dynBase)
+ })
+ })
+
+ // --- Step 4: InvalidConfig → fail-open, both nodes alive, traces retained ---
+ //
+ // US-009 AC(c): an invalid config (missing .so) leaves the prior set (empty after
+ // Remove) active on both data nodes, retaining all traces. Both nodes stay alive.
+ ginkgo.Describe("InvalidConfig: missing .so → fail-open, both nodes alive, traces retained", ginkgo.Ordered, func() {
+ const invalidRetainedID = "dist-invalid-retained-100ms"
+
+ ginkgo.BeforeAll(func() {
+ ctx := context.Background()
+
+ // Register with a .so path that does not exist in the trusted dir.
+ // Both data nodes must fail-open: keep prior set (empty), retain traces.
+ badCfg := tracepipeline.NewBasePipelineConfig(
+ "nonexistent_sampler_dist.so",
+ tracepipeline.DefaultMergeGrace,
+ )
+ badCtx, badCancel := context.WithTimeout(ctx, 30*time.Second)
+ defer badCancel()
+ tracepipeline.RegisterSamplerRuntime(badCtx, connection,
+ tracepipeline.PipelineGroup, badCfg)
+
+ // Write a trace that the base sampler would drop (dur=100, success).
+ // Because load failed (prior set = empty), no sampler active → retained.
+ writeTwoPartMerge(connection,
+ makeDynRow(invalidRetainedID, 100, "success"),
+ makeDynRow("dist-invalid-keep-companion", 800, "success"),
+ dynBase.Add(dynSeedOffset-3*time.Second))
+ })
+
+ ginkgo.It("both nodes alive after invalid config (GroupRegistryService.List succeeds)", func() {
+ c := databasev1.NewGroupRegistryServiceClient(connection)
+ ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+ defer cancel()
+ resp, listErr := c.List(ctx, &databasev1.GroupRegistryServiceListRequest{})
+ gomega.Expect(listErr).NotTo(gomega.HaveOccurred(),
+ "cluster must remain queryable after invalid config load on data nodes")
+ gomega.Expect(resp.GetGroup()).NotTo(gomega.BeEmpty(),
+ "at least the pipeline group must be listed")
+ })
+
+ ginkgo.It("traces are retained (fail-open) when .so load fails on both nodes", func() {
+ assertTracePresent(connection, invalidRetainedID, dynBase)
+ })
+ })
+
+ // --- Step 5: Per-node restart → config re-applied from schema store ---
+ //
+ // US-009 AC(d): each restarted data node re-applies pipeline config via the
+ // initial KindGroup syncLoop reconcile on boot — no re-write needed.
+ // We restart data node 0 (one at a time), assert it rejoins and drops correctly.
+ ginkgo.Describe("Restart: data node 0 re-applies config from schema store", ginkgo.Ordered, func() {
+ const (
+ dropAfterRestartID = "dist-restart-drop-100ms"
+ keepAfterRestartID = "dist-restart-keep-900ms"
+ )
+
+ var restartConn *grpc.ClientConn
+
+ ginkgo.BeforeAll(func() {
+ // First restore a valid sampler config (the InvalidConfig step left an empty
+ // set via a failed load). Re-register the base config so restart has something
+ // to replay.
+ regCtx, regCancel := context.WithTimeout(context.Background(), 30*time.Second)
+ defer regCancel()
+ tracepipeline.RegisterSamplerRuntime(regCtx, connection,
+ tracepipeline.PipelineGroup,
+ tracepipeline.NewBasePipelineConfig(clusterSoPath, tracepipeline.DefaultMergeGrace))
+
+ // Stop data node 0, then relaunch it from its existing data dir.
+ // The restarted node must re-apply config from the schema store without
+ // a second RegisterSamplerRuntime call.
+ closeDataNode0()
+
+ // Build the same flags used at first boot.
+ dn0RelaunchFlags := append(
+ setup.BuildDataNodeFlags(clusterConfig, clusterDataDir0, clusterDN0Ports),
+ "--trace-pipeline-native-plugin-enabled=true",
+ "--trace-pipeline-trusted-plugin-dir="+clusterTrustedDir,
+ "--trace-pipeline-merge-grace-default=0",
+ "--trace-max-merge-parts=2",
+ "--trace-flush-timeout=500ms",
+ )
+ dn0LogPath := clusterLogDir + fmt.Sprintf("/data-%d-restart.log", clusterDN0Ports[0])
+ newDN0Close, launchErr := setup.ExternalCMD(clusterBinPath, dn0LogPath,
+ append([]string{"data"}, dn0RelaunchFlags...)...)
+ gomega.Expect(launchErr).NotTo(gomega.HaveOccurred())
+ closeDataNode0 = newDN0Close
+
+ // Wait for the restarted node to pass gRPC health check.
+ dn0GRPCAddr := fmt.Sprintf("localhost:%d", clusterDN0Ports[0])
+ gomega.Eventually(
+ helpers.HealthCheck(dn0GRPCAddr, 10*time.Second, 10*time.Second,
+ grpc.WithTransportCredentials(insecure.NewCredentials())),
+ flags.EventuallyTimeout).Should(gomega.Succeed())
+
+ // Connect directly to data node 0 to write restart-specific traces.
+ // The node re-applies config from the schema store via syncLoop; once
+ // schema is visible its sampler registry is active.
+ var connErr error
+ restartConn, connErr = grpchelper.Conn(dn0GRPCAddr, 10*time.Second,
+ grpc.WithTransportCredentials(insecure.NewCredentials()))
+ gomega.Expect(connErr).NotTo(gomega.HaveOccurred())
+
+ // Wait for the trace schema to be visible on the restarted node.
+ waitForTraceSchemaOnAddr(dn0GRPCAddr)
+
+ // Write test traces via the liaison (both nodes get them).
+ writeTwoPartMerge(connection,
+ makeDynRow(dropAfterRestartID, 100, "success"),
+ makeDynRow(keepAfterRestartID, 900, "success"),
+ dynBase.Add(dynSeedOffset-4*time.Second))
+ })
+
+ ginkgo.AfterAll(func() {
+ if restartConn != nil {
+ _ = restartConn.Close()
+ }
+ })
+
+ ginkgo.It("drop-eligible trace is DROPPED after node 0 restart (sampler from store)", func() {
+ assertTraceAbsent(connection, dropAfterRestartID, dynBase)
+ })
+
+ ginkgo.It("keep-eligible trace is PRESENT after node 0 restart", func() {
+ assertTracePresent(connection, keepAfterRestartID, dynBase)
+ })
+ })
+
+ // --- Step 6: Late-join convergence ---
+ //
+ // US-009 AC(d): a new data node added to the cluster after the config was
+ // written converges via the syncLoop reconcile (30s backstop) or the
+ // WatchSchemas push stream. We boot a third data node, wait for schema sync,
+ // then assert it drops traces correctly.
+ ginkgo.Describe("LateJoin: new data node converges via schema-sync", ginkgo.Ordered, func() {
+ const (
+ lateDropID = "dist-latejoin-drop-100ms"
+ lateKeepID = "dist-latejoin-keep-900ms"
+ )
+
+ var (
+ lateJoinConn *grpc.ClientConn
+ closeLateNode func()
+ )
+
+ ginkgo.BeforeAll(func() {
+ // Allocate 4 ports for the late-join data node.
+ ljPorts, portsErr := test.AllocateFreePorts(4)
+ gomega.Expect(portsErr).NotTo(gomega.HaveOccurred())
+
+ ljDataDir, _, dataErr := test.NewSpace()
+ gomega.Expect(dataErr).NotTo(gomega.HaveOccurred())
+
+ // Launch the late-join data node into the existing cluster.
+ ljGRPCAddr, ljClose := setup.ExternalDataNode(
+ clusterConfig,
+ clusterBinPath,
+ ljDataDir,
+ clusterLogDir,
+ ljPorts,
+ "--trace-pipeline-native-plugin-enabled=true",
+ "--trace-pipeline-trusted-plugin-dir="+clusterTrustedDir,
+ "--trace-pipeline-merge-grace-default=0",
+ "--trace-max-merge-parts=2",
+ "--trace-flush-timeout=500ms",
+ )
+ closeLateNode = ljClose
+
+ // Wait for the late-joining node to receive schema via WatchSchemas stream
+ // or the 30s syncLoop reconcile.
+ waitForTraceSchemaOnAddr(ljGRPCAddr)
+
+ // Connect directly to the late-join node for this spec.
+ var connErr error
+ lateJoinConn, connErr = grpchelper.Conn(ljGRPCAddr, 10*time.Second,
+ grpc.WithTransportCredentials(insecure.NewCredentials()))
+ gomega.Expect(connErr).NotTo(gomega.HaveOccurred())
+
+ // Write traces via the liaison; they route to the cluster and the
+ // late-join node applies the sampler on the next merge.
+ writeTwoPartMerge(connection,
+ makeDynRow(lateDropID, 100, "success"),
+ makeDynRow(lateKeepID, 900, "success"),
+ dynBase.Add(dynSeedOffset-5*time.Second))
+ })
+
+ ginkgo.AfterAll(func() {
+ if lateJoinConn != nil {
+ _ = lateJoinConn.Close()
+ }
+ if closeLateNode != nil {
+ closeLateNode()
+ }
+ })
+
+ ginkgo.It("drop-eligible trace is DROPPED on late-joining node (converged from schema store)", func() {
+ // Query via the liaison; the late-join node participates in fan-out queries.
+ assertTraceAbsent(connection, lateDropID, dynBase)
+ })
+
+ ginkgo.It("keep-eligible trace is PRESENT on late-joining node", func() {
+ assertTracePresent(connection, lateKeepID, dynBase)
+ })
+ })
+
+ // --- Step 7: LiaisonNoLoad — liaison sampler-load path is unreachable ---
+ //
+ // US-009 AC(e) / R8: the liaison's plugin-load branch is structurally gated off.
+ // The liaison's schemaRepo is constructed by newLiaisonSchemaRepo (metadata.go:100)
+ // which sets role=ROLE_LIAISON and does NOT set nativePipelineEnabled. The
+ // reconcilePipeline branch in OnAddOrUpdate is:
+ //
+ // if sr.role == databasev1.Role_ROLE_DATA && sr.nativePipelineEnabled { ... }
+ //
+ // Both conditions are false on the liaison — the load branch is doubly-gated off.
+ // We assert this behaviorally: after the sampler is active on data nodes, the
+ // liaison is still healthy (GroupRegistryService.List succeeds), confirming it
+ // processed the KindGroup event without attempting to load a .so.
+ ginkgo.Describe("LiaisonNoLoad: liaison sampler-load path is unreachable", func() {
+ ginkgo.It("liaison processes KindGroup pipeline event without loading a .so (structurally gated)", func() {
+ // The sampler is active on data nodes (RegisterSamplerRuntime was called).
+ // If the liaison tried to load the .so it would fail (no trusted-dir flag)
+ // and the node would either crash or return an error. Asserting the liaison
+ // is alive and serving schema RPCs proves the load branch was not reached.
+ lnConn, connErr := grpchelper.Conn(liaisonGRPCAddr, 10*time.Second,
+ grpc.WithTransportCredentials(insecure.NewCredentials()))
+ gomega.Expect(connErr).NotTo(gomega.HaveOccurred())
+ defer func() { _ = lnConn.Close() }()
+
+ c := databasev1.NewGroupRegistryServiceClient(lnConn)
+ ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+ defer cancel()
+ resp, listErr := c.List(ctx, &databasev1.GroupRegistryServiceListRequest{})
+ gomega.Expect(listErr).NotTo(gomega.HaveOccurred(),
+ "liaison must be alive and serving schema RPCs (load branch not reached)")
+ gomega.Expect(resp.GetGroup()).NotTo(gomega.BeEmpty(),
+ "liaison must list at least the pipeline group")
+ })
+ })
+})
+
+// waitForTraceSchemaOnAddr polls the given gRPC address until the "filter" trace
+// schema in the pipeline group is visible, proving schema sync has completed.
+// Used for per-node assertions (restart + late-join specs).
+func waitForTraceSchemaOnAddr(grpcAddr string) {
+ conn, connErr := grpchelper.Conn(grpcAddr, 10*time.Second,
+ grpc.WithTransportCredentials(insecure.NewCredentials()))
+ gomega.Expect(connErr).NotTo(gomega.HaveOccurred())
+ defer func() { _ = conn.Close() }()
+
+ traceClient := databasev1.NewTraceRegistryServiceClient(conn)
+ gomega.Eventually(func(innerGm gomega.Gomega) {
+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+ defer cancel()
+ _, getErr := traceClient.Get(ctx, &databasev1.TraceRegistryServiceGetRequest{
+ Metadata: &commonv1.Metadata{
+ Name: "filter",
+ Group: tracepipeline.PipelineGroup,
+ },
+ })
+ innerGm.Expect(getErr).NotTo(gomega.HaveOccurred(),
+ "trace schema filter/%s not yet visible on %s", tracepipeline.PipelineGroup, grpcAddr)
+ }, flags.EventuallyTimeout, 500*time.Millisecond).Should(gomega.Succeed())
+}
diff --git a/test/integration/standalone/pipeline/common.go b/test/integration/standalone/pipeline/common.go
index e4d4d8eca..a309a0a91 100644
--- a/test/integration/standalone/pipeline/common.go
+++ b/test/integration/standalone/pipeline/common.go
@@ -33,12 +33,9 @@ import (
"github.com/onsi/gomega/gleak"
"google.golang.org/grpc"
"google.golang.org/grpc/credentials/insecure"
- "google.golang.org/protobuf/encoding/protojson"
- "google.golang.org/protobuf/types/known/structpb"
commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
- pipelinev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/pipeline/v1"
"github.com/apache/skywalking-banyandb/pkg/grpchelper"
"github.com/apache/skywalking-banyandb/pkg/logger"
"github.com/apache/skywalking-banyandb/pkg/pool"
@@ -56,6 +53,20 @@ var (
goods []gleak.Goroutine
closeSvr func()
now time.Time
+
+ // svrSoPath is the bare filename of the staged .so (trusted-dir-relative path).
+ // The validator requires a relative path; the data node resolves it against
+ // --trace-pipeline-trusted-plugin-dir at load time.
+ // Exposed for dynamic_test.go (Update/Remove/InvalidConfig/Restart specs).
+ svrSoPath string
+
+ // svrBinPath is the banyand binary used to boot the external process.
+ // Exposed for the Restart spec so it can relaunch the same binary.
+ svrBinPath string
+
+ // svrTrustedDir is the trusted-plugin directory passed to the server.
+ // The Restart spec re-uses it when relaunching.
+ svrTrustedDir string
)
var _ = ginkgo.SynchronizedBeforeSuite(func() []byte {
@@ -89,13 +100,9 @@ var _ = ginkgo.SynchronizedBeforeSuite(func() []byte {
gomega.Expect(trustedErr).NotTo(gomega.HaveOccurred())
// Copy the .so into the trusted dir under its bare filename.
- soPath := filepath.Join(trustedDir, "latencystatussampler.so")
+ soPath := filepath.Join(trustedDir, tracepipeline.PluginSOName)
gomega.Expect(copyFile(pluginSrc, soPath)).To(gomega.Succeed())
- // Write the TracePipelineConfig protojson file.
- configFile := filepath.Join(trustedDir, "pipeline.json")
- gomega.Expect(writePipelineConfig(configFile)).To(gomega.Succeed())
-
// Allocate ports and build the cluster config (property-based schema registry).
ports, portsErr := test.AllocateFreePorts(5)
gomega.Expect(portsErr).NotTo(gomega.HaveOccurred())
@@ -105,6 +112,8 @@ var _ = ginkgo.SynchronizedBeforeSuite(func() []byte {
config := setup.PropertyClusterConfig(dfWriter)
// Launch the external standalone server.
+ // No static --trace-pipeline-config: the sampler is activated dynamically
+ // via RegisterSamplerRuntime (UpdateGroup) after schema preload (US-CLEANUP-standalone).
// ExternalStandalone calls config.AddSchemaServerAddr so that
// PreloadSchemaViaProperty can reach the schema server after boot.
grpcAddr, _, svClose := setup.ExternalStandalone(
@@ -115,13 +124,19 @@ var _ = ginkgo.SynchronizedBeforeSuite(func() []byte {
ports,
"--trace-pipeline-native-plugin-enabled=true",
"--trace-pipeline-trusted-plugin-dir="+trustedDir,
- "--trace-pipeline-config="+configFile,
"--trace-pipeline-merge-grace-default=0",
"--trace-max-merge-parts=2",
"--trace-flush-timeout=500ms",
)
closeSvr = svClose
+ // Populate package-level vars for use by other test files (dynamic_test.go).
+ // svrSoPath is the bare (trusted-dir-relative) filename — the validator requires
+ // a relative path; the data node resolves it against the trusted dir at load time.
+ svrSoPath = tracepipeline.PluginSOName
+ svrBinPath = binPath
+ svrTrustedDir = trustedDir
+
// Preload trace schemas (group test-trace-pipeline + trace filter + index rules/bindings)
// into the running server via the property schema gRPC endpoint. The fixtures are
// owned by the tracepipeline case package so they do not pollute the shared
@@ -131,6 +146,19 @@ var _ = ginkgo.SynchronizedBeforeSuite(func() []byte {
// Wait until the schema has propagated to the data node.
waitForSchemaSync(grpcAddr)
+ // Activate the sampler pipeline dynamically via UpdateGroup (US-CLEANUP-standalone).
+ // The data node's KindGroup watch fires and loads the .so plugin into the registry.
+ // All specs in this suite assume the base config (thresholdMs=500) is active.
+ activateCtx, activateCancel := context.WithTimeout(context.Background(), 30*time.Second)
+ defer activateCancel()
+ activateConn, activateConnErr := grpchelper.Conn(grpcAddr, 10*time.Second,
+ grpc.WithTransportCredentials(insecure.NewCredentials()))
+ gomega.Expect(activateConnErr).NotTo(gomega.HaveOccurred())
+ defer func() { _ = activateConn.Close() }()
+ tracepipeline.RegisterSamplerRuntime(activateCtx, activateConn,
+ tracepipeline.PipelineGroup,
+ tracepipeline.NewBasePipelineConfig(tracepipeline.PluginSOName, tracepipeline.DefaultMergeGrace))
+
ns := timestamp.NowMilli().UnixNano()
now = time.Unix(0, ns-ns%int64(time.Minute))
return []byte(grpcAddr)
@@ -232,48 +260,6 @@ func copyFile(src, dst string) error {
return nil
}
-// writePipelineConfig marshals a TracePipelineConfig (protojson) to cfgPath.
-// The plugin path uses only the bare filename; the server resolves it against
-// the trusted dir passed via --trace-pipeline-trusted-plugin-dir.
-func writePipelineConfig(cfgPath string) error {
- pluginConfig, structErr := structpb.NewStruct(map[string]interface{}{
- "thresholdMs": float64(500),
- "successValue": "success",
- })
- if structErr != nil {
- return fmt.Errorf("build plugin config struct: %w", structErr)
- }
-
- cfg := &pipelinev1.TracePipelineConfig{
- Metadata: &commonv1.Metadata{
- Group: "test-trace-pipeline",
- },
- Enabled: true,
- EnabledEvents: []pipelinev1.PipelineEvent{
- pipelinev1.PipelineEvent_PIPELINE_EVENT_MERGE,
- },
- Plugins: []*pipelinev1.Plugin{
- {
- Name: "latencystatussampler",
- Kind: &pipelinev1.Plugin_Sampler{
- Sampler: &pipelinev1.SamplerPlugin{
- Path: "latencystatussampler.so",
- Symbol: "NewSampler",
- AbiVersion: 1,
- Config: pluginConfig,
- },
- },
- },
- },
- }
-
- data, marshalErr := protojson.Marshal(cfg)
- if marshalErr != nil {
- return fmt.Errorf("marshal pipeline config: %w", marshalErr)
- }
- return os.WriteFile(cfgPath, data, 0o600)
-}
-
// waitForSchemaSync polls the gRPC endpoint until at least one group is listed,
// confirming the property→data-node schema sync has completed.
func waitForSchemaSync(grpcAddr string) {
@@ -291,3 +277,28 @@ func waitForSchemaSync(grpcAddr string) {
innerGm.Expect(resp.GetGroup()).NotTo(gomega.BeEmpty(), "no groups visible on standalone yet")
}, flags.EventuallyTimeout, 500*time.Millisecond).Should(gomega.Succeed())
}
+
+// waitForTraceSchema polls until the "filter" trace schema in the pipeline group
+// is visible on the node. WriteBatchEntries calls GetTrace internally, so this
+// gate must pass before seeding data — especially after a restart where the
+// trace schema re-syncs after the group.
+func waitForTraceSchema(grpcAddr string) {
+ conn, connErr := grpchelper.Conn(grpcAddr, 10*time.Second,
+ grpc.WithTransportCredentials(insecure.NewCredentials()))
+ gomega.Expect(connErr).NotTo(gomega.HaveOccurred())
+ defer func() { _ = conn.Close() }()
+
+ traceClient := databasev1.NewTraceRegistryServiceClient(conn)
+ gomega.Eventually(func(innerGm gomega.Gomega) {
+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+ defer cancel()
+ _, getErr := traceClient.Get(ctx, &databasev1.TraceRegistryServiceGetRequest{
+ Metadata: &commonv1.Metadata{
+ Name: "filter",
+ Group: tracepipeline.PipelineGroup,
+ },
+ })
+ innerGm.Expect(getErr).NotTo(gomega.HaveOccurred(),
+ "trace schema filter/%s not yet visible after restart", tracepipeline.PipelineGroup)
+ }, flags.EventuallyTimeout, 500*time.Millisecond).Should(gomega.Succeed())
+}
diff --git a/test/integration/standalone/pipeline/dynamic_test.go b/test/integration/standalone/pipeline/dynamic_test.go
new file mode 100644
index 000000000..fec9a5b83
--- /dev/null
+++ b/test/integration/standalone/pipeline/dynamic_test.go
@@ -0,0 +1,440 @@
+// 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.
+
+//go:build trace_pipeline
+
+package pipeline_test
+
+import (
+ "context"
+ "fmt"
+ "time"
+
+ "github.com/onsi/ginkgo/v2"
+ "github.com/onsi/gomega"
+ "google.golang.org/grpc"
+ "google.golang.org/grpc/credentials/insecure"
+ "google.golang.org/protobuf/types/known/timestamppb"
+
+ databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+ modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
+ tracev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/trace/v1"
+ "github.com/apache/skywalking-banyandb/pkg/grpchelper"
+ "github.com/apache/skywalking-banyandb/pkg/test"
+ "github.com/apache/skywalking-banyandb/pkg/test/flags"
+ "github.com/apache/skywalking-banyandb/pkg/test/helpers"
+ "github.com/apache/skywalking-banyandb/pkg/test/setup"
+ tracepipeline "github.com/apache/skywalking-banyandb/test/cases/tracepipeline"
+ tracepipelinedata "github.com/apache/skywalking-banyandb/test/cases/tracepipeline/data"
+)
+
+// dynSeedOffset places dynamic-spec timestamps well in the past so
+// isMergeHot (grace=0) returns false at merge time.
+const dynSeedOffset = -2 * time.Hour
+
+// queryByTraceID queries the "filter" trace for a single trace_id over a window
+// covering [baseTime+dynSeedOffset-1m, now+1s] (millisecond-truncated).
+func queryByTraceID(innerGm gomega.Gomega, conn *grpc.ClientConn, traceID string, baseTime time.Time) *tracev1.QueryResponse {
+ windowStart := baseTime.Add(dynSeedOffset - time.Minute).Truncate(time.Millisecond)
+ windowEnd := time.Now().Add(time.Second).Truncate(time.Millisecond)
+
+ req := &tracev1.QueryRequest{
+ Groups: []string{tracepipeline.PipelineGroup},
+ Name: "filter",
+ TimeRange: &modelv1.TimeRange{
+ Begin: timestamppb.New(windowStart),
+ End: timestamppb.New(windowEnd),
+ },
+ Criteria: &modelv1.Criteria{
+ Exp: &modelv1.Criteria_Condition{
+ Condition: &modelv1.Condition{
+ Name: "trace_id",
+ Op: modelv1.Condition_BINARY_OP_EQ,
+ Value: &modelv1.TagValue{
+ Value: &modelv1.TagValue_Str{Str: &modelv1.Str{Value: traceID}},
+ },
+ },
+ },
+ },
+ Limit: 10,
+ }
+
+ c := tracev1.NewTraceServiceClient(conn)
+ ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+ defer cancel()
+ resp, queryErr := c.Query(ctx, req)
+ innerGm.Expect(queryErr).NotTo(gomega.HaveOccurred())
+ return resp
+}
+
+// assertTraceAbsent asserts the named trace_id is absent (dropped by sampler).
+func assertTraceAbsent(conn *grpc.ClientConn, traceID string, baseTime time.Time) {
+ gomega.Eventually(func(innerGm gomega.Gomega) {
+ resp := queryByTraceID(innerGm, conn, traceID, baseTime)
+ innerGm.Expect(resp.GetTraces()).To(gomega.BeEmpty(),
+ "trace %q should be absent (filtered by sampler)", traceID)
+ }, flags.EventuallyTimeout, 500*time.Millisecond).Should(gomega.Succeed())
+}
+
+// assertTracePresent asserts the named trace_id is present (kept / not dropped).
+func assertTracePresent(conn *grpc.ClientConn, traceID string, baseTime time.Time) {
+ gomega.Eventually(func(innerGm gomega.Gomega) {
+ resp := queryByTraceID(innerGm, conn, traceID, baseTime)
+ innerGm.Expect(resp.GetTraces()).NotTo(gomega.BeEmpty(),
+ "trace %q should be present (not filtered)", traceID)
+ }, flags.EventuallyTimeout, 500*time.Millisecond).Should(gomega.Succeed())
+}
+
+// writeTwoPartMerge writes two batches to trigger a filtering merge
+// (max-merge-parts=2, grace=0). It writes rowA as part-1, asserts rowA is
+// visible (one part, no merge yet), then writes rowB as part-2 to trigger
+// the merge. Both rows share the same base time offset so isMergeHot is
+// false.
+func writeTwoPartMerge(
+ conn *grpc.ClientConn,
+ rowA tracepipelinedata.TraceRow, rowB tracepipelinedata.TraceRow,
+ baseTime time.Time,
+) {
+ // Part 1: rowA at baseTime.
+ tracepipelinedata.WriteBatchEntries(conn, tracepipeline.PipelineGroup,
+ baseTime, time.Millisecond, []tracepipelinedata.TraceRow{rowA})
+
+ // Assert rowA is visible before any merge (proves server received it).
+ traceIDA := rowA.Tags[0].GetStr().GetValue()
+ gomega.Eventually(func(innerGm gomega.Gomega) {
+ resp := queryByTraceID(innerGm, conn, traceIDA, baseTime)
+ innerGm.Expect(resp.GetTraces()).NotTo(gomega.BeEmpty(),
+ "part-1 trace %q must be visible before merge", traceIDA)
+ }, flags.EventuallyTimeout, 500*time.Millisecond).Should(gomega.Succeed())
+
+ // Part 2: rowB at baseTime+10ms — triggers the merge.
+ tracepipelinedata.WriteBatchEntries(conn, tracepipeline.PipelineGroup,
+ baseTime.Add(10*time.Millisecond), time.Millisecond, []tracepipelinedata.TraceRow{rowB})
+}
+
+// makeDynRow is a convenience builder for TraceRow entries in dynamic specs.
+func makeDynRow(traceID string, durationMs int64, status string) tracepipelinedata.TraceRow {
+ return tracepipelinedata.TraceRow{
+ Span: "span-" + traceID,
+ Tags: []*modelv1.TagValue{
+ {Value: &modelv1.TagValue_Str{Str: &modelv1.Str{Value: traceID}}},
+ {Value: &modelv1.TagValue_Str{Str: &modelv1.Str{Value: "span-" + traceID}}},
+ {Value: &modelv1.TagValue_Str{Str: &modelv1.Str{Value: "svc-dyn"}}},
+ {Value: &modelv1.TagValue_Int{Int: &modelv1.Int{Value: durationMs}}},
+ {Value: &modelv1.TagValue_Str{Str: &modelv1.Str{Value: status}}},
+ },
+ }
+}
+
+// Dynamic sampler registration lifecycle specs (US-008).
+//
+// These specs run Ordered so each step builds on the previous state.
+// The base config (thresholdMs=500) is activated in common.go's BeforeSuite
+// via RegisterSamplerRuntime; the RegisterMergeFilterTable Describe in
+// common.go verifies the base drop/keep assertions.
+//
+// Steps here:
+// 2. Update (variant, thresholdMs=200): assert drop-set changes.
+// 3. Remove: assert previously-dropped traces are retained (Correction-C AC).
+// 4. InvalidConfig: assert fail-open — node alive, traces retained.
+// 5. Restart: assert config re-applied from schema store on reboot.
+var _ = ginkgo.Describe("Dynamic sampler registration (standalone)", ginkgo.Ordered, func() {
+ // dynBase anchors all dynamic-spec timestamps in the past (grace=0 → isMergeHot=false).
+ var dynBase time.Time
+
+ ginkgo.BeforeAll(func() {
+ dynBase = now
+ })
+
+ // --- Step 2: Update (variant thresholdMs=200) ---
+ //
+ // Sampler decision: drop if dur < thresholdMs AND status == "success".
+ // base (500): dur=300 success → DROP (300 < 500).
+ // variant (200): dur=300 success → KEEP (300 >= 200).
+ // So dur=300 success is the pivot trace that changes verdict between configs.
+ // dur=100 success → dropped under both.
+ ginkgo.Describe("Update: variant threshold (thresholdMs=200)", ginkgo.Ordered, func() {
+ const (
+ pivotID = "dyn-upd-pivot-300ms" // dropped under base, kept under variant
+ bothDropID = "dyn-upd-both-drop-100ms" // dropped under both configs
+ )
+
+ ginkgo.BeforeAll(func() {
+ ctx := context.Background()
+
+ // Switch to variant config (thresholdMs=200) via UpdateSamplerRuntime.
+ upCtx, upCancel := context.WithTimeout(ctx, 30*time.Second)
+ defer upCancel()
+ tracepipeline.UpdateSamplerRuntime(upCtx, connection,
+ tracepipeline.PipelineGroup,
+ tracepipeline.NewVariantPipelineConfig(svrSoPath, tracepipeline.DefaultMergeGrace))
+
+ // Write pivot trace (dur=300 success) + keep companion → two-part merge.
+ // Under variant (thresholdMs=200): 300 >= 200 → KEPT.
+ writeTwoPartMerge(connection,
+ makeDynRow(pivotID, 300, "success"),
+ makeDynRow("dyn-upd-keep-companion1", 900, "success"),
+ dynBase.Add(dynSeedOffset))
+
+ // Write both-drop trace (dur=100 success) + keep companion → two-part merge.
+ // Under variant (thresholdMs=200): 100 < 200 → DROPPED.
+ writeTwoPartMerge(connection,
+ makeDynRow(bothDropID, 100, "success"),
+ makeDynRow("dyn-upd-keep-companion2", 800, "error"),
+ dynBase.Add(dynSeedOffset-time.Second))
+ })
+
+ ginkgo.It("dur=300 success: KEPT under variant (threshold=200)", func() {
+ // Variant keeps 300ms traces (300 >= 200); base would have dropped them.
+ assertTracePresent(connection, pivotID, dynBase)
+ })
+
+ ginkgo.It("dur=100 success: still DROPPED under variant (100 < 200)", func() {
+ assertTraceAbsent(connection, bothDropID, dynBase)
+ })
+ })
+
+ // --- Step 3: Remove → stop dropping (Correction-C measurement AC) ---
+ //
+ // After RemoveSamplerRuntime the registry for the group is empty.
+ // Traces written after removal that WOULD be dropped (dur < 200, success)
+ // must be retained in the next merge. With grace=0, STOP latency is at most
+ // one merge pass — no nudge needed (Correction C).
+ ginkgo.Describe("Remove: stop dropping within merge_grace+one cycle", ginkgo.Ordered, func() {
+ const retainedID = "dyn-rem-retained-100ms"
+
+ ginkgo.BeforeAll(func() {
+ ctx := context.Background()
+
+ // Remove the sampler — registry becomes empty for the group.
+ rmCtx, rmCancel := context.WithTimeout(ctx, 30*time.Second)
+ defer rmCancel()
+ tracepipeline.RemoveSamplerRuntime(rmCtx, connection, tracepipeline.PipelineGroup)
+
+ // Write a trace that WOULD be dropped by the variant (dur=100, success).
+ // After removal it must be retained (fail-safe: no sampler → retain all).
+ writeTwoPartMerge(connection,
+ makeDynRow(retainedID, 100, "success"),
+ makeDynRow("dyn-rem-keep-companion", 800, "error"),
+ dynBase.Add(dynSeedOffset-2*time.Second))
+ })
+
+ ginkgo.It("previously-dropped shape (dur=100 success) is RETAINED after remove", func() {
+ // Correction-C measurement AC: STOP latency ≤ merge_grace + one merge cycle.
+ // With grace=0: after the first merge pass following removal, the trace must
+ // be present. We assert eventual presence (no nudge, no time bound beyond
+ // EventuallyTimeout which exceeds the merge cycle).
+ assertTracePresent(connection, retainedID, dynBase)
+ })
+ })
+
+ // --- Step 4: Invalid config → fail-open, node alive, traces retained ---
+ //
+ // US-008 AC(c): an invalid config (missing .so) leaves prior set (empty after
+ // Remove) active, retains all traces, node stays alive.
+ ginkgo.Describe("InvalidConfig: missing .so → fail-open, node alive, traces retained", ginkgo.Ordered, func() {
+ const invalidRetainedID = "dyn-invalid-retained-100ms"
+
+ ginkgo.BeforeAll(func() {
+ ctx := context.Background()
+
+ // Register with a path that does not exist inside the trusted dir.
+ // The data node must reject/fail-open: no .so at that path → load error.
+ badCfg := tracepipeline.NewBasePipelineConfig(
+ "nonexistent_sampler_dyn.so", // not staged in the trusted dir
+ tracepipeline.DefaultMergeGrace,
+ )
+ badCtx, badCancel := context.WithTimeout(ctx, 30*time.Second)
+ defer badCancel()
+ tracepipeline.RegisterSamplerRuntime(badCtx, connection,
+ tracepipeline.PipelineGroup, badCfg)
+
+ // Write a trace that the base sampler would drop (dur=100, success).
+ // Because load failed (fail-open: previous set = empty), no sampler
+ // is active → trace retained.
+ writeTwoPartMerge(connection,
+ makeDynRow(invalidRetainedID, 100, "success"),
+ makeDynRow("dyn-invalid-keep-companion", 800, "success"),
+ dynBase.Add(dynSeedOffset-3*time.Second))
+ })
+
+ ginkgo.It("node is alive after invalid config", func() {
+ // Prove the node is still serving by listing groups — a schema RPC that
+ // does not require Order/TraceIDs like a trace query.
+ c := databasev1.NewGroupRegistryServiceClient(connection)
+ ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+ defer cancel()
+ resp, listErr := c.List(ctx, &databasev1.GroupRegistryServiceListRequest{})
+ gomega.Expect(listErr).NotTo(gomega.HaveOccurred(),
+ "node must remain queryable (GroupRegistryService.List) after invalid config load")
+ gomega.Expect(resp.GetGroup()).NotTo(gomega.BeEmpty(),
+ "at least the pipeline group must be listed")
+ })
+
+ ginkgo.It("traces are retained (fail-open) when .so load fails", func() {
+ assertTracePresent(connection, invalidRetainedID, dynBase)
+ })
+ })
+
+ // --- Step 5: Restart → config re-applied from schema store ---
+ //
+ // US-008 AC(d): the node re-applies pipeline config from the schema store via
+ // the initial KindGroup syncLoop reconcile on boot — no re-write needed.
+ //
+ // Design: a dedicated second server is booted on its own ports so the primary
+ // server (used by all other specs) is not disturbed. The restart server:
+ // 1. Boots with no static config (dynamic-only).
+ // 2. Receives a RegisterSamplerRuntime via UpdateGroup (schema store write).
+ // 3. Is stopped then relaunched from the SAME data dir.
+ // 4. Is asserted to drop traces without a second RegisterSamplerRuntime call.
+ ginkgo.Describe("Restart: config re-applied from schema store", ginkgo.Ordered, func() {
+ const (
+ dropAfterRestartID = "dyn-restart-drop-100ms"
+ keepAfterRestartID = "dyn-restart-keep-900ms"
+ )
+
+ var (
+ restartConn *grpc.ClientConn
+ stopRestartSvr func()
+ )
+
+ ginkgo.BeforeAll(func() {
+ // Allocate a fresh set of ports for the restart server so it does not
+ // conflict with the primary server (still running on svrPorts).
+ rPorts, portsErr := test.AllocateFreePorts(5)
+ gomega.Expect(portsErr).NotTo(gomega.HaveOccurred())
+
+ // Fresh data directory — schema is seeded below via PreloadSchemaViaProperty.
+ rDataPath, _, dataErr := test.NewSpace()
+ gomega.Expect(dataErr).NotTo(gomega.HaveOccurred())
+ rLogDir, _, logErr := test.NewSpace()
+ gomega.Expect(logErr).NotTo(gomega.HaveOccurred())
+
+ // Build a cluster config for the restart server.
+ rTmpDir, _, tmpErr := test.NewSpace()
+ gomega.Expect(tmpErr).NotTo(gomega.HaveOccurred())
+ rDFWriter := setup.NewDiscoveryFileWriter(rTmpDir)
+ rConfig := setup.PropertyClusterConfig(rDFWriter)
+
+ // Boot the restart server (first boot).
+ // --schema-server-flush-timeout=500ms ensures schemas written via
+ // PreloadSchemaViaProperty and RegisterSamplerRuntime are persisted to
+ // disk within the flush interval so the restarted node can replay them.
+ rGRPCAddr, _, rClose := setup.ExternalStandalone(
+ rConfig,
+ svrBinPath,
+ rDataPath,
+ rLogDir,
+ rPorts,
+ "--trace-pipeline-native-plugin-enabled=true",
+ "--trace-pipeline-trusted-plugin-dir="+svrTrustedDir,
+ "--trace-pipeline-merge-grace-default=0",
+ "--trace-max-merge-parts=2",
+ "--trace-flush-timeout=500ms",
+ "--schema-server-flush-timeout=500ms",
+ )
+ stopRestartSvr = rClose
+
+ // Load the pipeline group + trace schema onto the restart server.
+ setup.PreloadSchemaViaProperty(rConfig, tracepipeline.PreloadSchema)
+ waitForSchemaSync(rGRPCAddr)
+
+ // Open a connection and register the base sampler config.
+ rConn, connErr := grpchelper.Conn(rGRPCAddr, 10*time.Second,
+ grpc.WithTransportCredentials(insecure.NewCredentials()))
+ gomega.Expect(connErr).NotTo(gomega.HaveOccurred())
+
+ regCtx, regCancel := context.WithTimeout(context.Background(), 30*time.Second)
+ defer regCancel()
+ tracepipeline.RegisterSamplerRuntime(regCtx, rConn,
+ tracepipeline.PipelineGroup,
+ tracepipeline.NewBasePipelineConfig(svrSoPath, tracepipeline.DefaultMergeGrace))
+
+ // Wait for the trace schema to be visible via TraceRegistryService.Get —
+ // this proves the schema has been written to the property store (not just
+ // the in-memory watch cache). Then wait an extra second so the 500ms
+ // flush interval has time to persist the data to disk before we stop.
+ waitForTraceSchema(rGRPCAddr)
+ time.Sleep(time.Second)
+ _ = rConn.Close()
+
+ // Stop the first boot — simulates the restart.
+ stopRestartSvr()
+
+ // Relaunch from the same data dir using captured flags (no duplicate
+ // AddSchemaServerAddr — BuildStandaloneFlags does not call it).
+ rRelaunchFlags := append(
+ setup.BuildStandaloneFlags(rConfig, rDataPath, rPorts),
+ "--trace-pipeline-native-plugin-enabled=true",
+ "--trace-pipeline-trusted-plugin-dir="+svrTrustedDir,
+ "--trace-pipeline-merge-grace-default=0",
+ "--trace-max-merge-parts=2",
+ "--trace-flush-timeout=500ms",
+ "--schema-server-flush-timeout=500ms",
+ )
+ rRestartLogPath := rLogDir + "/standalone-restart.log"
+ newTeardown, launchErr := setup.ExternalCMD(svrBinPath, rRestartLogPath,
+ append([]string{"standalone"}, rRelaunchFlags...)...)
+ gomega.Expect(launchErr).NotTo(gomega.HaveOccurred())
+ stopRestartSvr = newTeardown
+
+ // Wait for restarted node health checks.
+ rNewGRPCAddr := fmt.Sprintf("localhost:%d", rPorts[0])
+ rNewHTTPAddr := fmt.Sprintf("localhost:%d", rPorts[1])
+ gomega.Eventually(
+ helpers.HealthCheck(rNewGRPCAddr, 10*time.Second, 10*time.Second,
+ grpc.WithTransportCredentials(insecure.NewCredentials())),
+ flags.EventuallyTimeout).Should(gomega.Succeed())
+ gomega.Eventually(helpers.HTTPHealthCheck(rNewHTTPAddr, ""),
+ flags.EventuallyTimeout).Should(gomega.Succeed())
+
+ // Wait for schema re-sync (groups + trace) on the restarted node.
+ waitForSchemaSync(rNewGRPCAddr)
+ waitForTraceSchema(rNewGRPCAddr)
+
+ // Connect to the restarted node.
+ restartConn, connErr = grpchelper.Conn(rNewGRPCAddr, 10*time.Second,
+ grpc.WithTransportCredentials(insecure.NewCredentials()))
+ gomega.Expect(connErr).NotTo(gomega.HaveOccurred())
+
+ // Write test traces — sampler must be active from the schema store alone.
+ // dropAfterRestartID: dur=100 success → DROP (100 < 500).
+ // keepAfterRestartID: dur=900 success → KEEP (900 ≥ 500).
+ writeTwoPartMerge(restartConn,
+ makeDynRow(dropAfterRestartID, 100, "success"),
+ makeDynRow(keepAfterRestartID, 900, "success"),
+ dynBase.Add(dynSeedOffset-4*time.Second))
+ })
+
+ ginkgo.AfterAll(func() {
+ if restartConn != nil {
+ _ = restartConn.Close()
+ }
+ if stopRestartSvr != nil {
+ stopRestartSvr()
+ }
+ })
+
+ ginkgo.It("drop-eligible trace is DROPPED after restart (sampler from store)", func() {
+ assertTraceAbsent(restartConn, dropAfterRestartID, dynBase)
+ })
+
+ ginkgo.It("keep-eligible trace is PRESENT after restart", func() {
+ assertTracePresent(restartConn, keepAfterRestartID, dynBase)
+ })
+ })
+})
diff --git a/test/integration/standalone/pipeline/soak_test.go b/test/integration/standalone/pipeline/soak_test.go
index cdeb7b176..f9761f6a4 100644
--- a/test/integration/standalone/pipeline/soak_test.go
+++ b/test/integration/standalone/pipeline/soak_test.go
@@ -23,7 +23,7 @@ import (
tracepipeline "github.com/apache/skywalking-banyandb/test/cases/tracepipeline"
)
-// RegisterSoak wires the opt-in soak into the standalone pipeline suite.
+// RegisterSoak wires the opt-in basic soak into the standalone pipeline suite.
//
// The soak reuses the SharedContext (and therefore the same external standalone server)
// set up by common.go's SynchronizedBeforeSuite. The instant suite configures the server
@@ -40,3 +40,27 @@ import (
// BANYAND_TRACE_PLUGIN=<path>/latencystatussampler.so \
// go test ./test/integration/standalone/pipeline/... -run Soak -timeout 300s
var _ = tracepipeline.RegisterSoak("Standalone (.so plugin): In-Merge Filter Soak")
+
+// RegisterSoakDynamic wires the US-010 dynamic soak into the standalone pipeline suite.
+//
+// The dynamic soak extends the basic write→merge→verify loop with:
+// - Periodic Register/Update/Remove sampler churn (every 15 iterations), rotating
+// through base (thresholdMs=500) → variant (thresholdMs=200) → remove → base.
+// - Mid-run panicking-sampler injection (iteration 5): registers the latencystatussampler
+// with "panic":true in its config so Decide panics unconditionally. The engine's
+// fail-open recover wrapper absorbs the panic; the test asserts the node is still
+// reachable (GroupRegistryService.List), traces are retained (no sampler active →
+// retain all), and subsequent merges are not stalled.
+// - Bounded test-driver heap growth gate (3× initial HeapInuse).
+// - No close-balance / ref-count gate (R4): plugins are immortal; Close() is never
+// called on registry mutation.
+//
+// The spec is Serial because it mutates the group's pipeline config and must not
+// race with the basic soak or the instant-suite specs.
+//
+// Enable with: TRACE_PIPELINE_SOAK=1 (same gate as the basic soak).
+// Override duration: TRACE_PIPELINE_SOAK_DURATION=20s (default: 3 m).
+var _ = tracepipeline.RegisterSoakDynamic(
+ "Standalone (.so plugin): Dynamic Sampler Registration Soak (US-010)",
+ func() string { return svrSoPath },
+)
diff --git a/test/plugins/_latencystatussampler/main.go b/test/plugins/_latencystatussampler/main.go
index efcf13877..f856de682 100644
--- a/test/plugins/_latencystatussampler/main.go
+++ b/test/plugins/_latencystatussampler/main.go
@@ -52,6 +52,10 @@ var ABIVersion = sdk.ABIVersion
type samplerConfig struct {
ThresholdMs int64 `json:"thresholdMs"`
SuccessValue string `json:"successValue"`
+ // Panic causes Decide to panic unconditionally. Used by the US-010 soak to
+ // verify that the engine's fail-open recover wrapper absorbs the panic and
+ // the node continues processing without crashing or stalling merges.
+ Panic bool `json:"panic"`
}
// latencyStatusSampler drops a trace when duration < thresholdMs && status ==
@@ -59,6 +63,9 @@ type samplerConfig struct {
type latencyStatusSampler struct {
thresholdMs int64
successValue string
+ // panicOnDecide causes every Decide call to panic. The engine's recover
+ // wrapper catches the panic and retains all traces (fail-open).
+ panicOnDecide bool
}
// NewSampler is the constructor symbol the engine looks up. It parses the
@@ -82,8 +89,9 @@ func NewSampler(configJSON []byte) (sdk.Sampler, error) {
}
}
return &latencyStatusSampler{
- thresholdMs: cfg.ThresholdMs,
- successValue: cfg.SuccessValue,
+ thresholdMs: cfg.ThresholdMs,
+ successValue: cfg.SuccessValue,
+ panicOnDecide: cfg.Panic,
}, nil
}
@@ -106,7 +114,14 @@ func (s *latencyStatusSampler) Close() error { return nil }
//
// Any failure to satisfy either condition results in the trace being kept
// (fail-open).
+//
+// When panicOnDecide is true the function panics immediately; the engine's
+// recover wrapper catches the panic and retains all traces (fail-open). This
+// mode is used exclusively by the US-010 soak to verify engine resilience.
func (s *latencyStatusSampler) Decide(batch *sdk.TraceBatch) (sdk.Verdict, error) {
+ if s.panicOnDecide {
+ panic("latencystatussampler: intentional panic for US-010 soak (panicOnDecide=true)")
+ }
keep := make([]bool, len(batch.Traces))
for i := range batch.Traces {
keep[i] = s.keepTrace(&batch.Traces[i])