diff --git a/pkg/storage/unified/search/bleve.go b/pkg/storage/unified/search/bleve.go index 8134b3145a3..df37a406a29 100644 --- a/pkg/storage/unified/search/bleve.go +++ b/pkg/storage/unified/search/bleve.go @@ -83,6 +83,11 @@ type BleveOptions struct { // If nil, all indexes are owned by the current instance. OwnsIndex func(key resource.NamespacedResource) (bool, error) + // OwnsSubIndex is called to check whether a specific sub-index is owned by the current instance. + // This function considers the sub-index ID when determining ownership via ring hash. + // If nil, falls back to OwnsIndex behavior with subIndexID=0. + OwnsSubIndex func(key resource.NamespacedResource, subIndexID int) (bool, error) + // SubIndexCount is the number of sub-indexes per (namespace, group, resource). // When > 0, documents are distributed across sub-indexes using consistent hashing. // This enables horizontal scaling for large namespaces (1M+ documents). @@ -102,6 +107,10 @@ type bleveBackend struct { // set from opts.OwnsIndex, always non-nil ownsIndexFn func(key resource.NamespacedResource) (bool, error) + // set from opts.OwnsSubIndex, always non-nil + // Used for checking ownership of sub-indexes when sharding is enabled + ownsSubIndexFn func(key resource.NamespacedResource, subIndexID int) (bool, error) + cacheMx sync.RWMutex cache map[resource.NamespacedResource]*bleveIndex @@ -152,13 +161,23 @@ func NewBleveBackend(opts BleveOptions, indexMetrics *resource.BleveIndexMetrics ownFn = func(key resource.NamespacedResource) (bool, error) { return true, nil } } + ownSubFn := opts.OwnsSubIndex + if ownSubFn == nil { + // By default, fall back to OwnsIndex behavior (ignore subIndexID). + // This maintains backward compatibility when sub-index sharding is not enabled. + ownSubFn = func(key resource.NamespacedResource, subIndexID int) (bool, error) { + return ownFn(key) + } + } + be := &bleveBackend{ - log: l, - cache: map[resource.NamespacedResource]*bleveIndex{}, - subIndexCache: map[resource.SubIndexKey]*bleveIndex{}, - opts: opts, - ownsIndexFn: ownFn, - indexMetrics: indexMetrics, + log: l, + cache: map[resource.NamespacedResource]*bleveIndex{}, + subIndexCache: map[resource.SubIndexKey]*bleveIndex{}, + opts: opts, + ownsIndexFn: ownFn, + ownsSubIndexFn: ownSubFn, + indexMetrics: indexMetrics, } ctx, cancel := context.WithCancel(context.Background()) @@ -292,7 +311,13 @@ func (b *bleveBackend) runEvictExpiredOrUnownedIndexes(now time.Time) { unowned := map[resource.NamespacedResource]*bleveIndex{} ownCheckErrors := map[resource.NamespacedResource]error{} + // For sub-indexes + expiredSubIndexes := map[resource.SubIndexKey]*bleveIndex{} + unownedSubIndexes := map[resource.SubIndexKey]*bleveIndex{} + ownSubCheckErrors := map[resource.SubIndexKey]error{} + b.cacheMx.Lock() + // Process main cache (non-sharded indexes) for key, idx := range b.cache { // Check if index has expired. if !idx.expiration.IsZero() && now.After(idx.expiration) { @@ -312,21 +337,75 @@ func (b *bleveBackend) runEvictExpiredOrUnownedIndexes(now time.Time) { } } } + + // Process sub-index cache (sharded indexes) + for subKey, idx := range b.subIndexCache { + // Check if sub-index has expired. + if !idx.expiration.IsZero() && now.After(idx.expiration) { + delete(b.subIndexCache, subKey) + expiredSubIndexes[subKey] = idx + continue + } + + // Check if sub-index is owned by this instance using OwnsSubIndex. + // This considers the subIndexID when determining ownership via ring hash. + if cacheTTLMillis > 0 { + owned, err := b.ownsSubIndexFn(subKey.NamespacedResource, subKey.SubIndexID) + if err != nil { + ownSubCheckErrors[subKey] = err + } else if !owned && now.UnixMilli()-idx.lastFetchedFromCache.Load() > cacheTTLMillis { + delete(b.subIndexCache, subKey) + unownedSubIndexes[subKey] = idx + } + } + } b.cacheMx.Unlock() + // Log errors for main cache ownership checks for key, err := range ownCheckErrors { b.log.Warn("failed to check if index belongs to this instance", "key", key, "err", err) } + // Log errors for sub-index ownership checks + for subKey, err := range ownSubCheckErrors { + b.log.Warn("failed to check if sub-index belongs to this instance", "subKey", subKey, "err", err) + } + + // Evict unowned main indexes for key, idx := range unowned { b.log.Info("index evicted from cache", "reason", "unowned", "key", key, "storage", idx.indexStorage) b.closeIndex(idx, key) } + // Evict unowned sub-indexes + for subKey, idx := range unownedSubIndexes { + b.log.Info("sub-index evicted from cache", "reason", "unowned", "subKey", subKey, "storage", idx.indexStorage) + b.closeSubIndex(idx, subKey) + } + + // Evict expired main indexes for key, idx := range expired { b.log.Info("index evicted from cache", "reason", "expired", "key", key, "storage", idx.indexStorage) b.closeIndex(idx, key) } + + // Evict expired sub-indexes + for subKey, idx := range expiredSubIndexes { + b.log.Info("sub-index evicted from cache", "reason", "expired", "subKey", subKey, "storage", idx.indexStorage) + b.closeSubIndex(idx, subKey) + } +} + +// closeSubIndex closes a sub-index and updates metrics. +func (b *bleveBackend) closeSubIndex(idx *bleveIndex, key resource.SubIndexKey) { + err := idx.stopUpdaterAndCloseIndex() + if err != nil { + b.log.Error("failed to close sub-index", "key", key, "err", err) + } + + if b.indexMetrics != nil { + b.indexMetrics.OpenIndexes.WithLabelValues(idx.indexStorage).Dec() + } } // updateIndexSizeMetric sets the total size of all file-based indices metric. @@ -742,6 +821,7 @@ func (b *bleveBackend) closeAllIndexes() { b.cacheMx.Lock() defer b.cacheMx.Unlock() + // Close main indexes for key, idx := range b.cache { if err := idx.stopUpdaterAndCloseIndex(); err != nil { b.log.Error("Failed to close index", "err", err) @@ -752,6 +832,18 @@ func (b *bleveBackend) closeAllIndexes() { b.indexMetrics.OpenIndexes.WithLabelValues(idx.indexStorage).Dec() } } + + // Close sub-indexes + for subKey, idx := range b.subIndexCache { + if err := idx.stopUpdaterAndCloseIndex(); err != nil { + b.log.Error("Failed to close sub-index", "subKey", subKey, "err", err) + } + delete(b.subIndexCache, subKey) + + if b.indexMetrics != nil { + b.indexMetrics.OpenIndexes.WithLabelValues(idx.indexStorage).Dec() + } + } } type updateRequest struct { diff --git a/pkg/storage/unified/search/options.go b/pkg/storage/unified/search/options.go index d450e9ae24b..b3b1729c8d9 100644 --- a/pkg/storage/unified/search/options.go +++ b/pkg/storage/unified/search/options.go @@ -17,6 +17,7 @@ func NewSearchOptions( docs resource.DocumentBuilderSupplier, indexMetrics *resource.BleveIndexMetrics, ownsIndexFn func(key resource.NamespacedResource) (bool, error), + ownsSubIndexFn ...func(key resource.NamespacedResource, subIndexID int) (bool, error), ) (resource.SearchOptions, error) { //nolint:staticcheck // not yet migrated to OpenFeature if cfg.EnableSearch || features.IsEnabledGlobally(featuremgmt.FlagProvisioning) { @@ -39,13 +40,22 @@ func NewSearchOptions( } } + // Get OwnsSubIndex function if provided + var ownsSubIdx func(key resource.NamespacedResource, subIndexID int) (bool, error) + if len(ownsSubIndexFn) > 0 && ownsSubIndexFn[0] != nil { + ownsSubIdx = ownsSubIndexFn[0] + } + bleve, err := NewBleveBackend(BleveOptions{ Root: root, FileThreshold: int64(cfg.IndexFileThreshold), // fewer than X items will use a memory index IndexCacheTTL: cfg.IndexCacheTTL, // How long to keep the index cache in memory BuildVersion: cfg.BuildVersion, OwnsIndex: ownsIndexFn, + OwnsSubIndex: ownsSubIdx, IndexMinUpdateInterval: cfg.IndexMinUpdateInterval, + SubIndexCount: cfg.SubIndexesPerNamespace, + LargeFolderThreshold: cfg.LargeFolderThreshold, }, indexMetrics) if err != nil { diff --git a/pkg/storage/unified/sql/service.go b/pkg/storage/unified/sql/service.go index 06275c8754c..17bdaa55e4d 100644 --- a/pkg/storage/unified/sql/service.go +++ b/pkg/storage/unified/sql/service.go @@ -226,6 +226,18 @@ var ( ) func (s *service) OwnsIndex(key resource.NamespacedResource) (bool, error) { + // When sub-index sharding is enabled, use OwnsSubIndex with subIndexID=0 + // to maintain backward compatibility. OwnsIndex is used for the main index. + return s.OwnsSubIndex(key, 0) +} + +// OwnsSubIndex checks if the current instance owns a specific sub-index. +// The sub-index ID is included in the ring hash to distribute sub-indexes +// across nodes in the ring. This enables horizontal scaling for large namespaces. +// +// When subIndexID is 0 and SubIndexesPerNamespace is 0 (disabled), this behaves +// exactly like the original OwnsIndex - maintaining backward compatibility. +func (s *service) OwnsSubIndex(key resource.NamespacedResource, subIndexID int) (bool, error) { if s.searchRing == nil { return true, nil } @@ -235,9 +247,22 @@ func (s *service) OwnsIndex(key resource.NamespacedResource) (bool, error) { } ringHasher := fnv.New32a() - _, err := ringHasher.Write([]byte(key.Namespace)) - if err != nil { - return false, fmt.Errorf("error hashing namespace: %w", err) + + // When sub-index sharding is enabled (SubIndexesPerNamespace > 0), + // include the subIndexID in the hash to distribute sub-indexes across nodes. + // This allows different sub-indexes of the same namespace to be owned by + // different nodes, enabling horizontal scaling. + if s.cfg.SubIndexesPerNamespace > 0 { + _, err := ringHasher.Write([]byte(fmt.Sprintf("%s/%d", key.Namespace, subIndexID))) + if err != nil { + return false, fmt.Errorf("error hashing namespace with sub-index: %w", err) + } + } else { + // Original behavior: hash only the namespace + _, err := ringHasher.Write([]byte(key.Namespace)) + if err != nil { + return false, fmt.Errorf("error hashing namespace: %w", err) + } } rs, err := s.searchRing.GetWithOptions(ringHasher.Sum32(), searchOwnerRead, ring.WithReplicationFactor(s.searchRing.ReplicationFactor())) @@ -261,7 +286,7 @@ func (s *service) starting(ctx context.Context) error { return err } - searchOptions, err := search.NewSearchOptions(s.features, s.cfg, s.docBuilders, s.indexMetrics, s.OwnsIndex) + searchOptions, err := search.NewSearchOptions(s.features, s.cfg, s.docBuilders, s.indexMetrics, s.OwnsIndex, s.OwnsSubIndex) if err != nil { return err }