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:59 UTC
(skywalking-banyandb) 01/01: Resolve lifecycle stage opts on group update (fix short warm/cold segments) + schema delete-path fixes
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.