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/08 01:18:38 UTC
(skywalking-banyandb) branch main updated: feat(trace): bounded meter + logger telemetry for the sampler plugin SDK (#1206)
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 1a2696b8a feat(trace): bounded meter + logger telemetry for the sampler plugin SDK (#1206)
1a2696b8a is described below
commit 1a2696b8a7c79a0fb9421af48dbfc8c031d07140
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Wed Jul 8 09:18:33 2026 +0800
feat(trace): bounded meter + logger telemetry for the sampler plugin SDK (#1206)
* feat(trace): bounded meter+logger telemetry for the sampler plugin SDK
Give trace-pipeline sampler plugins metrics and logging under a bounded,
host-enforced scope, delivered opt-in via a frozen SDK façade.
- SDK façade (pkg/pipeline/sdk/telemetry.go): Meter/Counter/Gauge/Histogram,
Logger, Host, and the optional HostAware{UseHost(Host)} — stdlib-only
signatures, no ABIVersion bump (delivery is a type assertion).
- Per-group instance for HostAware plugins: the loader cache is keyed by
(path,symbol,configHash,group) for HostAware samplers (shared otherwise), and
UseHost is called exactly once under pluginCache.mu before the instance is
published, so attribution is correct by construction and there is no
UseHost/Decide data race (proven under -race with a Go fake).
- Host adapters (banyand/trace/plugin_telemetry.go): forced
banyandb_trace_pipeline_plugin_* prefix, {group,plugin_name} variable labels,
per-plugin cardinality cap (100) with overflow, recover-guarded registration
(never panics the host); rate-limited logger (module TRACE.PLUGIN.<G>.<P>,
INFO cap, 50 lines/s burst 100, 1 KiB msg, 16 fields); Delete-based series
reclaim on remove/replace/OnDelete; never-Close (R4). Self-observability:
plugin_telemetry_series_rejected_total / plugin_log_dropped_total /
plugin_telemetry_panic_total. reconcile idempotent-skips unchanged configs.
- Make the SDK genuinely zerolog-free: extract ValueType to the dependency-free
leaf pkg/pb/v1/valuetype and the var-array codec to pkg/encoding/vararray;
pkg/pb/v1 and pkg/encoding alias/forward so all existing callers are
unchanged. go list -deps of the SDK now pulls no zerolog/zap/logger/meter.
- Tests: adapter units, Go-fake integration (-race: cross-group + same-group +
distinct-config once-only, teardown reclaim, panic isolation), a real-.so
end-to-end test asserting metrics land in a prometheus registry and logs flow
through the standard logger, reference _telemetrysampler/_faultysampler .so
plugins + build/load smoke, docs and CHANGES.
---
CHANGES.md | 1 +
Makefile | 18 +-
banyand/trace/metadata.go | 115 +++-
banyand/trace/pipeline_loader.go | 46 +-
banyand/trace/pipeline_loader_test.go | 22 +-
banyand/trace/pipeline_metrics.go | 44 +-
banyand/trace/pipeline_metrics_test.go | 102 ++-
banyand/trace/pipeline_registry.go | 47 +-
banyand/trace/pipeline_watch_test.go | 18 +-
banyand/trace/plugin_telemetry.go | 712 ++++++++++++++++++++
banyand/trace/plugin_telemetry_integration_test.go | 550 +++++++++++++++
banyand/trace/plugin_telemetry_realso_test.go | 746 +++++++++++++++++++++
banyand/trace/plugin_telemetry_test.go | 378 +++++++++++
docs/design/post-trace-pipeline.md | 58 ++
pkg/encoding/array.go | 80 +--
pkg/encoding/{array.go => vararray/vararray.go} | 9 +-
pkg/pb/v1/value.go | 27 +-
pkg/pb/v1/valuetype/valuetype.go | 39 ++
pkg/pipeline/sdk/decode.go | 28 +-
pkg/pipeline/sdk/importgraph_test.go | 87 +++
pkg/pipeline/sdk/sdk.go | 21 +-
pkg/pipeline/sdk/telemetry.go | 104 +++
test/plugins/_faultysampler/main.go | 159 +++++
test/plugins/_latencystatussampler/main.go | 6 +-
test/plugins/_telemetrysampler/main.go | 128 ++++
test/plugins/telemetry_smoke_test.go | 171 +++++
26 files changed, 3527 insertions(+), 189 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index 0884bae7e..6614f06d8 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -74,6 +74,7 @@ Release Notes.
- 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 [...]
- Add FODC memory-pressure pprof capture: agent grabs heap/goroutine profiles when RSS nears the cgroup limit, served via the proxy.
+- Add bounded plugin telemetry (meter + logger) to the trace-pipeline sampler SDK. A sampler plugin opts in by implementing `sdk.HostAware`; the engine calls `UseHost(h)` once per group before the first `Decide`, delivering a scoped `sdk.Host` the plugin may cache and use in `Decide`. The host enforces all resource limits: metric names are forced under the `banyandb_trace_pipeline_plugin_` prefix; `{group, plugin_name}` labels are always present; per-plugin cardinality is capped at 100 s [...]
- Add a Claude/Codex plugin packaging the BanyanDB MCP server with a parse-only `validate_bydbql` tool (backed by the `mcp/tools/bydbql-parse` Go validator) and a BydbQL skill for read-only natural-language-to-BydbQL generation over STREAM/MEASURE/TRACE/PROPERTY resources.
### Bug Fixes
diff --git a/Makefile b/Makefile
index 26fbfc5e6..f9cf37bd7 100644
--- a/Makefile
+++ b/Makefile
@@ -134,6 +134,22 @@ build-trace-pipeline-plugin: ## Build the latencystatussampler.so plugin (Linux/
./test/plugins/_latencystatussampler
@echo "Built $(PLUGIN_OUTPUT_DIR)/latencystatussampler.so"
+.PHONY: build-trace-pipeline-telemetry-plugins
+build-trace-pipeline-telemetry-plugins: ## Build telemetrysampler.so and faultysampler.so reference plugins (Linux/macOS only; requires a C toolchain)
+ @if ! command -v gcc > /dev/null 2>&1 && ! command -v clang > /dev/null 2>&1; then \
+ echo "ERROR: build-trace-pipeline-telemetry-plugins requires a C toolchain (gcc or clang) but neither was found in PATH."; \
+ exit 1; \
+ fi
+ @mkdir -p $(PLUGIN_OUTPUT_DIR)
+ CGO_ENABLED=1 go build -buildmode=plugin -trimpath \
+ -o $(PLUGIN_OUTPUT_DIR)/telemetrysampler.so \
+ ./test/plugins/_telemetrysampler
+ @echo "Built $(PLUGIN_OUTPUT_DIR)/telemetrysampler.so"
+ CGO_ENABLED=1 go build -buildmode=plugin -trimpath \
+ -o $(PLUGIN_OUTPUT_DIR)/faultysampler.so \
+ ./test/plugins/_faultysampler
+ @echo "Built $(PLUGIN_OUTPUT_DIR)/faultysampler.so"
+
.PHONY: build-trace-pipeline-server
build-trace-pipeline-server: ## Build banyand-server with explicit CGO_ENABLED=1 for plugin hosting (Linux/macOS only)
@if ! command -v gcc > /dev/null 2>&1 && ! command -v clang > /dev/null 2>&1; then \
@@ -335,7 +351,7 @@ release-push-candidate: ## Push release candidate
.PHONY: all $(PROJECTS) clean build default nuke
.PHONY: lint check tidy format pre-push generate-test-cases capture-test-cases generate-trace-test-cases capture-trace-test-cases check-import-boundaries
.PHONY: test test-race test-coverage test-ci test-docker
-.PHONY: build-trace-pipeline-plugin build-trace-pipeline-server test-trace-pipeline
+.PHONY: build-trace-pipeline-plugin build-trace-pipeline-telemetry-plugins build-trace-pipeline-server test-trace-pipeline
.PHONY: license-check license-fix license-dep
.PHONY: release release-binary release-source release-sign release-assembly
.PHONY: vendor-update
diff --git a/banyand/trace/metadata.go b/banyand/trace/metadata.go
index fb4c15888..8a5fa1a20 100644
--- a/banyand/trace/metadata.go
+++ b/banyand/trace/metadata.go
@@ -66,27 +66,30 @@ type SchemaService interface {
type schemaRepo struct {
resourceSchema.Repository
- onGroupDelete func(groupName string)
- l *logger.Logger
- metadata metadata.Repo
- samplerMeter *samplerMetrics
- path string
- nodeID string
- trustedPluginDir string
- role databasev1.Role
- nativePipelineEnabled bool
+ onGroupDelete func(groupName string)
+ l *logger.Logger
+ metadata metadata.Repo
+ samplerMeter *samplerMetrics
+ pluginTelemetryFactory observability.Factory
+ path string
+ nodeID string
+ trustedPluginDir string
+ role databasev1.Role
+ nativePipelineEnabled bool
}
func newSchemaRepo(path string, svc *standalone, nodeLabels map[string]string, nodeID string) schemaRepo {
+ pipelineFactory := svc.omr.With(pipelineScope)
sr := schemaRepo{
- 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)),
+ 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(pipelineFactory),
+ pluginTelemetryFactory: pipelineFactory,
Repository: resourceSchema.NewRepository(
svc.metadata,
svc.l,
@@ -100,12 +103,14 @@ func newSchemaRepo(path string, svc *standalone, nodeLabels map[string]string, n
}
func newLiaisonSchemaRepo(path string, svc *liaison, traceDataNodeRegistry grpc.NodeRegistry) schemaRepo {
+ pipelineFactory := svc.omr.With(pipelineScope)
sr := schemaRepo{
- l: svc.l,
- path: path,
- metadata: svc.metadata,
- role: databasev1.Role_ROLE_LIAISON,
- samplerMeter: newSamplerMetrics(svc.omr.With(pipelineScope)),
+ l: svc.l,
+ path: path,
+ metadata: svc.metadata,
+ role: databasev1.Role_ROLE_LIAISON,
+ samplerMeter: newSamplerMetrics(pipelineFactory),
+ pluginTelemetryFactory: pipelineFactory,
Repository: resourceSchema.NewRepository(
svc.metadata,
svc.l,
@@ -246,6 +251,8 @@ func (sr *schemaRepo) OnDelete(metadata schema.Metadata) {
})
if sr.role == databasev1.Role_ROLE_DATA && sr.nativePipelineEnabled {
removeSamplersForGroup(g.Metadata.Name)
+ teardownGroupTelemetry(g.Metadata.Name)
+ setMergeGraceForGroup(g.Metadata.Name, 0)
sr.samplerMeter.setActiveCount(g.Metadata.Name, 0)
sr.samplerMeter.incRemoveTotal(g.Metadata.Name)
}
@@ -333,6 +340,7 @@ func (sr *schemaRepo) reconcilePipeline(group string, cfg *commonv1.TracePipelin
if cfg == nil || !cfg.GetEnabled() || !mergeEventEnabled(cfg) {
hadSamplers := len(lookupSamplers(group)) > 0
removeSamplersForGroup(group)
+ teardownGroupTelemetry(group)
setMergeGraceForGroup(group, 0)
sr.samplerMeter.setActiveCount(group, 0)
if hadSamplers {
@@ -340,23 +348,77 @@ func (sr *schemaRepo) reconcilePipeline(group string, cfg *commonv1.TracePipelin
}
return
}
+ // Idempotent skip: if the desired identity list (name+configHash in order) equals
+ // the currently registered one, this is a redundant watch-replay re-apply. Return
+ // early with no reload, no UseHost, no teardown, no swap, and no metrics churn.
+ // This is defense-in-depth for the same-group re-reconcile race on top of the
+ // once-only-UseHost invariant enforced inside loadSamplerPlugin.
+ desired := make([]nameHash, 0, len(cfg.GetPlugins()))
+ for _, p := range cfg.GetPlugins() {
+ sp := p.GetSampler()
+ if sp == nil {
+ continue
+ }
+ desired = append(desired, nameHash{
+ name: p.GetName(),
+ configHash: computeConfigHash(sp),
+ path: sp.GetPath(),
+ symbol: sp.GetSymbol(),
+ abiVersion: sp.GetAbiVersion(),
+ })
+ }
+ current := currentSamplerIdentity(group)
+ if len(desired) == len(current) {
+ identical := true
+ for idx := range desired {
+ if desired[idx] != current[idx] {
+ identical = false
+ break
+ }
+ }
+ if identical {
+ return
+ }
+ }
+ // Config actually changed: tear down the old adapters' series before rebuilding
+ // them for the new instances.
+ teardownGroupTelemetry(group)
// Distinguish register (no previous set) from update (previous set exists).
- isUpdate := len(lookupSamplers(group)) > 0
+ isUpdate := len(current) > 0
var newSet []namedSampler
for _, p := range cfg.GetPlugins() {
sp := p.GetSampler()
if sp == nil {
continue
}
+ pluginName := p.GetName()
+ bindHost := func(s sdk.Sampler) {
+ ha, ok := s.(sdk.HostAware)
+ if !ok {
+ return
+ }
+ tel := newPluginTelemetry(sr.pluginTelemetryFactory, logger.GetLogger("trace"), sr.samplerMeter, group, pluginName)
+ func() {
+ defer func() {
+ if r := recover(); r != nil {
+ sr.samplerMeter.incTelemetryPanic(group, pluginName)
+ sr.l.Warn().Str("group", group).Str("plugin", pluginName).Interface("panic", r).
+ Msg("plugin UseHost panicked; telemetry disabled for this plugin")
+ }
+ }()
+ ha.UseHost(tel)
+ }()
+ registerGroupTelemetry(group, tel)
+ }
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)
+ loadErr = fmt.Errorf("panic loading plugin %q: %v", pluginName, rec)
}
}()
- sampler, loadErr = loadSamplerPlugin(sp, sr.trustedPluginDir)
+ sampler, loadErr = loadSamplerPlugin(sp, sr.trustedPluginDir, group, bindHost)
}()
if loadErr != nil {
sr.samplerMeter.incLoadFailed(group, p.GetName(), samplerLoadFailReason(loadErr))
@@ -376,6 +438,9 @@ func (sr *schemaRepo) reconcilePipeline(group string, cfg *commonv1.TracePipelin
name: p.GetName(),
configHash: computeConfigHash(sp),
sampler: sampler,
+ path: sp.GetPath(),
+ symbol: sp.GetSymbol(),
+ abiVersion: sp.GetAbiVersion(),
})
}
replaceSamplersForGroup(group, newSet)
diff --git a/banyand/trace/pipeline_loader.go b/banyand/trace/pipeline_loader.go
index 1337b146c..9a0285551 100644
--- a/banyand/trace/pipeline_loader.go
+++ b/banyand/trace/pipeline_loader.go
@@ -75,6 +75,7 @@ type pluginCacheKey struct {
path string
symbol string
configHash string // sha256 hex of the protojson-marshaled config (or of "{}" when nil)
+ group string // "" for a shared (non-HostAware) instance; group name for a per-group HostAware instance
}
type loadedPluginCache struct {
@@ -84,7 +85,12 @@ 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 *commonv1.SamplerPlugin, trustedDir string) (_ sdk.Sampler, retErr error) {
+// group scopes a HostAware instance to one group so its injected telemetry is per-group; a
+// non-HostAware sampler is cached under the shared key (group=="") and re-used across groups.
+// bindHost, when non-nil, is invoked EXACTLY ONCE on a freshly constructed HostAware sampler,
+// under pluginCache.mu and BEFORE the instance is stored, so no concurrent cache-hit can hand a
+// not-yet-hosted instance to a merge. It is never called on a cache hit. bindHost may be nil.
+func loadSamplerPlugin(sp *commonv1.SamplerPlugin, trustedDir, group string, bindHost func(sdk.Sampler)) (_ sdk.Sampler, retErr error) {
if trustedDir == "" {
return nil, fmt.Errorf("trusted plugin dir not configured")
}
@@ -134,15 +140,51 @@ func loadSamplerPlugin(sp *commonv1.SamplerPlugin, trustedDir string) (_ sdk.Sam
key := pluginCacheKey{path: resolvedPath, symbol: symbol, configHash: configHash}
pluginCache.mu.Lock()
defer pluginCache.mu.Unlock()
+ // 3-step lookup, all under pluginCache.mu:
+ // 1. shared key (group=="") — a non-HostAware instance re-used across groups.
+ // 2. group key — an already-hosted HostAware instance; return WITHOUT calling bindHost.
+ // 3. miss — fall through to construct (ABI check, plugin.Open, ctor) under the lock.
if cached, ok := pluginCache.m[key]; ok {
return cached, nil
}
+ groupKey := key
+ groupKey.group = group
+ if cached, ok := pluginCache.m[groupKey]; ok {
+ return cached, nil
+ }
if sp.GetAbiVersion() != uint32(sdk.ABIVersion) {
return nil, fmt.Errorf("plugin ABI version %d does not match engine ABI version %d",
sp.GetAbiVersion(), sdk.ABIVersion)
}
+ sampler, err := newSamplerFromPlugin(resolvedPath, symbol, cfgJSON)
+ if err != nil {
+ return nil, err
+ }
+
+ // Classify the freshly constructed instance and key it accordingly. For a
+ // HostAware sampler, inject the host via bindHost while still holding
+ // pluginCache.mu and BEFORE storing, so no concurrent cache-hit can observe a
+ // not-yet-hosted instance. UseHost/telemetry setup thus runs under the same
+ // lock that already guards plugin.Open and the constructor; keep it fast.
+ if _, isHostAware := sampler.(sdk.HostAware); isHostAware {
+ key.group = group
+ if bindHost != nil {
+ bindHost(sampler)
+ }
+ }
+ pluginCache.m[key] = sampler
+ return sampler, nil
+}
+
+// newSamplerFromPlugin opens the .so, verifies its ABIVersion symbol, and calls
+// the constructor. It is a package var so tests can substitute a Go sampler and
+// exercise the cache/keying/bindHost logic under -race without a real .so. The
+// default implementation is behavior-preserving: it is the exact plugin.Open +
+// ABI check + constructor sequence that loadSamplerPlugin ran inline before the
+// seam was introduced.
+var newSamplerFromPlugin = func(resolvedPath, symbol string, cfgJSON []byte) (sdk.Sampler, error) {
var p *plugin.Plugin
var openErr error
func() {
@@ -194,7 +236,5 @@ func loadSamplerPlugin(sp *commonv1.SamplerPlugin, trustedDir string) (_ sdk.Sam
if sampler == nil {
return nil, fmt.Errorf("plugin %q constructor returned nil sampler", resolvedPath)
}
-
- pluginCache.m[key] = sampler
return sampler, nil
}
diff --git a/banyand/trace/pipeline_loader_test.go b/banyand/trace/pipeline_loader_test.go
index d3af540a3..f17b76c05 100644
--- a/banyand/trace/pipeline_loader_test.go
+++ b/banyand/trace/pipeline_loader_test.go
@@ -144,7 +144,7 @@ func TestLoadSamplerPlugin(t *testing.T) {
t.Run("happy path", func(t *testing.T) {
resetPluginCache()
- cfgStruct, cfgErr := structpb.NewStruct(map[string]interface{}{
+ cfgStruct, cfgErr := structpb.NewStruct(map[string]any{
"thresholdMs": float64(300),
"successValue": "ok",
})
@@ -157,7 +157,7 @@ func TestLoadSamplerPlugin(t *testing.T) {
Config: cfgStruct,
}
- sampler, loadErr := loadSamplerPlugin(sp, tmpDir)
+ sampler, loadErr := loadSamplerPlugin(sp, tmpDir, "", nil)
require.NoError(t, loadErr)
require.NotNil(t, sampler)
@@ -172,13 +172,13 @@ func TestLoadSamplerPlugin(t *testing.T) {
// 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{}{
+ cfgLow, cfgLowErr := structpb.NewStruct(map[string]any{
"thresholdMs": float64(100),
"successValue": "ok",
})
require.NoError(t, cfgLowErr)
- cfgHigh, cfgHighErr := structpb.NewStruct(map[string]interface{}{
+ cfgHigh, cfgHighErr := structpb.NewStruct(map[string]any{
"thresholdMs": float64(500),
"successValue": "ok",
})
@@ -200,11 +200,11 @@ func TestLoadSamplerPlugin(t *testing.T) {
t.Run("cache same config deduplicates", func(t *testing.T) {
resetPluginCache()
- first, firstErr := loadSamplerPlugin(spLow, tmpDir)
+ first, firstErr := loadSamplerPlugin(spLow, tmpDir, "", nil)
require.NoError(t, firstErr)
require.NotNil(t, first)
- second, secondErr := loadSamplerPlugin(spLow, tmpDir)
+ second, secondErr := loadSamplerPlugin(spLow, tmpDir, "", nil)
require.NoError(t, secondErr)
require.NotNil(t, second)
@@ -215,11 +215,11 @@ func TestLoadSamplerPlugin(t *testing.T) {
t.Run("cache changed config produces fresh instance", func(t *testing.T) {
resetPluginCache()
- instanceLow, errLow := loadSamplerPlugin(spLow, tmpDir)
+ instanceLow, errLow := loadSamplerPlugin(spLow, tmpDir, "", nil)
require.NoError(t, errLow)
require.NotNil(t, instanceLow)
- instanceHigh, errHigh := loadSamplerPlugin(spHigh, tmpDir)
+ instanceHigh, errHigh := loadSamplerPlugin(spHigh, tmpDir, "", nil)
require.NoError(t, errHigh)
require.NotNil(t, instanceHigh)
@@ -236,7 +236,7 @@ func TestLoadSamplerPlugin(t *testing.T) {
AbiVersion: 999, // proto-level ABI mismatch
Config: cfgLow,
}
- sampler, loadErr := loadSamplerPlugin(spBadABI, tmpDir)
+ sampler, loadErr := loadSamplerPlugin(spBadABI, tmpDir, "", nil)
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")
@@ -302,7 +302,7 @@ func TestLoadSamplerPlugin(t *testing.T) {
t.Run(tc.name, func(t *testing.T) {
resetPluginCache()
- sampler, loadErr := loadSamplerPlugin(tc.sp, tc.trustedDir)
+ sampler, loadErr := loadSamplerPlugin(tc.sp, tc.trustedDir, "", nil)
require.Error(t, loadErr, "expected an error for guard %q", tc.name)
assert.Nil(t, sampler)
assert.Contains(t, loadErr.Error(), tc.errContains,
@@ -328,7 +328,7 @@ func TestLoadSamplerPlugin_DotDotPrefixChildNotEscape(t *testing.T) {
Symbol: "NewSampler",
AbiVersion: uint32(sdk.ABIVersion),
}
- _, loadErr := loadSamplerPlugin(sp, tmpDir)
+ _, loadErr := loadSamplerPlugin(sp, tmpDir, "", nil)
require.Error(t, loadErr, "opening a nonexistent file must still error")
assert.NotContains(t, loadErr.Error(), "escapes trusted directory",
"a '..'-prefixed child path inside the trusted dir must not be rejected as an escape")
diff --git a/banyand/trace/pipeline_metrics.go b/banyand/trace/pipeline_metrics.go
index fe877182c..f27b675d0 100644
--- a/banyand/trace/pipeline_metrics.go
+++ b/banyand/trace/pipeline_metrics.go
@@ -40,6 +40,16 @@ type samplerMetrics struct {
// samplerLoadFailed{group,name,reason} counts individual plugin load failures
// on the fail-open path.
samplerLoadFailed meter.Counter
+ // pluginTelemetrySeriesRejectedTotal{group,plugin_name} counts metric series
+ // registrations rejected by the host (reserved labels, cardinality overflow,
+ // or registration panic).
+ pluginTelemetrySeriesRejectedTotal meter.Counter
+ // pluginLogDroppedTotal{group,plugin_name} counts log lines dropped by the
+ // host rate-limiter on behalf of a plugin.
+ pluginLogDroppedTotal meter.Counter
+ // pluginTelemetryPanicTotal{group,plugin_name} counts panics recovered inside
+ // the host telemetry adapter.
+ pluginTelemetryPanicTotal meter.Counter
}
// newSamplerMetrics creates a samplerMetrics wired to factory. If factory is
@@ -49,11 +59,14 @@ func newSamplerMetrics(factory observability.Factory) *samplerMetrics {
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"),
+ 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"),
+ pluginTelemetrySeriesRejectedTotal: factory.NewCounter("plugin_telemetry_series_rejected_total", "group", "plugin_name"),
+ pluginLogDroppedTotal: factory.NewCounter("plugin_log_dropped_total", "group", "plugin_name"),
+ pluginTelemetryPanicTotal: factory.NewCounter("plugin_telemetry_panic_total", "group", "plugin_name"),
}
}
@@ -91,3 +104,24 @@ func (m *samplerMetrics) incLoadFailed(group, name, reason string) {
}
m.samplerLoadFailed.Inc(1, group, name, reason)
}
+
+func (m *samplerMetrics) incSeriesRejected(group, plugin string) {
+ if m == nil {
+ return
+ }
+ m.pluginTelemetrySeriesRejectedTotal.Inc(1, group, plugin)
+}
+
+func (m *samplerMetrics) incLogDropped(group, plugin string) {
+ if m == nil {
+ return
+ }
+ m.pluginLogDroppedTotal.Inc(1, group, plugin)
+}
+
+func (m *samplerMetrics) incTelemetryPanic(group, plugin string) {
+ if m == nil {
+ return
+ }
+ m.pluginTelemetryPanicTotal.Inc(1, group, plugin)
+}
diff --git a/banyand/trace/pipeline_metrics_test.go b/banyand/trace/pipeline_metrics_test.go
index 0391fde39..550593e1a 100644
--- a/banyand/trace/pipeline_metrics_test.go
+++ b/banyand/trace/pipeline_metrics_test.go
@@ -41,8 +41,9 @@ type labeledCall struct {
}
type fakeMetricCounter struct {
- calls []labeledCall
- mu sync.Mutex
+ calls []labeledCall
+ deletes [][]string
+ mu sync.Mutex
}
func (f *fakeMetricCounter) Inc(delta float64, labels ...string) {
@@ -53,7 +54,38 @@ func (f *fakeMetricCounter) Inc(delta float64, labels ...string) {
f.mu.Unlock()
}
-func (f *fakeMetricCounter) Delete(_ ...string) bool { return true }
+func (f *fakeMetricCounter) Delete(labels ...string) bool {
+ cp := make([]string, len(labels))
+ copy(cp, labels)
+ f.mu.Lock()
+ f.deletes = append(f.deletes, cp)
+ f.mu.Unlock()
+ return true
+}
+
+// deletesWithPrefix counts recorded Delete calls whose label slice starts with
+// the given prefix.
+func (f *fakeMetricCounter) deletesWithPrefix(prefix ...string) int {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ count := 0
+ for _, d := range f.deletes {
+ if len(d) < len(prefix) {
+ continue
+ }
+ match := true
+ for idx, lv := range prefix {
+ if d[idx] != lv {
+ match = false
+ break
+ }
+ }
+ if match {
+ count++
+ }
+ }
+ return count
+}
func (f *fakeMetricCounter) callsWithLabels(labels ...string) int {
f.mu.Lock()
@@ -150,6 +182,13 @@ func newFakeMetricFactory() *fakeMetricFactory {
func (f *fakeMetricFactory) NewCounter(name string, _ ...string) meter.Counter {
f.mu.Lock()
defer f.mu.Unlock()
+ // Memoize by name so repeated NewCounter for the same series name returns the
+ // same handle, mirroring the real registry: two groups sharing a plugin metric
+ // name wrap ONE physical series and disambiguate via their prepended group
+ // label value. This is what the cross-group attribution assertion depends on.
+ if c, ok := f.counters[name]; ok {
+ return c
+ }
c := &fakeMetricCounter{}
f.counters[name] = c
return c
@@ -158,6 +197,9 @@ func (f *fakeMetricFactory) NewCounter(name string, _ ...string) meter.Counter {
func (f *fakeMetricFactory) NewGauge(name string, _ ...string) meter.Gauge {
f.mu.Lock()
defer f.mu.Unlock()
+ if g, ok := f.gauges[name]; ok {
+ return g
+ }
g := &fakeMetricGauge{}
f.gauges[name] = g
return g
@@ -278,50 +320,54 @@ func TestSamplerMetrics_RegisterUpdateRemove(t *testing.T) {
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) {
+// TestSamplerMetrics_SuccessUpdateAndIdempotentSkip exercises the success metric
+// path via a real config change and verifies the idempotent-skip: an
+// empty-plugins config on an empty registry is a no-op (no metric churn), while a
+// transition from a previously registered set to the empty set is a genuine
+// update that emits update-success and sets active_count=0. A second identical
+// empty reconcile is then skipped.
+func TestSamplerMetrics_SuccessUpdateAndIdempotentSkip(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.
+ // Empty-plugins config on an empty registry: desired == current == [] →
+ // idempotent skip. No register/update/active metric is emitted.
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}")
+ if regTotal != nil {
+ assert.Equal(t, 0, regTotal.callsWithLabels(group, "success"),
+ "empty-plugins config on an empty registry must be idempotently skipped")
+ }
+
+ // Seed a previously registered set, then reconcile to the empty-plugins
+ // config. desired ([]) != current (one sampler) → genuine update → success.
+ dummy := &dummySampler{}
+ replaceSamplersForGroup(group, []namedSampler{{name: "d", sampler: dummy}})
+ sr.reconcilePipeline(group, emptyCfg)
+
+ updTotal := factory.counter("sampler_update_total")
+ require.NotNil(t, updTotal)
+ assert.Equal(t, 1, updTotal.callsWithLabels(group, "success"),
+ "a set→empty transition must emit sampler_update_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")
+ assert.Equal(t, float64(0), lastVal, "active_count must be 0 after clearing to the empty set")
- // Second call with empty plugins → update success.
+ // Second identical empty reconcile: desired == current == [] → skipped, so
+ // update-success stays at exactly 1.
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)")
+ assert.Equal(t, 1, updTotal.callsWithLabels(group, "success"),
+ "a redundant identical empty reconcile must be idempotently skipped")
}
// TestSamplerMetrics_OnDeleteKindGroup verifies that the OnDelete KindGroup
diff --git a/banyand/trace/pipeline_registry.go b/banyand/trace/pipeline_registry.go
index a0f71e75e..eb2beb015 100644
--- a/banyand/trace/pipeline_registry.go
+++ b/banyand/trace/pipeline_registry.go
@@ -49,13 +49,19 @@ func lookupMergeGrace(group string) int64 {
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.
+// namedSampler pairs a sampler instance with its stable identity. The identity is the
+// plugin name plus everything that selects which code runs: the config hash AND the
+// plugin reference (path, symbol, ABI version). reconcilePipeline compares this identity
+// to decide whether a re-apply is redundant, so all fields that change the loaded plugin
+// must be captured here — otherwise a path/symbol/ABI swap under an unchanged config JSON
+// would be skipped and keep the stale plugin.
type namedSampler struct {
sampler sdk.Sampler
name string
configHash string
+ path string
+ symbol string
+ abiVersion uint32
}
// localSamplerRegistry maps group → ordered named samplers for in-process, flag-gated
@@ -110,6 +116,41 @@ func registerSampler(group string, s sdk.Sampler) func() {
}
}
+// nameHash is the stable identity of one registered sampler: its plugin name, config
+// hash, and plugin reference (path, symbol, ABI version). reconcilePipeline compares the
+// desired list against currentSamplerIdentity to skip redundant re-applies; including the
+// plugin reference ensures a path/symbol/ABI change (even with an unchanged config JSON)
+// triggers a rebuild.
+type nameHash struct {
+ name string
+ configHash string
+ path string
+ symbol string
+ abiVersion uint32
+}
+
+// currentSamplerIdentity returns the (name, configHash) identity of group's
+// registered samplers in declared order, or nil if none.
+func currentSamplerIdentity(group string) []nameHash {
+ localSamplerRegistry.mu.RLock()
+ defer localSamplerRegistry.mu.RUnlock()
+ ns := localSamplerRegistry.m[group]
+ if len(ns) == 0 {
+ return nil
+ }
+ out := make([]nameHash, len(ns))
+ for idx, item := range ns {
+ out[idx] = nameHash{
+ name: item.name,
+ configHash: item.configHash,
+ path: item.path,
+ symbol: item.symbol,
+ abiVersion: item.abiVersion,
+ }
+ }
+ return out
+}
+
// 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.
diff --git a/banyand/trace/pipeline_watch_test.go b/banyand/trace/pipeline_watch_test.go
index 09873ab25..b6ff5a7b8 100644
--- a/banyand/trace/pipeline_watch_test.go
+++ b/banyand/trace/pipeline_watch_test.go
@@ -47,6 +47,10 @@ func resetRegistries() {
localMergeGraceRegistry.m = make(map[string]int64)
localMergeGraceRegistry.mu.Unlock()
+ pluginTelemetryRegistry.mu.Lock()
+ pluginTelemetryRegistry.m = make(map[string][]*pluginTelemetry)
+ pluginTelemetryRegistry.mu.Unlock()
+
resetPluginCache()
}
@@ -65,7 +69,7 @@ func makeDataSchemaRepo(trustedDir string) schemaRepo {
// makeSamplerPlugin builds a SamplerPlugin for soName inside trustedDir.
func makeSamplerPlugin(soName string, thresholdMs float64) (*commonv1.SamplerPlugin, error) {
- cfgStruct, cfgErr := structpb.NewStruct(map[string]interface{}{
+ cfgStruct, cfgErr := structpb.NewStruct(map[string]any{
"thresholdMs": thresholdMs,
"successValue": "ok",
})
@@ -404,12 +408,10 @@ func TestReconcilePipeline_ConcurrentRace(t *testing.T) {
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 {
+ for wID := range workers {
+ wg.Go(func() {
+ for i := range iterations {
+ switch (wID + i) % 3 {
case 0:
replaceSamplersForGroup(group, []namedSampler{{name: "d", sampler: dummy}})
setMergeGraceForGroup(group, int64(i+1)*int64(time.Second))
@@ -421,7 +423,7 @@ func TestReconcilePipeline_ConcurrentRace(t *testing.T) {
_ = lookupMergeGrace(group)
}
}
- }(wID)
+ })
}
wg.Wait()
}
diff --git a/banyand/trace/plugin_telemetry.go b/banyand/trace/plugin_telemetry.go
new file mode 100644
index 000000000..e831887dd
--- /dev/null
+++ b/banyand/trace/plugin_telemetry.go
@@ -0,0 +1,712 @@
+// 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 (
+ "fmt"
+ "slices"
+ "strings"
+ "sync"
+ "time"
+ "unicode"
+
+ "github.com/rs/zerolog"
+
+ "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"
+)
+
+const (
+ // maxSeriesPerInstrument is the cardinality cap for custom label-value tuples
+ // per instrument instance.
+ maxSeriesPerInstrument = 100
+ // maxMetricNameLen is the maximum sanitized metric name length.
+ maxMetricNameLen = 64
+ // logRatePerSec is the sustained token-bucket refill rate (tokens/second).
+ logRatePerSec = 50.0
+ // logBurstSize is the maximum burst capacity of the log token bucket.
+ logBurstSize = 100.0
+ // maxMsgLen is the maximum log message length in bytes.
+ maxMsgLen = 1024
+ // maxLogFields is the maximum number of structured key/value pairs emitted.
+ maxLogFields = 16
+ // suppressWarnInterval is the minimum interval between rate-limiter summary WARNs.
+ suppressWarnInterval = 5 * time.Second
+ // overflowSentinel is the fixed label value substituted when cardinality is exceeded.
+ overflowSentinel = "__overflow__"
+)
+
+// reservedLabelSet contains label names plugins may not declare; the host
+// always prepends group and plugin_name itself.
+var reservedLabelSet = map[string]struct{}{
+ "group": {},
+ "plugin_name": {},
+}
+
+// noopCounter is a zero-allocation sdk.Counter for the bypass path.
+type noopCounter struct{}
+
+func (noopCounter) Inc(_ float64, _ ...string) {}
+
+// noopGauge is a zero-allocation sdk.Gauge for the bypass path.
+type noopGauge struct{}
+
+func (noopGauge) Set(_ float64, _ ...string) {}
+func (noopGauge) Add(_ float64, _ ...string) {}
+
+// noopHistogram is a zero-allocation sdk.Histogram for the bypass path.
+type noopHistogram struct{}
+
+func (noopHistogram) Observe(_ float64, _ ...string) {}
+
+// noopMeter is a zero-allocation sdk.Meter for the bypass path.
+type noopMeter struct{}
+
+func (noopMeter) Counter(_ string, _ ...string) sdk.Counter { return noopCounter{} }
+func (noopMeter) Gauge(_ string, _ ...string) sdk.Gauge { return noopGauge{} }
+func (noopMeter) Histogram(_ string, _ []float64, _ ...string) sdk.Histogram { return noopHistogram{} }
+
+// instrumentKey identifies a registered instrument in the cache by its full
+// metric name and joined label names.
+type instrumentKey struct {
+ fullName string
+ labelNames string // "\x00"-joined
+}
+
+// boundInstrumentKind distinguishes the three instrument kinds.
+type boundInstrumentKind uint8
+
+const (
+ kindCounter boundInstrumentKind = iota
+ kindGauge boundInstrumentKind = iota
+ kindHistogram boundInstrumentKind = iota
+)
+
+// cachedInstrument holds one registered instrument together with its kind tag.
+type cachedInstrument struct {
+ counter *boundCounter
+ gauge *boundGauge
+ histogram *boundHistogram
+ kind boundInstrumentKind
+}
+
+// boundCounter wraps a meter.Counter as sdk.Counter, enforcing cardinality.
+type boundCounter struct {
+ handle meter.Counter
+ sm *samplerMetrics
+ seen map[string]struct{}
+ group string
+ pluginName string
+ emitted [][]string
+ arity int
+ mu sync.Mutex
+}
+
+// Inc adds delta for the given custom label values, enforcing the cardinality cap.
+func (c *boundCounter) Inc(delta float64, customValues ...string) {
+ full := c.resolve(customValues)
+ c.handle.Inc(delta, full...)
+}
+
+func (c *boundCounter) resolve(customValues []string) []string {
+ custom := padOrTrim(customValues, c.arity)
+ key := strings.Join(custom, "\x00")
+ c.mu.Lock()
+ _, known := c.seen[key]
+ if !known {
+ if len(c.seen) >= maxSeriesPerInstrument {
+ c.mu.Unlock()
+ c.sm.incSeriesRejected(c.group, c.pluginName)
+ return makeOverflowValues(c.group, c.pluginName, c.arity)
+ }
+ c.seen[key] = struct{}{}
+ full := prependHostLabels(c.group, c.pluginName, custom)
+ c.emitted = append(c.emitted, full)
+ c.mu.Unlock()
+ return full
+ }
+ c.mu.Unlock()
+ return prependHostLabels(c.group, c.pluginName, custom)
+}
+
+// boundGauge wraps a meter.Gauge as sdk.Gauge, enforcing cardinality.
+type boundGauge struct {
+ handle meter.Gauge
+ sm *samplerMetrics
+ seen map[string]struct{}
+ group string
+ pluginName string
+ emitted [][]string
+ arity int
+ mu sync.Mutex
+}
+
+// Set replaces the gauge value for the given custom label values.
+func (g *boundGauge) Set(value float64, customValues ...string) {
+ full := g.resolve(customValues)
+ g.handle.Set(value, full...)
+}
+
+// Add adds delta to the gauge for the given custom label values.
+func (g *boundGauge) Add(delta float64, customValues ...string) {
+ full := g.resolve(customValues)
+ g.handle.Add(delta, full...)
+}
+
+func (g *boundGauge) resolve(customValues []string) []string {
+ custom := padOrTrim(customValues, g.arity)
+ key := strings.Join(custom, "\x00")
+ g.mu.Lock()
+ _, known := g.seen[key]
+ if !known {
+ if len(g.seen) >= maxSeriesPerInstrument {
+ g.mu.Unlock()
+ g.sm.incSeriesRejected(g.group, g.pluginName)
+ return makeOverflowValues(g.group, g.pluginName, g.arity)
+ }
+ g.seen[key] = struct{}{}
+ full := prependHostLabels(g.group, g.pluginName, custom)
+ g.emitted = append(g.emitted, full)
+ g.mu.Unlock()
+ return full
+ }
+ g.mu.Unlock()
+ return prependHostLabels(g.group, g.pluginName, custom)
+}
+
+// boundHistogram wraps a meter.Histogram as sdk.Histogram, enforcing cardinality.
+type boundHistogram struct {
+ handle meter.Histogram
+ group string
+ pluginName string
+ sm *samplerMetrics
+ seen map[string]struct{}
+ emitted [][]string
+ mu sync.Mutex
+ arity int
+}
+
+// Observe records a single observation for the given custom label values.
+func (h *boundHistogram) Observe(value float64, customValues ...string) {
+ full := h.resolve(customValues)
+ h.handle.Observe(value, full...)
+}
+
+func (h *boundHistogram) resolve(customValues []string) []string {
+ custom := padOrTrim(customValues, h.arity)
+ key := strings.Join(custom, "\x00")
+ h.mu.Lock()
+ _, known := h.seen[key]
+ if !known {
+ if len(h.seen) >= maxSeriesPerInstrument {
+ h.mu.Unlock()
+ h.sm.incSeriesRejected(h.group, h.pluginName)
+ return makeOverflowValues(h.group, h.pluginName, h.arity)
+ }
+ h.seen[key] = struct{}{}
+ full := prependHostLabels(h.group, h.pluginName, custom)
+ h.emitted = append(h.emitted, full)
+ h.mu.Unlock()
+ return full
+ }
+ h.mu.Unlock()
+ return prependHostLabels(h.group, h.pluginName, custom)
+}
+
+// pluginMeter is the sdk.Meter adapter for one (group, pluginName) pair.
+type pluginMeter struct {
+ factory observability.Factory
+ sm *samplerMetrics
+ cache map[instrumentKey]*cachedInstrument
+ group string
+ pluginName string
+ mu sync.Mutex
+ bypass bool
+}
+
+// Counter returns or creates a bounded sdk.Counter for the plugin.
+func (m *pluginMeter) Counter(name string, labelNames ...string) sdk.Counter {
+ if m.bypass {
+ return noopCounter{}
+ }
+ clean, filtered, ok := m.prepare(name, labelNames)
+ if !ok {
+ return noopCounter{}
+ }
+ key := instrumentKey{fullName: clean, labelNames: strings.Join(filtered, "\x00")}
+ m.mu.Lock()
+ if ci, ok2 := m.cache[key]; ok2 {
+ m.mu.Unlock()
+ return ci.counter
+ }
+ m.mu.Unlock()
+
+ allLabels := prependHostLabels("group", "plugin_name", filtered)
+ var handle meter.Counter
+ func() {
+ defer func() {
+ if r := recover(); r != nil {
+ m.sm.incSeriesRejected(m.group, m.pluginName)
+ m.sm.incTelemetryPanic(m.group, m.pluginName)
+ }
+ }()
+ handle = m.factory.NewCounter(clean, allLabels...)
+ }()
+ if handle == nil {
+ return noopCounter{}
+ }
+ bc := &boundCounter{
+ handle: handle,
+ group: m.group,
+ pluginName: m.pluginName,
+ sm: m.sm,
+ seen: make(map[string]struct{}),
+ arity: len(filtered),
+ }
+ m.mu.Lock()
+ if existing, ok2 := m.cache[key]; ok2 {
+ m.mu.Unlock()
+ return existing.counter
+ }
+ m.cache[key] = &cachedInstrument{kind: kindCounter, counter: bc}
+ m.mu.Unlock()
+ return bc
+}
+
+// Gauge returns or creates a bounded sdk.Gauge for the plugin.
+func (m *pluginMeter) Gauge(name string, labelNames ...string) sdk.Gauge {
+ if m.bypass {
+ return noopGauge{}
+ }
+ clean, filtered, ok := m.prepare(name, labelNames)
+ if !ok {
+ return noopGauge{}
+ }
+ key := instrumentKey{fullName: clean, labelNames: strings.Join(filtered, "\x00")}
+ m.mu.Lock()
+ if ci, ok2 := m.cache[key]; ok2 {
+ m.mu.Unlock()
+ return ci.gauge
+ }
+ m.mu.Unlock()
+
+ allLabels := prependHostLabels("group", "plugin_name", filtered)
+ var handle meter.Gauge
+ func() {
+ defer func() {
+ if r := recover(); r != nil {
+ m.sm.incSeriesRejected(m.group, m.pluginName)
+ m.sm.incTelemetryPanic(m.group, m.pluginName)
+ }
+ }()
+ handle = m.factory.NewGauge(clean, allLabels...)
+ }()
+ if handle == nil {
+ return noopGauge{}
+ }
+ bg := &boundGauge{
+ handle: handle,
+ group: m.group,
+ pluginName: m.pluginName,
+ sm: m.sm,
+ seen: make(map[string]struct{}),
+ arity: len(filtered),
+ }
+ m.mu.Lock()
+ if existing, ok2 := m.cache[key]; ok2 {
+ m.mu.Unlock()
+ return existing.gauge
+ }
+ m.cache[key] = &cachedInstrument{kind: kindGauge, gauge: bg}
+ m.mu.Unlock()
+ return bg
+}
+
+// Histogram returns or creates a bounded sdk.Histogram for the plugin.
+func (m *pluginMeter) Histogram(name string, buckets []float64, labelNames ...string) sdk.Histogram {
+ if m.bypass {
+ return noopHistogram{}
+ }
+ clean, filtered, ok := m.prepare(name, labelNames)
+ if !ok {
+ return noopHistogram{}
+ }
+ key := instrumentKey{fullName: clean, labelNames: strings.Join(filtered, "\x00")}
+ m.mu.Lock()
+ if ci, ok2 := m.cache[key]; ok2 {
+ m.mu.Unlock()
+ return ci.histogram
+ }
+ m.mu.Unlock()
+
+ allLabels := prependHostLabels("group", "plugin_name", filtered)
+ var handle meter.Histogram
+ func() {
+ defer func() {
+ if r := recover(); r != nil {
+ m.sm.incSeriesRejected(m.group, m.pluginName)
+ m.sm.incTelemetryPanic(m.group, m.pluginName)
+ }
+ }()
+ handle = m.factory.NewHistogram(clean, meter.Buckets(buckets), allLabels...)
+ }()
+ if handle == nil {
+ return noopHistogram{}
+ }
+ bh := &boundHistogram{
+ handle: handle,
+ group: m.group,
+ pluginName: m.pluginName,
+ sm: m.sm,
+ seen: make(map[string]struct{}),
+ arity: len(filtered),
+ }
+ m.mu.Lock()
+ if existing, ok2 := m.cache[key]; ok2 {
+ m.mu.Unlock()
+ return existing.histogram
+ }
+ m.cache[key] = &cachedInstrument{kind: kindHistogram, histogram: bh}
+ m.mu.Unlock()
+ return bh
+}
+
+// prepare sanitizes the name and filters reserved label names. Returns the full
+// metric name, filtered label names, and true on success; false means the caller
+// must return a no-op instrument.
+func (m *pluginMeter) prepare(name string, labelNames []string) (string, []string, bool) {
+ sanitized := sanitizeMetricName(name)
+ if sanitized == "" {
+ m.sm.incSeriesRejected(m.group, m.pluginName)
+ return "", nil, false
+ }
+ full := "plugin_" + sanitized
+ filtered := make([]string, 0, len(labelNames))
+ for _, ln := range labelNames {
+ if _, reserved := reservedLabelSet[ln]; reserved {
+ m.sm.incSeriesRejected(m.group, m.pluginName)
+ continue
+ }
+ filtered = append(filtered, ln)
+ }
+ return full, filtered, true
+}
+
+// sanitizeMetricName lowercases the name, replaces characters outside
+// [a-z0-9_] with '_', and truncates to maxMetricNameLen.
+func sanitizeMetricName(name string) string {
+ var b strings.Builder
+ b.Grow(len(name))
+ for _, r := range name {
+ switch {
+ case r >= 'a' && r <= 'z', r >= '0' && r <= '9', r == '_':
+ b.WriteRune(r)
+ case unicode.IsUpper(r):
+ b.WriteRune(unicode.ToLower(r))
+ default:
+ b.WriteByte('_')
+ }
+ if b.Len() >= maxMetricNameLen {
+ break
+ }
+ }
+ return b.String()
+}
+
+// makeOverflowValues builds the full label-values slice for the overflow sentinel.
+func makeOverflowValues(group, pluginName string, customArity int) []string {
+ vals := make([]string, 2+customArity)
+ vals[0] = group
+ vals[1] = pluginName
+ for i := range customArity {
+ vals[2+i] = overflowSentinel
+ }
+ return vals
+}
+
+// prependHostLabels returns a new slice with the two extra elements prepended to src.
+func prependHostLabels(a, b string, src []string) []string {
+ out := make([]string, 2+len(src))
+ out[0] = a
+ out[1] = b
+ copy(out[2:], src)
+ return out
+}
+
+// padOrTrim ensures customValues has exactly arity elements, padding with ""
+// or trimming as needed.
+func padOrTrim(customValues []string, arity int) []string {
+ if len(customValues) == arity {
+ return customValues
+ }
+ out := make([]string, arity)
+ copy(out, customValues)
+ return out
+}
+
+// pluginLogger is the sdk.Logger adapter for one (group, pluginName) pair.
+// It enforces a token-bucket rate limit and caps message length and field count.
+type pluginLogger struct {
+ lastRefill time.Time
+ lastWarnEmit time.Time
+ inner *logger.Logger
+ sm *samplerMetrics
+ group string
+ pluginName string
+ tokens float64
+ suppressed int64
+ mu sync.Mutex
+}
+
+func newPluginLogger(baseLogger *logger.Logger, sm *samplerMetrics, group, pluginName string) *pluginLogger {
+ inner := baseLogger.Named("plugin", strings.ToUpper(group), strings.ToUpper(pluginName))
+ return &pluginLogger{
+ inner: inner,
+ group: group,
+ pluginName: pluginName,
+ sm: sm,
+ tokens: logBurstSize,
+ lastRefill: time.Now(),
+ }
+}
+
+// acquire attempts to consume one token. Returns true if the caller may log.
+// If a suppression summary is due it is emitted before returning true.
+func (l *pluginLogger) acquire() bool {
+ now := time.Now()
+ l.mu.Lock()
+ elapsed := now.Sub(l.lastRefill).Seconds()
+ l.lastRefill = now
+ l.tokens += elapsed * logRatePerSec
+ if l.tokens > logBurstSize {
+ l.tokens = logBurstSize
+ }
+ if l.tokens < 1 {
+ l.suppressed++
+ l.mu.Unlock()
+ l.sm.incLogDropped(l.group, l.pluginName)
+ return false
+ }
+ l.tokens--
+ n := l.suppressed
+ emitSummary := n > 0 && now.Sub(l.lastWarnEmit) >= suppressWarnInterval
+ if emitSummary {
+ l.suppressed = 0
+ l.lastWarnEmit = now
+ }
+ l.mu.Unlock()
+ if emitSummary {
+ l.inner.Warn().Int64("suppressed", n).Msg("plugin log lines suppressed by rate limiter")
+ }
+ return true
+}
+
+// emitEvent attaches keysAndValues to ev and sends it.
+func emitEvent(ev *zerolog.Event, msg string, keysAndValues []any) {
+ // Truncate message.
+ if len(msg) > maxMsgLen {
+ msg = msg[:maxMsgLen]
+ }
+ // Cap fields to maxLogFields pairs.
+ if len(keysAndValues) > maxLogFields*2 {
+ keysAndValues = keysAndValues[:maxLogFields*2]
+ }
+ // Ensure even length (trailing odd key gets value "(MISSING)").
+ if len(keysAndValues)%2 != 0 {
+ keysAndValues = append(keysAndValues, "(MISSING)")
+ }
+ for i := 0; i < len(keysAndValues); i += 2 {
+ k := anyToString(keysAndValues[i])
+ v := keysAndValues[i+1]
+ if sv, ok := v.(string); ok {
+ ev = ev.Str(k, sv)
+ } else {
+ ev = ev.Interface(k, v)
+ }
+ }
+ ev.Msg(msg)
+}
+
+// anyToString converts an any key to string, falling back to fmt.Sprintf.
+func anyToString(v any) string {
+ if s, ok := v.(string); ok {
+ return s
+ }
+ return fmt.Sprintf("%v", v)
+}
+
+// Debug is a no-op — the adapter caps at INFO level.
+func (l *pluginLogger) Debug(_ string, _ ...any) {}
+
+// Info logs at INFO level with rate-limiting and field/message caps.
+func (l *pluginLogger) Info(msg string, keysAndValues ...any) {
+ if !l.acquire() {
+ return
+ }
+ emitEvent(l.inner.Info(), msg, keysAndValues)
+}
+
+// Warn logs at WARN level with rate-limiting and field/message caps.
+func (l *pluginLogger) Warn(msg string, keysAndValues ...any) {
+ if !l.acquire() {
+ return
+ }
+ emitEvent(l.inner.Warn(), msg, keysAndValues)
+}
+
+// Error logs at ERROR level with rate-limiting and field/message caps.
+func (l *pluginLogger) Error(msg string, keysAndValues ...any) {
+ if !l.acquire() {
+ return
+ }
+ emitEvent(l.inner.Error(), msg, keysAndValues)
+}
+
+// pluginTelemetryRegistry tracks the live pluginTelemetry adapters per group so a
+// group's plugin metric series can be Deleted when the group's samplers are
+// removed or replaced. It is a leaf lock: it never calls back into pluginCache.
+var pluginTelemetryRegistry = struct {
+ m map[string][]*pluginTelemetry
+ mu sync.Mutex
+}{m: map[string][]*pluginTelemetry{}}
+
+// registerGroupTelemetry appends t to group's adapter slice, deduping by pointer
+// so a re-registration of an already-tracked adapter cannot accumulate
+// duplicates. In normal flow bindHost only fires on a loader cache-miss (a fresh
+// instance ⇒ a fresh adapter), so each distinct adapter is registered once.
+func registerGroupTelemetry(group string, t *pluginTelemetry) {
+ pluginTelemetryRegistry.mu.Lock()
+ defer pluginTelemetryRegistry.mu.Unlock()
+ if slices.Contains(pluginTelemetryRegistry.m[group], t) {
+ return
+ }
+ pluginTelemetryRegistry.m[group] = append(pluginTelemetryRegistry.m[group], t)
+}
+
+// teardownGroupTelemetry Deletes the metric series of every adapter tracked for
+// group, but RETAINS the registry entries. An adapter's lifetime equals its
+// immortal, never-Closed sampler instance's (both keyed by
+// (path,symbol,configHash,group) in the loader cache): a config revert re-serves
+// the cached instance, which re-emits through the SAME retained adapter, so the
+// adapter must stay findable for a subsequent teardown — otherwise those
+// re-created series would leak. Deleting an already-Deleted tuple is a harmless
+// no-op, so re-tearing-down a retained adapter is safe. Retained adapters are
+// bounded by the distinct (configHash,group) pairs seen over the process
+// lifetime — the same bound as the immortal instances themselves.
+func teardownGroupTelemetry(group string) {
+ pluginTelemetryRegistry.mu.Lock()
+ adapters := slices.Clone(pluginTelemetryRegistry.m[group])
+ pluginTelemetryRegistry.mu.Unlock()
+ for _, t := range adapters {
+ t.teardown()
+ }
+}
+
+// pluginTelemetry is the sdk.Host implementation for one (group, pluginName) pair.
+type pluginTelemetry struct {
+ meterAdapter *pluginMeter
+ loggerAdapter *pluginLogger
+ sm *samplerMetrics
+}
+
+// newPluginTelemetry constructs a pluginTelemetry adapter. If factory is nil
+// the bypass path is used (no-op metrics, zero allocations on Inc/Set/Observe).
+func newPluginTelemetry(factory observability.Factory, baseLogger *logger.Logger, sm *samplerMetrics, group, pluginName string) *pluginTelemetry {
+ bypass := factory == nil
+ if bypass {
+ factory = observability.BypassRegistry.With(pipelineScope)
+ }
+ if baseLogger == nil {
+ baseLogger = logger.GetLogger("trace")
+ }
+ return &pluginTelemetry{
+ meterAdapter: &pluginMeter{
+ factory: factory,
+ group: group,
+ pluginName: pluginName,
+ sm: sm,
+ bypass: bypass,
+ cache: make(map[instrumentKey]*cachedInstrument),
+ },
+ loggerAdapter: newPluginLogger(baseLogger, sm, group, pluginName),
+ sm: sm,
+ }
+}
+
+// Meter returns the metric surface scoped to this plugin.
+func (t *pluginTelemetry) Meter() sdk.Meter {
+ if t.meterAdapter.bypass {
+ return noopMeter{}
+ }
+ return t.meterAdapter
+}
+
+// Logger returns the logging surface scoped to this plugin.
+func (t *pluginTelemetry) Logger() sdk.Logger {
+ return t.loggerAdapter
+}
+
+// teardown deletes all metric series registered by this plugin and clears the
+// cache. It is idempotent and panic-safe.
+func (t *pluginTelemetry) teardown() {
+ defer func() {
+ if r := recover(); r != nil {
+ t.sm.incTelemetryPanic(t.meterAdapter.group, t.meterAdapter.pluginName)
+ }
+ }()
+
+ t.meterAdapter.mu.Lock()
+ cache := t.meterAdapter.cache
+ t.meterAdapter.cache = make(map[instrumentKey]*cachedInstrument)
+ t.meterAdapter.mu.Unlock()
+
+ for _, ci := range cache {
+ switch ci.kind {
+ case kindCounter:
+ ci.counter.mu.Lock()
+ emitted := ci.counter.emitted
+ ci.counter.mu.Unlock()
+ for _, labelValues := range emitted {
+ safeDelete(func() { ci.counter.handle.Delete(labelValues...) })
+ }
+ case kindGauge:
+ ci.gauge.mu.Lock()
+ emitted := ci.gauge.emitted
+ ci.gauge.mu.Unlock()
+ for _, labelValues := range emitted {
+ safeDelete(func() { ci.gauge.handle.Delete(labelValues...) })
+ }
+ case kindHistogram:
+ ci.histogram.mu.Lock()
+ emitted := ci.histogram.emitted
+ ci.histogram.mu.Unlock()
+ for _, labelValues := range emitted {
+ safeDelete(func() { ci.histogram.handle.Delete(labelValues...) })
+ }
+ }
+ }
+}
+
+// safeDelete calls fn recovering any panic so a single bad Delete cannot abort
+// teardown of other series.
+func safeDelete(fn func()) {
+ defer func() { recover() }() //nolint:errcheck
+ fn()
+}
diff --git a/banyand/trace/plugin_telemetry_integration_test.go b/banyand/trace/plugin_telemetry_integration_test.go
new file mode 100644
index 000000000..767c47a2b
--- /dev/null
+++ b/banyand/trace/plugin_telemetry_integration_test.go
@@ -0,0 +1,550 @@
+// 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"
+ "path/filepath"
+ "sync"
+ "sync/atomic"
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+ "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"
+)
+
+// fakeHostAwareSampler is a Go-level sdk.Sampler + sdk.HostAware fake used to
+// exercise the cache/keying/bindHost/telemetry-attribution logic under -race
+// without a real .so. It is the "Host-caching plugin" the once-only invariant
+// hinges on: it caches the injected Host in UseHost and emits attribution
+// telemetry through it in Decide.
+type fakeHostAwareSampler struct {
+ host sdk.Host
+ useHostCount int64
+ panicOnUseHost bool
+ panicOnDecide bool
+ mu sync.Mutex
+}
+
+// UseHost caches the engine-owned Host and counts invocations. If configured to
+// panic it does so here to prove the engine-side recover still holds.
+func (f *fakeHostAwareSampler) UseHost(h sdk.Host) {
+ atomic.AddInt64(&f.useHostCount, 1)
+ if f.panicOnUseHost {
+ panic("fakeHostAwareSampler: UseHost panic")
+ }
+ f.mu.Lock()
+ f.host = h
+ f.mu.Unlock()
+}
+
+// Decide emits attribution telemetry through the cached Host (proving the emit
+// carries this instance's group) and retains the whole batch.
+func (f *fakeHostAwareSampler) Decide(batch *sdk.TraceBatch) (sdk.Verdict, error) {
+ if f.panicOnDecide {
+ panic("fakeHostAwareSampler: Decide panic")
+ }
+ f.mu.Lock()
+ h := f.host
+ f.mu.Unlock()
+ if h != nil {
+ h.Meter().Counter("decides").Inc(1)
+ h.Logger().Info("decide", "n", len(batch.Traces))
+ }
+ keep := make([]bool, len(batch.Traces))
+ for idx := range keep {
+ keep[idx] = true
+ }
+ return sdk.Verdict{Keep: keep}, nil
+}
+
+func (f *fakeHostAwareSampler) Kind() sdk.Kind { return sdk.KindSampler }
+func (f *fakeHostAwareSampler) Project() sdk.Projection { return sdk.Projection{} }
+func (f *fakeHostAwareSampler) Close() error { return nil }
+
+// UseHostCount returns the number of UseHost invocations observed so far.
+func (f *fakeHostAwareSampler) UseHostCount() int64 { return atomic.LoadInt64(&f.useHostCount) }
+
+// fakeNonHostAware implements only sdk.Sampler (no UseHost) to verify the
+// non-HostAware path is left untouched by telemetry wiring.
+type fakeNonHostAware struct{}
+
+func (fakeNonHostAware) Kind() sdk.Kind { return sdk.KindSampler }
+func (fakeNonHostAware) Project() sdk.Projection { return sdk.Projection{} }
+func (fakeNonHostAware) 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 (fakeNonHostAware) Close() error { return nil }
+
+// makeTelemetrySchemaRepo builds a ROLE_DATA schemaRepo wired to factory for both
+// samplerMeter and pluginTelemetryFactory, with trustedPluginDir pointing at a
+// real temp dir. It also creates the dummy file named by pluginPath inside that
+// dir so the path-within/symlink guards in loadSamplerPlugin pass before the
+// newSamplerFromPlugin seam (which the test overrides) is reached.
+func makeTelemetrySchemaRepo(t *testing.T, factory *fakeMetricFactory) (schemaRepo, string) {
+ t.Helper()
+ trustedDir := t.TempDir()
+ const pluginPath = "fake.so"
+ dummyFile := filepath.Join(trustedDir, pluginPath)
+ require.NoError(t, os.WriteFile(dummyFile, []byte("not a real plugin"), 0o600))
+ sr := schemaRepo{
+ l: logger.GetLogger("plugin-telemetry-integration-test"),
+ role: databasev1.Role_ROLE_DATA,
+ nativePipelineEnabled: true,
+ trustedPluginDir: trustedDir,
+ samplerMeter: newSamplerMetrics(factory),
+ pluginTelemetryFactory: factory,
+ }
+ return sr, pluginPath
+}
+
+// makeTelemetryPipelineCfg builds an enabled, merge-event pipeline config with a
+// single sampler plugin named pluginName at pluginPath.
+func makeTelemetryPipelineCfg(pluginName, pluginPath string) *commonv1.TracePipelineConfig {
+ return &commonv1.TracePipelineConfig{
+ Enabled: true,
+ EnabledEvents: []commonv1.PipelineEvent{commonv1.PipelineEvent_PIPELINE_EVENT_MERGE},
+ Plugins: []*commonv1.Plugin{
+ {
+ Name: pluginName,
+ Kind: &commonv1.Plugin_Sampler{Sampler: &commonv1.SamplerPlugin{
+ Path: pluginPath,
+ Symbol: "NewSampler",
+ AbiVersion: uint32(sdk.ABIVersion),
+ }},
+ },
+ },
+ }
+}
+
+// singleTraceBatch returns a one-trace read-only batch for driving Decide.
+func singleTraceBatch() *sdk.TraceBatch {
+ return &sdk.TraceBatch{Traces: []sdk.TraceBlock{{TraceID: "t"}}}
+}
+
+// TestPluginTelemetryIntegration_CrossGroupAttribution asserts that two groups
+// sharing the same plugin path+config each construct a DISTINCT HostAware
+// instance, each receives exactly one UseHost, and each instance's Decide emits
+// telemetry carrying ITS OWN group label — no cross-group label clobber.
+func TestPluginTelemetryIntegration_CrossGroupAttribution(t *testing.T) {
+ resetRegistries()
+ defer resetRegistries()
+
+ factory := newFakeMetricFactory()
+ sr, pluginPath := makeTelemetrySchemaRepo(t, factory)
+ const pluginName = "lss"
+
+ var constructed []*fakeHostAwareSampler
+ var mu sync.Mutex
+ saved := newSamplerFromPlugin
+ newSamplerFromPlugin = func(_, _ string, _ []byte) (sdk.Sampler, error) {
+ s := &fakeHostAwareSampler{}
+ mu.Lock()
+ constructed = append(constructed, s)
+ mu.Unlock()
+ return s, nil
+ }
+ defer func() { newSamplerFromPlugin = saved }()
+
+ cfg := makeTelemetryPipelineCfg(pluginName, pluginPath)
+ sr.reconcilePipeline("A", cfg)
+ sr.reconcilePipeline("B", cfg)
+
+ samplersA := lookupSamplers("A")
+ samplersB := lookupSamplers("B")
+ require.Len(t, samplersA, 1)
+ require.Len(t, samplersB, 1)
+
+ instA, okA := samplersA[0].(*fakeHostAwareSampler)
+ instB, okB := samplersB[0].(*fakeHostAwareSampler)
+ require.True(t, okA)
+ require.True(t, okB)
+ assert.NotSame(t, instA, instB, "each group must construct a distinct instance")
+ assert.Equal(t, int64(1), instA.UseHostCount(), "group A instance must be hosted exactly once")
+ assert.Equal(t, int64(1), instB.UseHostCount(), "group B instance must be hosted exactly once")
+
+ // Drive each group's Decide; the emit must carry that group's label.
+ _, decideErrA := samplersA[0].Decide(singleTraceBatch())
+ require.NoError(t, decideErrA)
+ _, decideErrB := samplersB[0].Decide(singleTraceBatch())
+ require.NoError(t, decideErrB)
+
+ decides := factory.counter("plugin_decides")
+ require.NotNil(t, decides, "the plugin 'decides' counter must have been created")
+ assert.Equal(t, 1, decides.callsWithLabels("A", pluginName),
+ "group A Decide must emit decides{group=A,plugin_name=lss}")
+ assert.Equal(t, 1, decides.callsWithLabels("B", pluginName),
+ "group B Decide must emit decides{group=B,plugin_name=lss}")
+}
+
+// TestPluginTelemetryIntegration_ReconcileIdempotentOnceOnly asserts that a
+// second identical reconcile reuses the same instance and does NOT re-run
+// UseHost or reconstruct (idempotent-skip + once-only).
+func TestPluginTelemetryIntegration_ReconcileIdempotentOnceOnly(t *testing.T) {
+ resetRegistries()
+ defer resetRegistries()
+
+ factory := newFakeMetricFactory()
+ sr, pluginPath := makeTelemetrySchemaRepo(t, factory)
+ const pluginName = "lss"
+
+ var constructCount int64
+ saved := newSamplerFromPlugin
+ newSamplerFromPlugin = func(_, _ string, _ []byte) (sdk.Sampler, error) {
+ atomic.AddInt64(&constructCount, 1)
+ return &fakeHostAwareSampler{}, nil
+ }
+ defer func() { newSamplerFromPlugin = saved }()
+
+ cfg := makeTelemetryPipelineCfg(pluginName, pluginPath)
+ sr.reconcilePipeline("A", cfg)
+ first := lookupSamplers("A")
+ require.Len(t, first, 1)
+ inst, ok := first[0].(*fakeHostAwareSampler)
+ require.True(t, ok)
+ require.Equal(t, int64(1), inst.UseHostCount())
+ require.Equal(t, int64(1), atomic.LoadInt64(&constructCount))
+
+ sr.reconcilePipeline("A", cfg)
+ second := lookupSamplers("A")
+ require.Len(t, second, 1)
+ assert.Same(t, inst, second[0], "identical reconcile must reuse the same instance")
+ assert.Equal(t, int64(1), inst.UseHostCount(), "once-only: UseHost must not re-run on idempotent reconcile")
+ assert.Equal(t, int64(1), atomic.LoadInt64(&constructCount), "idempotent reconcile must not reconstruct")
+}
+
+// TestPluginTelemetryIntegration_ConcurrentReconcileDecideRace drives concurrent
+// redundant reconcile replays and concurrent Decide calls under the race
+// detector. The once-only invariant (UseHostCount == 1) must hold even under
+// concurrent replay, with no data race.
+func TestPluginTelemetryIntegration_ConcurrentReconcileDecideRace(t *testing.T) {
+ resetRegistries()
+ defer resetRegistries()
+
+ factory := newFakeMetricFactory()
+ sr, pluginPath := makeTelemetrySchemaRepo(t, factory)
+ const pluginName = "lss"
+
+ saved := newSamplerFromPlugin
+ newSamplerFromPlugin = func(_, _ string, _ []byte) (sdk.Sampler, error) {
+ return &fakeHostAwareSampler{}, nil
+ }
+ defer func() { newSamplerFromPlugin = saved }()
+
+ cfg := makeTelemetryPipelineCfg(pluginName, pluginPath)
+ // Seed the instance once so lookups below always find a sampler.
+ sr.reconcilePipeline("A", cfg)
+ seeded := lookupSamplers("A")
+ require.Len(t, seeded, 1)
+ inst, ok := seeded[0].(*fakeHostAwareSampler)
+ require.True(t, ok)
+
+ var wg sync.WaitGroup
+ const reconcilers = 4
+ const deciders = 4
+ const iterations = 50
+ for range reconcilers {
+ wg.Go(func() {
+ for range iterations {
+ sr.reconcilePipeline("A", cfg)
+ }
+ })
+ }
+ for range deciders {
+ wg.Go(func() {
+ for range iterations {
+ samplers := lookupSamplers("A")
+ if len(samplers) > 0 {
+ _, _ = samplers[0].Decide(singleTraceBatch())
+ }
+ }
+ })
+ }
+ wg.Wait()
+
+ assert.Equal(t, int64(1), inst.UseHostCount(),
+ "once-only: UseHost must run exactly once even under concurrent reconcile replay")
+ assert.Same(t, inst, lookupSamplers("A")[0], "the same instance must remain registered")
+}
+
+// TestPluginTelemetryIntegration_TeardownDeletesOnlyThatGroup asserts that
+// removing group A deletes only A's plugin metric series, leaving B intact.
+func TestPluginTelemetryIntegration_TeardownDeletesOnlyThatGroup(t *testing.T) {
+ resetRegistries()
+ defer resetRegistries()
+
+ factory := newFakeMetricFactory()
+ sr, pluginPath := makeTelemetrySchemaRepo(t, factory)
+ const pluginName = "lss"
+
+ saved := newSamplerFromPlugin
+ newSamplerFromPlugin = func(_, _ string, _ []byte) (sdk.Sampler, error) {
+ return &fakeHostAwareSampler{}, nil
+ }
+ defer func() { newSamplerFromPlugin = saved }()
+
+ cfg := makeTelemetryPipelineCfg(pluginName, pluginPath)
+ sr.reconcilePipeline("A", cfg)
+ sr.reconcilePipeline("B", cfg)
+
+ // Emit a series for each group so there is something to delete.
+ _, decideErrA := lookupSamplers("A")[0].Decide(singleTraceBatch())
+ require.NoError(t, decideErrA)
+ _, decideErrB := lookupSamplers("B")[0].Decide(singleTraceBatch())
+ require.NoError(t, decideErrB)
+
+ // Remove group A.
+ sr.reconcilePipeline("A", nil)
+
+ decides := factory.counter("plugin_decides")
+ require.NotNil(t, decides)
+ assert.Equal(t, 1, decides.deletesWithPrefix("A", pluginName),
+ "teardown must Delete A's series")
+ assert.Equal(t, 0, decides.deletesWithPrefix("B", pluginName),
+ "teardown must NOT Delete B's series")
+
+ assert.Empty(t, lookupSamplers("A"), "group A must be removed")
+ require.Len(t, lookupSamplers("B"), 1, "group B must remain intact")
+}
+
+// TestPluginTelemetryIntegration_NonHostAwareUntouched asserts a non-HostAware
+// sampler is registered and runnable with no telemetry, no UseHost, and no panic.
+func TestPluginTelemetryIntegration_NonHostAwareUntouched(t *testing.T) {
+ resetRegistries()
+ defer resetRegistries()
+
+ factory := newFakeMetricFactory()
+ sr, pluginPath := makeTelemetrySchemaRepo(t, factory)
+ const pluginName = "plain"
+
+ saved := newSamplerFromPlugin
+ newSamplerFromPlugin = func(_, _ string, _ []byte) (sdk.Sampler, error) {
+ return fakeNonHostAware{}, nil
+ }
+ defer func() { newSamplerFromPlugin = saved }()
+
+ cfg := makeTelemetryPipelineCfg(pluginName, pluginPath)
+ require.NotPanics(t, func() { sr.reconcilePipeline("A", cfg) })
+
+ samplers := lookupSamplers("A")
+ require.Len(t, samplers, 1)
+ _, ok := samplers[0].(*fakeHostAwareSampler)
+ assert.False(t, ok, "registered sampler must be the non-HostAware fake")
+ verdict, decideErr := samplers[0].Decide(singleTraceBatch())
+ require.NoError(t, decideErr)
+ assert.Equal(t, []bool{true}, verdict.Keep, "non-HostAware sampler must still run")
+
+ // No plugin telemetry series must have been created for the non-HostAware path.
+ assert.Nil(t, factory.counter("plugin_decides"),
+ "non-HostAware sampler must not create plugin telemetry series")
+}
+
+// TestPluginTelemetryIntegration_UseHostPanicIsolation asserts that a UseHost
+// panic is isolated: the host does not panic, the sampler is still registered
+// and its Decide runs (telemetry disabled), and plugin_telemetry_panic_total is
+// incremented for {group, plugin}.
+func TestPluginTelemetryIntegration_UseHostPanicIsolation(t *testing.T) {
+ resetRegistries()
+ defer resetRegistries()
+
+ factory := newFakeMetricFactory()
+ sr, pluginPath := makeTelemetrySchemaRepo(t, factory)
+ const pluginName = "lss"
+
+ saved := newSamplerFromPlugin
+ newSamplerFromPlugin = func(_, _ string, _ []byte) (sdk.Sampler, error) {
+ return &fakeHostAwareSampler{panicOnUseHost: true}, nil
+ }
+ defer func() { newSamplerFromPlugin = saved }()
+
+ cfg := makeTelemetryPipelineCfg(pluginName, pluginPath)
+ require.NotPanics(t, func() { sr.reconcilePipeline("A", cfg) },
+ "a UseHost panic must not propagate to the host")
+
+ samplers := lookupSamplers("A")
+ require.Len(t, samplers, 1, "sampler must still be registered despite the UseHost panic")
+ verdict, decideErr := samplers[0].Decide(singleTraceBatch())
+ require.NoError(t, decideErr)
+ assert.Equal(t, []bool{true}, verdict.Keep, "sampler Decide must still run with telemetry disabled")
+
+ panicCounter := factory.counter("plugin_telemetry_panic_total")
+ require.NotNil(t, panicCounter)
+ assert.Equal(t, 1, panicCounter.callsWithLabels("A", pluginName),
+ "a UseHost panic must increment plugin_telemetry_panic_total{group=A,plugin_name=lss}")
+}
+
+// TestPluginTelemetryIntegration_TeardownRetainsAdapterForReuse proves that
+// teardownGroupTelemetry RETAINS registry entries so a config-revert that
+// re-serves the immortal cached instance can re-emit series and have them
+// Deleted again on a subsequent teardown. Under the old delete(m,group)
+// behavior the second teardown would find no adapters and the re-created
+// series would leak; under the RETAIN behavior the Delete count for the group
+// tuple reaches >= 2.
+func TestPluginTelemetryIntegration_TeardownRetainsAdapterForReuse(t *testing.T) {
+ resetRegistries()
+ defer resetRegistries()
+
+ factory := newFakeMetricFactory()
+ sr, pluginPath := makeTelemetrySchemaRepo(t, factory)
+ const pluginName = "lss"
+
+ saved := newSamplerFromPlugin
+ newSamplerFromPlugin = func(_, _ string, _ []byte) (sdk.Sampler, error) {
+ return &fakeHostAwareSampler{}, nil
+ }
+ defer func() { newSamplerFromPlugin = saved }()
+
+ cfg := makeTelemetryPipelineCfg(pluginName, pluginPath)
+ sr.reconcilePipeline("A", cfg)
+
+ // Step 3: drive Decide to emit the "decides" series.
+ samplersFirst := lookupSamplers("A")
+ require.Len(t, samplersFirst, 1, "group A must have exactly one sampler after first reconcile")
+ inst, ok := samplersFirst[0].(*fakeHostAwareSampler)
+ require.True(t, ok)
+ _, decideErr := inst.Decide(singleTraceBatch())
+ require.NoError(t, decideErr)
+
+ // Step 4: first teardown — must Delete the emitted "A" tuple.
+ teardownGroupTelemetry("A")
+ decides := factory.counter("plugin_decides")
+ require.NotNil(t, decides, "plugin_decides counter must exist after Decide emits it")
+ deletesAfterFirst := decides.deletesWithPrefix("A", pluginName)
+ assert.GreaterOrEqual(t, deletesAfterFirst, 1,
+ "first teardown must Delete the A-scoped series")
+
+ // Step 5: the SAME sampler Decide again (simulates a revert re-serving the
+ // immortal cached instance): this re-creates the series in the adapter.
+ _, reDecideErr := inst.Decide(singleTraceBatch())
+ require.NoError(t, reDecideErr)
+
+ // Step 6: second teardown — the retained adapter must still be findable and
+ // Delete the re-created tuple a second time. This is the invariant that would
+ // FAIL if teardown used delete(m,group) instead of retaining entries.
+ teardownGroupTelemetry("A")
+ deletesAfterSecond := decides.deletesWithPrefix("A", pluginName)
+ assert.GreaterOrEqual(t, deletesAfterSecond, 2,
+ "second teardown must Delete the re-created A-scoped series (adapter retained)")
+}
+
+// TestPluginTelemetryIntegration_ConcurrentDistinctConfigRace proves that
+// racing reconcile calls for two DISTINCT configs on the same group each
+// construct-and-host their respective instance exactly once, even under the
+// race detector. The once-only UseHost invariant must hold for both instances
+// across concurrent construction.
+func TestPluginTelemetryIntegration_ConcurrentDistinctConfigRace(t *testing.T) {
+ resetRegistries()
+ defer resetRegistries()
+
+ factory := newFakeMetricFactory()
+ sr, pluginPath := makeTelemetrySchemaRepo(t, factory)
+ const pluginName = "lss"
+
+ // instanceByJSON maps cfgJSON (string) to the stable fakeHostAwareSampler
+ // allocated on first-miss for that config. Concurrent construction of the
+ // same cfgJSON always returns the pointer stored on the first call.
+ var instanceMu sync.Mutex
+ instanceByJSON := make(map[string]*fakeHostAwareSampler)
+
+ saved := newSamplerFromPlugin
+ newSamplerFromPlugin = func(_, _ string, cfgJSON []byte) (sdk.Sampler, error) {
+ key := string(cfgJSON)
+ instanceMu.Lock()
+ defer instanceMu.Unlock()
+ if s, exists := instanceByJSON[key]; exists {
+ return s, nil
+ }
+ s := &fakeHostAwareSampler{}
+ instanceByJSON[key] = s
+ return s, nil
+ }
+ defer func() { newSamplerFromPlugin = saved }()
+
+ // Build two configs that differ in their structpb config so computeConfigHash
+ // returns distinct hashes, causing two distinct pluginCacheKey entries.
+ makeCfgWithThreshold := func(thresholdMs float64) *commonv1.TracePipelineConfig {
+ cfgStruct, structErr := structpb.NewStruct(map[string]any{
+ "thresholdMs": thresholdMs,
+ })
+ require.NoError(t, structErr)
+ return &commonv1.TracePipelineConfig{
+ Enabled: true,
+ EnabledEvents: []commonv1.PipelineEvent{commonv1.PipelineEvent_PIPELINE_EVENT_MERGE},
+ Plugins: []*commonv1.Plugin{
+ {
+ Name: pluginName,
+ Kind: &commonv1.Plugin_Sampler{Sampler: &commonv1.SamplerPlugin{
+ Path: pluginPath,
+ Symbol: "NewSampler",
+ AbiVersion: uint32(sdk.ABIVersion),
+ Config: cfgStruct,
+ }},
+ },
+ },
+ }
+ }
+ cfg1 := makeCfgWithThreshold(100)
+ cfg2 := makeCfgWithThreshold(200)
+
+ // Launch two goroutine groups racing reconcile for cfg1 and cfg2 on group A,
+ // plus a third goroutine that reads and Decides concurrently.
+ const iterations = 50
+ var wg sync.WaitGroup
+
+ wg.Go(func() {
+ for range iterations {
+ sr.reconcilePipeline("A", cfg1)
+ }
+ })
+
+ wg.Go(func() {
+ for range iterations {
+ sr.reconcilePipeline("A", cfg2)
+ }
+ })
+
+ wg.Go(func() {
+ for range iterations {
+ samplers := lookupSamplers("A")
+ if len(samplers) > 0 {
+ _, _ = samplers[0].Decide(singleTraceBatch())
+ }
+ }
+ })
+
+ wg.Wait()
+
+ // Verify: no data race (the race detector enforces this implicitly), and
+ // each distinct constructed instance was hosted exactly once.
+ instanceMu.Lock()
+ defer instanceMu.Unlock()
+ for cfgKey, inst := range instanceByJSON {
+ assert.Equal(t, int64(1), inst.UseHostCount(),
+ "each distinct constructed instance must be hosted exactly once (cfgKey=%q)", cfgKey)
+ }
+}
diff --git a/banyand/trace/plugin_telemetry_realso_test.go b/banyand/trace/plugin_telemetry_realso_test.go
new file mode 100644
index 000000000..f54b5d369
--- /dev/null
+++ b/banyand/trace/plugin_telemetry_realso_test.go
@@ -0,0 +1,746 @@
+// 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 trace
+
+import (
+ "io"
+ "os"
+ "os/exec"
+ "path/filepath"
+ "plugin"
+ "runtime"
+ "strings"
+ "sync"
+ "syscall"
+ "testing"
+
+ "github.com/prometheus/client_golang/prometheus"
+ dto "github.com/prometheus/client_model/go"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+ "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/banyand/observability/services"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+ "github.com/apache/skywalking-banyandb/pkg/meter/prom"
+ "github.com/apache/skywalking-banyandb/pkg/pipeline/sdk"
+)
+
+// telemetryPluginPkgPath is the path to the telemetrysampler plugin source
+// relative to this file's directory (banyand/trace).
+const telemetryPluginPkgPath = "../../test/plugins/_telemetrysampler"
+
+// captureStderr redirects the process-level stderr file descriptor around fn,
+// capturing all bytes written to fd 2 during the call. It reads the pipe after
+// closing the write end, so small output (a few log lines) is collected without
+// risk of deadlock.
+func captureStderr(t *testing.T, fn func()) string {
+ t.Helper()
+
+ r, w, pipeErr := os.Pipe()
+ require.NoError(t, pipeErr, "os.Pipe for stderr capture must succeed")
+
+ savedFd, dupErr := syscall.Dup(int(os.Stderr.Fd()))
+ require.NoError(t, dupErr, "syscall.Dup of stderr must succeed")
+
+ dup2Err := syscall.Dup2(int(w.Fd()), int(os.Stderr.Fd()))
+ require.NoError(t, dup2Err, "syscall.Dup2 redirecting stderr to pipe must succeed")
+
+ fn()
+
+ // Close the write end so the reader sees EOF.
+ w.Close()
+
+ // Restore the original stderr fd before reading (keeps test output visible).
+ _ = syscall.Dup2(savedFd, int(os.Stderr.Fd()))
+ _ = syscall.Close(savedFd)
+
+ captured, readErr := io.ReadAll(r)
+ r.Close()
+ require.NoError(t, readErr, "reading captured stderr must succeed")
+
+ return string(captured)
+}
+
+// labelValue returns the value of label name in m, or "" if not found.
+func labelValue(m *dto.Metric, name string) string {
+ for _, lp := range m.GetLabel() {
+ if lp.GetName() == name {
+ return lp.GetValue()
+ }
+ }
+ return ""
+}
+
+// TestPluginTelemetry_RealSO_MetricsAndLogging loads a REAL .so built from
+// ../../test/plugins/_telemetrysampler and verifies that:
+// - reconcilePipeline wires the plugin into the real prom-backed meter;
+// - one Decide call on a 3-trace batch increments the
+// banyandb_trace_pipeline_plugin_decisions{verdict="keep"} counter by 3;
+// - the plugin emits a structured log line through the global logger (→ stderr).
+//
+// The test skips when the race detector is active (a non-race .so cannot be
+// loaded into a race host), when CGO is unavailable, or when the toolchain
+// versions diverge.
+func TestPluginTelemetry_RealSO_MetricsAndLogging(t *testing.T) {
+ if raceDetectorEnabled {
+ t.Skip("TestPluginTelemetry_RealSO_MetricsAndLogging skipped: race detector active; " +
+ "a non-race .so cannot be loaded into a race-instrumented host")
+ }
+
+ resetRegistries()
+ defer resetRegistries()
+
+ // newRealSOSchemaRepo handles build, probe, and skip-on-failure.
+ // All real-.so tests share a single .so (see newRealSOSchemaRepo).
+ sr, soName, reg := newRealSOSchemaRepo(t)
+
+ const group = "realso-group"
+ const pluginName = "tel"
+
+ // logEvery=100 (the plugin default): batchN==1 satisfies 1%100==1 → logs on
+ // the very first batch. logEvery=1 would give 1%1==0 which never matches.
+ cfgStruct, structErr := structpb.NewStruct(map[string]any{"logEvery": float64(100)})
+ require.NoError(t, structErr, "structpb.NewStruct for plugin config must succeed")
+
+ cfg := &commonv1.TracePipelineConfig{
+ Enabled: true,
+ Plugins: []*commonv1.Plugin{
+ {
+ Name: pluginName,
+ Kind: &commonv1.Plugin_Sampler{
+ Sampler: &commonv1.SamplerPlugin{
+ Path: soName,
+ Symbol: "NewSampler",
+ AbiVersion: uint32(sdk.ABIVersion),
+ Config: cfgStruct,
+ },
+ },
+ },
+ },
+ }
+
+ sr.reconcilePipeline(group, cfg)
+
+ samplers := lookupSamplers(group)
+ require.NotEmpty(t, samplers, "reconcilePipeline must register at least one sampler for group %q", group)
+
+ // Invoke Decide while capturing stderr to collect the structured log line.
+ batch := &sdk.TraceBatch{Traces: make([]sdk.TraceBlock, 3)}
+ var verdict sdk.Verdict
+ var decideErr error
+ captured := captureStderr(t, func() {
+ verdict, decideErr = samplers[0].Decide(batch)
+ })
+
+ require.NoError(t, decideErr, "Decide must not return an error")
+ require.Len(t, verdict.Keep, 3, "Decide must return a Keep slice of length 3")
+ for idx, keep := range verdict.Keep {
+ assert.True(t, keep, "verdict.Keep[%d] must be true (sampler keeps all traces)", idx)
+ }
+
+ // --- Metrics assertion ---
+ // The host meter adapter names the metric "plugin_decisions"; prom.NewProvider
+ // prefixes with the pipeline scope namespace producing:
+ // banyandb_trace_pipeline_plugin_decisions{group, plugin_name, verdict}
+ mfs, gatherErr := reg.Gather()
+ require.NoError(t, gatherErr, "prometheus registry Gather must succeed")
+
+ const wantFamily = "banyandb_trace_pipeline_plugin_decisions"
+ var foundFamily *dto.MetricFamily
+ for _, mf := range mfs {
+ if mf.GetName() == wantFamily {
+ foundFamily = mf
+ break
+ }
+ }
+ require.NotNilf(t, foundFamily,
+ "metric family %q not found in gathered families; got: %v",
+ wantFamily, metricFamilyNames(mfs))
+
+ var foundMetric *dto.Metric
+ for _, m := range foundFamily.GetMetric() {
+ if labelValue(m, "group") == group &&
+ labelValue(m, "plugin_name") == pluginName &&
+ labelValue(m, "verdict") == "keep" {
+ foundMetric = m
+ break
+ }
+ }
+ require.NotNilf(t, foundMetric,
+ "metric series {group=%q, plugin_name=%q, verdict=%q} not found in family %q",
+ group, pluginName, "keep", wantFamily)
+
+ assert.Equal(t, float64(3), foundMetric.GetCounter().GetValue(),
+ "plugin_decisions counter must equal the number of traces in the batch (3)")
+
+ // --- Logging assertion ---
+ // The host logger adapter names the module TRACE.PLUGIN.<UPPER(GROUP)>.<UPPER(PLUGIN)>.
+ wantModule := "TRACE.PLUGIN." + strings.ToUpper(group) + "." + strings.ToUpper(pluginName)
+ wantMsg := "batch decided"
+ assert.True(t, strings.Contains(captured, wantModule),
+ "captured stderr must contain the plugin logger module name %q; got:\n%s", wantModule, captured)
+ assert.True(t, strings.Contains(captured, wantMsg),
+ "captured stderr must contain the log message %q; got:\n%s", wantMsg, captured)
+}
+
+// metricFamilyNames returns the names of all metric families for use in
+// diagnostic failure messages.
+func metricFamilyNames(mfs []*dto.MetricFamily) []string {
+ names := make([]string, 0, len(mfs))
+ for _, mf := range mfs {
+ names = append(names, mf.GetName())
+ }
+ return names
+}
+
+// sharedSO holds the result of the once-per-process .so build and probe.
+// Go deduplicates plugins by their internal package path (not by filesystem
+// path), so all real-.so tests must share a single .so file: a second
+// plugin.Open on a different filesystem path but the same internal plugin
+// path causes a "previous failure" error in the Go runtime.
+var sharedSO struct {
+ dir string
+ soPath string
+ skipReason string // non-empty → tests must skip
+ sync.Once
+}
+
+// initSharedSO builds the telemetrysampler .so exactly once into a
+// process-scoped temp directory and probes that the host can load it.
+// Results are stored in sharedSO; the caller must check sharedSO.skipReason.
+func initSharedSO() {
+ // os.MkdirTemp creates a directory that persists for the process lifetime
+ // (not cleaned by t.Cleanup), which is intentional: the .so must outlive
+ // any individual test's cleanup to prevent the path from being invalidated
+ // while the process-level plugin registration still references it.
+ dir, mkErr := os.MkdirTemp("", "realso-shared-*")
+ if mkErr != nil {
+ sharedSO.skipReason = "initSharedSO: os.MkdirTemp failed: " + mkErr.Error()
+ return
+ }
+ sharedSO.dir = dir
+
+ goExe, lookErr := exec.LookPath("go")
+ if lookErr != nil {
+ sharedSO.skipReason = "initSharedSO: 'go' binary not found in PATH"
+ return
+ }
+ soPath := filepath.Join(dir, "telemetrysampler.so")
+
+ _, thisFile, _, ok := runtime.Caller(0)
+ if !ok {
+ sharedSO.skipReason = "initSharedSO: cannot determine source file path via runtime.Caller"
+ return
+ }
+ absPluginDir := filepath.Clean(filepath.Join(filepath.Dir(thisFile), telemetryPluginPkgPath))
+
+ cmd := exec.Command(goExe, "build", "-buildmode=plugin", "-o", soPath, absPluginDir)
+ cmd.Env = append(os.Environ(), "CGO_ENABLED=1")
+ out, buildErr := cmd.CombinedOutput()
+ if buildErr != nil {
+ sharedSO.skipReason = "initSharedSO: cannot build .so (CGO may be unavailable): " +
+ buildErr.Error() + "\n" + string(out)
+ return
+ }
+ sharedSO.soPath = soPath
+
+ _, probeErr := plugin.Open(soPath)
+ if isPluginToolchainMismatch(probeErr) {
+ sharedSO.skipReason = "initSharedSO: host cannot load freshly built plugin (toolchain mismatch): " +
+ probeErr.Error()
+ return
+ }
+ if probeErr != nil {
+ sharedSO.skipReason = "initSharedSO: plugin.Open probe failed: " + probeErr.Error()
+ }
+}
+
+// newRealSOSchemaRepo is a shared setup helper for real-.so tests: it ensures
+// the telemetrysampler .so is built and probed (skipping on CGO unavailability
+// or toolchain mismatch), and returns a ready-to-use schemaRepo together with
+// the .so file name (for Plugin.Path) and the prometheus registry.
+//
+// All real-.so tests share a single .so file because Go deduplicates plugins
+// by their internal package path: opening the same plugin package from two
+// different filesystem paths in the same process causes a "previous failure"
+// error on the second open.
+func newRealSOSchemaRepo(t *testing.T) (sr schemaRepo, soName string, reg *prometheus.Registry) {
+ t.Helper()
+ sharedSO.Do(initSharedSO)
+ if sharedSO.skipReason != "" {
+ t.Skip(sharedSO.skipReason)
+ }
+
+ reg = prometheus.NewRegistry()
+ promProvider := prom.NewProvider(pipelineScope, reg)
+ factory := services.NewFactory(promProvider, nil, nil)
+
+ sr = schemaRepo{
+ l: logger.GetLogger("trace"),
+ role: databasev1.Role_ROLE_DATA,
+ nativePipelineEnabled: true,
+ trustedPluginDir: sharedSO.dir,
+ samplerMeter: newSamplerMetrics(factory),
+ pluginTelemetryFactory: factory,
+ }
+ soName = filepath.Base(sharedSO.soPath)
+ return sr, soName, reg
+}
+
+// faultyPluginPkgPath is the path to the faultysampler plugin source relative to
+// this file's directory (banyand/trace). It is a distinct Go package from
+// _telemetrysampler, so its .so can be loaded in the same process without the
+// Go plugin-dedup "previous failure" error.
+const faultyPluginPkgPath = "../../test/plugins/_faultysampler"
+
+// sharedFaultySO holds the result of the once-per-process .so build and probe
+// for the _faultysampler plugin. Strings/pointer fields come first so the
+// compiler can satisfy fieldalignment before the embedded sync.Once.
+var sharedFaultySO struct {
+ dir string
+ soPath string
+ skipReason string // non-empty → tests must skip
+ sync.Once
+}
+
+// initSharedFaultySO builds the faultysampler .so exactly once into a
+// process-scoped temp directory and probes that the host can load it.
+func initSharedFaultySO() {
+ dir, mkErr := os.MkdirTemp("", "realso-faulty-*")
+ if mkErr != nil {
+ sharedFaultySO.skipReason = "initSharedFaultySO: os.MkdirTemp failed: " + mkErr.Error()
+ return
+ }
+ sharedFaultySO.dir = dir
+
+ goExe, lookErr := exec.LookPath("go")
+ if lookErr != nil {
+ sharedFaultySO.skipReason = "initSharedFaultySO: 'go' binary not found in PATH"
+ return
+ }
+ soPath := filepath.Join(dir, "faultysampler.so")
+
+ _, thisFile, _, ok := runtime.Caller(0)
+ if !ok {
+ sharedFaultySO.skipReason = "initSharedFaultySO: cannot determine source file path via runtime.Caller"
+ return
+ }
+ absPluginDir := filepath.Clean(filepath.Join(filepath.Dir(thisFile), faultyPluginPkgPath))
+
+ cmd := exec.Command(goExe, "build", "-buildmode=plugin", "-o", soPath, absPluginDir)
+ cmd.Env = append(os.Environ(), "CGO_ENABLED=1")
+ out, buildErr := cmd.CombinedOutput()
+ if buildErr != nil {
+ sharedFaultySO.skipReason = "initSharedFaultySO: cannot build .so (CGO may be unavailable): " +
+ buildErr.Error() + "\n" + string(out)
+ return
+ }
+ sharedFaultySO.soPath = soPath
+
+ _, probeErr := plugin.Open(soPath)
+ if isPluginToolchainMismatch(probeErr) {
+ sharedFaultySO.skipReason = "initSharedFaultySO: host cannot load freshly built plugin (toolchain mismatch): " +
+ probeErr.Error()
+ return
+ }
+ if probeErr != nil {
+ sharedFaultySO.skipReason = "initSharedFaultySO: plugin.Open probe failed: " + probeErr.Error()
+ }
+}
+
+// newRealSOFaultyRepo sets up the faultysampler .so and returns a schemaRepo,
+// the .so base name (for Plugin.Path), and a fresh prometheus.Registry. It
+// skips the test when CGO is unavailable or there is a toolchain mismatch.
+func newRealSOFaultyRepo(t *testing.T) (sr schemaRepo, soName string, reg *prometheus.Registry) {
+ t.Helper()
+ sharedFaultySO.Do(initSharedFaultySO)
+ if sharedFaultySO.skipReason != "" {
+ t.Skip(sharedFaultySO.skipReason)
+ }
+
+ reg = prometheus.NewRegistry()
+ promProvider := prom.NewProvider(pipelineScope, reg)
+ factory := services.NewFactory(promProvider, nil, nil)
+
+ sr = schemaRepo{
+ l: logger.GetLogger("trace"),
+ role: databasev1.Role_ROLE_DATA,
+ nativePipelineEnabled: true,
+ trustedPluginDir: sharedFaultySO.dir,
+ samplerMeter: newSamplerMetrics(factory),
+ pluginTelemetryFactory: factory,
+ }
+ soName = filepath.Base(sharedFaultySO.soPath)
+ return sr, soName, reg
+}
+
+// makeFaultyCfg builds a TracePipelineConfig with one faultysampler plugin entry.
+// faults is marshaled into structpb and passed as the plugin config.
+func makeFaultyCfg(soName, pluginName string, faults map[string]any) *commonv1.TracePipelineConfig {
+ cfgStruct, structErr := structpb.NewStruct(faults)
+ if structErr != nil {
+ panic("makeFaultyCfg: structpb.NewStruct failed: " + structErr.Error())
+ }
+ return &commonv1.TracePipelineConfig{
+ Enabled: true,
+ Plugins: []*commonv1.Plugin{
+ {
+ Name: pluginName,
+ Kind: &commonv1.Plugin_Sampler{
+ Sampler: &commonv1.SamplerPlugin{
+ Path: soName,
+ Symbol: "NewSampler",
+ AbiVersion: uint32(sdk.ABIVersion),
+ Config: cfgStruct,
+ },
+ },
+ },
+ },
+ }
+}
+
+// gatherCounterValue returns the counter value for the first metric in family
+// familyName whose labels match all entries in wantLabels, or 0 if not found.
+func gatherCounterValue(mfs []*dto.MetricFamily, familyName string, wantLabels map[string]string) float64 {
+ for _, mf := range mfs {
+ if mf.GetName() != familyName {
+ continue
+ }
+ for _, m := range mf.GetMetric() {
+ match := true
+ for k, v := range wantLabels {
+ if labelValue(m, k) != v {
+ match = false
+ break
+ }
+ }
+ if match {
+ return m.GetCounter().GetValue()
+ }
+ }
+ }
+ return 0
+}
+
+// countFamilySeriesForGroup counts how many metric series in familyName have a
+// "group" label equal to group.
+func countFamilySeriesForGroup(mfs []*dto.MetricFamily, familyName, group string) int {
+ for _, mf := range mfs {
+ if mf.GetName() != familyName {
+ continue
+ }
+ count := 0
+ for _, m := range mf.GetMetric() {
+ if labelValue(m, "group") == group {
+ count++
+ }
+ }
+ return count
+ }
+ return 0
+}
+
+// TestPluginTelemetry_RealSO_FaultInjection loads the _faultysampler .so and
+// verifies that the host telemetry bounds hold under adversarial plugin behavior:
+//
+// - flood_logs_are_rate_limited: 500 Logger.Info calls per Decide are clamped
+// by the 50/s burst-100 token bucket; logged + dropped == 500.
+// - cardinality_is_capped: 130 Decide calls each registering a unique Counter
+// label value are capped at 100 kept + 1 overflow series; 30 rejected.
+// - usehost_panic_is_isolated: a panic in UseHost is recovered; the sampler is
+// still registered, keeps all traces, and the panic counter is incremented.
+func TestPluginTelemetry_RealSO_FaultInjection(t *testing.T) {
+ if raceDetectorEnabled {
+ t.Skip("TestPluginTelemetry_RealSO_FaultInjection skipped: race detector active; " +
+ "a non-race .so cannot be loaded into a race-instrumented host")
+ }
+
+ resetRegistries()
+ defer resetRegistries()
+
+ sr, soName, reg := newRealSOFaultyRepo(t)
+
+ t.Run("flood_logs_are_rate_limited", func(t *testing.T) {
+ const group = "fault-flood"
+ cfg := makeFaultyCfg(soName, "fault", map[string]any{
+ "floodLogs": true,
+ "floodCount": float64(500),
+ })
+ sr.reconcilePipeline(group, cfg)
+
+ samplers := lookupSamplers(group)
+ require.NotEmpty(t, samplers, "reconcilePipeline must register at least one sampler for group %q", group)
+
+ var logged int
+ batch := &sdk.TraceBatch{Traces: make([]sdk.TraceBlock, 1)}
+ captured := captureStderr(t, func() {
+ _, decideErr := samplers[0].Decide(batch)
+ require.NoError(t, decideErr, "Decide must not return an error")
+ })
+ logged = strings.Count(captured, "flood log line")
+
+ mfs, gatherErr := reg.Gather()
+ require.NoError(t, gatherErr, "prometheus registry Gather must succeed")
+
+ dropped := gatherCounterValue(mfs,
+ "banyandb_trace_pipeline_plugin_log_dropped_total",
+ map[string]string{"group": group, "plugin_name": "fault"})
+
+ assert.Greater(t, logged, 0,
+ "at least one log line must pass the rate limiter; got 0")
+ assert.Less(t, logged, 500,
+ "rate limiter must clamp output below 500; got %d", logged)
+ assert.Greater(t, dropped, float64(0),
+ "plugin_log_dropped_total must be >0 when 500 lines are emitted at once; got %v", dropped)
+ assert.Equal(t, 500, logged+int(dropped),
+ "every Info call must either be logged or counted as dropped: logged=%d dropped=%v", logged, dropped)
+ })
+
+ t.Run("cardinality_is_capped", func(t *testing.T) {
+ const group = "fault-card"
+ cfg := makeFaultyCfg(soName, "fault", map[string]any{
+ "explodeCardinality": true,
+ })
+ sr.reconcilePipeline(group, cfg)
+
+ samplers := lookupSamplers(group)
+ require.NotEmpty(t, samplers, "reconcilePipeline must register at least one sampler for group %q", group)
+
+ batch := &sdk.TraceBatch{Traces: make([]sdk.TraceBlock, 1)}
+ for range 130 {
+ _, decideErr := samplers[0].Decide(batch)
+ require.NoError(t, decideErr, "Decide must not return an error")
+ }
+
+ mfs, gatherErr := reg.Gather()
+ require.NoError(t, gatherErr, "prometheus registry Gather must succeed")
+
+ rejected := gatherCounterValue(mfs,
+ "banyandb_trace_pipeline_plugin_telemetry_series_rejected_total",
+ map[string]string{"group": group, "plugin_name": "fault"})
+ assert.Equal(t, float64(30), rejected,
+ "series_rejected must equal 130 calls − 100 cap = 30; got %v", rejected)
+
+ boomSeries := countFamilySeriesForGroup(mfs, "banyandb_trace_pipeline_plugin_boom", group)
+ assert.Equal(t, 101, boomSeries,
+ "boom family must have 101 series for group %q (100 kept + 1 __overflow__); got %d", group, boomSeries)
+ })
+
+ t.Run("usehost_panic_is_isolated", func(t *testing.T) {
+ const group = "fault-panic"
+ cfg := makeFaultyCfg(soName, "fault", map[string]any{
+ "panicInUseHost": true,
+ })
+
+ require.NotPanics(t, func() { sr.reconcilePipeline(group, cfg) },
+ "reconcilePipeline must not panic when the plugin's UseHost panics")
+
+ samplers := lookupSamplers(group)
+ require.NotEmpty(t, samplers,
+ "sampler must still be registered after a UseHost panic (fail-open)")
+
+ batch := &sdk.TraceBatch{Traces: make([]sdk.TraceBlock, 2)}
+ verdict, decideErr := samplers[0].Decide(batch)
+ require.NoError(t, decideErr, "Decide must not return an error after a UseHost panic")
+ require.Len(t, verdict.Keep, 2, "Decide must return a Keep slice of length 2 (fail-open)")
+ for idx, keep := range verdict.Keep {
+ assert.True(t, keep, "verdict.Keep[%d] must be true (sampler keeps all traces)", idx)
+ }
+
+ mfs, gatherErr := reg.Gather()
+ require.NoError(t, gatherErr, "prometheus registry Gather must succeed")
+
+ panicCount := gatherCounterValue(mfs,
+ "banyandb_trace_pipeline_plugin_telemetry_panic_total",
+ map[string]string{"group": group, "plugin_name": "fault"})
+ assert.Equal(t, float64(1), panicCount,
+ "plugin_telemetry_panic_total must equal 1 after one UseHost panic; got %v", panicCount)
+ })
+}
+
+// makeSamplerCfg builds a TracePipelineConfig with a single telemetrysampler
+// plugin entry using the given soName, logEvery value, and plugin name.
+func makeSamplerCfg(soName, pluginName string, logEvery float64) *commonv1.TracePipelineConfig {
+ cfgStruct, structErr := structpb.NewStruct(map[string]any{"logEvery": logEvery})
+ if structErr != nil {
+ panic("makeSamplerCfg: structpb.NewStruct failed: " + structErr.Error())
+ }
+ return &commonv1.TracePipelineConfig{
+ Enabled: true,
+ Plugins: []*commonv1.Plugin{
+ {
+ Name: pluginName,
+ Kind: &commonv1.Plugin_Sampler{
+ Sampler: &commonv1.SamplerPlugin{
+ Path: soName,
+ Symbol: "NewSampler",
+ AbiVersion: uint32(sdk.ABIVersion),
+ Config: cfgStruct,
+ },
+ },
+ },
+ },
+ }
+}
+
+// TestPluginTelemetry_RealSO_ConfigParamLogEvery proves that the logEvery config
+// parameter reaches the real .so and changes the log-emission frequency.
+//
+// The plugin logs "batch decided" when batchN % logEvery == 1 (batchN starts at 1
+// and increments per Decide call on the instance).
+// - Sub-case A: logEvery=2 → batches 1,3,5 satisfy n%2==1 → 3 log lines in 5 calls.
+// - Sub-case B: logEvery=1000 → only batch 1 satisfies n%1000==1 → 1 log line in 5 calls.
+func TestPluginTelemetry_RealSO_ConfigParamLogEvery(t *testing.T) {
+ if raceDetectorEnabled {
+ t.Skip("TestPluginTelemetry_RealSO_ConfigParamLogEvery skipped: race detector active; " +
+ "a non-race .so cannot be loaded into a race-instrumented host")
+ }
+
+ resetRegistries()
+ defer resetRegistries()
+
+ sr, soName, _ := newRealSOSchemaRepo(t)
+
+ const pluginName = "tel"
+
+ // --- Sub-case A: logEvery=2 ---
+ // batchN%2==1 for batchN in {1,3,5}: expect exactly 3 "batch decided" lines.
+ const groupA = "cfgparam-a"
+ cfgA := makeSamplerCfg(soName, pluginName, float64(2))
+ sr.reconcilePipeline(groupA, cfgA)
+
+ samplersA := lookupSamplers(groupA)
+ require.NotEmpty(t, samplersA, "reconcilePipeline must register at least one sampler for group %q", groupA)
+
+ batchSingle := &sdk.TraceBatch{Traces: make([]sdk.TraceBlock, 1)}
+ var countA int
+ capturedA := captureStderr(t, func() {
+ for range 5 {
+ _, decideErr := samplersA[0].Decide(batchSingle)
+ require.NoError(t, decideErr, "Decide must not error (group %q)", groupA)
+ }
+ })
+ countA = strings.Count(capturedA, "batch decided")
+
+ // --- Sub-case B: logEvery=1000 ---
+ // batchN%1000==1 only for batchN==1: expect exactly 1 "batch decided" line.
+ const groupB = "cfgparam-b"
+ cfgB := makeSamplerCfg(soName, pluginName, float64(1000))
+ sr.reconcilePipeline(groupB, cfgB)
+
+ samplersB := lookupSamplers(groupB)
+ require.NotEmpty(t, samplersB, "reconcilePipeline must register at least one sampler for group %q", groupB)
+
+ var countB int
+ capturedB := captureStderr(t, func() {
+ for range 5 {
+ _, decideErr := samplersB[0].Decide(batchSingle)
+ require.NoError(t, decideErr, "Decide must not error (group %q)", groupB)
+ }
+ })
+ countB = strings.Count(capturedB, "batch decided")
+
+ // logEvery=2: batches 1,3,5 satisfy batchN%2==1 → 3 log lines.
+ assert.Equal(t, 3, countA,
+ "logEvery=2: expected 3 log lines (batches 1,3,5 satisfy batchN%%2==1); got %d\nstderr:\n%s", countA, capturedA)
+ // logEvery=1000: only batch 1 satisfies batchN%1000==1 → 1 log line.
+ assert.Equal(t, 1, countB,
+ "logEvery=1000: expected 1 log line (only batch 1 satisfies batchN%%1000==1); got %d\nstderr:\n%s", countB, capturedB)
+ assert.Greater(t, countA, countB, "logEvery=2 must produce more log lines than logEvery=1000")
+}
+
+// TestPluginTelemetry_RealSO_MalformedConfigFailOpen proves that a malformed
+// config value (a string where the plugin's NewSampler expects an int64) causes
+// the real .so's constructor to return an error, and that the engine handles this
+// fail-open: no sampler is registered and the engine is not wedged.
+//
+// Note: structpb always produces valid JSON, but the TYPE mismatch
+// ({"logEvery":"not-a-number"} where the plugin wants int64) causes
+// json.Unmarshal into samplerConfig to fail → NewSampler returns
+// "telemetrysampler: invalid config JSON" → reconcilePipeline keeps previous
+// (empty) set → lookupSamplers returns nil (fail-open: all traces retained).
+func TestPluginTelemetry_RealSO_MalformedConfigFailOpen(t *testing.T) {
+ if raceDetectorEnabled {
+ t.Skip("TestPluginTelemetry_RealSO_MalformedConfigFailOpen skipped: race detector active; " +
+ "a non-race .so cannot be loaded into a race-instrumented host")
+ }
+
+ resetRegistries()
+ defer resetRegistries()
+
+ sr, soName, _ := newRealSOSchemaRepo(t)
+
+ const (
+ group = "malformed-cfg"
+ pluginName = "tel"
+ )
+
+ // Build a config whose logEvery value is a STRING — valid JSON from structpb's
+ // perspective, but the plugin's json.Unmarshal into int64 will fail, causing
+ // NewSampler to return "telemetrysampler: invalid config JSON".
+ badCfgStruct, badStructErr := structpb.NewStruct(map[string]any{"logEvery": "not-a-number"})
+ require.NoError(t, badStructErr, "structpb.NewStruct for bad config must succeed (structpb always accepts string values)")
+
+ badCfg := &commonv1.TracePipelineConfig{
+ Enabled: true,
+ Plugins: []*commonv1.Plugin{
+ {
+ Name: pluginName,
+ Kind: &commonv1.Plugin_Sampler{
+ Sampler: &commonv1.SamplerPlugin{
+ Path: soName,
+ Symbol: "NewSampler",
+ AbiVersion: uint32(sdk.ABIVersion),
+ Config: badCfgStruct,
+ },
+ },
+ },
+ },
+ }
+
+ // The engine must not panic when the plugin constructor returns an error.
+ require.NotPanics(t, func() { sr.reconcilePipeline(group, badCfg) })
+
+ // Fail-open: the constructor error leaves the group with no samplers
+ // (the previous set was empty), so all traces are retained.
+ assert.Empty(t, lookupSamplers(group),
+ "lookupSamplers must be empty after a constructor-error reconcile (fail-open path)")
+
+ // Prove the engine is not wedged: a valid config for the same group must
+ // register successfully and produce working Decide calls.
+ goodCfg := makeSamplerCfg(soName, pluginName, float64(100))
+ require.NotPanics(t, func() { sr.reconcilePipeline(group, goodCfg) })
+
+ samplersAfterRecovery := lookupSamplers(group)
+ require.NotEmpty(t, samplersAfterRecovery,
+ "lookupSamplers must be non-empty after a valid-config reconcile (engine must recover from bad-config)")
+
+ // Verify the recovered sampler works: a 2-trace batch must return keep-all.
+ batch := &sdk.TraceBatch{Traces: make([]sdk.TraceBlock, 2)}
+ verdict, decideErr := samplersAfterRecovery[0].Decide(batch)
+ require.NoError(t, decideErr, "Decide must not error after recovery")
+ require.Len(t, verdict.Keep, 2, "Decide must return a Keep slice of length 2 after recovery")
+ for idx, keep := range verdict.Keep {
+ assert.True(t, keep, "verdict.Keep[%d] must be true after recovery (sampler keeps all traces)", idx)
+ }
+}
diff --git a/banyand/trace/plugin_telemetry_test.go b/banyand/trace/plugin_telemetry_test.go
new file mode 100644
index 000000000..72a20c148
--- /dev/null
+++ b/banyand/trace/plugin_telemetry_test.go
@@ -0,0 +1,378 @@
+// 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 (
+ "bytes"
+ "encoding/json"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/rs/zerolog"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+ "github.com/apache/skywalking-banyandb/pkg/pipeline/sdk"
+)
+
+// newTestSamplerMetrics builds a samplerMetrics backed by the fakeMetricFactory.
+func newTestSamplerMetrics() (*samplerMetrics, *fakeMetricFactory) {
+ f := newFakeMetricFactory()
+ sm := newSamplerMetrics(f)
+ return sm, f
+}
+
+// TestPluginTelemetry_MetricNamePrefix verifies that plugin metric names are
+// always prefixed with "plugin_" regardless of what the plugin requests.
+func TestPluginTelemetry_MetricNamePrefix(t *testing.T) {
+ sm, f := newTestSamplerMetrics()
+ baseLogger := logger.GetLogger("trace")
+ pt := newPluginTelemetry(f, baseLogger, sm, "grp", "plug")
+ m := pt.Meter()
+ require.NotNil(t, m)
+
+ _ = m.Counter("my_counter")
+ _ = m.Gauge("my_gauge")
+
+ // The factory should have received names prefixed with "plugin_".
+ f.mu.Lock()
+ _, hasCounter := f.counters["plugin_my_counter"]
+ _, hasGauge := f.gauges["plugin_my_gauge"]
+ f.mu.Unlock()
+
+ assert.True(t, hasCounter, "counter name must be prefixed with plugin_")
+ assert.True(t, hasGauge, "gauge name must be prefixed with plugin_")
+}
+
+// TestPluginTelemetry_MetricNameSanitization verifies that mixed-case and
+// special-character names are lowercased and sanitized.
+func TestPluginTelemetry_MetricNameSanitization(t *testing.T) {
+ sm, f := newTestSamplerMetrics()
+ baseLogger := logger.GetLogger("trace")
+ pt := newPluginTelemetry(f, baseLogger, sm, "grp", "plug")
+ m := pt.Meter()
+
+ _ = m.Counter("MyCounter-V2")
+
+ f.mu.Lock()
+ _, ok := f.counters["plugin_mycounter_v2"]
+ f.mu.Unlock()
+ assert.True(t, ok, "sanitized counter name must be plugin_mycounter_v2")
+}
+
+// TestPluginTelemetry_ReservedLabelRejected verifies that reserved label names
+// {group, plugin_name} are silently removed and series_rejected incremented.
+func TestPluginTelemetry_ReservedLabelRejected(t *testing.T) {
+ sm, f := newTestSamplerMetrics()
+ baseLogger := logger.GetLogger("trace")
+ pt := newPluginTelemetry(f, baseLogger, sm, "grp", "plug")
+ m := pt.Meter()
+
+ // Both "group" and "plugin_name" are reserved — they should be stripped.
+ c := m.Counter("hits", "group", "plugin_name", "env")
+ require.NotNil(t, c)
+
+ rejectedCounter := f.counter("plugin_telemetry_series_rejected_total")
+ require.NotNil(t, rejectedCounter)
+ // Two reserved labels → two increments.
+ assert.GreaterOrEqual(t, len(rejectedCounter.calls), 2,
+ "two reserved label names must each increment series_rejected")
+}
+
+// TestPluginTelemetry_CardinalityCap verifies that emitting more than
+// maxSeriesPerInstrument distinct custom-value tuples routes extras to the
+// overflow sentinel and increments series_rejected for each excess tuple.
+func TestPluginTelemetry_CardinalityCap(t *testing.T) {
+ sm, f := newTestSamplerMetrics()
+ baseLogger := logger.GetLogger("trace")
+ pt := newPluginTelemetry(f, baseLogger, sm, "grp", "plug")
+ m := pt.Meter()
+
+ c := m.Counter("reqs", "svc")
+ require.NotNil(t, c)
+
+ // Emit exactly maxSeriesPerInstrument distinct tuples.
+ for i := range maxSeriesPerInstrument {
+ c.Inc(1, strings.Repeat("x", i+1))
+ }
+
+ rejBefore := len(f.counter("plugin_telemetry_series_rejected_total").calls)
+
+ // Emit K extra distinct tuples — each should be rejected.
+ const extraK = 5
+ for i := range extraK {
+ c.Inc(1, "extra_"+strings.Repeat("y", i+1))
+ }
+
+ rejCounter := f.counter("plugin_telemetry_series_rejected_total")
+ rejAfter := len(rejCounter.calls)
+ assert.Equal(t, extraK, rejAfter-rejBefore,
+ "each excess distinct tuple must increment series_rejected exactly once")
+
+ // Verify the underlying counter received the overflow sentinel label values.
+ underlying := f.counter("plugin_reqs")
+ require.NotNil(t, underlying)
+ overflowCount := underlying.callsWithLabels("grp", "plug", overflowSentinel)
+ assert.Equal(t, extraK, overflowCount, "overflow tuples must use the sentinel value")
+}
+
+// TestPluginTelemetry_BypassZeroAllocs verifies that Inc/Set/Observe on a
+// bypass-backed adapter perform zero heap allocations for no-label invocations.
+// The BenchmarkPluginTelemetry_BypassInc benchmark covers the labeled case;
+// the Go compiler is unable to avoid a []string backing allocation when a
+// variadic call crosses an interface boundary in a test context, but the
+// adapter itself builds no internal slices.
+func TestPluginTelemetry_BypassZeroAllocs(t *testing.T) {
+ sm, _ := newTestSamplerMetrics()
+ baseLogger := logger.GetLogger("trace")
+ // nil factory → bypass path.
+ pt := newPluginTelemetry(nil, baseLogger, sm, "grp", "plug")
+ m := pt.Meter()
+
+ c := m.Counter("hits")
+ g := m.Gauge("lat")
+ h := m.Histogram("size", []float64{1, 5, 10})
+
+ allocs := testing.AllocsPerRun(100, func() {
+ c.Inc(1)
+ g.Set(3.14)
+ h.Observe(7.0)
+ })
+ assert.Equal(t, float64(0), allocs, "bypass path must have 0 allocs/op on no-label calls")
+}
+
+// TestPluginTelemetry_LoggerInfoWarnError verifies that Info/Warn/Error calls
+// go through the rate limiter without panicking.
+func TestPluginTelemetry_LoggerInfoWarnError(t *testing.T) {
+ sm, _ := newTestSamplerMetrics()
+ baseLogger := logger.GetLogger("trace")
+ pt := newPluginTelemetry(nil, baseLogger, sm, "grp", "plug")
+ l := pt.Logger()
+
+ require.NotPanics(t, func() {
+ l.Info("hello", "k1", "v1")
+ l.Warn("warn msg", "k2", 42)
+ l.Error("err msg")
+ })
+}
+
+// TestPluginTelemetry_LoggerDebugDropped verifies that Debug calls are silently
+// dropped under the INFO level cap.
+func TestPluginTelemetry_LoggerDebugDropped(t *testing.T) {
+ sm, _ := newTestSamplerMetrics()
+ baseLogger := logger.GetLogger("trace")
+ pt := newPluginTelemetry(nil, baseLogger, sm, "grp", "plug")
+ l := pt.Logger()
+
+ // This must not panic and must not count against the rate limiter.
+ require.NotPanics(t, func() {
+ for range 200 {
+ l.Debug("debug msg", "k", "v")
+ }
+ })
+ // Rate limiter tokens must be unaffected by Debug calls.
+ pl := pt.loggerAdapter
+ pl.mu.Lock()
+ tok := pl.tokens
+ pl.mu.Unlock()
+ assert.Equal(t, logBurstSize, tok, "Debug must not consume rate-limiter tokens")
+}
+
+// TestPluginTelemetry_LogRateLimit verifies that emitting many lines quickly
+// drops some and increments plugin_log_dropped_total.
+func TestPluginTelemetry_LogRateLimit(t *testing.T) {
+ sm, f := newTestSamplerMetrics()
+ baseLogger := logger.GetLogger("trace")
+ pt := newPluginTelemetry(f, baseLogger, sm, "grp", "plug")
+ l := pt.Logger()
+
+ const burst = 200
+ for range burst {
+ l.Info("rapid fire")
+ }
+
+ droppedCounter := f.counter("plugin_log_dropped_total")
+ require.NotNil(t, droppedCounter)
+ // We emitted 200 lines but burst capacity is 100 — at least some must drop.
+ assert.Greater(t, len(droppedCounter.calls), 0,
+ "emitting beyond burst capacity must increment plugin_log_dropped_total")
+}
+
+// TestPluginTelemetry_LogMsgTruncated verifies that the production emit path
+// (emitEvent) truncates a message longer than maxMsgLen before it is written. It
+// drives the real emitEvent with a zerolog event wired to a buffer and asserts the
+// emitted "message" field, rather than re-implementing the truncation in the test.
+func TestPluginTelemetry_LogMsgTruncated(t *testing.T) {
+ longMsg := strings.Repeat("A", maxMsgLen+100)
+
+ var buf bytes.Buffer
+ lg := zerolog.New(&buf)
+ emitEvent(lg.Info(), longMsg, nil)
+
+ var out map[string]any
+ require.NoError(t, json.Unmarshal(buf.Bytes(), &out))
+ msg, ok := out[zerolog.MessageFieldName].(string)
+ require.True(t, ok, "emitted log must carry a message field")
+ assert.Len(t, msg, maxMsgLen, "emitEvent must truncate the message to maxMsgLen bytes")
+ assert.Equal(t, strings.Repeat("A", maxMsgLen), msg, "truncated message must be the maxMsgLen-byte prefix")
+}
+
+// TestPluginTelemetry_LogFieldsCapped verifies that more than maxLogFields pairs
+// do not panic and that trailing odd key gets "(MISSING)".
+func TestPluginTelemetry_LogFieldsCapped(t *testing.T) {
+ sm, _ := newTestSamplerMetrics()
+ baseLogger := logger.GetLogger("trace")
+ pt := newPluginTelemetry(nil, baseLogger, sm, "grp", "plug")
+ l := pt.Logger()
+
+ // Build 20 pairs (exceeds maxLogFields=16) and one trailing odd key.
+ kvs := make([]any, 0, 41)
+ for i := range 20 {
+ kvs = append(kvs, "k", i)
+ }
+ kvs = append(kvs, "trailing_odd_key")
+
+ require.NotPanics(t, func() { l.Info("fields test", kvs...) })
+}
+
+// TestPluginTelemetry_Teardown verifies that teardown Deletes all emitted tuples
+// and is safe to call multiple times.
+func TestPluginTelemetry_Teardown(t *testing.T) {
+ sm, f := newTestSamplerMetrics()
+ baseLogger := logger.GetLogger("trace")
+ pt := newPluginTelemetry(f, baseLogger, sm, "grp", "plug")
+ m := pt.Meter()
+
+ c := m.Counter("evt", "env")
+ c.Inc(1, "prod")
+ c.Inc(1, "staging")
+
+ require.NotPanics(t, func() { pt.teardown() })
+ require.NotPanics(t, func() { pt.teardown() }) // idempotent
+
+ underlying := f.counter("plugin_evt")
+ require.NotNil(t, underlying)
+ // After teardown the cache is cleared; the instrument received the calls.
+ assert.GreaterOrEqual(t, len(underlying.calls), 2,
+ "underlying counter must have recorded the Inc calls")
+}
+
+// TestPluginTelemetry_TeardownPanicsafe verifies that teardown does not
+// propagate panics from Delete to the caller.
+func TestPluginTelemetry_TeardownPanicsafe(t *testing.T) {
+ sm, f := newTestSamplerMetrics()
+ baseLogger := logger.GetLogger("trace")
+ pt := newPluginTelemetry(f, baseLogger, sm, "grp", "plug")
+ m := pt.Meter()
+
+ // Register a counter so there is something to tear down.
+ c := m.Counter("panic_evt")
+ c.Inc(1)
+
+ require.NotPanics(t, func() { pt.teardown() },
+ "teardown must not propagate panics from Delete")
+}
+
+// TestPluginTelemetry_NilSamplerMetrics verifies that a nil *samplerMetrics
+// does not cause panics in the adapter paths.
+func TestPluginTelemetry_NilSamplerMetrics(t *testing.T) {
+ baseLogger := logger.GetLogger("trace")
+ // nil sm is valid — all inc helpers guard against nil receiver.
+ pt := newPluginTelemetry(nil, baseLogger, nil, "grp", "plug")
+ m := pt.Meter()
+
+ require.NotPanics(t, func() {
+ c := m.Counter("safe")
+ c.Inc(1)
+ })
+}
+
+// BenchmarkPluginTelemetry_BypassInc measures allocations on the bypass path.
+// The no-label variant must produce 0 allocs/op; the labeled variant shows 1
+// alloc/op from the Go compiler materializing the variadic []string when
+// crossing the sdk.Counter interface boundary — that allocation is in the
+// caller, not in the noop adapter itself.
+func BenchmarkPluginTelemetry_BypassInc(b *testing.B) {
+ sm, _ := newTestSamplerMetrics()
+ baseLogger := logger.GetLogger("trace")
+ pt := newPluginTelemetry(nil, baseLogger, sm, "grp", "plug")
+ // No-label counter: must be 0 allocs/op.
+ c := pt.Meter().Counter("hits")
+ b.ReportAllocs()
+ b.ResetTimer()
+ for range b.N {
+ c.Inc(1)
+ }
+}
+
+// TestPluginTelemetry_SameInstrumentCached verifies that calling Counter with
+// the same name+labels twice returns the same underlying sdk.Counter instance.
+func TestPluginTelemetry_SameInstrumentCached(t *testing.T) {
+ sm, f := newTestSamplerMetrics()
+ baseLogger := logger.GetLogger("trace")
+ pt := newPluginTelemetry(f, baseLogger, sm, "grp", "plug")
+ m := pt.Meter()
+
+ c1 := m.Counter("evt")
+ c2 := m.Counter("evt")
+ assert.Equal(t, c1, c2, "repeated Counter() calls must return the same handle")
+
+ f.mu.Lock()
+ _, ok := f.counters["plugin_evt"]
+ f.mu.Unlock()
+ assert.True(t, ok)
+}
+
+// TestPluginTelemetry_MeterReturnedHostInterface verifies that pluginTelemetry
+// satisfies sdk.Host.
+func TestPluginTelemetry_MeterReturnedHostInterface(t *testing.T) {
+ sm, _ := newTestSamplerMetrics()
+ baseLogger := logger.GetLogger("trace")
+ pt := newPluginTelemetry(nil, baseLogger, sm, "grp", "plug")
+
+ var host sdk.Host = pt
+ require.NotNil(t, host.Meter())
+ require.NotNil(t, host.Logger())
+}
+
+// TestPluginTelemetry_LogSuppressionSummary verifies that the suppression
+// summary is emitted at most once per suppressWarnInterval.
+func TestPluginTelemetry_LogSuppressionSummary(t *testing.T) {
+ sm, f := newTestSamplerMetrics()
+ baseLogger := logger.GetLogger("trace")
+ pt := newPluginTelemetry(f, baseLogger, sm, "grp", "plug")
+ l := pt.loggerAdapter
+
+ // Drain the bucket so subsequent calls will drop.
+ for range int(logBurstSize) + 10 {
+ l.Info("drain")
+ }
+
+ droppedBefore := len(f.counter("plugin_log_dropped_total").calls)
+ assert.Greater(t, droppedBefore, 0, "some lines must have been dropped")
+
+ // Advance time so the suppression-summary interval is exceeded.
+ l.mu.Lock()
+ l.lastWarnEmit = time.Now().Add(-suppressWarnInterval * 2)
+ l.mu.Unlock()
+
+ // Next Info call should emit the summary and not panic.
+ require.NotPanics(t, func() { l.Info("trigger summary") })
+}
diff --git a/docs/design/post-trace-pipeline.md b/docs/design/post-trace-pipeline.md
index 663510622..999d1735b 100644
--- a/docs/design/post-trace-pipeline.md
+++ b/docs/design/post-trace-pipeline.md
@@ -256,6 +256,64 @@ The sampler is the only kind this design ships, but `Plugin` is a generic envelo
The same three-step pattern (proto arm → SDK sub-interface + constructor → engine dispatch) covers any future kind (exporter, router, …); the discriminator is always the set `oneof` arm, cross-checked by `Plugin.Kind()`.
+### 2.7 Plugin Telemetry (Meter + Logger)
+
+A plugin opts into bounded telemetry by implementing `sdk.HostAware` in addition to `sdk.Sampler`. The engine type-asserts the constructed sampler to `sdk.HostAware` and, if present, calls `UseHost(h)` **exactly once per group** before the plugin's first `Decide`. Plugins that do not implement `sdk.HostAware` are unaffected; there is no ABI change — delivery is via type assertion, not a new constructor argument.
+
+**Frozen SDK façade** (`pkg/pipeline/sdk/telemetry.go`):
+
+```go
+// Host is the telemetry context the engine injects into a HostAware plugin.
+type Host interface {
+ Meter() Meter // metric surface scoped to this plugin's group
+ Logger() Logger // logging surface scoped to this plugin's group
+}
+
+// HostAware is the optional interface a plugin implements to receive a Host.
+type HostAware interface {
+ UseHost(Host) // called once, before the first Decide
+}
+
+// Meter is the bounded metric surface.
+type Meter interface {
+ Counter(name string, labelNames ...string) Counter
+ Gauge(name string, labelNames ...string) Gauge
+ Histogram(name string, buckets []float64, labelNames ...string) Histogram
+}
+
+// Logger is the bounded logging surface (no third-party logger types cross the .so boundary).
+type Logger interface {
+ Debug(msg string, keysAndValues ...any)
+ Info(msg string, keysAndValues ...any)
+ Warn(msg string, keysAndValues ...any)
+ Error(msg string, keysAndValues ...any)
+}
+```
+
+**Host-enforced bounds.** The engine owns and enforces every resource limit — the plugin can request arbitrarily but the host silently clamps:
+
+| Resource | Bound |
+|---|---|
+| Metric name prefix | Forced to `banyandb_trace_pipeline_plugin_` |
+| Labels | `{group, plugin_name}` always present; plugin-supplied labels appended |
+| Cardinality cap | **100** distinct label-value series per plugin; overflow series are counted under an `_overflow` sentinel and not registered in Prometheus |
+| Log rate | **50 lines/s**, burst **100** |
+| Log message size | **1 KiB** (truncated if exceeded) |
+| Structured fields per log call | **16** key/value pairs (excess pairs dropped) |
+| Default log level | **INFO** (Debug calls are no-ops unless the data node's `--log-level=debug` is set) |
+
+**Per-group instance delivery.** Because `UseHost` is called once *per group*, a `HostAware` plugin receives a distinct `Host` (and therefore a distinct `Meter` scope) for every group it is registered under. Metrics emitted from group `sw_trace` and group `sw_zipkinTrace` carry different `{group}` label values and are counted against independent cardinality budgets. Non-`HostAware` plugins share a single sampler instance across groups and receive no Host.
+
+**Plugin may cache the Host.** `UseHost` is called before the first `Decide` and not called again for the lifetime of the group config. A plugin should cache `h` in a struct field and use it freely in `Decide`; there is no need to call `UseHost` again or to check for a new Host on each batch.
+
+**Series reclaim on teardown.** The host never calls `Close` on `sdk.Counter`/`sdk.Gauge`/`sdk.Histogram` objects (Go has no finalizer contract here). When a pipeline config is removed, the host calls `sdk.Plugin.Close()` and then deletes every metric series registered by that plugin instance, reclaiming Prometheus memory without a process restart.
+
+**No ABI bump.** The addition of `sdk.HostAware` / `sdk.Host` / `sdk.Meter` / `sdk.Logger` to `pkg/pipeline/sdk` is backward-compatible: old plugins that do not implement `sdk.HostAware` simply fail the type assertion and receive no Host. `sdk.ABIVersion` remains **1**.
+
+**D7 caveat — transitive zerolog dependency.** The `.so` already links `zerolog` transitively via `pbv1` (`pkg/pb/v1`). The `sdk.Logger` façade avoids widening that coupling to plugin authors (no `zerolog.Logger` or `zerolog.Event` crosses the `.so` boundary) rather than eliminating the transitive link. Plugin authors should not rely on `zerolog` being present; use only the `sdk.Logger` interface.
+
+Reference plugins demonstrating well-behaved and adversarial use of the façade live at [`test/plugins/_telemetrysampler`](../../test/plugins/_telemetrysampler) (keeps all traces, emits bounded counter + sparse Info log) and [`test/plugins/_faultysampler`](../../test/plugins/_faultysampler) (panic in UseHost, log flood, cardinality explosion — verifies host bounds hold). Build them with `make build-trace-pipeline-telemetry-plugins`.
+
## Retention-Decision Timing & Trace Completeness
Both the plugin gating policy (§1.1) and the per-stage retention plugin (§1.2) operate on the **whole** trace, not on individual spans as they arrive: `D_total` needs the earliest start and latest end, `has_error` scans all spans, and tag matchers scan all spans. Spans of one trace, however, arrive at the storage node anywhere from milliseconds to hours apart, and BanyanDB writes each span into the segment selected by its **event time** (the `start_time` tag), not its arrival time (`bany [...]
diff --git a/pkg/encoding/array.go b/pkg/encoding/array.go
index 888862d51..2a77eecd7 100644
--- a/pkg/encoding/array.go
+++ b/pkg/encoding/array.go
@@ -18,83 +18,29 @@
package encoding
import (
- "bytes"
- "errors"
+ "github.com/apache/skywalking-banyandb/pkg/encoding/vararray"
)
+// EntityDelimiter and Escape are re-exported from the dependency-free vararray
+// leaf so existing encoding.EntityDelimiter / encoding.Escape usages are
+// unchanged, while consumers that need only the var-array codec (e.g. the
+// plugin SDK) can import the leaf without pulling in this package's logging deps.
const (
// EntityDelimiter is the delimiter for entities in a variable-length array.
- EntityDelimiter = '|'
+ EntityDelimiter = vararray.EntityDelimiter
// Escape is the escape character for entities in a variable-length array.
- Escape = '\\'
+ Escape = vararray.Escape
)
-// MarshalVarArray marshals a byte slice into a variable-length array format.
-// It escapes delimiter and escape characters within the source slice.
+// MarshalVarArray marshals a byte slice into a variable-length array format,
+// escaping delimiter and escape characters. It forwards to the vararray leaf.
func MarshalVarArray(dest, src []byte) []byte {
- if bytes.IndexByte(src, EntityDelimiter) < 0 && bytes.IndexByte(src, Escape) < 0 {
- dest = append(dest, src...)
- dest = append(dest, EntityDelimiter)
- return dest
- }
- for _, b := range src {
- if b == EntityDelimiter || b == Escape {
- dest = append(dest, Escape)
- }
- dest = append(dest, b)
- }
- dest = append(dest, EntityDelimiter)
- return dest
+ return vararray.MarshalVarArray(dest, src)
}
// UnmarshalVarArray unmarshals a variable-length array from src starting at idx.
-//
-// WARNING: This function mutates src. Decoding is performed in-place by
-// overwriting bytes in src[idx:next) to remove escape characters. The decoded
-// value is the view src[idx:end] into the same backing array; copy it (for
-// example, with bytes.Clone or append([]byte(nil), ...)) if you need to
-// preserve the original encoded buffer or keep the decoded value independent of
-// subsequent in-place decoding on the same buffer.
-//
-// It returns:
-// - end: the index of the first byte after the decoded value (exclusive)
-// - next: the index of the next element (the byte after the delimiter)
-//
-// The caller can iterate without creating subslices by tracking indices:
-//
-// for idx < len(src) {
-// end, next, err := UnmarshalVarArray(src, idx)
-// // use src[idx:end]
-// idx = next
-// }
+// It forwards to the vararray leaf; see vararray.UnmarshalVarArray for the
+// in-place-mutation contract and return semantics.
func UnmarshalVarArray(src []byte, idx int) (int, int, error) {
- if idx >= len(src) {
- return 0, 0, errors.New("empty entity value")
- }
- if src[idx] == EntityDelimiter {
- // Empty value; value is src[idx:idx], next starts after the delimiter.
- return idx, idx + 1, nil
- }
- // Decode in-place: read index i, write index j.
- writeIdx := idx
- for readIdx := idx; readIdx < len(src); readIdx++ {
- b := src[readIdx]
- switch {
- case b == Escape:
- // Escape must be followed by at least one more byte.
- if readIdx+1 >= len(src) {
- return 0, 0, errors.New("invalid escape character")
- }
- readIdx++
- src[writeIdx] = src[readIdx]
- writeIdx++
- case b == EntityDelimiter:
- // Return end index of decoded value and index after delimiter.
- return writeIdx, readIdx + 1, nil
- default:
- src[writeIdx] = b
- writeIdx++
- }
- }
- return 0, 0, errors.New("invalid variable array")
+ return vararray.UnmarshalVarArray(src, idx)
}
diff --git a/pkg/encoding/array.go b/pkg/encoding/vararray/vararray.go
similarity index 86%
copy from pkg/encoding/array.go
copy to pkg/encoding/vararray/vararray.go
index 888862d51..242d27194 100644
--- a/pkg/encoding/array.go
+++ b/pkg/encoding/vararray/vararray.go
@@ -15,7 +15,14 @@
// specific language governing permissions and limitations
// under the License.
-package encoding
+// Package vararray holds the escaped, delimiter-separated variable-length array
+// codec (MarshalVarArray / UnmarshalVarArray and the delimiter/escape bytes). It
+// is a dependency-free leaf (imports only stdlib), so consumers that need only
+// this codec — e.g. the plugin SDK at pkg/pipeline/sdk — can use it without
+// transitively pulling in pkg/encoding's logging dependencies. pkg/encoding
+// forwards to these functions, so existing encoding.*VarArray callers are
+// unaffected.
+package vararray
import (
"bytes"
diff --git a/pkg/pb/v1/value.go b/pkg/pb/v1/value.go
index bcd314fe3..c0f318711 100644
--- a/pkg/pb/v1/value.go
+++ b/pkg/pb/v1/value.go
@@ -30,21 +30,26 @@ import (
"github.com/apache/skywalking-banyandb/pkg/convert"
"github.com/apache/skywalking-banyandb/pkg/encoding"
"github.com/apache/skywalking-banyandb/pkg/logger"
+ "github.com/apache/skywalking-banyandb/pkg/pb/v1/valuetype"
)
-// ValueType is the type of the tag and field value.
-type ValueType byte
+// ValueType is the type of the tag and field value. It is aliased to the
+// dependency-free valuetype leaf package so consumers that need only the type
+// tag (e.g. the plugin SDK) can avoid pulling in this package's heavier deps;
+// the constants below re-export the leaf's values so pbv1.ValueType* usages are
+// unchanged.
+type ValueType = valuetype.ValueType
-// ValueType constants.
+// ValueType constants (re-exported from the valuetype leaf).
const (
- ValueTypeUnknown ValueType = iota
- ValueTypeStr
- ValueTypeInt64
- ValueTypeFloat64
- ValueTypeBinaryData
- ValueTypeStrArr
- ValueTypeInt64Arr
- ValueTypeTimestamp
+ ValueTypeUnknown = valuetype.ValueTypeUnknown
+ ValueTypeStr = valuetype.ValueTypeStr
+ ValueTypeInt64 = valuetype.ValueTypeInt64
+ ValueTypeFloat64 = valuetype.ValueTypeFloat64
+ ValueTypeBinaryData = valuetype.ValueTypeBinaryData
+ ValueTypeStrArr = valuetype.ValueTypeStrArr
+ ValueTypeInt64Arr = valuetype.ValueTypeInt64Arr
+ ValueTypeTimestamp = valuetype.ValueTypeTimestamp
)
// MustTagValueToValueType converts modelv1.TagValue to ValueType.
diff --git a/pkg/pb/v1/valuetype/valuetype.go b/pkg/pb/v1/valuetype/valuetype.go
new file mode 100644
index 000000000..0517796eb
--- /dev/null
+++ b/pkg/pb/v1/valuetype/valuetype.go
@@ -0,0 +1,39 @@
+// 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 valuetype holds the ValueType byte enum used to tag encoded tag and
+// field values. It is a dependency-free leaf: it imports nothing, so packages
+// that need only the type tag (e.g. the plugin SDK at pkg/pipeline/sdk) can
+// reference it without transitively pulling in the heavier pkg/pb/v1 package
+// and its logging/encoding dependencies. pkg/pb/v1 re-exports these symbols as
+// aliases, so existing pbv1.ValueType / pbv1.ValueType* usages are unaffected.
+package valuetype
+
+// ValueType is the type of the tag and field value.
+type ValueType byte
+
+// ValueType constants.
+const (
+ ValueTypeUnknown ValueType = iota
+ ValueTypeStr
+ ValueTypeInt64
+ ValueTypeFloat64
+ ValueTypeBinaryData
+ ValueTypeStrArr
+ ValueTypeInt64Arr
+ ValueTypeTimestamp
+)
diff --git a/pkg/pipeline/sdk/decode.go b/pkg/pipeline/sdk/decode.go
index 101cf695f..8d43e69ba 100644
--- a/pkg/pipeline/sdk/decode.go
+++ b/pkg/pipeline/sdk/decode.go
@@ -21,8 +21,8 @@ import (
"fmt"
"github.com/apache/skywalking-banyandb/pkg/convert"
- "github.com/apache/skywalking-banyandb/pkg/encoding"
- pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
+ "github.com/apache/skywalking-banyandb/pkg/encoding/vararray"
+ "github.com/apache/skywalking-banyandb/pkg/pb/v1/valuetype"
)
// Value is a single decoded tag value. The accessor matching ValueType returns
@@ -35,7 +35,7 @@ type Value struct {
intArr []int64
int64Val int64
floatVal float64
- valueType pbv1.ValueType
+ valueType valuetype.ValueType
null bool
}
@@ -43,7 +43,7 @@ type Value struct {
func (v Value) IsNull() bool { return v.null }
// ValueType returns the type tag of the decoded value.
-func (v Value) ValueType() pbv1.ValueType { return v.valueType }
+func (v Value) ValueType() valuetype.ValueType { return v.valueType }
// Str returns the string value (valid for ValueTypeStr).
func (v Value) Str() string { return v.str }
@@ -77,31 +77,31 @@ func (c *TagColumn) At(row int) (Value, error) {
// block, into a typed Value. It mirrors the engine's own per-row decode so a
// plugin never needs to import banyand/trace internals. A nil raw value yields
// a null Value.
-func DecodeTagValue(valueType pbv1.ValueType, raw []byte) (Value, error) {
+func DecodeTagValue(valueType valuetype.ValueType, raw []byte) (Value, error) {
if raw == nil {
return Value{valueType: valueType, null: true}, nil
}
switch valueType {
- case pbv1.ValueTypeStr:
+ case valuetype.ValueTypeStr:
return Value{valueType: valueType, str: string(raw)}, nil
- case pbv1.ValueTypeInt64:
+ case valuetype.ValueTypeInt64:
if len(raw) != 8 {
return Value{}, fmt.Errorf("int64: expected 8 bytes, got %d", len(raw))
}
return Value{valueType: valueType, int64Val: convert.BytesToInt64(raw)}, nil
- case pbv1.ValueTypeFloat64:
+ case valuetype.ValueTypeFloat64:
if len(raw) != 8 {
return Value{}, fmt.Errorf("float64: expected 8 bytes, got %d", len(raw))
}
return Value{valueType: valueType, floatVal: convert.BytesToFloat64(raw)}, nil
- case pbv1.ValueTypeBinaryData:
+ case valuetype.ValueTypeBinaryData:
return Value{valueType: valueType, bytes: raw}, nil
- case pbv1.ValueTypeTimestamp:
+ case valuetype.ValueTypeTimestamp:
if len(raw) != 8 {
return Value{}, fmt.Errorf("timestamp: expected 8 bytes, got %d", len(raw))
}
return Value{valueType: valueType, int64Val: convert.BytesToInt64(raw)}, nil
- case pbv1.ValueTypeInt64Arr:
+ case valuetype.ValueTypeInt64Arr:
if len(raw)%8 != 0 {
return Value{}, fmt.Errorf("int64 array: length %d is not a multiple of 8", len(raw))
}
@@ -110,10 +110,10 @@ func DecodeTagValue(valueType pbv1.ValueType, raw []byte) (Value, error) {
values = append(values, convert.BytesToInt64(raw[i:i+8]))
}
return Value{valueType: valueType, intArr: values}, nil
- case pbv1.ValueTypeStrArr:
+ case valuetype.ValueTypeStrArr:
var values []string
for idx := 0; idx < len(raw); {
- end, next, err := encoding.UnmarshalVarArray(raw, idx)
+ end, next, err := vararray.UnmarshalVarArray(raw, idx)
if err != nil {
return Value{}, fmt.Errorf("str array: %w", err)
}
@@ -121,7 +121,7 @@ func DecodeTagValue(valueType pbv1.ValueType, raw []byte) (Value, error) {
idx = next
}
return Value{valueType: valueType, strArr: values}, nil
- case pbv1.ValueTypeUnknown:
+ case valuetype.ValueTypeUnknown:
return Value{valueType: valueType, null: true}, nil
default:
return Value{}, fmt.Errorf("unsupported value type: %d", valueType)
diff --git a/pkg/pipeline/sdk/importgraph_test.go b/pkg/pipeline/sdk/importgraph_test.go
new file mode 100644
index 000000000..14c5d1edb
--- /dev/null
+++ b/pkg/pipeline/sdk/importgraph_test.go
@@ -0,0 +1,87 @@
+// 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 sdk_test
+
+import (
+ "bufio"
+ "bytes"
+ "os/exec"
+ "strings"
+ "testing"
+)
+
+// TestSDKImportGraph asserts that pkg/pipeline/sdk — the pinned surface plugin
+// authors build against — has NO transitive dependency on any logging, meter, or
+// heavy banyand-internal package. The SDK depends only on dependency-free leaves
+// (pkg/convert, pkg/encoding/vararray, pkg/pb/v1/valuetype). This is an absolute
+// check (not relative to pkg/pb/v1): a plugin built against this package must not
+// link zerolog/zap, so a host zerolog bump can never break the plugin ABI.
+func TestSDKImportGraph(t *testing.T) {
+ goPath, lookErr := exec.LookPath("go")
+ if lookErr != nil {
+ t.Skip("go binary not on PATH; skipping import-graph check")
+ }
+
+ forbidden := []string{
+ "github.com/rs/zerolog",
+ "go.uber.org/zap",
+ "github.com/apache/skywalking-banyandb/pkg/logger",
+ "github.com/apache/skywalking-banyandb/pkg/meter",
+ "github.com/apache/skywalking-banyandb/banyand/observability",
+ // pkg/pb/v1 is heavy (it imports pkg/logger); the SDK must reach ValueType
+ // via the pkg/pb/v1/valuetype leaf instead, so pbv1 itself is forbidden.
+ "github.com/apache/skywalking-banyandb/pkg/pb/v1",
+ }
+
+ sdkDeps := listDeps(t, goPath, "github.com/apache/skywalking-banyandb/pkg/pipeline/sdk")
+
+ for _, banned := range forbidden {
+ if sdkDeps[banned] {
+ t.Errorf("pkg/pipeline/sdk transitively imports forbidden package %q; the plugin SDK must stay"+
+ " dependency-pure (stdlib + leaf types only) so plugins are not version-locked to host internals", banned)
+ }
+ // Also check subpaths (e.g. github.com/rs/zerolog/internal/json), but do
+ // not let a leaf subpackage (pkg/pb/v1/valuetype) trip the pkg/pb/v1 prefix.
+ for dep := range sdkDeps {
+ if dep == "github.com/apache/skywalking-banyandb/pkg/pb/v1/valuetype" {
+ continue
+ }
+ if strings.HasPrefix(dep, banned+"/") {
+ t.Errorf("pkg/pipeline/sdk transitively imports forbidden package %q; the plugin SDK must stay pure", dep)
+ }
+ }
+ }
+}
+
+// listDeps returns the full transitive import set of pkg as a set (map to bool).
+func listDeps(t *testing.T, goPath, pkg string) map[string]bool {
+ t.Helper()
+ cmd := exec.Command(goPath, "list", "-deps", pkg)
+ out, runErr := cmd.Output()
+ if runErr != nil {
+ t.Fatalf("go list -deps %s failed: %v", pkg, runErr)
+ }
+ deps := make(map[string]bool)
+ sc := bufio.NewScanner(bytes.NewReader(out))
+ for sc.Scan() {
+ if line := strings.TrimSpace(sc.Text()); line != "" {
+ deps[line] = true
+ }
+ }
+ return deps
+}
diff --git a/pkg/pipeline/sdk/sdk.go b/pkg/pipeline/sdk/sdk.go
index 0871e0da8..2cec256ed 100644
--- a/pkg/pipeline/sdk/sdk.go
+++ b/pkg/pipeline/sdk/sdk.go
@@ -42,12 +42,13 @@
// chain this is the conjunction of the links' keep/drop verdicts.
//
// The boundary deliberately crosses only stdlib slices plus the engine's
-// stable, byte-sized pbv1.ValueType enum, so no banyand/trace-internal struct
-// is shared across the .so boundary.
+// stable, byte-sized valuetype.ValueType enum (a dependency-free leaf package,
+// re-exported by pkg/pb/v1 as pbv1.ValueType), so no banyand/trace-internal
+// struct — and no logging/meter dependency — is shared across the .so boundary.
package sdk
import (
- pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
+ "github.com/apache/skywalking-banyandb/pkg/pb/v1/valuetype"
)
// ABIVersion is compiled into the host and must be re-exported, unchanged, by
@@ -81,11 +82,13 @@ type Plugin interface {
// plugin's lifetime.
Kind() Kind
- // Project is the column-selection handshake, called once at load. The
- // engine honors it for the plugin's lifetime: Tags drives the native tag
- // projection (only those tag columns are decoded); SpanIDs and Spans gate
- // the spans stream. Intrinsic columns (TraceID, MinTS, MaxTS) are always
- // present regardless of the projection.
+ // Project is the column-selection handshake called when a merge chain is
+ // built (once per merge, not necessarily once per process load). The engine
+ // unions the projections from all samplers in the chain and honors the
+ // result for that chain's lifetime: Tags drives the native tag projection
+ // (only those tag columns are decoded); SpanIDs and Spans gate the spans
+ // stream. Intrinsic columns (TraceID, MinTS, MaxTS) are always present
+ // regardless of the projection.
Project() Projection
// Close releases any resources the plugin holds. It is called once when the
@@ -196,7 +199,7 @@ type TagColumn struct {
Values [][]byte
// ValueType is the engine's stable, byte-sized type tag for every value in
// the column.
- ValueType pbv1.ValueType
+ ValueType valuetype.ValueType
}
// Verdict is the per-trace decision, aligned to TraceBatch.Traces.
diff --git a/pkg/pipeline/sdk/telemetry.go b/pkg/pipeline/sdk/telemetry.go
new file mode 100644
index 000000000..d7b70326c
--- /dev/null
+++ b/pkg/pipeline/sdk/telemetry.go
@@ -0,0 +1,104 @@
+// 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.
+
+// Freeze note: this file defines the plugin-facing telemetry façade. It is
+// FROZEN: adding or removing methods on any interface declared here is a
+// breaking .so ABI change and requires an ABIVersion bump. The boundary
+// deliberately crosses only stdlib types so that no zerolog, zap, or
+// banyand-internal meter type leaks across the .so boundary.
+
+package sdk
+
+// Meter is the bounded metric surface handed to a plugin via Host. The host
+// enforces naming, labels, cardinality, and lifecycle; the plugin only requests.
+// Methods are safe to call from multiple goroutines concurrently.
+type Meter interface {
+ // Counter returns or creates a counter with the given name and label names.
+ Counter(name string, labelNames ...string) Counter
+ // Gauge returns or creates a gauge with the given name and label names.
+ Gauge(name string, labelNames ...string) Gauge
+ // Histogram returns or creates a histogram with the given name, explicit
+ // bucket boundaries, and label names.
+ Histogram(name string, buckets []float64, labelNames ...string) Histogram
+}
+
+// Counter is a monotonically increasing metric. It must not be decremented.
+type Counter interface {
+ // Inc adds delta to the counter for the given label values. Label values
+ // must align positionally with the label names declared in Meter.Counter.
+ // Delta must be non-negative.
+ Inc(delta float64, labelValues ...string)
+}
+
+// Gauge is a metric whose value can go up or down freely.
+type Gauge interface {
+ // Set replaces the gauge value for the given label values.
+ Set(value float64, labelValues ...string)
+ // Add adds delta (positive or negative) to the gauge for the given label values.
+ Add(delta float64, labelValues ...string)
+}
+
+// Histogram records observations into pre-declared buckets. Use it to track
+// distributions such as latency or payload size.
+type Histogram interface {
+ // Observe records a single observation for the given label values.
+ Observe(value float64, labelValues ...string)
+}
+
+// Logger is the bounded logging surface handed to a plugin via Host. It exposes
+// no third-party types (e.g. zerolog, zap) — only a message plus alternating
+// key/value pairs so no external logger dependency crosses the .so boundary.
+// Implementations are expected to be safe for concurrent use.
+type Logger interface {
+ // Debug logs a debug-level message with optional structured key/value pairs.
+ Debug(msg string, keysAndValues ...any)
+ // Info logs an info-level message with optional structured key/value pairs.
+ Info(msg string, keysAndValues ...any)
+ // Warn logs a warning-level message with optional structured key/value pairs.
+ Warn(msg string, keysAndValues ...any)
+ // Error logs an error-level message with optional structured key/value pairs.
+ Error(msg string, keysAndValues ...any)
+}
+
+// Host is the telemetry context the engine injects into a HostAware plugin.
+// It provides scoped metric and logging surfaces that the engine owns; the
+// plugin must not cache a Host beyond the lifetime of the pipeline config that
+// created it. Instrument handles obtained from Meter() are valid for the
+// lifetime of that Host: do not cache a Counter, Gauge, or Histogram across
+// distinct UseHost calls, because the engine may reclaim and rebuild a group's
+// metric series on reconcile or teardown — a handle from a prior Host would
+// emit into a reclaimed series.
+type Host interface {
+ // Meter returns the metric surface scoped to this plugin's group.
+ Meter() Meter
+ // Logger returns the logging surface scoped to this plugin's group.
+ Logger() Logger
+}
+
+// HostAware is the optional interface a plugin implements to receive a Host.
+// The engine type-asserts the constructed sampler to HostAware and, if present,
+// calls UseHost exactly once before the plugin's first Decide call. Plugins
+// that do not implement HostAware are unaffected — there is no ABI change.
+// A plugin may cache the Host and use it for the duration of Decide; the
+// engine guarantees a distinct Host per group so metrics and logs are
+// automatically namespaced.
+type HostAware interface {
+ // UseHost delivers the engine-owned telemetry context to the plugin. It is
+ // called once, before the first Decide, with a Host scoped to the plugin's
+ // group.
+ UseHost(Host)
+}
diff --git a/test/plugins/_faultysampler/main.go b/test/plugins/_faultysampler/main.go
new file mode 100644
index 000000000..4f8940baa
--- /dev/null
+++ b/test/plugins/_faultysampler/main.go
@@ -0,0 +1,159 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// Command faultysampler is an adversarial reference post-trace sampler plugin
+// that verifies the engine's host-side bounds hold under hostile plugin
+// behaviour. It implements both sdk.Sampler and sdk.HostAware, but its config
+// selects one or more fault modes that deliberately abuse the telemetry façade:
+//
+// - panicInUseHost: UseHost panics immediately; the engine must isolate the
+// panic so telemetry is disabled for this plugin but the plugin continues
+// to run (fail-open) and does not crash the data node.
+//
+// - floodLogs: Decide emits thousands of Logger.Info calls per batch; the
+// host's rate-limiter must clamp throughput to its declared budget (50
+// lines/s burst 100) so the data node's log stream is not flooded.
+//
+// - explodeCardinality: Decide calls Meter.Counter with a unique label value
+// per invocation, producing unbounded distinct series; the host's
+// cardinality cap (100 series per plugin) must clamp the overflow so
+// Prometheus scrape memory does not grow without bound.
+//
+// Decision logic: always retain all traces (fail-open) regardless of fault
+// mode, so liveness is not affected by the adversarial telemetry behaviour.
+// Nil-safe: if UseHost was never called (host is nil — e.g. because the engine
+// isolated a panic in UseHost) Decide proceeds without emitting any telemetry.
+//
+// Config JSON (from SamplerPlugin.config Struct):
+//
+// {
+// "panicInUseHost": false,
+// "floodLogs": false,
+// "explodeCardinality": false,
+// "floodCount": 10000
+// }
+//
+// All fields default to false / 10000. Multiple fault modes may be combined.
+//
+// Build it as a Go plugin (the leading-underscore directory is intentionally
+// excluded from `go build ./...` and linters, exactly like the reference
+// _example directory, yet remains buildable by explicit path):
+//
+// CGO_ENABLED=1 go build -buildmode=plugin -trimpath \
+// -o faultysampler.so \
+// ./test/plugins/_faultysampler
+package main
+
+import (
+ "encoding/json"
+ "fmt"
+ "sync/atomic"
+
+ "github.com/apache/skywalking-banyandb/pkg/pipeline/sdk"
+)
+
+// ABIVersion re-exports the SDK ABI version. The engine refuses to load the
+// plugin unless this equals its own compiled sdk.ABIVersion.
+var ABIVersion = sdk.ABIVersion
+
+// samplerConfig is the JSON shape the operator sets in SamplerPlugin.config.
+type samplerConfig struct {
+ // PanicInUseHost causes UseHost to panic; the engine must recover.
+ PanicInUseHost bool `json:"panicInUseHost"`
+ // FloodLogs causes Decide to emit FloodCount Logger.Info calls per batch.
+ FloodLogs bool `json:"floodLogs"`
+ // ExplodeCardinality causes Decide to register a new Counter with a unique
+ // label value on every call, exhausting the host cardinality cap.
+ ExplodeCardinality bool `json:"explodeCardinality"`
+ // FloodCount is the number of log lines emitted per batch when FloodLogs is
+ // true; defaults to 10000.
+ FloodCount int `json:"floodCount"`
+}
+
+// faultySampler is a deliberately adversarial plugin for host-bounds testing.
+type faultySampler struct {
+ cfg samplerConfig
+ cachedHost sdk.Host
+ // callSeq is a monotonically increasing call counter used to generate
+ // unique cardinality-explosion label values.
+ callSeq atomic.Int64
+}
+
+// NewSampler is the constructor symbol the engine looks up. It parses the
+// operator config and returns a ready-to-use Sampler.
+func NewSampler(configJSON []byte) (sdk.Sampler, error) {
+ cfg := samplerConfig{FloodCount: 10000}
+ if len(configJSON) > 0 {
+ if unmarshalErr := json.Unmarshal(configJSON, &cfg); unmarshalErr != nil {
+ return nil, fmt.Errorf("faultysampler: invalid config JSON: %w", unmarshalErr)
+ }
+ if cfg.FloodCount <= 0 {
+ cfg.FloodCount = 10000
+ }
+ }
+ return &faultySampler{cfg: cfg}, nil
+}
+
+// UseHost satisfies sdk.HostAware. When panicInUseHost is set it panics
+// immediately; the engine must recover from the panic so the plugin continues
+// to run and the data node does not crash. Otherwise the Host is cached for
+// use in Decide.
+func (s *faultySampler) UseHost(h sdk.Host) {
+ if s.cfg.PanicInUseHost {
+ panic("faultysampler: intentional panic in UseHost (panicInUseHost=true)")
+ }
+ s.cachedHost = h
+}
+
+// Kind reports the sampler kind.
+func (s *faultySampler) Kind() sdk.Kind { return sdk.KindSampler }
+
+// Project declares no tag columns — this sampler keeps everything regardless.
+func (s *faultySampler) Project() sdk.Projection { return sdk.Projection{} }
+
+// Close releases resources; this sampler holds none.
+func (s *faultySampler) Close() error { return nil }
+
+// Decide always retains every trace (fail-open) while exercising the selected
+// host-bounds fault modes. All telemetry calls are nil-safe: they are skipped
+// when cachedHost is nil (e.g. because the engine isolated a UseHost panic).
+func (s *faultySampler) Decide(batch *sdk.TraceBatch) (sdk.Verdict, error) {
+ keep := make([]bool, len(batch.Traces))
+ for idx := range keep {
+ keep[idx] = true
+ }
+ if s.cachedHost == nil {
+ return sdk.Verdict{Keep: keep}, nil
+ }
+
+ if s.cfg.FloodLogs {
+ logger := s.cachedHost.Logger()
+ for i := 0; i < s.cfg.FloodCount; i++ {
+ logger.Info("flood log line", "i", i, "traces", len(batch.Traces))
+ }
+ }
+
+ if s.cfg.ExplodeCardinality {
+ seq := s.callSeq.Add(1)
+ uniqueID := fmt.Sprintf("uid-%d", seq)
+ s.cachedHost.Meter().Counter("boom", "id").Inc(1, uniqueID)
+ }
+
+ return sdk.Verdict{Keep: keep}, nil
+}
+
+func main() {}
diff --git a/test/plugins/_latencystatussampler/main.go b/test/plugins/_latencystatussampler/main.go
index f856de682..3476a25ec 100644
--- a/test/plugins/_latencystatussampler/main.go
+++ b/test/plugins/_latencystatussampler/main.go
@@ -40,7 +40,7 @@ import (
"encoding/json"
"fmt"
- pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
+ "github.com/apache/skywalking-banyandb/pkg/pb/v1/valuetype"
"github.com/apache/skywalking-banyandb/pkg/pipeline/sdk"
)
@@ -160,7 +160,7 @@ func (s *latencyStatusSampler) keepTrace(b *sdk.TraceBlock) bool {
// Decode duration as int64 milliseconds.
var durationMs int64
switch durVal.ValueType() {
- case pbv1.ValueTypeInt64:
+ case valuetype.ValueTypeInt64:
durationMs = durVal.Int64()
default:
return true // fail open: unexpected duration type
@@ -169,7 +169,7 @@ func (s *latencyStatusSampler) keepTrace(b *sdk.TraceBlock) bool {
// Decode status as string.
var statusStr string
switch statusVal.ValueType() {
- case pbv1.ValueTypeStr:
+ case valuetype.ValueTypeStr:
statusStr = statusVal.Str()
default:
return true // fail open: unexpected status type
diff --git a/test/plugins/_telemetrysampler/main.go b/test/plugins/_telemetrysampler/main.go
new file mode 100644
index 000000000..afa717fd7
--- /dev/null
+++ b/test/plugins/_telemetrysampler/main.go
@@ -0,0 +1,128 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// Command telemetrysampler is a reference post-trace sampler plugin that
+// demonstrates well-behaved use of the sdk.HostAware telemetry façade. It
+// implements both sdk.Sampler and sdk.HostAware: the engine calls UseHost once
+// (per group, before the first Decide) and the sampler caches the Host to emit
+// bounded metrics and structured logs from Decide.
+//
+// Decision logic: keep all traces, but increment a "decisions" counter (with
+// a "verdict" label set to "keep") and emit a sparse Info log so the host's
+// rate-limiter and cardinality cap are exercised in a benign way. The sampler
+// is nil-safe: if UseHost was never called (host is nil) Decide proceeds
+// silently without emitting any telemetry.
+//
+// Config JSON (from SamplerPlugin.config Struct):
+//
+// { "logEvery": 100 }
+//
+// logEvery controls how many batches are processed between Info log emissions;
+// the default is 100. 0 or a negative value is treated as 100.
+//
+// Build it as a Go plugin (the leading-underscore directory is intentionally
+// excluded from `go build ./...` and linters, exactly like the reference
+// _example directory, yet remains buildable by explicit path):
+//
+// CGO_ENABLED=1 go build -buildmode=plugin -trimpath \
+// -o telemetrysampler.so \
+// ./test/plugins/_telemetrysampler
+package main
+
+import (
+ "encoding/json"
+ "fmt"
+ "sync/atomic"
+
+ "github.com/apache/skywalking-banyandb/pkg/pipeline/sdk"
+)
+
+// ABIVersion re-exports the SDK ABI version. The engine refuses to load the
+// plugin unless this equals its own compiled sdk.ABIVersion.
+var ABIVersion = sdk.ABIVersion
+
+// samplerConfig is the JSON shape the operator sets in SamplerPlugin.config.
+type samplerConfig struct {
+ // LogEvery controls the inter-batch gap between Info log emissions.
+ LogEvery int64 `json:"logEvery"`
+}
+
+// telemetrySampler keeps all traces while emitting bounded metrics and logs
+// via the injected Host.
+type telemetrySampler struct {
+ logEvery int64
+ // cachedHost is set by UseHost; may be nil if the engine did not call it.
+ cachedHost sdk.Host
+ // decisions is the cached Counter used across Decide calls.
+ decisions sdk.Counter
+ // batchCount tracks how many batches have been processed since last log.
+ batchCount atomic.Int64
+}
+
+// NewSampler is the constructor symbol the engine looks up. It parses the
+// operator config and returns a ready-to-use Sampler. An empty or missing
+// config is accepted and yields the default settings.
+func NewSampler(configJSON []byte) (sdk.Sampler, error) {
+ cfg := samplerConfig{LogEvery: 100}
+ if len(configJSON) > 0 {
+ if unmarshalErr := json.Unmarshal(configJSON, &cfg); unmarshalErr != nil {
+ return nil, fmt.Errorf("telemetrysampler: invalid config JSON: %w", unmarshalErr)
+ }
+ if cfg.LogEvery <= 0 {
+ cfg.LogEvery = 100
+ }
+ }
+ return &telemetrySampler{logEvery: cfg.LogEvery}, nil
+}
+
+// UseHost satisfies sdk.HostAware. The engine calls this exactly once (per
+// group) before the first Decide. The plugin caches h and initializes its
+// Counter here so Decide never races on first use.
+func (s *telemetrySampler) UseHost(h sdk.Host) {
+ s.cachedHost = h
+ s.decisions = h.Meter().Counter("decisions", "verdict")
+}
+
+// Kind reports the sampler kind.
+func (s *telemetrySampler) Kind() sdk.Kind { return sdk.KindSampler }
+
+// Project declares no tag columns — this sampler keeps everything and only
+// needs the intrinsic TraceID/MinTS/MaxTS columns (always present).
+func (s *telemetrySampler) Project() sdk.Projection { return sdk.Projection{} }
+
+// Close releases resources; this sampler holds none beyond the injected Host.
+func (s *telemetrySampler) Close() error { return nil }
+
+// Decide retains every trace in the batch, emitting a counter increment and a
+// sparse Info log via the injected Host. Nil-safe: no telemetry is emitted
+// when UseHost was never called.
+func (s *telemetrySampler) Decide(batch *sdk.TraceBatch) (sdk.Verdict, error) {
+ keep := make([]bool, len(batch.Traces))
+ for idx := range keep {
+ keep[idx] = true
+ }
+ if s.cachedHost != nil {
+ s.decisions.Inc(float64(len(batch.Traces)), "keep")
+ batchN := s.batchCount.Add(1)
+ if batchN%s.logEvery == 1 {
+ s.cachedHost.Logger().Info("batch decided", "traces", len(batch.Traces), "batchN", batchN)
+ }
+ }
+ return sdk.Verdict{Keep: keep}, nil
+}
+
+func main() {}
diff --git a/test/plugins/telemetry_smoke_test.go b/test/plugins/telemetry_smoke_test.go
new file mode 100644
index 000000000..2ebfe8954
--- /dev/null
+++ b/test/plugins/telemetry_smoke_test.go
@@ -0,0 +1,171 @@
+// 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 && !race
+
+// Package plugins_test is a build/load smoke test for the reference telemetry
+// plugins. It compiles each plugin as a .so in a temporary directory, opens
+// it with plugin.Open, and verifies the exported symbols and the sdk.HostAware
+// type assertion. The test is gated behind the trace_pipeline build tag so it
+// only runs when the C toolchain is present (the same gate used by the
+// integration suite and the Makefile target).
+//
+// NOTE: this test must NOT be run with -race. A race-instrumented host cannot
+// load a .so built without -race; the make target (test-trace-pipeline) omits
+// -race for exactly this reason. See the comment above PLUGIN_OUTPUT_DIR in
+// the root Makefile for the full rationale.
+package plugins_test
+
+import (
+ "os"
+ "os/exec"
+ "path/filepath"
+ "plugin"
+ "runtime"
+ "strings"
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+
+ "github.com/apache/skywalking-banyandb/pkg/pipeline/sdk"
+)
+
+// isToolchainMismatch reports whether err is the Go runtime's rejection of a
+// .so whose build context differs from the host's (compiler version skew).
+// plugin.Open surfaces this as "different version of package ...".
+func isToolchainMismatch(err error) bool {
+ if err == nil {
+ return false
+ }
+ msg := err.Error()
+ return strings.Contains(msg, "different version of package") ||
+ strings.Contains(msg, "plugin was built with a different version")
+}
+
+// buildPlugin compiles the plugin at srcDir into outDir/<name>.so and returns
+// the .so path. The test is skipped if the build fails (CGO unavailable, no
+// C toolchain, or `go` not found in PATH).
+func buildPlugin(t *testing.T, outDir, name, srcDir string) string {
+ t.Helper()
+ if runtime.GOOS == "windows" {
+ t.Skip("Go plugins are not supported on Windows")
+ }
+
+ goExe, lookErr := exec.LookPath("go")
+ if lookErr != nil {
+ t.Skip("telemetry plugin smoke test skipped: 'go' binary not found in PATH")
+ }
+
+ // Resolve the plugin source directory from this file's location so the path
+ // is correct regardless of the working directory when tests are run.
+ _, thisFile, _, ok := runtime.Caller(0)
+ if !ok {
+ t.Skip("telemetry plugin smoke test skipped: cannot determine source file path via runtime.Caller")
+ }
+ absPluginDir := filepath.Clean(filepath.Join(filepath.Dir(thisFile), srcDir))
+
+ soPath := filepath.Join(outDir, name+".so")
+
+ // Do NOT pass -trimpath or -race: go test is not built with -trimpath, and
+ // plugin.Open requires the host and plugin to share the same package path
+ // encoding. This test is excluded from -race runs by the trace_pipeline
+ // build tag and the Makefile comment; do not add -race here.
+ cmd := exec.Command(goExe, "build", "-buildmode=plugin", "-o", soPath, absPluginDir)
+ cmd.Env = append(os.Environ(), "CGO_ENABLED=1")
+ out, buildErr := cmd.CombinedOutput()
+ if buildErr != nil {
+ t.Skipf("telemetry plugin smoke test skipped: cannot build %s.so (CGO may be unavailable): %v\n%s",
+ name, buildErr, out)
+ }
+ return soPath
+}
+
+// openPlugin opens soPath and skips the test on a toolchain mismatch.
+func openPlugin(t *testing.T, soPath string) *plugin.Plugin {
+ t.Helper()
+ p, openErr := plugin.Open(soPath)
+ if isToolchainMismatch(openErr) {
+ t.Skipf("telemetry plugin smoke test skipped: host cannot load a freshly built plugin (toolchain mismatch): %v", openErr)
+ }
+ require.NoError(t, openErr, "plugin.Open(%s)", soPath)
+ return p
+}
+
+// TestTelemetrySamplerSmoke builds and loads _telemetrysampler, then checks
+// ABIVersion, NewSampler, and the sdk.HostAware type assertion.
+func TestTelemetrySamplerSmoke(t *testing.T) {
+ tmpDir, mkdirErr := os.MkdirTemp("", "telemetrysampler-smoke-*")
+ require.NoError(t, mkdirErr)
+ defer os.RemoveAll(tmpDir)
+
+ soPath := buildPlugin(t, tmpDir, "telemetrysampler", "_telemetrysampler")
+ p := openPlugin(t, soPath)
+
+ // ABIVersion must equal sdk.ABIVersion.
+ abiSym, lookErr := p.Lookup("ABIVersion")
+ require.NoError(t, lookErr, "ABIVersion symbol must be exported")
+ abiPtr, ok := abiSym.(*int)
+ require.True(t, ok, "ABIVersion must be *int, got %T", abiSym)
+ assert.Equal(t, sdk.ABIVersion, *abiPtr, "ABIVersion mismatch")
+
+ // NewSampler must be present and callable.
+ ctorSym, lookErr := p.Lookup("NewSampler")
+ require.NoError(t, lookErr, "NewSampler symbol must be exported")
+ ctor, ok := ctorSym.(func([]byte) (sdk.Sampler, error))
+ require.True(t, ok, "NewSampler must have signature func([]byte)(sdk.Sampler,error), got %T", ctorSym)
+
+ sampler, ctorErr := ctor(nil)
+ require.NoError(t, ctorErr, "NewSampler(nil) must succeed")
+ require.NotNil(t, sampler)
+
+ // The sampler must implement sdk.HostAware.
+ _, isHostAware := sampler.(sdk.HostAware)
+ assert.True(t, isHostAware, "telemetrysampler must implement sdk.HostAware")
+}
+
+// TestFaultySamplerSmoke builds and loads _faultysampler, then checks
+// ABIVersion, NewSampler, and the sdk.HostAware type assertion.
+func TestFaultySamplerSmoke(t *testing.T) {
+ tmpDir, mkdirErr := os.MkdirTemp("", "faultysampler-smoke-*")
+ require.NoError(t, mkdirErr)
+ defer os.RemoveAll(tmpDir)
+
+ soPath := buildPlugin(t, tmpDir, "faultysampler", "_faultysampler")
+ p := openPlugin(t, soPath)
+
+ // ABIVersion must equal sdk.ABIVersion.
+ abiSym, lookErr := p.Lookup("ABIVersion")
+ require.NoError(t, lookErr, "ABIVersion symbol must be exported")
+ abiPtr, ok := abiSym.(*int)
+ require.True(t, ok, "ABIVersion must be *int, got %T", abiSym)
+ assert.Equal(t, sdk.ABIVersion, *abiPtr, "ABIVersion mismatch")
+
+ // NewSampler must be present and callable.
+ ctorSym, lookErr := p.Lookup("NewSampler")
+ require.NoError(t, lookErr, "NewSampler symbol must be exported")
+ ctor, ok := ctorSym.(func([]byte) (sdk.Sampler, error))
+ require.True(t, ok, "NewSampler must have signature func([]byte)(sdk.Sampler,error), got %T", ctorSym)
+
+ sampler, ctorErr := ctor(nil)
+ require.NoError(t, ctorErr, "NewSampler(nil) must succeed")
+ require.NotNil(t, sampler)
+
+ // The sampler must implement sdk.HostAware.
+ _, isHostAware := sampler.(sdk.HostAware)
+ assert.True(t, isHostAware, "faultysampler must implement sdk.HostAware")
+}