[addon-operator] fix/add recovery to possibly deadlock (#687)

Signed-off-by: Pavel Okhlopkov <pavel.okhlopkov@flant.com>

Author: ldmonster

.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

@@ -10,7 +10,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.11.3
+	github.com/flant/shell-operator v1.11.4
 	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

@@ -164,8 +164,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.11.3 h1:Yp2N/cn/y1glCHKzfAKB8HPcIZ18TidGkgCaP38E6oc=
-github.com/flant/shell-operator v1.11.3/go.mod h1:TFTCgXpp+yrvSUQSQKgotJbRK720fiqwaQdFVU6dAlU=
+github.com/flant/shell-operator v1.11.4 h1:PfLAq11RG4+gmtr5/yxBSEeuFH6btDT87oO+usph4ts=
+github.com/flant/shell-operator v1.11.4/go.mod h1:TFTCgXpp+yrvSUQSQKgotJbRK720fiqwaQdFVU6dAlU=
 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/fluxcd/flagger v1.36.1 h1:X2PumtNwZz9YSGaOtZLFm2zAKLgHhFkbNv8beg7ifyc=

pkg/addon-operator/metrics.go

@@ -149,7 +149,7 @@ func StartTasksQueueLengthUpdater(metricStorage metricsstorage.Storage, tqs *que
 	go func() {
 		for {
 			// Gather task queues lengths.
-			tqs.Iterate(context.TODO(), func(_ context.Context, queue *queue.TaskQueue) {
+			tqs.IterateSnapshot(context.TODO(), 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

@@ -181,7 +181,8 @@ func convergeDone(op *AddonOperator) func(g Gomega) bool {
 		}
 		mainQueue := op.engine.TaskQueues.GetMain()
 		g.Expect(func() bool {
-			if mainQueue.IsEmpty() {
+			// If main queue is empty, converge is done.
+			if mainQueue.Length() == 0 {
 				return true
 			}
 			return mainQueue.GetFirst().GetFailureCount() >= 2

pkg/addon-operator/queue.go

@@ -95,7 +95,8 @@ func RemoveCurrentConvergeTasks(convergeQueues []*queue.TaskQueue, logLabels map
 	convergeDrained := false
 
 	for _, queue := range convergeQueues {
-		if queue == nil || queue.IsEmpty() {
+		// Skip empty queues.
+		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", 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", 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", 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", 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", 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", 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", 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", 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", 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", 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", 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", 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", 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", metricStorage)
 
 				Task := &sh_task.BaseTask{Type: task.GlobalHookRun, Id: "unknown"}
 				q.AddLast(Task.WithMetadata(task.HookMetadata{ModuleName: "unknown"}))

pkg/helm_resources_manager/helm_resources_manager.go

@@ -55,22 +55,23 @@ type helmResourcesManager struct {
 var _ HelmResourcesManager = &helmResourcesManager{}
 
 func NewHelmResourcesManager(ctx context.Context, kclient *klient.Client, logger *log.Logger) (HelmResourcesManager, error) {
-	//nolint:govet
 	cctx, cancel := context.WithCancel(ctx)
 	if kclient == nil {
-		//nolint:govet
+		cancel()
 		return nil, fmt.Errorf("kube client not set")
 	}
 
 	cfg := kclient.RestConfig()
 	defaultLabelSelector, err := labels.Parse(app.ExtraLabels)
 	if err != nil {
+		cancel()
 		return nil, err
 	}
 	cache, err := cr_cache.New(cfg, cr_cache.Options{
 		DefaultLabelSelector: defaultLabelSelector,
 	})
 	if err != nil {
+		cancel()
 		return nil, err
 	}
 
@@ -80,6 +81,7 @@ func NewHelmResourcesManager(ctx context.Context, kclient *klient.Client, logger
 
 	log.Debug("Helm resource manager: cache's been started")
 	if synced := cache.WaitForCacheSync(cctx); !synced {
+		cancel()
 		return nil, fmt.Errorf("Couldn't sync helm resource informer cache")
 	}
 	log.Debug("Helm resourcer manager: cache has been synced")