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/14 02:29:10 UTC
(skywalking-banyandb) branch main updated: feat(trace): enforce single-trace-per-group when a sampler pipeline is enabled (#1215)
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 2fbb65efa feat(trace): enforce single-trace-per-group when a sampler pipeline is enabled (#1215)
2fbb65efa is described below
commit 2fbb65efa68b60991fe824e764b85d9b3aef67df
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Tue Jul 14 10:29:04 2026 +0800
feat(trace): enforce single-trace-per-group when a sampler pipeline is enabled (#1215)
* feat(trace): enforce single-trace-per-group when a sampler pipeline is enabled
A trace-pipeline sampler applies to one trace schema per group, so a group
whose pipeline is enabled with a sampler plugin must contain exactly one
trace. Enforce this at registration time in both directions:
- CreateTrace/UpdateTrace reject introducing a second, differently-named
trace into such a group (the updated trace is excluded by name).
- CreateGroup/UpdateGroup reject enabling a sampler pipeline on a group
that already holds more than one trace.
The constraint triggers only when the pipeline is enabled and declares at
least one sampler plugin; disabled or sampler-less pipelines are unaffected.
Rejections use codes.FailedPrecondition, consistent with the neighboring
tombstone invariant.
Part of apache/skywalking#13634
* docs(trace): document plugin pipeline registration (the activation step)
The plugin development guide and plugins.md showed the TracePipelineConfig JSON
but never explained how to attach it to a group, which is the step that actually
activates the plugin (the data node reconciles the group's pipeline and calls
plugin.Open on the mounted .so). Document the supported operator paths — bydbctl
`group update -f`, REST `PUT /api/v1/group/schema/{group}`, and the gRPC
GroupRegistryService.Update (all write through the single schema registry and
preserve the nested pipeline field) — plus the sampler_active_count activation
check.
Also correct the plugin-sidecar register helper's comment: it claimed the HTTP
gateway drops the nested pipeline field and that the GroupRegistryService is not
watched. Both are wrong — protojson round-trips the field losslessly and the
public GroupRegistryService writes through the same property-based schema store.
The helper uses the property-schema client only because the standalone test pod
exposes just that schema-server address, not a liaison registry endpoint.
Part of apache/skywalking#13634
* fix(trace): run sampler single-trace check after existence is confirmed
Addresses Copilot review on #1215: UpdateTrace and UpdateGroup ran the
sampler single-trace check before updateResource loaded the existing
resource, so updating a non-existent trace/group in a sampler-enabled group
returned codes.FailedPrecondition instead of a not-found error.
Move both checks into updateResource validator callbacks, which run only
after the existing schema has been confirmed. Add tests asserting a missing
trace/group now surfaces not-found rather than FailedPrecondition.
Part of apache/skywalking#13634
* fix(lifecycle): resolve co-located node by --grpc-addr to de-flake sender tier
resolveSelfIdentity matched the sender only by pod hostname via
hostMatches, which strips the port. In the distributed lifecycle test
both data nodes register on 127.0.0.1 (differing only by port), so the
resolver returned whichever node the registry listing ordered first --
non-deterministically stamping remote_tier="warm" instead of "hot"
and failing the queue_sub_total_finished sender-label assertion.
Add a port-aware first pass that matches the lifecycle's --grpc-addr
against the registry GrpcAddress via grpcAddrEqual (full host:port),
deterministically pinning the co-located node even when nodes share a
host. Pod-hostname matching remains as the pass-2 fallback for the K8s
sidecar case. Plumb l.gRPCAddr through parseGroup and add a regression
test reproducing the shared-loopback ordering.
---
banyand/backup/lifecycle/service.go | 6 +-
banyand/backup/lifecycle/steps.go | 74 +++++---
banyand/backup/lifecycle/steps_test.go | 58 ++++--
banyand/metadata/schema/property/client.go | 83 ++++++++-
.../schema/property/pipeline_single_trace_test.go | 203 +++++++++++++++++++++
docs/operation/plugins-development.md | 30 ++-
docs/operation/plugins.md | 10 +
test/plugin-sidecar/register/main.go | 15 +-
8 files changed, 421 insertions(+), 58 deletions(-)
diff --git a/banyand/backup/lifecycle/service.go b/banyand/backup/lifecycle/service.go
index ed2433515..c4607a45c 100644
--- a/banyand/backup/lifecycle/service.go
+++ b/banyand/backup/lifecycle/service.go
@@ -1203,7 +1203,7 @@ func (l *lifecycleService) getGroupsToProcess(ctx context.Context, progress *Pro
func (l *lifecycleService) processStreamGroup(ctx context.Context, g *commonv1.Group,
streamDir string, nodes []*databasev1.Node, labels map[string]string, progress *Progress,
) {
- group, senderNode, senderRole, senderTier, err := parseGroup(g, labels, nodes, l.l, l.metadata, l.clusterStateMgr, l.omr)
+ group, senderNode, senderRole, senderTier, err := parseGroup(g, l.gRPCAddr, labels, nodes, l.l, l.metadata, l.clusterStateMgr, l.omr)
if err != nil {
l.l.Error().Err(err).Msgf("failed to parse group %s", g.Metadata.Name)
return
@@ -1323,7 +1323,7 @@ func (l *lifecycleService) deleteExpiredStreamSegments(ctx context.Context, g *c
func (l *lifecycleService) processMeasureGroup(ctx context.Context, g *commonv1.Group, measureDir string,
nodes []*databasev1.Node, labels map[string]string, progress *Progress,
) {
- group, senderNode, senderRole, senderTier, err := parseGroup(g, labels, nodes, l.l, l.metadata, l.clusterStateMgr, l.omr)
+ group, senderNode, senderRole, senderTier, err := parseGroup(g, l.gRPCAddr, labels, nodes, l.l, l.metadata, l.clusterStateMgr, l.omr)
if err != nil {
l.l.Error().Err(err).Msgf("failed to parse group %s", g.Metadata.Name)
return
@@ -1430,7 +1430,7 @@ func (l *lifecycleService) deleteExpiredTraceSegments(ctx context.Context, g *co
func (l *lifecycleService) processTraceGroup(ctx context.Context, g *commonv1.Group, traceDir string,
nodes []*databasev1.Node, labels map[string]string, progress *Progress,
) {
- group, senderNode, senderRole, senderTier, err := parseGroup(g, labels, nodes, l.l, l.metadata, l.clusterStateMgr, l.omr)
+ group, senderNode, senderRole, senderTier, err := parseGroup(g, l.gRPCAddr, labels, nodes, l.l, l.metadata, l.clusterStateMgr, l.omr)
if err != nil {
l.l.Error().Err(err).Msgf("failed to parse group %s", g.Metadata.Name)
return
diff --git a/banyand/backup/lifecycle/steps.go b/banyand/backup/lifecycle/steps.go
index 0c3f1d9a3..e0d7540ee 100644
--- a/banyand/backup/lifecycle/steps.go
+++ b/banyand/backup/lifecycle/steps.go
@@ -101,23 +101,39 @@ func (l *lifecycleService) getSnapshots(ctx context.Context, groups []*commonv1.
// its banyandb_queue_sub_total_* family with non-empty remote_node and
// remote_tier.
//
-// Resolution: the lifecycle sidecar's own pod hostname is the stable
-// identifier the receiver will see as the sender. It comes from
-// POD_NAME (K8s downward API) and falls back to os.Hostname() — the
-// same precedence as nativeNodeContext at service.go:160-165. The
-// function then looks the host up directly in the data-node registry,
-// matching against the host portion of GrpcAddress (the registry may
-// carry an IP, a headless-service FQDN, or a loopback alias, depending
-// on which bind address the data pod registered with). The first
-// registry entry whose host matches (with loopback-alias and
-// IP-literal normalization via hostMatches) is the co-located data
-// pod; its Metadata.Name is the SenderNode and its Labels["type"] is
-// the SenderTier.
+// Resolution runs in two passes and re-runs on every parseGroup call (no
+// caching), so a data-pod restart, re-registration, or new host is
+// picked up by the next cycle:
//
-// Re-runs on every parseGroup call (no caching) so a data-pod
-// restart, re-registration, or new host is picked up by the next
-// cycle. Returns ok=false when no registry entry matches.
-func resolveSelfIdentity(selfPodHost string, nodes []*databasev1.Node) (senderNode, senderTier string, ok bool) {
+// - Pass 1 (--grpc-addr): the lifecycle already knows the exact
+// co-located data node it snapshots from — its --grpc-addr. Matching
+// that against the registry's GrpcAddress via grpcAddrEqual compares
+// the full host:port, so when several data nodes share a host (e.g.
+// 127.0.0.1 in the integration harness, or a shared node IP) the
+// co-located node is still resolved unambiguously. Hostname matching
+// alone strips the port and would pick whichever same-host node the
+// registry listing happens to return first.
+// - Pass 2 (pod hostname): falls back to the sidecar's own pod
+// hostname (POD_NAME → os.Hostname(), same precedence as
+// nativeNodeContext at service.go:160-165) matched against the host
+// portion of GrpcAddress via hostMatches. This covers the K8s case
+// where --grpc-addr is a loopback or service alias that does not
+// string-match the pod's registered GrpcAddress.
+//
+// The first matching registry entry is the co-located data pod; its
+// Metadata.Name is the SenderNode and its Labels["type"] is the
+// SenderTier. Returns ok=false when no registry entry matches.
+func resolveSelfIdentity(selfGRPCAddr, selfPodHost string, nodes []*databasev1.Node) (senderNode, senderTier string, ok bool) {
+ if selfGRPCAddr != "" {
+ for _, n := range nodes {
+ if n == nil || n.Metadata == nil {
+ continue
+ }
+ if grpcAddrEqual(n.GrpcAddress, selfGRPCAddr) {
+ return n.Metadata.Name, n.Labels["type"], true
+ }
+ }
+ }
if selfPodHost == "" {
return "", "", false
}
@@ -269,7 +285,7 @@ func cloneIntervalRule(ir *commonv1.IntervalRule) *commonv1.IntervalRule {
//nolint:contextcheck // health check goroutine uses context.Background()
func parseGroup(
- g *commonv1.Group, nodeLabels map[string]string, nodes []*databasev1.Node,
+ g *commonv1.Group, selfGRPCAddr string, nodeLabels map[string]string, nodes []*databasev1.Node,
l *logger.Logger, metadata metadata.Repo, clusterStateMgr *clusterStateManager,
omr observability.MetricsRegistry,
) (group *GroupConfig, senderNode, senderRole, senderTier string, err error) {
@@ -348,16 +364,18 @@ func parseGroup(
// Stamp the lifecycle's self identity onto the publisher so the wire
// SenderNode / SenderRole / SenderTier fields and the parallel
// banyandb_lifecycle_migration_* labels are populated. The
- // resolveSelfIdentity algorithm matches the lifecycle's own pod
- // hostname (POD_NAME -> os.Hostname(), same precedence as
- // nativeNodeContext at service.go:160-165) against the
- // data-node registry's GrpcAddress with loopback-alias and
- // port-strip normalization. The first matching registry entry is
- // the co-located data pod; its Metadata.Name is the BanyanDB
- // NodeID the receiver records as remote_node, and its
- // Labels["type"] is the receiver's remote_tier. SenderRole is
- // hard-coded to "lifecycle" to mirror the liaison's
- // "liaison" pattern at pkg/cmdsetup/liaison.go:170-171.
+ // resolveSelfIdentity algorithm first matches the lifecycle's
+ // --grpc-addr against the data-node registry's GrpcAddress
+ // (full host:port, so co-located nodes sharing a host are still
+ // disambiguated), then falls back to the sidecar's own pod hostname
+ // (POD_NAME -> os.Hostname(), same precedence as nativeNodeContext
+ // at service.go:160-165) with loopback-alias and port-strip
+ // normalization. The first matching registry entry is the
+ // co-located data pod; its Metadata.Name is the BanyanDB NodeID the
+ // receiver records as remote_node, and its Labels["type"] is the
+ // receiver's remote_tier. SenderRole is hard-coded to "lifecycle" to
+ // mirror the liaison's "liaison" pattern at
+ // pkg/cmdsetup/liaison.go:170-171.
//
// The (senderNode, "lifecycle", senderTier) tuple returned here is
// consumed by three downstream emissions, all sharing the same
@@ -374,7 +392,7 @@ func parseGroup(
// the struct comment in service.go and CHANGES.md for the
// asymmetry.
selfHost := selfPodHostname()
- senderNode, senderTier, resolvedOK := resolveSelfIdentity(selfHost, nodes)
+ senderNode, senderTier, resolvedOK := resolveSelfIdentity(selfGRPCAddr, selfHost, nodes)
if resolvedOK {
senderRole = lifecycleRoleName
client.SetSelfNode(senderNode, senderRole, senderTier)
diff --git a/banyand/backup/lifecycle/steps_test.go b/banyand/backup/lifecycle/steps_test.go
index 04033ba0a..936cd9f21 100644
--- a/banyand/backup/lifecycle/steps_test.go
+++ b/banyand/backup/lifecycle/steps_test.go
@@ -86,7 +86,7 @@ func TestParseGroup_RejectsMissingIntervals(t *testing.T) {
for _, c := range cases {
t.Run(c.name, func(t *testing.T) {
g := makeGroup(c.mutate)
- _, _, _, _, err := parseGroup(g, map[string]string{"type": "warm"}, nil, nil, nil, nil, nil)
+ _, _, _, _, err := parseGroup(g, "", map[string]string{"type": "warm"}, nil, nil, nil, nil, nil)
require.Error(t, err)
assert.Contains(t, err.Error(), c.errFrag)
})
@@ -101,12 +101,13 @@ func TestParseGroup_RejectsMissingIntervals(t *testing.T) {
// unit test requires an integration harness — that coverage lives in
// test/cases/lifecycle/lifecycle.go.
-// TestResolveSelfIdentity exercises the post-fix pod-name primary lookup
-// against representative registry shapes. The first three cases prove
-// the bug fix: a host-portion match on the registry's GrpcAddress
-// (whether loopback, IP, or FQDN) resolves to the co-located data
-// node's Metadata.Name. The last two cases prove the no-match and
-// empty-input guards.
+// TestResolveSelfIdentity exercises both resolution passes against
+// representative registry shapes. The pod-hostname pass (pass 2, empty
+// selfGRPCAddr) is covered by the first block: a host-portion match on
+// the registry's GrpcAddress (whether loopback, IP, or FQDN) resolves to
+// the co-located data node's Metadata.Name, plus the no-match and
+// empty-input guards. The --grpc-addr pass (pass 1) is covered by the
+// TestResolveSelfIdentity_GRPCAddrDisambiguates regression below.
func TestResolveSelfIdentity(t *testing.T) {
nodes := []*databasev1.Node{
// Production-bug repro: lifecycle's --grpc-addr is
@@ -127,29 +128,29 @@ func TestResolveSelfIdentity(t *testing.T) {
{Metadata: &commonv1.Metadata{Name: "data-cold-0:17912"}, GrpcAddress: "10.116.3.84:17912", Labels: map[string]string{"type": "cold"}},
}
- node, tier, ok := resolveSelfIdentity("data-hot-0", nodes)
+ node, tier, ok := resolveSelfIdentity("", "data-hot-0", nodes)
assert.True(t, ok, "DNS-form GrpcAddress must match by host portion (the production-bug case)")
assert.Equal(t, "data-hot-0:17912", node)
assert.Equal(t, "hot", tier)
- node, tier, ok = resolveSelfIdentity("127.0.0.1", nodes)
+ node, tier, ok = resolveSelfIdentity("", "127.0.0.1", nodes)
assert.True(t, ok, "loopback GrpcAddress matches a loopback selfPodHost")
assert.Equal(t, "data-warm-0:17912", node)
assert.Equal(t, "warm", tier)
// selfPodHost="10.116.3.84" matches the IP-form entry.
- node, tier, ok = resolveSelfIdentity("10.116.3.84", nodes)
+ node, tier, ok = resolveSelfIdentity("", "10.116.3.84", nodes)
assert.True(t, ok, "IP-form GrpcAddress matches an IP selfPodHost")
assert.Equal(t, "data-cold-0:17912", node)
assert.Equal(t, "cold", tier)
// selfPodHost="data-warm-1" is not in the registry: no match.
- _, _, ok = resolveSelfIdentity("data-warm-1", nodes)
+ _, _, ok = resolveSelfIdentity("", "data-warm-1", nodes)
assert.False(t, ok, "selfPodHost not in registry must return ok=false (no wildcard)")
- // Empty selfPodHost: no match.
- _, _, ok = resolveSelfIdentity("", nodes)
- assert.False(t, ok, "empty selfPodHost must return ok=false (no panic)")
+ // Empty selfGRPCAddr and selfPodHost: no match.
+ _, _, ok = resolveSelfIdentity("", "", nodes)
+ assert.False(t, ok, "empty inputs must return ok=false (no panic)")
// grpcAddrEqual: post-fix also accepts host-portion exact match.
assert.True(t, grpcAddrEqual("data-x.headless:17912", "data-x:17912"),
@@ -162,6 +163,35 @@ func TestResolveSelfIdentity(t *testing.T) {
"different hosts with same port must not match")
}
+// TestResolveSelfIdentity_GRPCAddrDisambiguates is the regression for the
+// distributed-lifecycle CI flake: a hot and a warm data node both register
+// on 127.0.0.1, differing only by port. Hostname matching (pass 2) strips
+// the port, so both nodes match a loopback selfPodHost and the resolver
+// returns whichever the registry listing happens to order first — here the
+// warm node is listed first, which produced remote_tier="warm" instead of
+// "hot". The --grpc-addr pass (pass 1) compares the full host:port and must
+// pin the co-located hot node regardless of registry order.
+func TestResolveSelfIdentity_GRPCAddrDisambiguates(t *testing.T) {
+ // Warm node deliberately listed first to defeat any first-match bias.
+ nodes := []*databasev1.Node{
+ {Metadata: &commonv1.Metadata{Name: "127.0.0.1:41777"}, GrpcAddress: "127.0.0.1:41777", Labels: map[string]string{"type": "warm"}},
+ {Metadata: &commonv1.Metadata{Name: "127.0.0.1:17912"}, GrpcAddress: "127.0.0.1:17912", Labels: map[string]string{"type": "hot"}},
+ }
+
+ // Pass 1 (--grpc-addr set): resolves the exact co-located hot node.
+ node, tier, ok := resolveSelfIdentity("127.0.0.1:17912", "127.0.0.1", nodes)
+ assert.True(t, ok)
+ assert.Equal(t, "127.0.0.1:17912", node, "--grpc-addr pass must pin the co-located node by full host:port")
+ assert.Equal(t, "hot", tier, "must resolve the hot tier even though the warm node shares the host and is listed first")
+
+ // Pass 2 (--grpc-addr empty): hostname matching alone cannot tell the
+ // two loopback nodes apart and returns the first-listed warm node.
+ // This documents exactly why pass 1 is required.
+ _, tier, ok = resolveSelfIdentity("", "127.0.0.1", nodes)
+ assert.True(t, ok)
+ assert.Equal(t, "warm", tier, "hostname-only matching is order-dependent — the flake this fix eliminates")
+}
+
// TestSelfPodHostnamePrecedence checks that selfPodHostname() prefers
// POD_NAME (the K8s downward API value) and falls back to os.Hostname()
// when POD_NAME is unset.
diff --git a/banyand/metadata/schema/property/client.go b/banyand/metadata/schema/property/client.go
index 9bc106923..1558d5fcd 100644
--- a/banyand/metadata/schema/property/client.go
+++ b/banyand/metadata/schema/property/client.go
@@ -846,11 +846,79 @@ func (r *SchemaRegistry) ListTrace(ctx context.Context, opt schema.ListOpt) ([]*
return listResources[*databasev1.Trace](ctx, r, schema.KindTrace, opt.Group, true)
}
+// pipelineEnabledWithSampler reports whether an enabled trace pipeline declares
+// at least one sampler plugin. The single-trace-per-group constraint applies
+// only to such groups.
+func pipelineEnabledWithSampler(cfg *commonv1.TracePipelineConfig) bool {
+ if cfg == nil || !cfg.GetEnabled() {
+ return false
+ }
+ for _, plugin := range cfg.GetPlugins() {
+ if plugin.GetSampler() != nil {
+ return true
+ }
+ }
+ return false
+}
+
+// checkTraceAgainstPipeline enforces, at registration time, that a group whose
+// trace pipeline is enabled with a sampler plugin holds at most one trace. It
+// rejects a create/update that would introduce a second, differently-named
+// trace into such a group; the trace being updated is excluded by name.
+func (r *SchemaRegistry) checkTraceAgainstPipeline(ctx context.Context, trace *databasev1.Trace) error {
+ groupName := trace.GetMetadata().GetGroup()
+ g, groupErr := r.GetGroup(ctx, groupName)
+ if groupErr != nil {
+ // The group must exist for the trace create/update to succeed; let the
+ // downstream group validation surface a missing/broken group rather than
+ // masking it. Skip the cross-entity check when the group can't be read.
+ return nil
+ }
+ if !pipelineEnabledWithSampler(g.GetPipeline()) {
+ return nil
+ }
+ traces, listErr := r.ListTrace(ctx, schema.ListOpt{Group: groupName})
+ if listErr != nil {
+ return listErr
+ }
+ selfName := trace.GetMetadata().GetName()
+ for _, existing := range traces {
+ if existing.GetMetadata().GetName() != selfName {
+ return status.Errorf(codes.FailedPrecondition,
+ "group %q has an enabled sampler pipeline and may contain only one trace; trace %q already exists",
+ groupName, existing.GetMetadata().GetName())
+ }
+ }
+ return nil
+}
+
+// checkPipelineAgainstTraces enforces the inverse constraint: a sampler pipeline
+// may be enabled on a group only while the group holds at most one trace.
+func (r *SchemaRegistry) checkPipelineAgainstTraces(ctx context.Context, group *commonv1.Group) error {
+ if !pipelineEnabledWithSampler(group.GetPipeline()) {
+ return nil
+ }
+ groupName := group.GetMetadata().GetName()
+ traces, listErr := r.ListTrace(ctx, schema.ListOpt{Group: groupName})
+ if listErr != nil {
+ return listErr
+ }
+ if len(traces) > 1 {
+ return status.Errorf(codes.FailedPrecondition,
+ "group %q has %d traces; a sampler pipeline may only be enabled on a group with at most one trace",
+ groupName, len(traces))
+ }
+ return nil
+}
+
// CreateTrace creates a trace schema.
func (r *SchemaRegistry) CreateTrace(ctx context.Context, trace *databasev1.Trace) (int64, error) {
if validateErr := validate.Trace(trace); validateErr != nil {
return 0, validateErr
}
+ if pipelineErr := r.checkTraceAgainstPipeline(ctx, trace); pipelineErr != nil {
+ return 0, pipelineErr
+ }
now := time.Now().UnixNano()
trace.Metadata.ModRevision = now
trace.UpdatedAt = timestamppb.Now()
@@ -865,11 +933,14 @@ func (r *SchemaRegistry) UpdateTrace(ctx context.Context, trace *databasev1.Trac
now := time.Now().UnixNano()
trace.Metadata.ModRevision = now
trace.UpdatedAt = timestamppb.Now()
+ // The pipeline check runs as an updateResource validator so it executes only
+ // after the existing trace has been loaded — a missing trace surfaces as
+ // not-found rather than FailedPrecondition.
return updateResource(ctx, r, schema.KindTrace, trace, func(prev *databasev1.Trace) error {
if err := validate.TraceUpdate(prev, trace); err != nil {
return fmt.Errorf("validation failed: %w", err)
}
- return nil
+ return r.checkTraceAgainstPipeline(ctx, trace)
})
}
@@ -893,6 +964,9 @@ func (r *SchemaRegistry) CreateGroup(ctx context.Context, group *commonv1.Group)
if validateErr := validate.Group(group); validateErr != nil {
return 0, validateErr
}
+ if pipelineErr := r.checkPipelineAgainstTraces(ctx, group); pipelineErr != nil {
+ return 0, pipelineErr
+ }
now := time.Now().UnixNano()
group.Metadata.ModRevision = now
group.UpdatedAt = timestamppb.Now()
@@ -907,7 +981,12 @@ func (r *SchemaRegistry) UpdateGroup(ctx context.Context, group *commonv1.Group)
now := time.Now().UnixNano()
group.Metadata.ModRevision = now
group.UpdatedAt = timestamppb.Now()
- return updateResource(ctx, r, schema.KindGroup, group)
+ // The pipeline check runs as an updateResource validator so it executes only
+ // after the existing group has been confirmed — a missing group surfaces as
+ // not-found rather than FailedPrecondition.
+ return updateResource(ctx, r, schema.KindGroup, group, func(_ *commonv1.Group) error {
+ return r.checkPipelineAgainstTraces(ctx, group)
+ })
}
// DeleteGroup deletes a group and all its resources.
diff --git a/banyand/metadata/schema/property/pipeline_single_trace_test.go b/banyand/metadata/schema/property/pipeline_single_trace_test.go
new file mode 100644
index 000000000..91cee5390
--- /dev/null
+++ b/banyand/metadata/schema/property/pipeline_single_trace_test.go
@@ -0,0 +1,203 @@
+// 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 property_test
+
+import (
+ "context"
+ "testing"
+
+ "github.com/stretchr/testify/require"
+ "google.golang.org/grpc/codes"
+ "google.golang.org/grpc/status"
+ "google.golang.org/protobuf/types/known/timestamppb"
+
+ commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
+ databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+ "github.com/apache/skywalking-banyandb/banyand/metadata/schema/property"
+)
+
+// samplerPipeline builds a trace pipeline config with a single sampler plugin.
+// When enabled, it triggers the single-trace-per-group constraint.
+func samplerPipeline(enabled bool) *commonv1.TracePipelineConfig {
+ return &commonv1.TracePipelineConfig{
+ Enabled: enabled,
+ EnabledEvents: []commonv1.PipelineEvent{commonv1.PipelineEvent_PIPELINE_EVENT_MERGE},
+ Plugins: []*commonv1.Plugin{
+ {
+ Name: "latency-status",
+ Kind: &commonv1.Plugin_Sampler{
+ Sampler: &commonv1.SamplerPlugin{
+ Path: "latencystatussampler.so",
+ AbiVersion: 1,
+ },
+ },
+ },
+ },
+ }
+}
+
+// traceTestGroup is the group name used across the single-trace-per-group tests.
+const traceTestGroup = "tg"
+
+// traceGroup returns a CATALOG_TRACE group, optionally carrying a pipeline.
+func traceGroup(pipeline *commonv1.TracePipelineConfig) *commonv1.Group {
+ return &commonv1.Group{
+ Metadata: &commonv1.Metadata{Name: traceTestGroup},
+ Catalog: commonv1.Catalog_CATALOG_TRACE,
+ ResourceOpts: &commonv1.ResourceOpts{
+ ShardNum: 1,
+ SegmentInterval: &commonv1.IntervalRule{Unit: commonv1.IntervalRule_UNIT_DAY, Num: 1},
+ Ttl: &commonv1.IntervalRule{Unit: commonv1.IntervalRule_UNIT_DAY, Num: 3},
+ },
+ Pipeline: pipeline,
+ }
+}
+
+// traceNamed clones the standard test trace with the given name in traceTestGroup.
+func traceNamed(name string) *databasev1.Trace {
+ tr := testTrace(traceTestGroup)
+ tr.Metadata.Name = name
+ tr.UpdatedAt = timestamppb.Now()
+ return tr
+}
+
+func requireFailedPrecondition(t *testing.T, err error) {
+ t.Helper()
+ require.Error(t, err)
+ require.Equalf(t, codes.FailedPrecondition, status.Code(err), "expected FailedPrecondition, got: %v", err)
+}
+
+func newTraceRegistry(t *testing.T) (*property.SchemaRegistry, context.Context) {
+ addr := startTestSchemaServer(t)
+ return newTestRegistry(t, addr), context.Background()
+}
+
+// TestCreateTrace_RejectsSecondTraceWhenSamplerEnabled verifies direction 1:
+// a group whose pipeline is enabled with a sampler plugin may hold only one trace.
+func TestCreateTrace_RejectsSecondTraceWhenSamplerEnabled(t *testing.T) {
+ reg, ctx := newTraceRegistry(t)
+ _, groupErr := reg.CreateGroup(ctx, traceGroup(samplerPipeline(true)))
+ require.NoError(t, groupErr)
+
+ _, firstErr := reg.CreateTrace(ctx, traceNamed("trace-a"))
+ require.NoError(t, firstErr)
+
+ _, secondErr := reg.CreateTrace(ctx, traceNamed("trace-b"))
+ requireFailedPrecondition(t, secondErr)
+}
+
+// TestCreateTrace_AllowedWithoutSamplerPipeline verifies the constraint does not
+// apply when the group has no pipeline.
+func TestCreateTrace_AllowedWithoutSamplerPipeline(t *testing.T) {
+ reg, ctx := newTraceRegistry(t)
+ _, groupErr := reg.CreateGroup(ctx, traceGroup(nil))
+ require.NoError(t, groupErr)
+
+ _, firstErr := reg.CreateTrace(ctx, traceNamed("trace-a"))
+ require.NoError(t, firstErr)
+ _, secondErr := reg.CreateTrace(ctx, traceNamed("trace-b"))
+ require.NoError(t, secondErr)
+}
+
+// TestCreateTrace_AllowedWhenPipelineDisabled verifies the trigger is gated on
+// Enabled: a disabled sampler pipeline does not constrain the trace count.
+func TestCreateTrace_AllowedWhenPipelineDisabled(t *testing.T) {
+ reg, ctx := newTraceRegistry(t)
+ _, groupErr := reg.CreateGroup(ctx, traceGroup(samplerPipeline(false)))
+ require.NoError(t, groupErr)
+
+ _, firstErr := reg.CreateTrace(ctx, traceNamed("trace-a"))
+ require.NoError(t, firstErr)
+ _, secondErr := reg.CreateTrace(ctx, traceNamed("trace-b"))
+ require.NoError(t, secondErr)
+}
+
+// TestUpdateTrace_AllowsSelf verifies updating the single existing trace is not
+// rejected (self is excluded by name).
+func TestUpdateTrace_AllowsSelf(t *testing.T) {
+ reg, ctx := newTraceRegistry(t)
+ _, groupErr := reg.CreateGroup(ctx, traceGroup(samplerPipeline(true)))
+ require.NoError(t, groupErr)
+ _, createErr := reg.CreateTrace(ctx, traceNamed("trace-a"))
+ require.NoError(t, createErr)
+
+ _, updateErr := reg.UpdateTrace(ctx, traceNamed("trace-a"))
+ require.NoError(t, updateErr)
+}
+
+// TestUpdateGroup_RejectsEnablingSamplerWithMultipleTraces verifies direction 2:
+// a sampler pipeline cannot be enabled on a group that already holds >1 trace.
+func TestUpdateGroup_RejectsEnablingSamplerWithMultipleTraces(t *testing.T) {
+ reg, ctx := newTraceRegistry(t)
+ _, groupErr := reg.CreateGroup(ctx, traceGroup(nil))
+ require.NoError(t, groupErr)
+ _, aErr := reg.CreateTrace(ctx, traceNamed("trace-a"))
+ require.NoError(t, aErr)
+ _, bErr := reg.CreateTrace(ctx, traceNamed("trace-b"))
+ require.NoError(t, bErr)
+
+ _, updateErr := reg.UpdateGroup(ctx, traceGroup(samplerPipeline(true)))
+ requireFailedPrecondition(t, updateErr)
+}
+
+// TestUpdateGroup_AllowsEnablingSamplerWithOneTrace verifies enabling a sampler
+// pipeline on a single-trace group succeeds.
+func TestUpdateGroup_AllowsEnablingSamplerWithOneTrace(t *testing.T) {
+ reg, ctx := newTraceRegistry(t)
+ _, groupErr := reg.CreateGroup(ctx, traceGroup(nil))
+ require.NoError(t, groupErr)
+ _, aErr := reg.CreateTrace(ctx, traceNamed("trace-a"))
+ require.NoError(t, aErr)
+
+ _, updateErr := reg.UpdateGroup(ctx, traceGroup(samplerPipeline(true)))
+ require.NoError(t, updateErr)
+}
+
+// TestCreateGroup_WithSamplerPipelineNoTraces verifies a group may be created
+// with an enabled sampler pipeline while it holds no traces yet.
+func TestCreateGroup_WithSamplerPipelineNoTraces(t *testing.T) {
+ reg, ctx := newTraceRegistry(t)
+ _, groupErr := reg.CreateGroup(ctx, traceGroup(samplerPipeline(true)))
+ require.NoError(t, groupErr)
+}
+
+// TestUpdateTrace_MissingTraceReturnsNotFound verifies the pipeline check does
+// not mask a not-found error: updating a non-existent trace in a sampler-enabled
+// group reports not-found, because the check runs after existence is confirmed.
+func TestUpdateTrace_MissingTraceReturnsNotFound(t *testing.T) {
+ reg, ctx := newTraceRegistry(t)
+ _, groupErr := reg.CreateGroup(ctx, traceGroup(samplerPipeline(true)))
+ require.NoError(t, groupErr)
+ _, createErr := reg.CreateTrace(ctx, traceNamed("trace-a"))
+ require.NoError(t, createErr)
+
+ _, updateErr := reg.UpdateTrace(ctx, traceNamed("trace-b"))
+ require.Error(t, updateErr)
+ require.NotEqual(t, codes.FailedPrecondition, status.Code(updateErr))
+ require.Contains(t, updateErr.Error(), "not exist")
+}
+
+// TestUpdateGroup_MissingGroupReturnsNotFound verifies updating a non-existent
+// group with a sampler pipeline reports not-found rather than FailedPrecondition.
+func TestUpdateGroup_MissingGroupReturnsNotFound(t *testing.T) {
+ reg, ctx := newTraceRegistry(t)
+ _, updateErr := reg.UpdateGroup(ctx, traceGroup(samplerPipeline(true)))
+ require.Error(t, updateErr)
+ require.NotEqual(t, codes.FailedPrecondition, status.Code(updateErr))
+ require.Contains(t, updateErr.Error(), "not exist")
+}
diff --git a/docs/operation/plugins-development.md b/docs/operation/plugins-development.md
index c62b58166..b5c813aad 100644
--- a/docs/operation/plugins-development.md
+++ b/docs/operation/plugins-development.md
@@ -103,7 +103,16 @@ plugin-capable host image (`apache/skywalking-banyandb:<tag>-plugins`) at
-trace-pipeline-trusted-plugin-dir=/plugins
```
-Reference the plugin by filename in the group's `TracePipelineConfig`:
+Mount the carrier at `/plugins` via an **OCI image volume** (K8s ≥1.31) or an
+**initContainer + emptyDir** (portable) — full instructions, both mechanisms,
+third-party plugins, and Kubernetes manifests are in [`plugins.md`](plugins.md)
+and [`examples/kubernetes/plugins/`](../../examples/kubernetes/plugins).
+
+**Register the pipeline — this is the step that actually activates the plugin.**
+Mounting only delivers the `.so`. The data node *loads* it when a group's
+`TracePipelineConfig` (referencing the plugin by its trusted-dir-relative
+filename) is registered: the node watches the schema registry and, on the group
+add/update, reconciles the pipeline and calls `plugin.Open` on the mounted `.so`.
```json
{ "enabled": true, "enabledEvents": ["PIPELINE_EVENT_MERGE"],
@@ -112,10 +121,21 @@ Reference the plugin by filename in the group's `TracePipelineConfig`:
"config": { "thresholdMs": 500, "successValue": "success" } } } ] }
```
-Mount the carrier at `/plugins` via an **OCI image volume** (K8s ≥1.31) or an
-**initContainer + emptyDir** (portable) — full instructions, both mechanisms,
-third-party plugins, and Kubernetes manifests are in [`plugins.md`](plugins.md)
-and [`examples/kubernetes/plugins/`](../../examples/kubernetes/plugins).
+Attach it with any standard schema API — all write through the one schema
+registry (`GroupRegistryService`) and preserve the nested `pipeline` field:
+
+- **bydbctl:** add a `pipeline:` block to the group's YAML, then
+ `bydbctl group update -f group.yaml` (or `group create`).
+- **REST:** `PUT /api/v1/group/schema/{group}` with `{"group": {…, "pipeline": {…}}}`.
+- **gRPC:** `GroupRegistryService.Update` — `Get` the group, set `.pipeline`, then
+ `Update`. Reference: [`test/cases/tracepipeline/ops.go`](../../test/cases/tracepipeline/ops.go)
+ (`applyPipelineConfig`).
+
+**Verify activation** on the data node's metrics port (`2121`):
+`banyandb_trace_pipeline_sampler_active_count{group=…}` goes `>0` once the plugin
+loads. A bad/mismatched `.so` fails **open and loud** — an ERROR log plus
+`banyandb_trace_pipeline_sampler_load_failed{…}` — and the group simply keeps its
+previous sampler set; the node stays up.
## Step 5 — Validate the deployment shape in Kubernetes
diff --git a/docs/operation/plugins.md b/docs/operation/plugins.md
index 8155f9720..73775d1d1 100644
--- a/docs/operation/plugins.md
+++ b/docs/operation/plugins.md
@@ -94,6 +94,16 @@ group's pipeline config (`TracePipelineConfig` on the `Group`,
}
```
+Attaching this `pipeline` to the group is what **activates** the plugin: the data
+node watches the schema registry and, on the group add/update, reconciles the
+pipeline and calls `plugin.Open` on the mounted `.so`. Register it through any
+standard schema API — all go through `GroupRegistryService` and preserve the
+nested `pipeline` field: `bydbctl group update -f group.yaml` (with a `pipeline:`
+block), `PUT /api/v1/group/schema/{group}`, or the gRPC `GroupRegistryService.Update`
+(see `test/cases/tracepipeline/ops.go`). Verify with
+`banyandb_trace_pipeline_sampler_active_count{group=…}` `>0` (metrics port `2121`);
+a bad `.so` fails open + loud (`…_sampler_load_failed`), leaving the node healthy.
+
## Delivering the carrier (first-party) and third-party plugins
The loader supports exactly **one** trusted dir
diff --git a/test/plugin-sidecar/register/main.go b/test/plugin-sidecar/register/main.go
index 34b991dc7..ccc374908 100644
--- a/test/plugin-sidecar/register/main.go
+++ b/test/plugin-sidecar/register/main.go
@@ -20,12 +20,15 @@
// reconciles it and calls plugin.Open on the carrier-mounted .so. It is used
// by the plugin-sidecar kind ship gate to drive assertion (c).
//
-// It writes through the PROPERTY schema registry client (the same path
-// test/cases/tracepipeline's PreloadSchemaViaProperty uses), NOT the
-// GroupRegistryService HTTP/gRPC endpoint: only the property schema store is
-// watched by the data node's trace schemaRepo, so only writes through it fire
-// the KindGroup reconcile that loads the plugin. The property store also
-// round-trips the nested pipeline field (the HTTP gateway does not).
+// It writes directly to the property-based schema registry (the single schema
+// store — same path test/cases/tracepipeline's PreloadSchemaViaProperty uses)
+// via the schema server the standalone pod exposes on :17916. That is a
+// TEST-HARNESS convenience: this helper only has that schema-server address
+// wired, not a liaison registry endpoint. The supported OPERATOR path is the
+// public GroupRegistryService (gRPC / REST / bydbctl), which writes through the
+// same store and preserves the nested pipeline field — see
+// docs/operation/plugins-development.md. Either way, the group add/update fires
+// the data node's KindGroup reconcile, which calls plugin.Open on the .so.
package main
import (