chore(shard-distributor): did some formatting, and use current load instead of new ewma for determinig if it should persist

Signed-off-by: Theis Mathiassen <[email protected]>

Author: Theis-Mathiassen

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

@@ -31,13 +31,12 @@ var (
 )
 
 type executorStoreImpl struct {
-	client     *clientv3.Client
-	prefix     string
-	logger     log.Logger
-	shardCache *shardcache.ShardToExecutorCache
-	timeSource clock.TimeSource
-	// Max interval (seconds) before we force a shard-stat persist.
-	maxStatsPersistIntervalSeconds int64
+	client                         *clientv3.Client
+	prefix                         string
+	logger                         log.Logger
+	shardCache                     *shardcache.ShardToExecutorCache
+	timeSource                     clock.TimeSource
+	maxStatsPersistIntervalSeconds int64 // Max interval (seconds) before we force a shard-stat persist.
 }
 
 // Constants for gating shard statistics writes to reduce etcd load.
@@ -175,10 +174,7 @@ func (s *executorStoreImpl) RecordHeartbeat(ctx context.Context, namespace, exec
 
 func deriveStatsPersistInterval(shardStatsTTL time.Duration) int64 {
 	ttlSeconds := int64(shardStatsTTL.Seconds())
-	if ttlSeconds <= 1 {
-		return 1
-	}
-	return ttlSeconds - 1
+	return max(1, ttlSeconds-1)
 }
 
 func (s *executorStoreImpl) recordShardStatistics(ctx context.Context, namespace, executorID string, reported map[string]*types.ShardStatusReport) {
@@ -251,15 +247,14 @@ func (s *executorStoreImpl) recordShardStatistics(ctx context.Context, namespace
 		// Update smoothed load via EWMA.
 		prevSmoothed := stats.SmoothedLoad
 		prevUpdate := stats.LastUpdateTime
-		newSmoothed := ewmaSmoothedLoad(prevSmoothed, load, prevUpdate, now)
 
 		// Decide whether to persist this update. We always persist if this is the
 		// first observation (prevUpdate == 0). Otherwise, if the change is small
 		// and the previous persist is recent, skip the write to reduce etcd load.
 		shouldPersist := true
 		if prevUpdate > 0 {
 			age := now - prevUpdate
-			delta := math.Abs(newSmoothed - prevSmoothed)
+			delta := math.Abs(load - prevSmoothed)
 			if delta < shardStatsEpsilon && age < s.maxStatsPersistIntervalSeconds {
 				shouldPersist = false
 			}
@@ -270,6 +265,7 @@ func (s *executorStoreImpl) recordShardStatistics(ctx context.Context, namespace
 			continue
 		}
 
+		newSmoothed := ewmaSmoothedLoad(prevSmoothed, load, prevUpdate, now)
 		stats.SmoothedLoad = newSmoothed
 		stats.LastUpdateTime = now