controller: update reconcilers for changes in connection API

This patch updates the reconcilers to adapt to the ability of
handling errors returned by connection pool.

Signed-off-by: Niraj Yadav <[email protected]>

Author: black-dragon74

internal/controller/csiaddons/encryptionkeyrotationjob_controller.go

@@ -278,7 +278,10 @@ func (r *EncryptionKeyRotationJobReconciler) rotateEncryptionKey(
 	ctx context.Context,
 	logger *logr.Logger,
 	target *targetDetails) error {
-	clientName, client := r.getClientByNode(target.driverName, target.nodeID)
+	clientName, client, err := r.getClientByNode(target.driverName, target.nodeID)
+	if err != nil {
+		return err
+	}
 	if client == nil {
 		return fmt.Errorf("node client not found for node id: %s", target.nodeID)
 	}
@@ -292,7 +295,7 @@ func (r *EncryptionKeyRotationJobReconciler) rotateEncryptionKey(
 	timedCtx, cFunc := context.WithTimeout(ctx, target.timeout)
 	defer cFunc()
 
-	_, err := client.EncryptionKeyRotate(timedCtx, req)
+	_, err = client.EncryptionKeyRotate(timedCtx, req)
 	if err != nil {
 		if status.Code(err) == codes.Unimplemented {
 			logger.Info("encryptionkeyrotation not implemented by driver")
@@ -305,21 +308,24 @@ func (r *EncryptionKeyRotationJobReconciler) rotateEncryptionKey(
 	return nil
 }
 
-func (r *EncryptionKeyRotationJobReconciler) getClientByNode(driver, nodeID string) (string, proto.EncryptionKeyRotationClient) {
-	conns := r.ConnPool.GetByNodeID(driver, nodeID)
+func (r *EncryptionKeyRotationJobReconciler) getClientByNode(driver, nodeID string) (string, proto.EncryptionKeyRotationClient, error) {
+	conns, err := r.ConnPool.GetByNodeID(driver, nodeID)
+	if err != nil {
+		return "", nil, err
+	}
 	for k, v := range conns {
 		for _, cap := range v.Capabilities {
 			if cap.GetEncryptionKeyRotation() == nil {
 				continue
 			}
 
 			if cap.GetEncryptionKeyRotation().Type == identity.Capability_EncryptionKeyRotation_ENCRYPTIONKEYROTATION {
-				return k, proto.NewEncryptionKeyRotationClient(v.Client)
+				return k, proto.NewEncryptionKeyRotationClient(v.Client), nil
 			}
 		}
 	}
 
-	return "", nil
+	return "", nil, nil
 }
 
 // SetupWithManager sets up the controller with the Manager.

internal/controller/csiaddons/persistentvolumeclaim_controller.go

@@ -171,7 +171,7 @@ func (r *PersistentVolumeClaimReconciler) checkDriverSupportCapability(
 	logger *logr.Logger,
 	annotations map[string]string,
 	driverName string,
-	cap Operation) (bool, bool) {
+	cap Operation) (bool, bool, error) {
 	driverSupportsCap := false
 	capFound := false
 
@@ -183,14 +183,17 @@ func (r *PersistentVolumeClaimReconciler) checkDriverSupportCapability(
 		driverAnnotation = krCSIAddonsDriverAnnotation
 	default:
 		logger.Info("Unknown capability", "Capability", cap)
-		return false, false
+		return false, false, nil
 	}
 
 	if drivers, ok := annotations[driverAnnotation]; ok && slices.Contains(strings.Split(drivers, ","), driverAnnotation) {
 		driverSupportsCap = true
 	}
 
-	conns := r.ConnPool.GetByNodeID(driverName, "")
+	conns, err := r.ConnPool.GetByNodeID(driverName, "")
+	if err != nil {
+		return false, false, err
+	}
 	for _, conn := range conns {
 		for _, c := range conn.Capabilities {
 			switch cap {
@@ -203,20 +206,20 @@ func (r *PersistentVolumeClaimReconciler) checkDriverSupportCapability(
 			}
 
 			if capFound {
-				return false, true
+				return false, true, nil
 			}
 		}
 	}
 
 	// If the driver supports the capability but the capability is not found in connection pool,
 	if driverSupportsCap {
 		logger.Info(fmt.Sprintf("Driver supports %s but driver is not registered in the connection pool, Requeuing request", cap), "DriverName", driverName)
-		return true, false
+		return true, false, nil
 	}
 
 	// If the driver does not support the capability, skip requeue
 	logger.Info(fmt.Sprintf("Driver does not support %s, skip Requeue", cap), "DriverName", driverName)
-	return false, false
+	return false, false, nil
 }
 
 // determineScheduleAndRequeue determines the schedule from annotations.
@@ -957,15 +960,21 @@ func (r *PersistentVolumeClaimReconciler) getScheduleFromNS(
 		// Depending on requeue value, it will return ErrorConnNotFoundRequeueNeeded.
 		switch annotationKey {
 		case krcJobScheduleTimeAnnotation:
-			requeue, keyRotationSupported := r.checkDriverSupportCapability(logger, ns.Annotations, driverName, keyRotationOp)
+			requeue, keyRotationSupported, err := r.checkDriverSupportCapability(logger, ns.Annotations, driverName, keyRotationOp)
+			if err != nil {
+				return "", err
+			}
 			if keyRotationSupported {
 				return schedule, nil
 			}
 			if requeue {
 				return "", ErrConnNotFoundRequeueNeeded
 			}
 		case rsCronJobScheduleTimeAnnotation:
-			requeue, supportReclaimspace := r.checkDriverSupportCapability(logger, ns.Annotations, driverName, relciamSpaceOp)
+			requeue, supportReclaimspace, err := r.checkDriverSupportCapability(logger, ns.Annotations, driverName, relciamSpaceOp)
+			if err != nil {
+				return "", err
+			}
 			if supportReclaimspace {
 				// if driver supports space reclamation,
 				// return schedule from ns annotation.

internal/controller/csiaddons/reclaimspacejob_controller.go

@@ -366,20 +366,23 @@ func (r *ReclaimSpaceJobReconciler) getLeadingRSClient(
 
 // getRSClientByNode returns ReclaimSpaceClient given driverName and nodeID.
 func (r *ReclaimSpaceJobReconciler) getRSClientByNode(
-	driverName, nodeID string) (string, proto.ReclaimSpaceClient) {
-	conns := r.ConnPool.GetByNodeID(driverName, nodeID)
+	driverName, nodeID string) (string, proto.ReclaimSpaceClient, error) {
+	conns, err := r.ConnPool.GetByNodeID(driverName, nodeID)
+	if err != nil {
+		return "", nil, err
+	}
 	for k, v := range conns {
 		for _, cap := range v.Capabilities {
 			if cap.GetReclaimSpace() == nil {
 				continue
 			}
 			if cap.GetReclaimSpace().Type == identity.Capability_ReclaimSpace_ONLINE {
-				return k, proto.NewReclaimSpaceClient(v.Client)
+				return k, proto.NewReclaimSpaceClient(v.Client), nil
 			}
 		}
 	}
 
-	return "", nil
+	return "", nil, nil
 }
 
 // controllerReclaimSpace makes controller reclaim space request if controller client is found
@@ -426,9 +429,12 @@ func (r *ReclaimSpaceJobReconciler) nodeReclaimSpace(
 	ctx context.Context,
 	logger *logr.Logger,
 	target *targetDetails) (*int64, error) {
-	clientName, nodeClient := r.getRSClientByNode(
+	clientName, nodeClient, err := r.getRSClientByNode(
 		target.driverName,
 		target.nodeID)
+	if err != nil {
+		return nil, err
+	}
 	if nodeClient == nil {
 		return nil, fmt.Errorf("node Client not found for %q nodeID", target.nodeID)
 	}