fixup! [task-controller] Task Controller supports parallel reconciles

Author: justonedev1

control-operator/internal/controller/task_controller.go

@@ -61,7 +61,6 @@ type TaskReconciler struct {
 	MaxConcurrentReconciles int
 }
 
-// var clientsForContainers map[string]*OccClient = make(map[string]*OccClient)
 var clientsForContainers sync.Map
 
 const taskFinalizer string = "aliecs.alice.cern/finalizer"
@@ -139,7 +138,6 @@ func (r *TaskReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.
 		return ctrl.Result{}, nil
 	}
 
-	// if _, exists := clientsForContainers[t.Name]; !exists {
 	if _, exists := clientsForContainers.Load(t.Name); !exists {
 		if existingPod.Status.PodIP == "" {
 			log.Info("pod doesn't have IP yet, we wait for different event")
@@ -163,7 +161,6 @@ func (r *TaskReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.
 	// on them being implemented
 	if t.Status.State == "" {
 		log.V(1).Info("Status.State is empty, querying container")
-		// client, exists := clientsForContainers[t.Name]
 		client, exists := clientsForContainers.Load(t.Name)
 		if !exists {
 			return ctrl.Result{Requeue: true}, nil
@@ -193,17 +190,17 @@ func (r *TaskReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.
 
 	// Handle Spec -> gRPC State Sync
 	if t.Status.State != t.Spec.State {
-		// client, exists := clientsForContainers[t.Name]
-		client, exists := clientsForContainers.Load(t.Name)
+		clientAny, exists := clientsForContainers.Load(t.Name)
 		if !exists {
 			return ctrl.Result{Requeue: true}, nil
 		}
 
-		stateReply, err := client.(*OccClient).GetState(ctx)
+		client := clientAny.(*OccClient)
+
+		stateReply, err := client.GetState(ctx)
 		if err != nil {
 			log.Info("Failed to get state for sync, retrying in 5s", "error", err.Error())
-			client.(*OccClient).Close()
-			// delete(clientsForContainers, t.Name)
+			client.Close()
 			clientsForContainers.Delete(t.Name)
 			return ctrl.Result{RequeueAfter: 5 * time.Second}, nil
 		}
@@ -215,9 +212,9 @@ func (r *TaskReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.
 			)
 
 			if t.Spec.Control.Mode == "fairmq" {
-				newState, transErr = client.(*OccClient).FairMQTransitionRequest(ctx, stateReply.GetState(), t.Spec.State, t.Spec.Arguments)
+				newState, transErr = client.FairMQTransitionRequest(ctx, stateReply.GetState(), t.Spec.State, t.Spec.Arguments)
 			} else {
-				reply, err := client.(*OccClient).TransitionRequest(ctx, stateReply.GetState(), t.Spec.State, t.Spec.Arguments)
+				reply, err := client.TransitionRequest(ctx, stateReply.GetState(), t.Spec.State, t.Spec.Arguments)
 				transErr = err
 				if err == nil && reply.GetOk() {
 					newState = strings.ToLower(reply.GetState())
@@ -275,7 +272,6 @@ func (r *TaskReconciler) createGRPCConsumer(ctx context.Context, t *aliecsv1alph
 		return ctrl.Result{RequeueAfter: 5 * time.Second}, nil
 	}
 
-	// clientsForContainers[t.Name] = client
 	clientsForContainers.Store(t.Name, client)
 
 	if err := r.recordCondition(ctx, t, aliecsv1alpha1.ConditionGRPCConnected, metav1.ConditionTrue, "Connected", fmt.Sprintf("gRPC connection established to %s", addr)); err != nil {
@@ -285,7 +281,6 @@ func (r *TaskReconciler) createGRPCConsumer(ctx context.Context, t *aliecsv1alph
 }
 
 func (r *TaskReconciler) consumeGRPCConsumerIfReady(ctx context.Context, t *aliecsv1alpha1.Task, log logr.Logger) ctrl.Result {
-	// client, exists := clientsForContainers[t.Name]
 	client, exists := clientsForContainers.Load(t.Name)
 
 	if !exists {
@@ -351,13 +346,11 @@ func (r *TaskReconciler) deletePod(ctx context.Context, t *aliecsv1alpha1.Task,
 }
 
 func (*TaskReconciler) cleargRPC(t *aliecsv1alpha1.Task, log logr.Logger) {
-	// if client, exists := clientsForContainers[t.Name]; exists {
 	if client, exists := clientsForContainers.Load(t.Name); exists {
 		log.Info("Cleaning up gRPC connection")
 		if err := client.(*OccClient).Close(); err != nil {
 			log.Error(err, "Failed to close gRPC client during deletion")
 		}
-		// delete(clientsForContainers, t.Name)
 		clientsForContainers.Delete(t.Name)
 		log.Info("gRPC cleaned")
 	}