[shell-operator] fix/add recovery to possibly deadlock (#822)

Signed-off-by: Pavel Okhlopkov <[email protected]>
Co-authored-by: Copilot <[email protected]>

Author: ldmonster

.golangci.yaml

@@ -100,6 +100,7 @@ linters:
       - third_party$
       - builtin$
       - examples$
+      - _test\.go$
 issues:
   max-issues-per-linter: 0
   max-same-issues: 0

pkg/hook/controller/hook_controller.go

@@ -27,16 +27,16 @@ type BindingExecutionInfo struct {
 	KubernetesBinding   htypes.OnKubernetesEventConfig
 }
 
-// В каждый хук надо будет положить этот объект.
-// Предварительно позвав With*Bindings и InitBindingsControllers
+// This object needs to be placed in each hook.
+// After calling With*Bindings and InitBindingsControllers
 
-// Для kube надо будет сделать HandleEnableKubernetesBindings, чтобы получить списки существующих обхектов
-// и потом StartMonitors
+// For kube, HandleEnableKubernetesBindings needs to be called to get lists of existing objects
+// and then StartMonitors
 
-// Всё Handle* методы принимают callback, чтобы создавать задания независимо.
+// All Handle* methods accept a callback to create tasks independently.
 
-// методом KubernetesSnapshots можно достать все кубовые объекты, чтобы добавить
-// их в какой-то свой binding context
+// The KubernetesSnapshots method can be used to get all Kubernetes objects to add
+// them to some binding context
 
 func NewHookController() *HookController {
 	return &HookController{}

pkg/shell-operator/manager_events_handler.go

@@ -2,7 +2,6 @@ package shell_operator
 
 import (
 	"context"
-	"log/slog"
 
 	"github.com/deckhouse/deckhouse/pkg/log"
 
@@ -86,17 +85,7 @@ func (m *ManagerEventsHandler) Start() {
 				return
 			}
 
-			m.taskQueues.DoWithLock(func(tqs *queue.TaskQueueSet) {
-				for _, resTask := range tailTasks {
-					if q, ok := tqs.Queues.Get(resTask.GetQueueName()); !ok {
-						log.Error("Possible bug!!! Got task for queue but queue is not created yet.",
-							slog.String("queueName", resTask.GetQueueName()),
-							slog.String("description", resTask.GetDescription()))
-					} else {
-						q.AddLast(resTask)
-					}
-				}
-			})
+			m.taskQueues.AddTailTasks(tailTasks...)
 		}
 	}()
 }

pkg/shell-operator/operator.go

@@ -932,7 +932,7 @@ func (op *ShellOperator) runMetrics() {
 	// task queue length
 	go func() {
 		for {
-			op.TaskQueues.Iterate(context.TODO(), func(_ context.Context, queue *queue.TaskQueue) {
+			op.TaskQueues.IterateSnapshot(context.TODO(), func(_ context.Context, queue *queue.TaskQueue) {
 				queueLen := float64(queue.Length())
 				op.MetricStorage.GaugeSet(metrics.TasksQueueLength, queueLen, map[string]string{"queue": queue.Name})
 			})

pkg/task/dump/dump.go

@@ -79,51 +79,61 @@ func TaskQueues(tqs *queue.TaskQueueSet, format string, showEmpty bool) interfac
 	tasksCount := 0
 	mainTasksCount := 0
 
-	tqs.Iterate(context.TODO(), func(_ context.Context, queue *queue.TaskQueue) {
+	tqs.IterateSnapshot(context.TODO(), func(_ context.Context, queue *queue.TaskQueue) {
 		if queue == nil {
 			return
 		}
 
+		tasks := getTasksForQueue(queue)
+		length := queue.Length()
+		status := queue.GetStatus()
+		isEmpty := length == 0
+
 		if queue.Name == tqs.MainName {
-			mainTasksCount = queue.Length()
-			if queue.IsEmpty() {
+			mainTasksCount = length
+			if isEmpty {
 				mainQueue := dumpQueue{
 					Name:       queue.Name,
-					TasksCount: queue.Length(),
+					TasksCount: length,
 				}
+
 				result.Empty = append(result.Empty, mainQueue)
 				result.MainQueue = &mainQueue
-			} else {
-				tasks := getTasksForQueue(queue)
-				mainQueue := dumpQueue{
-					Name:       queue.Name,
-					TasksCount: queue.Length(),
-					Status:     queue.GetStatus(),
-					Tasks:      tasks,
-				}
-				result.Active = append(result.Active, mainQueue)
-				result.MainQueue = &mainQueue
+
+				return
+			}
+
+			mainQueue := dumpQueue{
+				Name:       queue.Name,
+				TasksCount: length,
+				Status:     status,
+				Tasks:      tasks,
 			}
+
+			result.Active = append(result.Active, mainQueue)
+			result.MainQueue = &mainQueue
+
 			return
 		}
 
 		otherQueuesCount++
-		if queue.IsEmpty() {
+		if isEmpty {
 			emptyQueues++
 			result.Empty = append(result.Empty, dumpQueue{
 				Name: queue.Name,
 			})
-		} else {
-			activeQueues++
-			tasksCount += queue.Length()
-			tasks := getTasksForQueue(queue)
-			result.Active = append(result.Active, dumpQueue{
-				Name:       queue.Name,
-				TasksCount: queue.Length(),
-				Status:     queue.GetStatus(),
-				Tasks:      tasks,
-			})
+
+			return
 		}
+
+		activeQueues++
+		tasksCount += length
+		result.Active = append(result.Active, dumpQueue{
+			Name:       queue.Name,
+			TasksCount: length,
+			Status:     status,
+			Tasks:      tasks,
+		})
 	})
 
 	result.SortByName()

pkg/task/dump/dump_test.go

@@ -88,7 +88,7 @@ func Test_Dump(t *testing.T) {
 				"queue_name":   "active-queue",
 			},
 			{
-				"queue_action": "IsEmpty",
+				"queue_action": "Length",
 				"queue_name":   "empty",
 			},
 		}