You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by li...@apache.org on 2026/07/14 09:19:58 UTC
(skywalking-banyandb) branch fix/schema-stage-opts-propagation created (now e95a49d01)
This is an automated email from the ASF dual-hosted git repository.
mrproliu pushed a change to branch fix/schema-stage-opts-propagation
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
at e95a49d01 Resolve lifecycle stage opts on group update (fix short warm/cold segments) + schema delete-path fixes
This branch includes the following new commits:
new e95a49d01 Resolve lifecycle stage opts on group update (fix short warm/cold segments) + schema delete-path fixes
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(skywalking-banyandb) 01/01: Resolve lifecycle stage opts on group update (fix short warm/cold segments) + schema delete-path fixes
Posted by li...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mrproliu pushed a commit to branch fix/schema-stage-opts-propagation
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit e95a49d01e6c825d4e21a81f4c95c2fa37162239
Author: mrproliu <74...@qq.com>
AuthorDate: Tue Jul 14 17:19:48 2026 +0800
Resolve lifecycle stage opts on group update (fix short warm/cold segments) + schema delete-path fixes
---
CHANGES.md | 5 +
banyand/internal/storage/segment.go | 7 ++
banyand/measure/metadata.go | 69 +++++---------
banyand/measure/metadata_resolve_test.go | 85 +++++++++++++++++
banyand/measure/topn.go | 47 ++++++++++
banyand/measure/topn_unregister_test.go | 105 +++++++++++++++++++++
banyand/queue/pub/stage.go | 156 +++++++++++++++++++++++++++++++
banyand/queue/pub/stage_test.go | 112 ++++++++++++++++++++++
banyand/stream/metadata.go | 65 +++++--------
banyand/trace/metadata.go | 68 +++++---------
banyand/trace/trace.go | 6 +-
pkg/schema/cache.go | 67 ++++++++++---
pkg/schema/cache_test.go | 99 ++++++++++++++++++++
pkg/schema/schema.go | 6 ++
14 files changed, 746 insertions(+), 151 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index 0de319786..95dbdbf6d 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -133,6 +133,11 @@ Release Notes.
- Fix block metadata reset before unmarshal for stream and trace.
- Lifecycle only handles stream/measure/trace snapshots, skipping other catalogs.
- Fix liaison goroutine leak from uncanceled gRPC SyncPart stream context in write-queue part sync.
+- Fix group update overwriting a warm/cold node's stage-resolved segment interval, TTL and shard count with the group default.
+- Re-index bound resources when an index rule or binding is deleted, clearing stale index config.
+- Deleting one TopN aggregation no longer tears down sibling aggregations on the same source measure.
+- Purge a deleted group's resource, index-rule and binding cache entries to avoid dangling references.
+- Clear a trace subject's index when its last index rule or binding is removed.
### Document
diff --git a/banyand/internal/storage/segment.go b/banyand/internal/storage/segment.go
index 965a6d470..00dd5b67d 100644
--- a/banyand/internal/storage/segment.go
+++ b/banyand/internal/storage/segment.go
@@ -597,6 +597,13 @@ func (sc *segmentController[T, O]) updateOptions(resourceOpts *commonv1.Resource
sc.l.Panic().Msg("segment interval unit cannot be changed")
return
}
+ // A live segment-interval change is legitimate (e.g. a stage was re-timed) but
+ // rare; surface it so an unexpected change -- such as a stage node being fed the
+ // group default -- is visible instead of silent.
+ if sc.opts.SegmentInterval.Num != si.Num {
+ sc.l.Info().Int("from", sc.opts.SegmentInterval.Num).Int("to", si.Num).
+ Msg("segment interval number changed on options update; existing segments keep their span, new segments use the new interval")
+ }
sc.opts.SegmentInterval = si
sc.opts.TTL = MustToIntervalRule(resourceOpts.Ttl)
sc.opts.ShardNum = resourceOpts.ShardNum
diff --git a/banyand/measure/metadata.go b/banyand/measure/metadata.go
index 550c697c6..a21fad2ba 100644
--- a/banyand/measure/metadata.go
+++ b/banyand/measure/metadata.go
@@ -355,7 +355,7 @@ func (sr *schemaRepo) OnDelete(metadata schema.Metadata) {
case schema.KindTopNAggregation:
if sr.pipeline != nil {
topNAggregation := metadata.Spec.(*databasev1.TopNAggregation)
- sr.stopSteamingManager(topNAggregation.SourceMeasure)
+ sr.removeTopNAggregation(topNAggregation.SourceMeasure, topNAggregation)
}
default:
}
@@ -729,60 +729,25 @@ func (s *supplier) OpenDB(groupSchema *commonv1.Group) (resourceSchema.DB, error
if ro == nil {
return nil, fmt.Errorf("no resource opts in group %s", name)
}
- shardNum := ro.ShardNum
- ttl := ro.Ttl
- segInterval := ro.SegmentInterval
- // Non-zero default so the idle-segment reclaimer ticker actually starts
- // (storage/rotation.go gates it on >=1s). Staged Close paths override below.
- segmentIdleTimeout := time.Hour
- disableRetention := false
- disableRotation := false
- if len(ro.Stages) > 0 && len(s.nodeLabels) > 0 {
- var ttlNum uint32
- foundMatched := false
- for i, st := range ro.Stages {
- if st.Ttl.Unit != ro.Ttl.Unit {
- return nil, fmt.Errorf("ttl unit %s is not consistent with stage %s", ro.Ttl.Unit, st.Ttl.Unit)
- }
- selector, err := pub.ParseLabelSelector(st.NodeSelector)
- if err != nil {
- return nil, errors.WithMessagef(err, "failed to parse node selector %s", st.NodeSelector)
- }
- ttlNum += st.Ttl.Num
- if !selector.Matches(s.nodeLabels) {
- continue
- }
- foundMatched = true
- ttl.Num += ttlNum
- shardNum = st.ShardNum
- segInterval = st.SegmentInterval
- if st.Close {
- segmentIdleTimeout = 5 * time.Minute
- }
- disableRetention = i+1 < len(ro.Stages)
- disableRotation = true
- break
- }
- if !foundMatched {
- disableRetention = true
- disableRotation = true
- }
+ res, err := pub.ResolveStage(s.l, name, ro, s.nodeLabels)
+ if err != nil {
+ return nil, err
}
group := groupSchema.Metadata.Name
opts := storage.TSDBOpts[*tsTable, option]{
- ShardNum: shardNum,
+ ShardNum: res.ResourceOpts.ShardNum,
Location: path.Join(s.path, group),
TSTableCreator: newTSTable,
TableMetrics: metrics,
- SegmentInterval: storage.MustToIntervalRule(segInterval),
- TTL: storage.MustToIntervalRule(ttl),
+ SegmentInterval: storage.MustToIntervalRule(res.ResourceOpts.SegmentInterval),
+ TTL: storage.MustToIntervalRule(res.ResourceOpts.Ttl),
Option: s.option,
SeriesIndexFlushTimeoutSeconds: s.option.flushTimeout.Nanoseconds() / int64(time.Second),
SeriesIndexCacheMaxBytes: int(s.option.seriesCacheMaxSize),
StorageMetricsFactory: factory,
- SegmentIdleTimeout: segmentIdleTimeout,
- DisableRetention: disableRetention,
- DisableRotation: disableRotation,
+ SegmentIdleTimeout: res.SegmentIdleTimeout,
+ DisableRetention: res.DisableRetention,
+ DisableRotation: res.DisableRotation,
MemoryLimit: s.pm.GetLimit(),
}
return storage.OpenTSDB(
@@ -792,6 +757,14 @@ func (s *supplier) OpenDB(groupSchema *commonv1.Group) (resourceSchema.DB, error
opts, s.c, group)
}
+// ResolveResourceOpts returns the stage-resolved ResourceOpts for this node so that a
+// group UpdateOptions applies the matched stage's interval/ttl/shardNum instead of the
+// group default. On resolution error it keeps the group default rather than failing the
+// update.
+func (s *supplier) ResolveResourceOpts(groupSchema *commonv1.Group) *commonv1.ResourceOpts {
+ return pub.ResolveResourceOptsForUpdate(s.l, groupSchema, s.nodeLabels)
+}
+
type queueSupplier struct {
metadata metadata.Repo
omr observability.MetricsRegistry
@@ -834,6 +807,12 @@ func (s *queueSupplier) ResourceSchema(md *commonv1.Metadata) (resourceSchema.Re
return s.metadata.MeasureRegistry().GetMeasure(ctx, md)
}
+// ResolveResourceOpts returns the group opts unchanged: the liaison write queue has no
+// lifecycle stages to resolve.
+func (s *queueSupplier) ResolveResourceOpts(groupSchema *commonv1.Group) *commonv1.ResourceOpts {
+ return groupSchema.GetResourceOpts()
+}
+
func (s *queueSupplier) OpenDB(groupSchema *commonv1.Group) (resourceSchema.DB, error) {
name := groupSchema.Metadata.Name
p := common.Position{
diff --git a/banyand/measure/metadata_resolve_test.go b/banyand/measure/metadata_resolve_test.go
new file mode 100644
index 000000000..3e7713fbb
--- /dev/null
+++ b/banyand/measure/metadata_resolve_test.go
@@ -0,0 +1,85 @@
+// 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 measure
+
+import (
+ "testing"
+
+ "github.com/stretchr/testify/require"
+
+ commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+)
+
+func resolveDay(n uint32) *commonv1.IntervalRule {
+ return &commonv1.IntervalRule{Unit: commonv1.IntervalRule_UNIT_DAY, Num: n}
+}
+
+func metricsHourGroup() *commonv1.Group {
+ return &commonv1.Group{
+ Metadata: &commonv1.Metadata{Name: "sw_metricsHour"},
+ ResourceOpts: &commonv1.ResourceOpts{
+ ShardNum: 1,
+ SegmentInterval: resolveDay(5),
+ Ttl: resolveDay(1),
+ Stages: []*commonv1.LifecycleStage{
+ {Name: "warm", ShardNum: 2, SegmentInterval: resolveDay(10), Ttl: resolveDay(7), NodeSelector: "type=warm"},
+ {Name: "cold", ShardNum: 3, SegmentInterval: resolveDay(20), Ttl: resolveDay(30), NodeSelector: "type=cold", Close: true},
+ },
+ },
+ }
+}
+
+// TestSupplierResolveResourceOpts_ColdNode exercises the real measure data supplier's
+// ResolveResourceOpts -- the value storeGroup feeds into DB.UpdateOptions -- and asserts
+// a type=cold node resolves the cold stage's interval/ttl/shardNum, not the group default.
+func TestSupplierResolveResourceOpts_ColdNode(t *testing.T) {
+ s := &supplier{l: logger.GetLogger("test"), nodeLabels: map[string]string{"type": "cold"}}
+ g := metricsHourGroup()
+
+ got := s.ResolveResourceOpts(g)
+ require.Equal(t, uint32(20), got.GetSegmentInterval().GetNum(), "cold segment interval, not group default 5")
+ require.Equal(t, uint32(3), got.GetShardNum(), "cold shard num")
+ require.Equal(t, uint32(38), got.GetTtl().GetNum(), "cumulative ttl 1+7+30")
+
+ // The shared group schema must not be mutated by resolution.
+ require.Equal(t, uint32(5), g.ResourceOpts.GetSegmentInterval().GetNum())
+ require.Equal(t, uint32(1), g.ResourceOpts.GetTtl().GetNum())
+ require.Equal(t, uint32(1), g.ResourceOpts.GetShardNum())
+}
+
+// TestSupplierResolveResourceOpts_UnlabeledNodeKeepsGroupDefault asserts a node with no
+// labels (e.g. the hot/initial tier) keeps the group default.
+func TestSupplierResolveResourceOpts_UnlabeledNodeKeepsGroupDefault(t *testing.T) {
+ s := &supplier{l: logger.GetLogger("test")}
+ got := s.ResolveResourceOpts(metricsHourGroup())
+ require.Equal(t, uint32(5), got.GetSegmentInterval().GetNum())
+ require.Equal(t, uint32(1), got.GetTtl().GetNum())
+}
+
+// TestSupplierResolveResourceOpts_ResolveErrorKeepsGroupDefault asserts that when stage
+// resolution errors (here a stage ttl unit inconsistent with the group), the update path
+// keeps the group default instead of failing.
+func TestSupplierResolveResourceOpts_ResolveErrorKeepsGroupDefault(t *testing.T) {
+ s := &supplier{l: logger.GetLogger("test"), nodeLabels: map[string]string{"type": "cold"}}
+ g := metricsHourGroup()
+ g.ResourceOpts.Stages[1].Ttl = &commonv1.IntervalRule{Unit: commonv1.IntervalRule_UNIT_HOUR, Num: 30}
+ got := s.ResolveResourceOpts(g)
+ require.Equal(t, uint32(5), got.GetSegmentInterval().GetNum(), "keeps group default on resolve error")
+ require.Equal(t, uint32(1), got.GetTtl().GetNum())
+}
diff --git a/banyand/measure/topn.go b/banyand/measure/topn.go
index 04a562253..bda30bf52 100644
--- a/banyand/measure/topn.go
+++ b/banyand/measure/topn.go
@@ -149,6 +149,28 @@ func (sr *schemaRepo) stopSteamingManager(sourceMeasure *commonv1.Metadata) {
}
}
+// removeTopNAggregation tears down a single top-n aggregation, keeping the other
+// aggregations on the same source measure alive. Only when the source measure has no
+// aggregation left is the whole manager closed and dropped. Deleting one aggregation
+// previously closed every sibling via stopSteamingManager.
+func (sr *schemaRepo) removeTopNAggregation(sourceMeasure *commonv1.Metadata, topNSchema *databasev1.TopNAggregation) {
+ if sourceMeasure == nil {
+ return
+ }
+ key := getKey(sourceMeasure)
+ v, ok := sr.topNProcessorMap.Load(key)
+ if !ok {
+ return
+ }
+ manager := v.(*topNProcessorManager)
+ if manager.unregister(topNSchema) {
+ if err := manager.Close(); err != nil {
+ sr.l.Err(err).Msg("fail to close the emptied top-n manager")
+ }
+ sr.topNProcessorMap.Delete(key)
+ }
+}
+
type dataPointWithEntityValues struct {
tagSpec logical.TagSpecRegistry
*measurev1.DataPointValue
@@ -745,6 +767,31 @@ func (manager *topNProcessorManager) start(ctx context.Context, topNSchema *data
return nil
}
+// unregister removes a single top-n aggregation's registered task and processors while
+// leaving sibling aggregations on the same source measure running. It returns true when
+// no registered task remains, so the caller can drop the whole manager.
+func (manager *topNProcessorManager) unregister(topNSchema *databasev1.TopNAggregation) (empty bool) {
+ manager.Lock()
+ if manager.closed {
+ manager.Unlock()
+ return true
+ }
+ manager.registeredTasks = slices.DeleteFunc(manager.registeredTasks, func(task *databasev1.TopNAggregation) bool {
+ return task.GetMetadata().GetName() == topNSchema.GetMetadata().GetName()
+ })
+ removed := manager.removeProcessors(topNSchema)
+ empty = len(manager.registeredTasks) == 0
+ manager.Unlock()
+ // Close outside the lock: processor.Close can block up to ~5s and would otherwise
+ // stall onMeasureWrite readers; the processors are already detached from the manager.
+ for _, processor := range removed {
+ if err := processor.Close(); err != nil {
+ manager.l.Err(err).Msg("fail to close the removed top-n processor")
+ }
+ }
+ return empty
+}
+
func (manager *topNProcessorManager) removeProcessors(topNSchema *databasev1.TopNAggregation) []topNProcessor {
var processors []topNProcessor
var newList []topNProcessor
diff --git a/banyand/measure/topn_unregister_test.go b/banyand/measure/topn_unregister_test.go
new file mode 100644
index 000000000..8cbdc2a05
--- /dev/null
+++ b/banyand/measure/topn_unregister_test.go
@@ -0,0 +1,105 @@
+// 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 measure
+
+import (
+ "context"
+ "testing"
+
+ "github.com/stretchr/testify/require"
+
+ commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
+ databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+ "github.com/apache/skywalking-banyandb/pkg/flow"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+)
+
+type fakeTopNProcessor struct {
+ schema *databasev1.TopNAggregation
+ closed *bool
+}
+
+func (f fakeTopNProcessor) In() chan<- flow.StreamRecord { return nil }
+func (f fakeTopNProcessor) Setup(context.Context) error { return nil }
+func (f fakeTopNProcessor) Teardown(context.Context) error { return nil }
+func (f fakeTopNProcessor) Close() error { *f.closed = true; return nil }
+func (f fakeTopNProcessor) Src() chan interface{} { return nil }
+func (f fakeTopNProcessor) TopNSchema() *databasev1.TopNAggregation { return f.schema }
+
+func agg(name string) *databasev1.TopNAggregation {
+ return &databasev1.TopNAggregation{Metadata: &commonv1.Metadata{Group: "g", Name: name}}
+}
+
+// TestTopNManager_UnregisterKeepsSiblings asserts that removing one top-n aggregation
+// closes only its own processor and leaves the sibling aggregation on the same source
+// measure running -- deleting one previously tore down every sibling.
+func TestTopNManager_UnregisterKeepsSiblings(t *testing.T) {
+ aggA, aggB := agg("a"), agg("b")
+ var closedA, closedB bool
+ mgr := &topNProcessorManager{
+ registeredTasks: []*databasev1.TopNAggregation{aggA, aggB},
+ processorList: []topNProcessor{
+ fakeTopNProcessor{schema: aggA, closed: &closedA},
+ fakeTopNProcessor{schema: aggB, closed: &closedB},
+ },
+ }
+
+ empty := mgr.unregister(aggA)
+ require.False(t, empty, "sibling b is still registered")
+ require.True(t, closedA, "a's processor must be closed")
+ require.False(t, closedB, "b's processor must stay alive")
+ require.Len(t, mgr.registeredTasks, 1)
+ require.Equal(t, "b", mgr.registeredTasks[0].GetMetadata().GetName())
+ require.Len(t, mgr.processorList, 1)
+
+ empty = mgr.unregister(aggB)
+ require.True(t, empty, "no aggregation left after removing b")
+ require.True(t, closedB, "b's processor closed on the last removal")
+ require.Empty(t, mgr.processorList)
+}
+
+// TestRemoveTopNAggregation_DropsManagerOnlyWhenEmpty asserts the schemaRepo-level
+// wrapper keeps the manager (and its siblings) alive until the last aggregation on a
+// source measure is removed, at which point the manager is closed and dropped from the map.
+func TestRemoveTopNAggregation_DropsManagerOnlyWhenEmpty(t *testing.T) {
+ l := logger.GetLogger("test")
+ sr := &schemaRepo{l: l}
+ source := &commonv1.Metadata{Group: "g", Name: "m"}
+ aggA, aggB := agg("a"), agg("b")
+ var closedA, closedB bool
+ mgr := &topNProcessorManager{
+ l: l,
+ registeredTasks: []*databasev1.TopNAggregation{aggA, aggB},
+ processorList: []topNProcessor{
+ fakeTopNProcessor{schema: aggA, closed: &closedA},
+ fakeTopNProcessor{schema: aggB, closed: &closedB},
+ },
+ }
+ sr.topNProcessorMap.Store(getKey(source), mgr)
+
+ sr.removeTopNAggregation(source, aggA)
+ _, ok := sr.topNProcessorMap.Load(getKey(source))
+ require.True(t, ok, "manager kept while sibling b remains")
+ require.True(t, closedA)
+ require.False(t, closedB)
+
+ sr.removeTopNAggregation(source, aggB)
+ _, ok = sr.topNProcessorMap.Load(getKey(source))
+ require.False(t, ok, "manager dropped from the map after the last aggregation is removed")
+ require.True(t, closedB)
+}
diff --git a/banyand/queue/pub/stage.go b/banyand/queue/pub/stage.go
new file mode 100644
index 000000000..c6f9440e1
--- /dev/null
+++ b/banyand/queue/pub/stage.go
@@ -0,0 +1,156 @@
+// 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 pub
+
+import (
+ "fmt"
+ "time"
+
+ "google.golang.org/protobuf/proto"
+
+ commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+)
+
+// ResolveStageResourceOpts resolves the effective ResourceOpts for a data node whose
+// lifecycle stage is selected by nodeLabels. It returns a deep clone of ro with
+// SegmentInterval, Ttl (cumulative through the matched stage) and ShardNum overridden
+// by the matched stage. matchedStage/matchedIdx identify the matched stage and are
+// nil/-1 when no stage applies (the group has no stages, the node has no labels, or no
+// stage selector matched the node's labels). The returned opts is always a fresh clone,
+// so callers never mutate the shared group schema.
+//
+// This is the single source of truth for the per-stage interval/ttl/shard resolution:
+// both the initial OpenDB and any later UpdateOptions must go through it, otherwise a
+// group update silently overwrites a stage node's interval/ttl with the group default.
+func ResolveStageResourceOpts(ro *commonv1.ResourceOpts, nodeLabels map[string]string) (
+ resolved *commonv1.ResourceOpts, matchedStage *commonv1.LifecycleStage, matchedIdx int, err error,
+) {
+ if ro == nil {
+ return nil, nil, -1, nil
+ }
+ resolved = proto.Clone(ro).(*commonv1.ResourceOpts)
+ if len(ro.GetStages()) == 0 || len(nodeLabels) == 0 {
+ return resolved, nil, -1, nil
+ }
+ // Reject malformed schemas up front so a missing interval/ttl returns an error here
+ // rather than panicking a data node later in storage.MustToIntervalRule.
+ if ro.GetTtl() == nil {
+ return nil, nil, -1, fmt.Errorf("group ttl must be set when lifecycle stages are configured")
+ }
+ var ttlNum uint32
+ for i, st := range ro.GetStages() {
+ if st.GetTtl() == nil || st.GetSegmentInterval() == nil {
+ return nil, nil, -1, fmt.Errorf("lifecycle stage %q must set both ttl and segmentInterval", st.GetName())
+ }
+ if st.GetTtl().GetUnit() != ro.GetTtl().GetUnit() {
+ return nil, nil, -1, fmt.Errorf("ttl unit %s is not consistent with stage %s ttl unit %s",
+ ro.GetTtl().GetUnit(), st.GetName(), st.GetTtl().GetUnit())
+ }
+ selector, parseErr := ParseLabelSelector(st.GetNodeSelector())
+ if parseErr != nil {
+ return nil, nil, -1, fmt.Errorf("failed to parse node selector %q: %w", st.GetNodeSelector(), parseErr)
+ }
+ ttlNum += st.GetTtl().GetNum()
+ if !selector.Matches(nodeLabels) {
+ continue
+ }
+ resolved.SegmentInterval = proto.Clone(st.GetSegmentInterval()).(*commonv1.IntervalRule)
+ resolved.ShardNum = st.GetShardNum()
+ // resolved.Ttl is the clone of the (validated non-nil) group ttl, so it keeps the
+ // group's unit; only the cumulative number changes: the group base plus every
+ // stage ttl up to and including the matched stage.
+ resolved.Ttl.Num = ro.GetTtl().GetNum() + ttlNum
+ return resolved, st, i, nil
+ }
+ return resolved, nil, -1, nil
+}
+
+// StageResolution is the fully-resolved storage configuration for a data node's tsdb,
+// derived from the group ResourceOpts and the node's matched lifecycle stage. It carries
+// everything OpenDB needs so the caller does no further stage logic.
+//
+// Only ResourceOpts is re-applied on a group UpdateOptions; the rotation/retention fields
+// are wired once at open time, so adding/removing/retiming a stage needs a node restart.
+type StageResolution struct {
+ // ResourceOpts holds the effective SegmentInterval/Ttl/ShardNum -- the matched stage's
+ // values (cumulative ttl) or the group default. It is always a fresh clone.
+ ResourceOpts *commonv1.ResourceOpts
+ SegmentIdleTimeout time.Duration
+ // Matched reports whether the node matched a lifecycle stage (a warm/cold tier); the
+ // hot/initial tier and unlabeled nodes are not matched.
+ Matched bool
+ DisableRetention bool
+ DisableRotation bool
+}
+
+// ResolveStage resolves the complete tsdb configuration for this node from the group opts
+// and the node's lifecycle stage. It warns when a labeled node whose group has stages
+// matches none of them and silently falls back to the group default -- the condition that
+// previously produced short segments. This is the single entry point OpenDB uses.
+func ResolveStage(l *logger.Logger, groupName string, ro *commonv1.ResourceOpts, nodeLabels map[string]string) (StageResolution, error) {
+ resolved, matchedStage, matchedIdx, err := ResolveStageResourceOpts(ro, nodeLabels)
+ if err != nil {
+ return StageResolution{}, err
+ }
+ staged := len(ro.GetStages()) > 0 && len(nodeLabels) > 0
+ if matchedStage == nil && staged {
+ l.Warn().Str("group", groupName).Interface("nodeLabels", nodeLabels).
+ Msg("no lifecycle stage matched this data node's labels; using the group-default segment interval/ttl/shardNum")
+ }
+ // A non-zero idle timeout so the reclaimer ticker starts (storage/rotation.go gates it
+ // on >=1s); a Close stage shortens it. Retention is disabled on non-terminal tiers
+ // (data migrates onward) and on staged-but-unmatched nodes; rotation is disabled on any
+ // staged node.
+ res := StageResolution{
+ ResourceOpts: resolved,
+ Matched: matchedStage != nil,
+ SegmentIdleTimeout: time.Hour,
+ }
+ switch {
+ case matchedStage != nil:
+ if matchedStage.GetClose() {
+ res.SegmentIdleTimeout = 5 * time.Minute
+ }
+ res.DisableRetention = matchedIdx+1 < len(ro.GetStages())
+ res.DisableRotation = true
+ case staged:
+ res.DisableRetention = true
+ res.DisableRotation = true
+ }
+ return res, nil
+}
+
+// ResolveResourceOptsForUpdate resolves the ResourceOpts a data node applies when a group
+// is updated: the matched stage's interval/ttl/shardNum, or the group default when no
+// stage applies. On a resolution error it keeps the group default rather than failing the
+// update, and returns nil when the group carries no ResourceOpts. This is the shared body
+// of every data supplier's ResolveResourceOpts.
+func ResolveResourceOptsForUpdate(l *logger.Logger, groupSchema *commonv1.Group, nodeLabels map[string]string) *commonv1.ResourceOpts {
+ ro := groupSchema.GetResourceOpts()
+ if ro == nil {
+ return nil
+ }
+ res, err := ResolveStage(l, groupSchema.GetMetadata().GetName(), ro, nodeLabels)
+ if err != nil {
+ l.Warn().Err(err).Str("group", groupSchema.GetMetadata().GetName()).
+ Msg("failed to resolve stage resource opts on update; keeping group-default opts")
+ return ro
+ }
+ return res.ResourceOpts
+}
diff --git a/banyand/queue/pub/stage_test.go b/banyand/queue/pub/stage_test.go
new file mode 100644
index 000000000..de5068b2c
--- /dev/null
+++ b/banyand/queue/pub/stage_test.go
@@ -0,0 +1,112 @@
+// 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 pub
+
+import (
+ "testing"
+
+ "github.com/stretchr/testify/require"
+
+ commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
+)
+
+func day(n uint32) *commonv1.IntervalRule {
+ return &commonv1.IntervalRule{Unit: commonv1.IntervalRule_UNIT_DAY, Num: n}
+}
+
+// stagedOpts mirrors the sw_metricsHour shape: group default 5d/ttl 1d, warm 10d/ttl 7d,
+// cold 20d/ttl 30d.
+func stagedOpts() *commonv1.ResourceOpts {
+ return &commonv1.ResourceOpts{
+ ShardNum: 1,
+ SegmentInterval: day(5),
+ Ttl: day(1),
+ Stages: []*commonv1.LifecycleStage{
+ {Name: "warm", ShardNum: 2, SegmentInterval: day(10), Ttl: day(7), NodeSelector: "type=warm"},
+ {Name: "cold", ShardNum: 3, SegmentInterval: day(20), Ttl: day(30), NodeSelector: "type=cold", Close: true},
+ },
+ }
+}
+
+func TestResolveStageResourceOpts_ColdNodeGetsColdStage(t *testing.T) {
+ resolved, matched, idx, err := ResolveStageResourceOpts(stagedOpts(), map[string]string{"type": "cold"})
+ require.NoError(t, err)
+ require.NotNil(t, matched)
+ require.Equal(t, "cold", matched.GetName())
+ require.Equal(t, 1, idx)
+ require.Equal(t, uint32(20), resolved.GetSegmentInterval().GetNum(), "cold segment interval")
+ require.Equal(t, uint32(3), resolved.GetShardNum(), "cold shard num")
+ // Cumulative ttl through cold: group 1 + warm 7 + cold 30 = 38.
+ require.Equal(t, uint32(38), resolved.GetTtl().GetNum(), "cumulative cold ttl")
+}
+
+func TestResolveStageResourceOpts_WarmNodeGetsWarmStage(t *testing.T) {
+ resolved, matched, idx, err := ResolveStageResourceOpts(stagedOpts(), map[string]string{"type": "warm"})
+ require.NoError(t, err)
+ require.NotNil(t, matched)
+ require.Equal(t, "warm", matched.GetName())
+ require.Equal(t, 0, idx)
+ require.Equal(t, uint32(10), resolved.GetSegmentInterval().GetNum())
+ require.Equal(t, uint32(2), resolved.GetShardNum())
+ require.Equal(t, uint32(8), resolved.GetTtl().GetNum(), "group 1 + warm 7")
+}
+
+func TestResolveStageResourceOpts_NoMatchFallsBackToGroupDefault(t *testing.T) {
+ // A node label matching no stage selector must fall back to the group default, and
+ // signal matched=nil so the caller can warn.
+ resolved, matched, idx, err := ResolveStageResourceOpts(stagedOpts(), map[string]string{"type": "hot"})
+ require.NoError(t, err)
+ require.Nil(t, matched)
+ require.Equal(t, -1, idx)
+ require.Equal(t, uint32(5), resolved.GetSegmentInterval().GetNum())
+ require.Equal(t, uint32(1), resolved.GetShardNum())
+ require.Equal(t, uint32(1), resolved.GetTtl().GetNum())
+}
+
+func TestResolveStageResourceOpts_NoLabelsOrNoStages(t *testing.T) {
+ resolved, matched, _, err := ResolveStageResourceOpts(stagedOpts(), nil)
+ require.NoError(t, err)
+ require.Nil(t, matched)
+ require.Equal(t, uint32(5), resolved.GetSegmentInterval().GetNum())
+
+ noStages := &commonv1.ResourceOpts{ShardNum: 1, SegmentInterval: day(5), Ttl: day(1)}
+ resolved, matched, _, err = ResolveStageResourceOpts(noStages, map[string]string{"type": "cold"})
+ require.NoError(t, err)
+ require.Nil(t, matched)
+ require.Equal(t, uint32(5), resolved.GetSegmentInterval().GetNum())
+}
+
+func TestResolveStageResourceOpts_DoesNotMutateInput(t *testing.T) {
+ // Regression for the pointer-aliasing bug: resolving must not mutate the shared
+ // group schema's Ttl/SegmentInterval.
+ ro := stagedOpts()
+ resolved, _, _, err := ResolveStageResourceOpts(ro, map[string]string{"type": "cold"})
+ require.NoError(t, err)
+ require.Equal(t, uint32(20), resolved.GetSegmentInterval().GetNum(), "resolved copy carries the cold interval")
+ require.Equal(t, uint32(1), ro.GetTtl().GetNum(), "group base ttl untouched")
+ require.Equal(t, uint32(5), ro.GetSegmentInterval().GetNum(), "group default interval untouched")
+ require.Equal(t, uint32(1), ro.GetShardNum(), "group shard num untouched")
+}
+
+func TestResolveStageResourceOpts_TTLUnitMismatchErrors(t *testing.T) {
+ ro := stagedOpts()
+ ro.Stages[1].Ttl = &commonv1.IntervalRule{Unit: commonv1.IntervalRule_UNIT_HOUR, Num: 30}
+ _, matched, _, err := ResolveStageResourceOpts(ro, map[string]string{"type": "cold"})
+ require.Error(t, err)
+ require.Nil(t, matched)
+}
diff --git a/banyand/stream/metadata.go b/banyand/stream/metadata.go
index 47001983c..6d0866faa 100644
--- a/banyand/stream/metadata.go
+++ b/banyand/stream/metadata.go
@@ -564,60 +564,25 @@ func (s *supplier) OpenDB(groupSchema *commonv1.Group) (resourceSchema.DB, error
if ro == nil {
return nil, fmt.Errorf("no resource opts in group %s", name)
}
- shardNum := ro.ShardNum
- ttl := ro.Ttl
- segInterval := ro.SegmentInterval
- // Non-zero default so the idle-segment reclaimer ticker actually starts
- // (storage/rotation.go gates it on >=1s). Staged Close paths override below.
- segmentIdleTimeout := time.Hour
- disableRetention := false
- disableRotation := false
- if len(ro.Stages) > 0 && len(s.nodeLabels) > 0 {
- var ttlNum uint32
- foundMatched := false
- for i, st := range ro.Stages {
- if st.Ttl.Unit != ro.Ttl.Unit {
- return nil, fmt.Errorf("ttl unit %s is not consistent with stage %s", ro.Ttl.Unit, st.Ttl.Unit)
- }
- selector, err := pub.ParseLabelSelector(st.NodeSelector)
- if err != nil {
- return nil, errors.WithMessagef(err, "failed to parse node selector %s", st.NodeSelector)
- }
- ttlNum += st.Ttl.Num
- if !selector.Matches(s.nodeLabels) {
- continue
- }
- foundMatched = true
- ttl.Num += ttlNum
- shardNum = st.ShardNum
- segInterval = st.SegmentInterval
- if st.Close {
- segmentIdleTimeout = 5 * time.Minute
- }
- disableRetention = i+1 < len(ro.Stages)
- disableRotation = true
- break
- }
- if !foundMatched {
- disableRetention = true
- disableRotation = true
- }
+ res, err := pub.ResolveStage(s.l, name, ro, s.nodeLabels)
+ if err != nil {
+ return nil, err
}
group := groupSchema.Metadata.Name
opts := storage.TSDBOpts[*tsTable, option]{
- ShardNum: shardNum,
+ ShardNum: res.ResourceOpts.ShardNum,
Location: path.Join(s.path, group),
TSTableCreator: newTSTable,
TableMetrics: s.newMetrics(p),
- SegmentInterval: storage.MustToIntervalRule(segInterval),
- TTL: storage.MustToIntervalRule(ttl),
+ SegmentInterval: storage.MustToIntervalRule(res.ResourceOpts.SegmentInterval),
+ TTL: storage.MustToIntervalRule(res.ResourceOpts.Ttl),
Option: s.option,
SeriesIndexFlushTimeoutSeconds: s.option.flushTimeout.Nanoseconds() / int64(time.Second),
SeriesIndexCacheMaxBytes: int(s.option.seriesCacheMaxSize),
StorageMetricsFactory: s.omr.With(storageScope.ConstLabels(meter.ToLabelPairs(common.DBLabelNames(), p.DBLabelValues()))),
- SegmentIdleTimeout: segmentIdleTimeout,
- DisableRetention: disableRetention,
- DisableRotation: disableRotation,
+ SegmentIdleTimeout: res.SegmentIdleTimeout,
+ DisableRetention: res.DisableRetention,
+ DisableRotation: res.DisableRotation,
MemoryLimit: s.pm.GetLimit(),
}
return storage.OpenTSDB(
@@ -628,6 +593,12 @@ func (s *supplier) OpenDB(groupSchema *commonv1.Group) (resourceSchema.DB, error
)
}
+// ResolveResourceOpts returns the stage-resolved ResourceOpts so a group UpdateOptions
+// applies the matched stage's interval/ttl/shardNum instead of the group default.
+func (s *supplier) ResolveResourceOpts(groupSchema *commonv1.Group) *commonv1.ResourceOpts {
+ return pub.ResolveResourceOptsForUpdate(s.l, groupSchema, s.nodeLabels)
+}
+
var _ resourceSchema.ResourceSupplier = (*queueSupplier)(nil)
type queueSupplier struct {
@@ -677,6 +648,12 @@ func (s *queueSupplier) ResourceSchema(md *commonv1.Metadata) (resourceSchema.Re
return s.metadata.StreamRegistry().GetStream(ctx, md)
}
+// ResolveResourceOpts returns the group opts unchanged: the liaison write queue has no
+// lifecycle stages to resolve.
+func (s *queueSupplier) ResolveResourceOpts(groupSchema *commonv1.Group) *commonv1.ResourceOpts {
+ return groupSchema.GetResourceOpts()
+}
+
func (s *queueSupplier) OpenDB(groupSchema *commonv1.Group) (resourceSchema.DB, error) {
name := groupSchema.Metadata.Name
p := common.Position{
diff --git a/banyand/trace/metadata.go b/banyand/trace/metadata.go
index efe68da17..622184ef4 100644
--- a/banyand/trace/metadata.go
+++ b/banyand/trace/metadata.go
@@ -817,65 +817,29 @@ func (s *supplier) OpenDB(groupSchema *commonv1.Group) (resourceSchema.DB, error
if ro == nil {
return nil, fmt.Errorf("no resource opts in group %s", name)
}
- shardNum := ro.ShardNum
- ttl := ro.Ttl
- segInterval := ro.SegmentInterval
- // Non-zero default so the idle-segment reclaimer ticker actually starts
- // (storage/rotation.go gates it on >=1s). Staged Close paths override below.
- segmentIdleTimeout := time.Hour
- disableRetention := false
- disableRotation := false
- foundMatched := false
- if len(ro.Stages) > 0 && len(s.nodeLabels) > 0 {
- var ttlNum uint32
- for i, st := range ro.Stages {
- if st.Ttl.Unit != ro.Ttl.Unit {
- return nil, fmt.Errorf("ttl unit %s is not consistent with stage %s", ro.Ttl.Unit, st.Ttl.Unit)
- }
- selector, err := pub.ParseLabelSelector(st.NodeSelector)
- if err != nil {
- return nil, errors.WithMessagef(err, "failed to parse node selector %s", st.NodeSelector)
- }
- ttlNum += st.Ttl.Num
- if !selector.Matches(s.nodeLabels) {
- continue
- }
- foundMatched = true
- ttl.Num += ttlNum
- shardNum = st.ShardNum
- segInterval = st.SegmentInterval
- if st.Close {
- segmentIdleTimeout = 5 * time.Minute
- }
- disableRetention = i+1 < len(ro.Stages)
- disableRotation = true
- break
- }
- if !foundMatched {
- disableRetention = true
- disableRotation = true
- }
+ res, err := pub.ResolveStage(s.l, name, ro, s.nodeLabels)
+ if err != nil {
+ return nil, err
}
// isHot marks the Hot stage: a group with no staging at all, a node with no
// labels (so staging cannot apply), or a node that matched no stage selector.
- isHot := len(ro.Stages) == 0 || len(s.nodeLabels) == 0 || !foundMatched
opt := s.option
- opt.isHot = isHot
+ opt.isHot = !res.Matched
group := groupSchema.Metadata.Name
opts := storage.TSDBOpts[*tsTable, option]{
- ShardNum: shardNum,
+ ShardNum: res.ResourceOpts.ShardNum,
Location: path.Join(s.path, group),
TSTableCreator: newTSTable,
TableMetrics: s.newMetrics(p),
- SegmentInterval: storage.MustToIntervalRule(segInterval),
- TTL: storage.MustToIntervalRule(ttl),
+ SegmentInterval: storage.MustToIntervalRule(res.ResourceOpts.SegmentInterval),
+ TTL: storage.MustToIntervalRule(res.ResourceOpts.Ttl),
Option: opt,
SeriesIndexFlushTimeoutSeconds: s.option.flushTimeout.Nanoseconds() / int64(time.Second),
SeriesIndexCacheMaxBytes: int(s.option.seriesCacheMaxSize),
StorageMetricsFactory: s.omr.With(storageScope.ConstLabels(meter.ToLabelPairs(common.DBLabelNames(), p.DBLabelValues()))),
- SegmentIdleTimeout: segmentIdleTimeout,
- DisableRetention: disableRetention,
- DisableRotation: disableRotation,
+ SegmentIdleTimeout: res.SegmentIdleTimeout,
+ DisableRetention: res.DisableRetention,
+ DisableRotation: res.DisableRotation,
MemoryLimit: s.pm.GetLimit(),
}
return storage.OpenTSDB(
@@ -886,6 +850,12 @@ func (s *supplier) OpenDB(groupSchema *commonv1.Group) (resourceSchema.DB, error
)
}
+// ResolveResourceOpts returns the stage-resolved ResourceOpts so a group UpdateOptions
+// applies the matched stage's interval/ttl/shardNum instead of the group default.
+func (s *supplier) ResolveResourceOpts(groupSchema *commonv1.Group) *commonv1.ResourceOpts {
+ return pub.ResolveResourceOptsForUpdate(s.l, groupSchema, s.nodeLabels)
+}
+
// queueSupplier is the supplier for liaison service.
type queueSupplier struct {
metadata metadata.Repo
@@ -933,6 +903,12 @@ func (qs *queueSupplier) ResourceSchema(md *commonv1.Metadata) (resourceSchema.R
return qs.metadata.TraceRegistry().GetTrace(ctx, md)
}
+// ResolveResourceOpts returns the group opts unchanged: the liaison write queue has no
+// lifecycle stages to resolve.
+func (qs *queueSupplier) ResolveResourceOpts(groupSchema *commonv1.Group) *commonv1.ResourceOpts {
+ return groupSchema.GetResourceOpts()
+}
+
func (qs *queueSupplier) OpenDB(groupSchema *commonv1.Group) (resourceSchema.DB, error) {
name := groupSchema.Metadata.Name
p := common.Position{
diff --git a/banyand/trace/trace.go b/banyand/trace/trace.go
index 3f9dc4cdc..7a6b127fe 100644
--- a/banyand/trace/trace.go
+++ b/banyand/trace/trace.go
@@ -133,9 +133,9 @@ func (t *trace) GetIndexRules() []*databasev1.IndexRule {
}
func (t *trace) OnIndexUpdate(index []*databasev1.IndexRule) {
- if len(index) == 0 {
- return
- }
+ // Store unconditionally (matching measure/stream): an empty index must clear a
+ // previously-applied set, otherwise deleting a subject's last index rule/binding
+ // would leave the stale rules in effect.
var is indexSchema
is.indexRules = index
is.parse(t.schema)
diff --git a/pkg/schema/cache.go b/pkg/schema/cache.go
index 7a4b92328..7ac5e1f1f 100644
--- a/pkg/schema/cache.go
+++ b/pkg/schema/cache.go
@@ -22,6 +22,7 @@ import (
"io"
"path"
"runtime/debug"
+ "strings"
"sync"
"sync/atomic"
"time"
@@ -192,6 +193,10 @@ func (sr *schemaRepo) processEvent(ctx context.Context, evt MetadataEvent) error
case EventKindIndexRule:
key := getKey(evt.Metadata.GetMetadata())
sr.indexRuleMap.Delete(key)
+ // Re-index resources bound to this rule so the deleted rule is dropped from
+ // their in-memory index; the add path (storeIndexRule) refreshes the same
+ // way, the delete path previously did not, leaving stale index config.
+ sr.reindexBoundSubjects(key)
case EventKindIndexRuleBinding:
indexRuleBinding := evt.Metadata.(*databasev1.IndexRuleBinding)
col, _ := sr.bindingForwardMap.Load(getKey(&commonv1.Metadata{
@@ -215,6 +220,9 @@ func (sr *schemaRepo) processEvent(ctx context.Context, evt MetadataEvent) error
tMap := col.(*sync.Map)
tMap.Delete(key)
}
+ // Refresh the subject's index from the remaining bindings so the unbound
+ // rules stop being applied (mirrors storeIndexRuleBinding's add-path refresh).
+ sr.updateIndex(indexRuleBinding)
}
}
return err
@@ -368,7 +376,10 @@ func (sr *schemaRepo) storeGroup(ctx context.Context, groupMeta *commonv1.Metada
return g, nil
}
sr.l.Info().Str("group", name).Msg("updating the group resource options")
- g.db.Load().(DB).UpdateOptions(groupSchema.ResourceOpts)
+ // Resolve through the supplier so a data node re-applies the matched lifecycle
+ // stage's interval/ttl/shardNum; passing the raw group default here would silently
+ // clobber a warm/cold node's stage-resolved values.
+ g.db.Load().(DB).UpdateOptions(g.resourceSupplier.ResolveResourceOpts(groupSchema))
return g, nil
}
@@ -384,10 +395,36 @@ func (sr *schemaRepo) deleteGroup(groupMeta *commonv1.Metadata) error {
if !loaded {
return nil
}
+ // Defensive cascade: the orchestrated deletion flow removes every child before the
+ // group, but if a group delete arrives without that (a crash mid-sequence, or a
+ // non-orchestrated caller) the child entries would dangle in these caches and keep
+ // serving a closed tsdb. Purge everything under this group.
+ sr.purgeGroupFromCaches(name)
grp := g.(*group)
return grp.close()
}
+// purgeGroupFromCaches removes every resource/index-rule/binding cache entry belonging
+// to the group. Keys are "group/name" (getKey), so the "group/" prefix cannot collide
+// with a differently named group.
+func (sr *schemaRepo) purgeGroupFromCaches(group string) {
+ prefix := group + "/"
+ deleteByPrefix := func(m *sync.Map) {
+ m.Range(func(k, _ any) bool {
+ if ks, ok := k.(string); ok && strings.HasPrefix(ks, prefix) {
+ m.Delete(k)
+ }
+ return true
+ })
+ }
+ sr.resourceMutex.Lock()
+ deleteByPrefix(&sr.resourceMap)
+ sr.resourceMutex.Unlock()
+ deleteByPrefix(&sr.indexRuleMap)
+ deleteByPrefix(&sr.bindingForwardMap)
+ deleteByPrefix(&sr.bindingBackwardMap)
+}
+
func (sr *schemaRepo) DropGroup(name string) error {
if g, ok := sr.groupMap.Load(name); ok {
return g.(*group).drop()
@@ -461,24 +498,28 @@ func (sr *schemaRepo) storeIndexRule(indexRule *databasev1.IndexRule) {
if prev.(*databasev1.IndexRule).GetMetadata().ModRevision <= indexRule.GetMetadata().ModRevision {
sr.indexRuleMap.Store(key, indexRule)
sr.updateLatestModRevision(indexRule.GetMetadata().GetModRevision())
- if col, _ := sr.bindingBackwardMap.Load(key); col != nil {
- col.(*sync.Map).Range(func(_, value any) bool {
- sr.updateIndex(value.(*databasev1.IndexRuleBinding))
- return true
- })
- }
+ sr.reindexBoundSubjects(key)
}
} else {
sr.updateLatestModRevision(indexRule.GetMetadata().GetModRevision())
- if col, _ := sr.bindingBackwardMap.Load(key); col != nil {
- col.(*sync.Map).Range(func(_, value any) bool {
- sr.updateIndex(value.(*databasev1.IndexRuleBinding))
- return true
- })
- }
+ sr.reindexBoundSubjects(key)
}
}
+// reindexBoundSubjects refreshes the in-memory index of every resource bound to the index
+// rule identified by ruleKey (through the backward binding map). Called whenever the rule
+// is added, updated, or deleted, so a stored resource always reflects the current rule set.
+func (sr *schemaRepo) reindexBoundSubjects(ruleKey string) {
+ col, _ := sr.bindingBackwardMap.Load(ruleKey)
+ if col == nil {
+ return
+ }
+ col.(*sync.Map).Range(func(_, value any) bool {
+ sr.updateIndex(value.(*databasev1.IndexRuleBinding))
+ return true
+ })
+}
+
func (sr *schemaRepo) storeIndexRuleBinding(indexRuleBinding *databasev1.IndexRuleBinding) {
var changed bool
col, _ := sr.bindingForwardMap.LoadOrStore(getKey(&commonv1.Metadata{
diff --git a/pkg/schema/cache_test.go b/pkg/schema/cache_test.go
index ebafce414..f0a8ae1dd 100644
--- a/pkg/schema/cache_test.go
+++ b/pkg/schema/cache_test.go
@@ -18,6 +18,7 @@
package schema
import (
+ "context"
"errors"
"testing"
@@ -208,6 +209,10 @@ func (s *fakeResourceSupplier) OpenDB(_ *commonv1.Group) (DB, error) {
return fakeDB{}, nil
}
+func (s *fakeResourceSupplier) ResolveResourceOpts(groupSchema *commonv1.Group) *commonv1.ResourceOpts {
+ return groupSchema.GetResourceOpts()
+}
+
func buildGroup(name string) *commonv1.Group {
return &commonv1.Group{
Metadata: &commonv1.Metadata{Name: name},
@@ -295,3 +300,97 @@ func TestInitGroup_AlreadyInit_NoReopen(t *testing.T) {
require.NoError(t, err)
assert.Equal(t, 1, supplier.openCalls, "already-initialized group must not re-invoke OpenDB")
}
+
+// recordingListener captures the rules from the most recent OnIndexUpdate so a test can
+// assert whether a delete re-indexed the resource.
+type recordingListener struct {
+ rules *[]*databasev1.IndexRule
+}
+
+func (r recordingListener) OnIndexUpdate(rules []*databasev1.IndexRule) { *r.rules = rules }
+
+type recordingSupplier struct {
+ lis recordingListener
+}
+
+func (recordingSupplier) ResourceSchema(_ *commonv1.Metadata) (ResourceSchema, error) {
+ return nil, nil
+}
+
+func (s recordingSupplier) OpenResource(_ Resource) (IndexListener, error) { return s.lis, nil }
+
+// TestDeleteIndexRule_ReindexesBoundResource asserts that deleting an index rule
+// re-indexes resources bound to it so the deleted rule is dropped from their in-memory
+// index. The add path already refreshed; the delete path previously left it stale.
+func TestDeleteIndexRule_ReindexesBoundResource(t *testing.T) {
+ var applied []*databasev1.IndexRule
+ sr := &schemaRepo{
+ l: testLogger(),
+ resourceSchemaSupplier: recordingSupplier{lis: recordingListener{rules: &applied}},
+ }
+ require.NoError(t, sr.storeResource(buildMeasure("m", 1)))
+ rule := &databasev1.IndexRule{Metadata: &commonv1.Metadata{Group: "g", Name: "r", ModRevision: 1}}
+ sr.storeIndexRule(rule)
+ sr.storeIndexRuleBinding(&databasev1.IndexRuleBinding{
+ Metadata: &commonv1.Metadata{Group: "g", Name: "b", ModRevision: 1},
+ Subject: &databasev1.Subject{Catalog: commonv1.Catalog_CATALOG_MEASURE, Name: "m"},
+ Rules: []string{"r"},
+ })
+ require.Len(t, applied, 1, "binding must apply the rule to the resource")
+
+ require.NoError(t, sr.processEvent(context.Background(), MetadataEvent{
+ Typ: EventDelete, Kind: EventKindIndexRule, Metadata: rule,
+ }))
+ require.Empty(t, applied, "deleting the index rule must re-index the resource without it")
+}
+
+// TestDeleteIndexRuleBinding_ReindexesSubject asserts that deleting a binding re-indexes
+// the subject so the now-unbound rule stops being applied.
+func TestDeleteIndexRuleBinding_ReindexesSubject(t *testing.T) {
+ var applied []*databasev1.IndexRule
+ sr := &schemaRepo{
+ l: testLogger(),
+ resourceSchemaSupplier: recordingSupplier{lis: recordingListener{rules: &applied}},
+ }
+ require.NoError(t, sr.storeResource(buildMeasure("m", 1)))
+ sr.storeIndexRule(&databasev1.IndexRule{Metadata: &commonv1.Metadata{Group: "g", Name: "r", ModRevision: 1}})
+ binding := &databasev1.IndexRuleBinding{
+ Metadata: &commonv1.Metadata{Group: "g", Name: "b", ModRevision: 1},
+ Subject: &databasev1.Subject{Catalog: commonv1.Catalog_CATALOG_MEASURE, Name: "m"},
+ Rules: []string{"r"},
+ }
+ sr.storeIndexRuleBinding(binding)
+ require.Len(t, applied, 1)
+
+ require.NoError(t, sr.processEvent(context.Background(), MetadataEvent{
+ Typ: EventDelete, Kind: EventKindIndexRuleBinding, Metadata: binding,
+ }))
+ require.Empty(t, applied, "deleting the binding must re-index the subject without the unbound rule")
+}
+
+// TestDeleteGroup_PurgesChildCaches asserts that deleting a group purges its child
+// resource/index-rule/binding entries so nothing dangles after the group is gone.
+func TestDeleteGroup_PurgesChildCaches(t *testing.T) {
+ sr := newTestSchemaRepo()
+ // A portable (nil supplier) group so deleteGroup proceeds past its lookup and
+ // close() is a no-op.
+ sr.groupMap.Store("g", newGroup(nil, nil, nil))
+ require.NoError(t, sr.storeResource(buildMeasure("m", 1)))
+ sr.storeIndexRule(&databasev1.IndexRule{Metadata: &commonv1.Metadata{Group: "g", Name: "r", ModRevision: 1}})
+ sr.storeIndexRuleBinding(&databasev1.IndexRuleBinding{
+ Metadata: &commonv1.Metadata{Group: "g", Name: "b", ModRevision: 1},
+ Subject: &databasev1.Subject{Catalog: commonv1.Catalog_CATALOG_MEASURE, Name: "m"},
+ Rules: []string{"r"},
+ })
+
+ require.NoError(t, sr.deleteGroup(&commonv1.Metadata{Name: "g"}))
+
+ _, hasResource := sr.resourceMap.Load("g/m")
+ _, hasRule := sr.indexRuleMap.Load("g/r")
+ _, hasForward := sr.bindingForwardMap.Load("g/m")
+ _, hasBackward := sr.bindingBackwardMap.Load("g/r")
+ require.False(t, hasResource, "resource entry must be purged")
+ require.False(t, hasRule, "index rule entry must be purged")
+ require.False(t, hasForward, "forward binding entry must be purged")
+ require.False(t, hasBackward, "backward binding entry must be purged")
+}
diff --git a/pkg/schema/schema.go b/pkg/schema/schema.go
index da301a039..05b738328 100644
--- a/pkg/schema/schema.go
+++ b/pkg/schema/schema.go
@@ -92,6 +92,12 @@ type ResourceSchemaSupplier interface {
type ResourceSupplier interface {
ResourceSchemaSupplier
OpenDB(groupSchema *commonv1.Group) (DB, error)
+ // ResolveResourceOpts returns the ResourceOpts that should drive the tsdb for this
+ // supplier's node: a data supplier resolves the matched lifecycle stage's
+ // interval/ttl/shardNum, a liaison/queue supplier returns the group default. It lets
+ // a group UpdateOptions apply the same stage-resolved values that OpenDB used, instead
+ // of silently overwriting them with the group default.
+ ResolveResourceOpts(groupSchema *commonv1.Group) *commonv1.ResourceOpts
}
// DB is the interface of a tsdb.