fixed a bug when a job deletes itself and it ends up with locked mutex

config/samples/clusters_v1beta1_cassandra.yaml

@@ -37,7 +37,7 @@ spec:
   pciCompliance: false
   luceneEnabled: false        # can be enabled only on 3.11.13 version of Cassandra
   passwordAndUserAuth: true
-#  userRef:
+#  userRefs:
 #    - namespace: default
 #      name: cassandrauser-sample
 #    - namespace: default

config/samples/clusters_v1beta1_postgresql.yaml

@@ -42,7 +42,7 @@ spec:
   #  - email: "rostyslp@netapp.com"
   #  description: "test 222"
   slaTier: "NON_PRODUCTION"
-#  userRef:
+#  userRefs:
 #    - namespace: default
 #      name: postgresqluser-sample
   privateNetworkCluster: false

controllers/clusterresources/awssecuritygroupfirewallrule_controller.go

@@ -311,7 +311,7 @@ func (r *AWSSecurityGroupFirewallRuleReconciler) newWatchStatusJob(firewallRule
 			}
 
 			if instaFirewallRuleStatus.Status == statusDELETED {
-				go r.Scheduler.RemoveJob(firewallRule.GetJobID(scheduler.StatusChecker))
+				r.Scheduler.RemoveJob(firewallRule.GetJobID(scheduler.StatusChecker))
 			}
 		}
 

controllers/clusterresources/awsvpcpeering_controller.go

@@ -396,7 +396,7 @@ func (r *AWSVPCPeeringReconciler) newWatchStatusJob(awsPeering *v1beta1.AWSVPCPe
 				"namespaced name", namespacedName,
 			)
 
-			go r.Scheduler.RemoveJob(awsPeering.GetJobID(scheduler.StatusChecker))
+			r.Scheduler.RemoveJob(awsPeering.GetJobID(scheduler.StatusChecker))
 
 			return nil
 		}
@@ -459,6 +459,7 @@ func (r *AWSVPCPeeringReconciler) newWatchStatusJob(awsPeering *v1beta1.AWSVPCPe
 	}
 }
 
+// SetupWithManager sets up the controller with the Manager.
 func (r *AWSVPCPeeringReconciler) SetupWithManager(mgr ctrl.Manager) error {
 	return ctrl.NewControllerManagedBy(mgr).
 		For(&v1beta1.AWSVPCPeering{}, builder.WithPredicates(predicate.Funcs{

controllers/clusters/cadence_controller.go

@@ -777,7 +777,7 @@ func (r *CadenceReconciler) newWatchStatusJob(cadence *v1beta1.Cadence) schedule
 		if k8serrors.IsNotFound(err) {
 			l.Info("Resource is not found in the k8s cluster. Closing Instaclustr status sync.",
 				"namespaced name", namespacedName)
-			go r.Scheduler.RemoveJob(cadence.GetJobID(scheduler.StatusChecker))
+			r.Scheduler.RemoveJob(cadence.GetJobID(scheduler.StatusChecker))
 			return nil
 		}
 		if err != nil {

controllers/clusters/cassandra_controller.go

@@ -1185,7 +1185,7 @@ func (r *CassandraReconciler) newUsersCreationJob(c *v1beta1.Cassandra) schedule
 		l.Info("User creation job successfully finished", "resource name", c.Name)
 		r.EventRecorder.Eventf(c, models.Normal, models.Created, "User creation job successfully finished")
 
-		go r.Scheduler.RemoveJob(c.GetJobID(scheduler.UserCreator))
+		r.Scheduler.RemoveJob(c.GetJobID(scheduler.UserCreator))
 
 		return nil
 	}

controllers/clusters/kafka_controller.go

@@ -862,7 +862,7 @@ func (r *KafkaReconciler) newUsersCreationJob(kafka *v1beta1.Kafka) scheduler.Jo
 			"User creation job successfully finished",
 		)
 
-		go r.Scheduler.RemoveJob(kafka.GetJobID(scheduler.UserCreator))
+		r.Scheduler.RemoveJob(kafka.GetJobID(scheduler.UserCreator))
 
 		return nil
 	}

controllers/clusters/opensearch_controller.go

@@ -907,7 +907,7 @@ func (r *OpenSearchReconciler) newUsersCreationJob(o *v1beta1.OpenSearch) schedu
 			"User creation job successfully finished",
 		)
 
-		go r.Scheduler.RemoveJob(o.GetJobID(scheduler.UserCreator))
+		r.Scheduler.RemoveJob(o.GetJobID(scheduler.UserCreator))
 
 		return nil
 	}

controllers/clusters/postgresql_controller.go

@@ -1353,7 +1353,7 @@ func (r *PostgreSQLReconciler) newUsersCreationJob(c *v1beta1.PostgreSQL) schedu
 		l.Info("User creation job successfully finished", "resource name", c.Name)
 		r.EventRecorder.Eventf(c, models.Normal, models.Created, "User creation job successfully finished")
 
-		go r.Scheduler.RemoveJob(c.GetJobID(scheduler.UserCreator))
+		r.Scheduler.RemoveJob(c.GetJobID(scheduler.UserCreator))
 
 		return nil
 	}

controllers/clusters/redis_controller.go

@@ -873,7 +873,7 @@ func (r *RedisReconciler) newUsersCreationJob(redis *v1beta1.Redis) scheduler.Jo
 			"User creation job successfully finished",
 		)
 
-		go r.Scheduler.RemoveJob(redis.GetJobID(scheduler.UserCreator))
+		r.Scheduler.RemoveJob(redis.GetJobID(scheduler.UserCreator))
 
 		return nil
 	}
@@ -889,7 +889,7 @@ func (r *RedisReconciler) newWatchStatusJob(redis *v1beta1.Redis) scheduler.Job
 				"namespaced name", namespacedName)
 			r.Scheduler.RemoveJob(redis.GetJobID(scheduler.UserCreator))
 			r.Scheduler.RemoveJob(redis.GetJobID(scheduler.BackupsChecker))
-			go r.Scheduler.RemoveJob(redis.GetJobID(scheduler.StatusChecker))
+			r.Scheduler.RemoveJob(redis.GetJobID(scheduler.StatusChecker))
 			return nil
 		}
 

pkg/scheduler/scheduler.go

@@ -139,5 +139,7 @@ func (s *scheduler) RemoveJob(jobID string) {
 	delete(s.workItems, jobID)
 	s.logger.Info("Job was removed", "job id", jobID)
 
-	v.done <- struct{}{}
+	go func() {
+		v.done <- struct{}{}
+	}()
 }