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

devanbenz · gwossum · web-flow · commit c168cc5aca39 · 2025-09-25T09:03:52.000-05:00
Co-authored-by: Geoffrey Wossum <gwossum@influxdata.com> closes: #26525
diff --git a/cmd/influxd/run/server.go b/cmd/influxd/run/server.go
@@ -299,7 +299,7 @@ func (s *Server) appendRetentionPolicyService(c retention.Config) {
 		return
 	}
 	srv := retention.NewService(c)
-	srv.MetaClient = s.MetaClient
+	srv.SetOSSMetaClient(s.MetaClient)
 	srv.TSDBStore = s.TSDBStore
 	srv.DropShardMetaRef = retention.OSSDropShardMetaRef(s.MetaClient)
 	s.Services = append(s.Services, srv)
diff --git a/services/retention/service.go b/services/retention/service.go
@@ -4,6 +4,8 @@ package retention // import "github.com/influxdata/influxdb/services/retention"
 import (
 	"errors"
 	"fmt"
+	"math"
+	"slices"
 	"sync"
 	"time"
 
@@ -13,13 +15,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() error {
 	if !s.config.Enabled || s.done != nil {
@@ -116,6 +150,10 @@ func (s *Service) run() {
 	}
 }
 
+func (s *Service) isOSS() bool {
+	return s.NodeID() == ossNodeID
+}
+
 func (s *Service) DeletionCheck() {
 	log, logEnd := logger.NewOperation(s.logger, "Retention policy deletion check", "retention_delete_check")
 	defer logEnd()
@@ -244,16 +282,21 @@ func (s *Service) DeletionCheck() {
 
 	// 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(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(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() {
diff --git a/services/retention/service_test.go b/services/retention/service_test.go
@@ -266,7 +266,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())
@@ -721,7 +721,7 @@ func NewService(c retention.Config) *Service {
 	l := logger.New(&s.LogBuf)
 	s.WithLogger(l)
 
-	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

Original file line number	Diff line number	Diff line change
`@@ -299,7 +299,7 @@ func (s *Server) appendRetentionPolicyService(c retention.Config) {`
`299`	`299`	`return`
`300`	`300`	`}`
`301`	`301`	`srv := retention.NewService(c)`
`302`		`- srv.MetaClient = s.MetaClient`
	`302`	`+ srv.SetOSSMetaClient(s.MetaClient)`
`303`	`303`	`srv.TSDBStore = s.TSDBStore`
`304`	`304`	`srv.DropShardMetaRef = retention.OSSDropShardMetaRef(s.MetaClient)`
`305`	`305`	`s.Services = append(s.Services, srv)`