feat(shard distributor): decouple shard stats write-throttling decision from heartbeat TTL

Signed-off-by: Andreas Holt <[email protected]>

Author: AndreasHolt

common/config/config.go

@@ -658,8 +658,9 @@ type (
 	}
 
 	LeaderProcess struct {
-		Period       time.Duration `yaml:"period"`
-		HeartbeatTTL time.Duration `yaml:"heartbeatTTL"`
+		Period        time.Duration `yaml:"period"`
+		HeartbeatTTL  time.Duration `yaml:"heartbeatTTL"`
+		ShardStatsTTL time.Duration `yaml:"shardStatsTTL"`
 	}
 )
 

config/development.yaml

@@ -186,3 +186,4 @@ shardDistribution:
   process:
     period: 1s
     heartbeatTTL: 2s
+    shardStatsTTL: 60s

service/sharddistributor/config/config.go

@@ -79,8 +79,9 @@ type (
 	}
 
 	LeaderProcess struct {
-		Period       time.Duration `yaml:"period"`
-		HeartbeatTTL time.Duration `yaml:"heartbeatTTL"`
+		Period        time.Duration `yaml:"period"`
+		HeartbeatTTL  time.Duration `yaml:"heartbeatTTL"`
+		ShardStatsTTL time.Duration `yaml:"shardStatsTTL"`
 	}
 )
 
@@ -97,6 +98,10 @@ const (
 	MigrationModeONBOARDED              = "onboarded"
 )
 
+const (
+	DefaultShardStatsTTL = time.Minute
+)
+
 // ConfigMode maps string migration mode values to types.MigrationMode
 var ConfigMode = map[string]types.MigrationMode{
 	MigrationModeINVALID:                types.MigrationModeINVALID,

service/sharddistributor/leader/process/processor.go

@@ -81,6 +81,9 @@ func NewProcessorFactory(
 	if cfg.Process.HeartbeatTTL == 0 {
 		cfg.Process.HeartbeatTTL = _defaultHearbeatTTL
 	}
+	if cfg.Process.ShardStatsTTL == 0 {
+		cfg.Process.ShardStatsTTL = config.DefaultShardStatsTTL
+	}
 
 	return &processorFactory{
 		logger:        logger,
@@ -277,7 +280,7 @@ func (p *namespaceProcessor) cleanupStaleShardStats(ctx context.Context, namespa
 
 	activeShards := make(map[string]struct{})
 	now := p.timeSource.Now().Unix()
-	shardStatsTTL := int64(p.cfg.HeartbeatTTL.Seconds())
+	shardStatsTTL := int64(p.cfg.ShardStatsTTL.Seconds())
 
 	// 1. build set of active executors
 

service/sharddistributor/leader/process/processor_test.go

@@ -44,8 +44,9 @@ func setupProcessorTest(t *testing.T, namespaceType string) *testDependencies {
 			mockedClock,
 			config.ShardDistribution{
 				Process: config.LeaderProcess{
-					Period:       time.Second,
-					HeartbeatTTL: time.Second,
+					Period:        time.Second,
+					HeartbeatTTL:  time.Second,
+					ShardStatsTTL: 10 * time.Second,
 				},
 			},
 		),
@@ -215,7 +216,7 @@ func TestCleanupStaleShardStats(t *testing.T) {
 		shardStats := map[string]store.ShardStatistics{
 			"shard-1": {SmoothedLoad: 1.0, LastUpdateTime: now.Unix(), LastMoveTime: now.Unix()},
 			"shard-2": {SmoothedLoad: 2.0, LastUpdateTime: now.Unix(), LastMoveTime: now.Unix()},
-			"shard-3": {SmoothedLoad: 3.0, LastUpdateTime: now.Add(-2 * time.Second).Unix(), LastMoveTime: now.Add(-2 * time.Second).Unix()},
+			"shard-3": {SmoothedLoad: 3.0, LastUpdateTime: now.Add(-11 * time.Second).Unix(), LastMoveTime: now.Add(-11 * time.Second).Unix()},
 		}
 
 		namespaceState := &store.NamespaceState{

service/sharddistributor/store/etcd/executorstore/etcdstore.go

@@ -96,13 +96,18 @@ func NewStore(p ExecutorStoreParams) (store.Store, error) {
 		timeSource = clock.NewRealTimeSource()
 	}
 
+	shardStatsTTL := p.Cfg.Process.ShardStatsTTL
+	if shardStatsTTL <= 0 {
+		shardStatsTTL = config.DefaultShardStatsTTL
+	}
+
 	store := &executorStoreImpl{
 		client:                         etcdClient,
 		prefix:                         etcdCfg.Prefix,
 		logger:                         p.Logger,
 		shardCache:                     shardCache,
 		timeSource:                     timeSource,
-		maxStatsPersistIntervalSeconds: deriveStatsPersistInterval(p.Cfg.Process.HeartbeatTTL),
+		maxStatsPersistIntervalSeconds: deriveStatsPersistInterval(shardStatsTTL),
 	}
 
 	p.Lifecycle.Append(fx.StartStopHook(store.Start, store.Stop))
@@ -168,13 +173,12 @@ func (s *executorStoreImpl) RecordHeartbeat(ctx context.Context, namespace, exec
 	return nil
 }
 
-func deriveStatsPersistInterval(heartbeatTTL time.Duration) int64 {
-	ttlSeconds := int64(heartbeatTTL.Seconds())
-	interval := ttlSeconds - 1
-	if interval < 1 {
-		interval = 1
+func deriveStatsPersistInterval(shardStatsTTL time.Duration) int64 {
+	ttlSeconds := int64(shardStatsTTL.Seconds())
+	if ttlSeconds <= 1 {
+		return 1
 	}
-	return interval
+	return ttlSeconds - 1
 }
 
 func (s *executorStoreImpl) recordShardStatistics(ctx context.Context, namespace, executorID string, reported map[string]*types.ShardStatusReport) {

service/sharddistributor/store/etcd/executorstore/etcdstore_test.go

@@ -182,6 +182,7 @@ func TestRecordHeartbeatSkipsShardStatisticsWithNilReport(t *testing.T) {
 func TestRecordHeartbeatShardStatisticsThrottlesWrites(t *testing.T) {
 	tc := testhelper.SetupStoreTestCluster(t)
 	tc.LeaderCfg.Process.HeartbeatTTL = 10 * time.Second
+	tc.LeaderCfg.Process.ShardStatsTTL = 10 * time.Second
 	mockTS := clock.NewMockedTimeSourceAt(time.Unix(1000, 0))
 	executorStore := createStoreWithTimeSource(t, tc, mockTS)
 	esImpl, ok := executorStore.(*executorStoreImpl)