Fix/1.11 possibly deadlock queue

.golangci.yaml

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

go.mod

@@ -9,7 +9,7 @@ require (
 	github.com/dominikbraun/graph v0.23.0
 	github.com/ettle/strcase v0.2.0
 	github.com/flant/kube-client v1.3.1
-	github.com/flant/shell-operator v1.10.6
+	github.com/flant/shell-operator v1.10.7
 	github.com/go-chi/chi/v5 v5.2.2
 	github.com/go-openapi/loads v0.19.5
 	github.com/go-openapi/spec v0.19.8

go.sum

@@ -144,8 +144,8 @@ github.com/flant/go-openapi-validate v0.19.12-flant.1 h1:GuB9XEfiLHq3M7fafRLq1AW
 github.com/flant/go-openapi-validate v0.19.12-flant.1/go.mod h1:Rzou8hA/CBw8donlS6WNEUQupNvUZ0waH08tGe6kAQ4=
 github.com/flant/kube-client v1.3.1 h1:1SdD799sujXNg2F6Z27le/+qkcKQaKf9Z492YGEhVhc=
 github.com/flant/kube-client v1.3.1/go.mod h1:mql6hsZMgBLAhdj3Emb8TrP5MVdXduFQ2NLjzn6IF0Y=
-github.com/flant/shell-operator v1.10.6 h1:LAIrxG6dLivJVOpn8ucwP0D8WzOBB7U+Fk7B9x8mcH0=
-github.com/flant/shell-operator v1.10.6/go.mod h1:El4fR63G/anIxQklPbK3CUiGTBcurUzf/4runR7FQhg=
+github.com/flant/shell-operator v1.10.7 h1:gyP8ncNnIdE4Ft8oKaG645pfA77W/Evj3zylxdpCl50=
+github.com/flant/shell-operator v1.10.7/go.mod h1:El4fR63G/anIxQklPbK3CUiGTBcurUzf/4runR7FQhg=
 github.com/flopp/go-findfont v0.1.0 h1:lPn0BymDUtJo+ZkV01VS3661HL6F4qFlkhcJN55u6mU=
 github.com/flopp/go-findfont v0.1.0/go.mod h1:wKKxRDjD024Rh7VMwoU90i6ikQRCr+JTHB5n4Ejkqvw=
 github.com/fogleman/gg v1.3.0 h1:/7zJX8F6AaYQc57WQCyN9cAIz+4bCJGO9B+dyW29am8=

pkg/addon-operator/metrics.go

@@ -148,7 +148,7 @@ func StartTasksQueueLengthUpdater(metricStorage metric.Storage, tqs *queue.TaskQ
 	go func() {
 		for {
 			// Gather task queues lengths.
-			tqs.Iterate(context.TODO(), func(_ context.Context, queue *queue.TaskQueue) {
+			tqs.IterateSnapshot(context.Background(), func(_ context.Context, queue *queue.TaskQueue) {
 				queueLen := float64(queue.Length())
 				metricStorage.GaugeSet("{PREFIX}tasks_queue_length", queueLen, map[string]string{"queue": queue.Name})
 			})

pkg/addon-operator/operator_test.go

@@ -168,7 +168,7 @@ func convergeDone(op *AddonOperator) func(g Gomega) bool {
 		}
 		mainQueue := op.engine.TaskQueues.GetMain()
 		g.Expect(func() bool {
-			if mainQueue.IsEmpty() {
+			if mainQueue.Length() == 0 {
 				return true
 			}
 			return mainQueue.GetFirst().GetFailureCount() >= 2

pkg/addon-operator/queue.go

@@ -95,7 +95,7 @@ func RemoveCurrentConvergeTasks(convergeQueues []*queue.TaskQueue, logLabels map
 	convergeDrained := false
 
 	for _, queue := range convergeQueues {
-		if queue == nil || queue.IsEmpty() {
+		if queue == nil || queue.Length() == 0 {
 			continue
 		}
 

pkg/addon-operator/queue_test.go

@@ -27,7 +27,7 @@ func Test_ModuleEnsureCRDsTasksInQueueAfterId(t *testing.T) {
 				metricStorage.HistogramObserveMock.Set(func(_ string, _ float64, _ map[string]string, _ []float64) {})
 				metricStorage.GaugeSetMock.Set(func(_ string, _ float64, _ map[string]string) {})
 
-				q := queue.NewTasksQueue(metricStorage)
+				q := queue.NewTasksQueue("test-queue", metricStorage)
 
 				Task := &sh_task.BaseTask{Type: task.ModuleEnsureCRDs, Id: currentTaskID}
 				q.AddLast(Task)
@@ -49,7 +49,7 @@ func Test_ModuleEnsureCRDsTasksInQueueAfterId(t *testing.T) {
 				metricStorage.HistogramObserveMock.Set(func(_ string, _ float64, _ map[string]string, _ []float64) {})
 				metricStorage.GaugeSetMock.Set(func(_ string, _ float64, _ map[string]string) {})
 
-				q := queue.NewTasksQueue(metricStorage)
+				q := queue.NewTasksQueue("test-queue", metricStorage)
 
 				Task := &sh_task.BaseTask{Type: task.ModuleEnsureCRDs, Id: currentTaskID}
 				q.AddLast(Task)
@@ -70,7 +70,7 @@ func Test_ModuleEnsureCRDsTasksInQueueAfterId(t *testing.T) {
 				metricStorage.HistogramObserveMock.Set(func(_ string, _ float64, _ map[string]string, _ []float64) {})
 				metricStorage.GaugeSetMock.Set(func(_ string, _ float64, _ map[string]string) {})
 
-				q := queue.NewTasksQueue(metricStorage)
+				q := queue.NewTasksQueue("test-queue", metricStorage)
 
 				Task := &sh_task.BaseTask{Type: task.ModuleRun, Id: currentTaskID}
 				q.AddLast(Task.WithMetadata(task.HookMetadata{ModuleName: "unknown"}))
@@ -107,7 +107,7 @@ func Test_QueueHasPendingModuleRunTask(t *testing.T) {
 				metricStorage.HistogramObserveMock.Set(func(_ string, _ float64, _ map[string]string, _ []float64) {})
 				metricStorage.GaugeSetMock.Set(func(_ string, _ float64, _ map[string]string) {})
 
-				q := queue.NewTasksQueue(metricStorage)
+				q := queue.NewTasksQueue("test-queue", metricStorage)
 
 				Task := &sh_task.BaseTask{Type: task.ModuleRun, Id: "unknown"}
 				q.AddLast(Task.WithMetadata(task.HookMetadata{ModuleName: "unknown"}))
@@ -128,7 +128,7 @@ func Test_QueueHasPendingModuleRunTask(t *testing.T) {
 				metricStorage.HistogramObserveMock.Set(func(_ string, _ float64, _ map[string]string, _ []float64) {})
 				metricStorage.GaugeSetMock.Set(func(_ string, _ float64, _ map[string]string) {})
 
-				q := queue.NewTasksQueue(metricStorage)
+				q := queue.NewTasksQueue("test-queue", metricStorage)
 
 				Task := &sh_task.BaseTask{Type: task.ModuleRun, Id: "unknown"}
 				q.AddLast(Task.WithMetadata(task.HookMetadata{ModuleName: "unknown"}))
@@ -152,7 +152,7 @@ func Test_QueueHasPendingModuleRunTask(t *testing.T) {
 				metricStorage.HistogramObserveMock.Set(func(_ string, _ float64, _ map[string]string, _ []float64) {})
 				metricStorage.GaugeSetMock.Set(func(_ string, _ float64, _ map[string]string) {})
 
-				q := queue.NewTasksQueue(metricStorage)
+				q := queue.NewTasksQueue("test-queue", metricStorage)
 
 				Task := &sh_task.BaseTask{Type: task.ModuleRun, Id: "test"}
 				q.AddLast(Task.WithMetadata(task.HookMetadata{ModuleName: "test"}))
@@ -173,7 +173,7 @@ func Test_QueueHasPendingModuleRunTask(t *testing.T) {
 				metricStorage.HistogramObserveMock.Set(func(_ string, _ float64, _ map[string]string, _ []float64) {})
 				metricStorage.GaugeSetMock.Set(func(_ string, _ float64, _ map[string]string) {})
 
-				q := queue.NewTasksQueue(metricStorage)
+				q := queue.NewTasksQueue("test-queue", metricStorage)
 
 				Task := &sh_task.BaseTask{Type: task.ModuleRun, Id: "unknown"}
 				q.AddLast(Task.WithMetadata(task.HookMetadata{ModuleName: "unknown"}))
@@ -294,7 +294,7 @@ func Test_RemoveAdjacentConvergeModules(t *testing.T) {
 			metricStorage.HistogramObserveMock.Set(func(_ string, _ float64, _ map[string]string, _ []float64) {})
 			metricStorage.GaugeSetMock.Set(func(_ string, _ float64, _ map[string]string) {})
 
-			q := queue.NewTasksQueue(metricStorage)
+			q := queue.NewTasksQueue("test-queue", metricStorage)
 
 			for i := range tt.in {
 				tsk := &tt.in[i]
@@ -385,7 +385,7 @@ func Test_ModulesWithPendingModuleRun(t *testing.T) {
 			metricStorage.HistogramObserveMock.Set(func(_ string, _ float64, _ map[string]string, _ []float64) {})
 			metricStorage.GaugeSetMock.Set(func(_ string, _ float64, _ map[string]string) {})
 
-			q := queue.NewTasksQueue(metricStorage)
+			q := queue.NewTasksQueue("test-queue", metricStorage)
 
 			for _, tsk := range tt.in {
 				q.AddLast(tsk)
@@ -537,7 +537,7 @@ func Test_RemoveCurrentConvergeTasks(t *testing.T) {
 				metricStorage.HistogramObserveMock.Set(func(_ string, _ float64, _ map[string]string, _ []float64) {})
 				metricStorage.GaugeSetMock.Optional().Set(func(_ string, _ float64, _ map[string]string) {})
 
-				q := queue.NewTasksQueue(metricStorage)
+				q := queue.NewTasksQueue("test-queue", metricStorage)
 
 				// Fill queue from the test case.
 				queues = append(queues, q)
@@ -666,7 +666,7 @@ func Test_RemoveCurrentConvergeTasksFromId(t *testing.T) {
 			metricStorage.HistogramObserveMock.Set(func(_ string, _ float64, _ map[string]string, _ []float64) {})
 			metricStorage.GaugeSetMock.Set(func(_ string, _ float64, _ map[string]string) {})
 
-			q := queue.NewTasksQueue(metricStorage)
+			q := queue.NewTasksQueue("test-queue", metricStorage)
 
 			// Fill queue from the test case.
 			for i := range tt.initialTasks {
@@ -715,7 +715,7 @@ func Test_ConvergeModulesInQueue(t *testing.T) {
 				metricStorage.HistogramObserveMock.Set(func(_ string, _ float64, _ map[string]string, _ []float64) {})
 				metricStorage.GaugeSetMock.Set(func(_ string, _ float64, _ map[string]string) {})
 
-				q := queue.NewTasksQueue(metricStorage)
+				q := queue.NewTasksQueue("test-queue", metricStorage)
 
 				Task := &sh_task.BaseTask{Type: task.ModuleRun, Id: "unknown"}
 				q.AddLast(Task.WithMetadata(task.HookMetadata{ModuleName: "unknown"}))
@@ -736,7 +736,7 @@ func Test_ConvergeModulesInQueue(t *testing.T) {
 				metricStorage.HistogramObserveMock.Set(func(_ string, _ float64, _ map[string]string, _ []float64) {})
 				metricStorage.GaugeSetMock.Set(func(_ string, _ float64, _ map[string]string) {})
 
-				q := queue.NewTasksQueue(metricStorage)
+				q := queue.NewTasksQueue("test-queue", metricStorage)
 
 				Task := &sh_task.BaseTask{Type: task.GlobalHookRun, Id: "unknown"}
 				q.AddLast(Task.WithMetadata(task.HookMetadata{ModuleName: "unknown"}))
@@ -764,7 +764,7 @@ func Test_ConvergeModulesInQueue(t *testing.T) {
 				metricStorage.HistogramObserveMock.Set(func(_ string, _ float64, _ map[string]string, _ []float64) {})
 				metricStorage.GaugeSetMock.Set(func(_ string, _ float64, _ map[string]string) {})
 
-				q := queue.NewTasksQueue(metricStorage)
+				q := queue.NewTasksQueue("test-queue", metricStorage)
 
 				Task := &sh_task.BaseTask{Type: task.GlobalHookRun, Id: "unknown"}
 				q.AddLast(Task.WithMetadata(task.HookMetadata{ModuleName: "unknown"}))