[yt/admin/cms] Add increasing timeout between node decomission and ban in CMS.

* Changelog entry
Type: feature
Component: go-sdk

Add increasing timeout between node decomission and ban in CMS.
commit_hash:7ebe1ccb8bbfa9cc232fbd614389761f77030b90

Author: p00p

.mapping.json

@@ -33,6 +33,7 @@
   "internal/cms/missing_chunks_throttler.go":"yt/admin/cms/internal/cms/missing_chunks_throttler.go",
   "internal/cms/mocks/storage_mock.go":"yt/admin/cms/internal/cms/mocks/storage_mock.go",
   "internal/cms/node.go":"yt/admin/cms/internal/cms/node.go",
+  "internal/cms/node_ban_limiter.go":"yt/admin/cms/internal/cms/node_ban_limiter.go",
   "internal/cms/proxy_cache_mock.go":"yt/admin/cms/internal/cms/proxy_cache_mock.go",
   "internal/cms/proxy_role_limits.go":"yt/admin/cms/internal/cms/proxy_role_limits.go",
   "internal/cms/queue_agent.go":"yt/admin/cms/internal/cms/queue_agent.go",

internal/cms/node.go

@@ -309,12 +309,8 @@ func (p *TaskProcessor) decommissionNode(ctx context.Context, r *models.Node) {
 	nodeWithoutChunks := storedChunkCount == 0 && node.State == ytsys.NodeStateOnline
 	nodeOffline := p.checkNodeOffline(node)
 
-	if !task.IsGroupTask && !nodeWithoutChunks && time.Since(p.lastNodeBanTime) < p.conf.BannedNodeHoldOffPeriod {
-		p.l.Info("can not ban node as another one was banned recently", p.nodeLogFields(task, r)...)
-		return
-	}
-
-	if task.IsGroupTask && task.TaskGroup != p.lastBannedNodeGroup && time.Since(p.lastNodeBanTime) < p.conf.BannedNodeHoldOffPeriod {
+	if !p.nodeBanLimiter.Allow() &&
+		((!task.IsGroupTask && !nodeWithoutChunks) || (task.IsGroupTask && task.TaskGroup != p.lastBannedNodeGroup)) {
 		p.l.Info("can not ban node as another one was banned recently", p.nodeLogFields(task, r)...)
 		return
 	}
@@ -348,7 +344,7 @@ func (p *TaskProcessor) decommissionNode(ctx context.Context, r *models.Node) {
 	}
 	p.l.Info("node banned", p.nodeLogFields(task, r)...)
 
-	p.lastNodeBanTime = time.Now()
+	p.nodeBanLimiter.LastBanTime = time.Now()
 	if task.IsGroupTask {
 		p.lastBannedNodeGroup = task.TaskGroup
 	}
@@ -539,13 +535,15 @@ func (p *TaskProcessor) nodeLogFields(t *models.Task, n *models.Node, extra ...l
 	return fields
 }
 
-func (p *TaskProcessor) unbanNodes(ctx context.Context) error {
+func (p *TaskProcessor) unbanNodesBecauseOfLVC(ctx context.Context) error {
 	tasks, err := p.storage.GetAll(ctx)
 	if err != nil {
 		return err
 	}
 
 	var firstError error
+	var unbanned bool
+
 	for _, t := range tasks {
 		if t.WalleStatus == walle.StatusOK {
 			continue
@@ -558,15 +556,25 @@ func (p *TaskProcessor) unbanNodes(ctx context.Context) error {
 				}
 
 				n := r.Role.(*models.Node)
+				if time.Since(time.Time(n.BanTime)) > p.conf.LVCUnbanWindow {
+					continue
+				}
+
 				p.l.Info("unbanning node", p.nodeLogFields(t, n)...)
 				err := p.unbanNode(ctx, t, n)
 				if err != nil && firstError == nil {
 					firstError = err
+				} else if err == nil {
+					unbanned = true
 				}
 			}
 		}
 	}
 
+	if unbanned {
+		p.nodeBanLimiter.Unbanned()
+	}
+
 	return firstError
 }
 

internal/cms/node_ban_limiter.go

@@ -0,0 +1,72 @@
+package cms
+
+import (
+	"time"
+)
+
+// NodeBanLimiter stores node ban limiter.
+//
+// Unbanned() method must be called when a node was unbanned because of LVC,
+// Allow() method can be called to check if another node can be banned.
+type NodeBanLimiter struct {
+	LastBanTime time.Time
+	minTimeout  time.Duration
+	maxTimeout  time.Duration
+	unbanTimes  []time.Time
+}
+
+// NewNodeBanLimiter creates and initializes new node ban limiter.
+func NewNodeBanLimiter(minTimeout, maxTimeout time.Duration) *NodeBanLimiter {
+	l := &NodeBanLimiter{
+		minTimeout: minTimeout,
+		maxTimeout: maxTimeout,
+	}
+	return l
+}
+
+// Allow returns true if another node can be banned.
+//
+// It checks that "timeout" has expired since LastBanTime,
+// "timeout" is calculated as follows:
+// if Unbanned() method wasn't called during 3*maxTimeout duration,
+// "timeout" is minTimeout, if Unbanned() method was called once
+// during 3*maxTimeout duration, "timeout" is (minTimeout + maxTimeout) / 2,
+// otherwise "timeout" is maxTimeout.
+func (l *NodeBanLimiter) Allow() bool {
+	return l.allow(l.LastBanTime, time.Now().UTC())
+}
+
+func (l *NodeBanLimiter) allow(lastBanTime, now time.Time) bool {
+	l.update(now)
+	var timeout = l.minTimeout
+	switch len(l.unbanTimes) {
+	case 0:
+		timeout = l.minTimeout
+	case 1:
+		timeout = (l.minTimeout + l.maxTimeout) / 2
+	default:
+		timeout = l.maxTimeout
+	}
+	return now.Sub(lastBanTime) >= timeout
+}
+
+// Unbanned must be called when a node was unbanned because of LVC.
+func (l *NodeBanLimiter) Unbanned() {
+	l.unbanned(time.Now().UTC())
+}
+
+func (l *NodeBanLimiter) unbanned(now time.Time) {
+	l.update(now)
+	l.unbanTimes = append(l.unbanTimes, now)
+}
+
+func (l *NodeBanLimiter) update(now time.Time) {
+	var newQueue []time.Time
+	for _, t := range l.unbanTimes {
+		if now.Sub(t) >= 3*l.maxTimeout {
+			continue
+		}
+		newQueue = append(newQueue, t)
+	}
+	l.unbanTimes = newQueue
+}

internal/cms/task_processor.go

@@ -26,6 +26,7 @@ const (
 	defaultMinGroupTaskSize                 = 2
 	defaultOfflineNodeRetentionPeriod       = time.Minute * 20
 	defaultBannedNodeHoldOffPeriod          = time.Minute * 3
+	defaultLVCUnbanWindow                   = time.Hour * 2
 	defaultDisabledSchedulerJobsWaitTimeout = time.Minute * 20
 	defaultDisabledWriteSessionsWaitTimeout = time.Minute * 20
 	defaultOfflineHTTPProxyRetentionPeriod  = time.Minute * 10
@@ -99,6 +100,10 @@ type TaskProcessorConfig struct {
 
 	OfflineNodeRetentionPeriod time.Duration `yaml:"offline_node_retention_period"`
 	BannedNodeHoldOffPeriod    time.Duration `yaml:"banned_node_hold_off_period"`
+	MaxBannedNodeHoldOffPeriod time.Duration `yaml:"max_banned_node_hold_off_period"`
+	// LVCUnbanWindow is a duration, after which node will not be unbanned because of LVC.
+	// It is assumed that all lost chunks will be found before LVCUnbanWindow has expired.
+	LVCUnbanWindow time.Duration `yaml:"lvc_unban_window"`
 	// DisabledSchedulerJobsWaitTimeout is a max time to wait for node's scheduler jobs to finish
 	// in fast decommission scenarios.
 	DisabledSchedulerJobsWaitTimeout time.Duration `yaml:"disabled_scheduler_jobs_wait_timeout"`
@@ -194,6 +199,14 @@ func (c *TaskProcessorConfig) UnmarshalYAML(unmarshal func(any) error) error {
 		c.BannedNodeHoldOffPeriod = defaultBannedNodeHoldOffPeriod
 	}
 
+	if c.MaxBannedNodeHoldOffPeriod <= 0 {
+		c.MaxBannedNodeHoldOffPeriod = 3 * c.BannedNodeHoldOffPeriod
+	}
+
+	if c.LVCUnbanWindow <= 0 {
+		c.LVCUnbanWindow = defaultLVCUnbanWindow
+	}
+
 	if c.DisabledSchedulerJobsWaitTimeout <= 0 {
 		c.DisabledSchedulerJobsWaitTimeout = defaultDisabledSchedulerJobsWaitTimeout
 	}
@@ -313,12 +326,12 @@ type TaskProcessor struct {
 	// hostAnnotations store additional host information retrieved from Wall-e.
 	hostAnnotations *HostAnnotations
 
-	// lastNodeBanTime stores a time of the latest node ban made by CMS.
-	// This value is used to limit the number of parallel node bans.
-	lastNodeBanTime time.Time
 	// lastBannedNodeGroup stores group name of a last banned node
 	// belonging to a group task.
 	lastBannedNodeGroup string
+	// nodeBanLimiter stores a time of the latest node ban and unbans made by CMS.
+	// It is used to limit the number of parallel node bans, increasing ban interval, if LVC found.
+	nodeBanLimiter *NodeBanLimiter
 
 	chunkIntegrity         *ytsys.ChunkIntegrity
 	missingChunksThrottler *MissingChunksThrottler
@@ -379,6 +392,7 @@ func (p *TaskProcessor) reset(tasks []*models.Task) {
 	p.rpcProxyRoleLimits = NewProxyRoleLimits(p.Conf().MaxRPCProxiesPerRole, &rpcProxyCache{c: p.cluster})
 	p.initRateLimiter(tasks)
 	p.initGPURateLimiter(tasks)
+	p.nodeBanLimiter = NewNodeBanLimiter(p.conf.BannedNodeHoldOffPeriod, p.conf.MaxBannedNodeHoldOffPeriod)
 	p.initLastNodeBanTime(tasks)
 }
 
@@ -429,24 +443,27 @@ func (p *TaskProcessor) initGPURateLimiter(tasks []*models.Task) {
 
 // initLastNodeBanTime initializes last node ban time.
 func (p *TaskProcessor) initLastNodeBanTime(tasks []*models.Task) {
-	if !p.lastNodeBanTime.IsZero() {
+	if !p.nodeBanLimiter.LastBanTime.IsZero() {
 		return
 	}
 
+	var lastBanTime time.Time
+
 	for _, t := range tasks {
 		for _, n := range t.GetNodes() {
 			banTime := time.Time(n.BanTime)
-			if n.Banned && banTime.After(p.lastNodeBanTime) {
-				p.lastNodeBanTime = banTime
+			if n.Banned && banTime.After(lastBanTime) {
+				lastBanTime = banTime
 				if t.IsGroupTask {
 					p.lastBannedNodeGroup = t.TaskGroup
 				}
 			}
 		}
 	}
 
+	p.nodeBanLimiter.LastBanTime = lastBanTime
 	p.l.Info("setting last node ban time",
-		log.Time("last_node_ban_time", p.lastNodeBanTime),
+		log.Time("last_node_ban_time", lastBanTime),
 		log.String("last_banned_node_group", p.lastBannedNodeGroup))
 }
 
@@ -656,10 +673,10 @@ func (p *TaskProcessor) checkChunkIntegrity(ctx context.Context) error {
 
 	if i.LVC > 0 || i.QMC > 0 {
 		p.l.Info("LVC > 0 or QMC > 0 -> unbanning nodes", log.Int64("lvc", i.LVC), log.Int64("qmc", i.QMC))
-		if err := p.unbanNodes(ctx); err != nil {
+		if err := p.unbanNodesBecauseOfLVC(ctx); err != nil {
 			p.l.Error("error unbanning nodes", log.Error(err))
 		} else {
-			p.lastNodeBanTime = time.Time{}
+			p.nodeBanLimiter.LastBanTime = time.Time{}
 		}
 	} else {
 		p.l.Info("LVC is 0")