fix: stop noisy logging about phantom shards that do not belong to node (#26565)

Stop noisy logging about phantom shards that do not belong to the
current node by checking the shard ownership before logging about the
phantom shard. Note that only the logging was inaccurate. This did not
accidentally remove shards from the metadata that weren't really phantom
shards due to checks in `DropShardMetaRef` implementations.

NOTE: OSS does not suffer from the noisy phantom shard logging issue.
This PR is keep the retention service in sync with 1.x.

This is a port from master-1.x.

closes: #26525
(cherry picked from commit 8ef2aca)

Author: gwossum

storage/engine.go

@@ -143,7 +143,7 @@ func NewEngine(path string, c Config, options ...Option) *Engine {
 
 	e.retentionService = retention.NewService(c.RetentionService)
 	e.retentionService.TSDBStore = e.tsdbStore
-	e.retentionService.MetaClient = e.metaClient
+	e.retentionService.SetOSSMetaClient(e.metaClient)
 	e.retentionService.DropShardMetaRef = retention.OSSDropShardMetaRef(e.MetaClient())
 
 	e.precreatorService = precreator.NewService(c.PrecreatorConfig)

v1/services/retention/service.go

@@ -5,6 +5,8 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"math"
+	"slices"
 	"sync"
 	"time"
 
@@ -15,13 +17,37 @@ import (
 	"go.uber.org/zap"
 )
 
-type MetaClient interface {
+type OSSMetaClient interface {
 	Databases() []meta.DatabaseInfo
 	DeleteShardGroup(database, policy string, id uint64) error
 	DropShard(id uint64) error
 	PruneShardGroups() error
 }
 
+type MetaClient interface {
+	OSSMetaClient
+	NodeID() uint64
+}
+
+const (
+	// ossNodeID is a special node ID for OSS nodes. No enterprise node will ever have this node ID.
+	// 0 can not be used because there is a brief period on startup before meta-client initialization
+	// and before joining a cluster that NodeID() == 0.
+	ossNodeID uint64 = math.MaxUint64
+)
+
+// ossMetaClientAdapter adds methods the retention service needs to the OSS meta.Client implementation.
+// OSSMetaClient is decorated with methods needed for the Enterprise retention service instead of adding
+// them to the OSS MetaClient to avoid polluting the OSS MetaClient namespace.
+type ossMetaClientAdapter struct {
+	OSSMetaClient
+}
+
+// NodeID returns the magic ossNodeID identifier.
+func (c *ossMetaClientAdapter) NodeID() uint64 {
+	return ossNodeID
+}
+
 // Service represents the retention policy enforcement service.
 type Service struct {
 	MetaClient
@@ -58,12 +84,20 @@ func NewService(c Config) *Service {
 }
 
 // OSSDropShardMetaRef creates a closure appropriate for OSS to use as DropShardMetaRef.
-func OSSDropShardMetaRef(mc MetaClient) func(uint64, []uint64) error {
+func OSSDropShardMetaRef(mc OSSMetaClient) func(uint64, []uint64) error {
 	return func(shardID uint64, owners []uint64) error {
 		return mc.DropShard(shardID)
 	}
 }
 
+func (s *Service) SetMetaClient(c MetaClient) {
+	s.MetaClient = c
+}
+
+func (s *Service) SetOSSMetaClient(c OSSMetaClient) {
+	s.SetMetaClient(&ossMetaClientAdapter{OSSMetaClient: c})
+}
+
 // Open starts retention policy enforcement.
 func (s *Service) Open(ctx context.Context) error {
 	if !s.config.Enabled || s.cancel != nil {
@@ -148,6 +182,10 @@ func (s *Service) run(ctx context.Context) {
 	}
 }
 
+func (s *Service) isOSS() bool {
+	return s.NodeID() == ossNodeID
+}
+
 func (s *Service) DeletionCheck(ctx context.Context) {
 	log, logEnd := logger.NewOperation(ctx, s.logger, "Retention policy deletion check", "retention_delete_check")
 	defer logEnd()
@@ -276,16 +314,21 @@ func (s *Service) DeletionCheck(ctx context.Context) {
 
 	// Check for expired phantom shards that exist in the metadata but not in the store.
 	for id, info := range deletedShardIDs {
-		func() {
-			log, logEnd := logger.NewOperation(ctx, log, "Drop phantom shard references", "retention_drop_phantom_refs",
-				logger.Database(info.db), logger.Shard(id), logger.RetentionPolicy(info.rp), zap.Uint64s("owners", info.owners))
-			defer logEnd()
-			log.Warn("Expired phantom shard detected during retention check, removing from metadata")
-			if err := s.DropShardMetaRef(id, info.owners); err != nil {
-				log.Error("Error dropping shard meta reference for phantom shard", zap.Error(err))
-				retryNeeded = true
-			}
-		}()
+		// Enterprise tracks shard ownership while OSS does not because it is single node. A shard not in the
+		// TSDB but in the metadata is always a phantom shard for OSS. For enterprise, it is only a phantom shard
+		// if this node is supposed to own the shard according to the metadata.
+		if s.isOSS() || slices.Contains(info.owners, s.NodeID()) {
+			func() {
+				log, logEnd := logger.NewOperation(ctx, log, "Drop phantom shard references", "retention_drop_phantom_refs",
+					logger.Database(info.db), logger.Shard(id), logger.RetentionPolicy(info.rp), zap.Uint64s("owners", info.owners))
+				defer logEnd()
+				log.Warn("Expired phantom shard detected during retention check, removing from metadata")
+				if err := s.DropShardMetaRef(id, info.owners); err != nil {
+					log.Error("Error dropping shard meta reference for phantom shard", zap.Error(err))
+					retryNeeded = true
+				}
+			}()
+		}
 	}
 
 	func() {

v1/services/retention/service_test.go

@@ -268,7 +268,7 @@ func TestRetention_DeletionCheck(t *testing.T) {
 	}
 
 	s := retention.NewService(cfg)
-	s.MetaClient = mc
+	s.SetOSSMetaClient(mc)
 	s.TSDBStore = store
 	s.DropShardMetaRef = retention.OSSDropShardMetaRef(s.MetaClient)
 	require.NoError(t, s.Open(context.Background()))
@@ -728,7 +728,7 @@ func NewService(tb testing.TB, c retention.Config) *Service {
 	s.LogBuf = logbuf
 	s.WithLogger(log)
 
-	s.Service.MetaClient = s.MetaClient
+	s.Service.SetOSSMetaClient(s.MetaClient)
 	s.Service.TSDBStore = s.TSDBStore
 	s.Service.DropShardMetaRef = retention.OSSDropShardMetaRef(s.Service.MetaClient)
 	return s