fix: minor bugfix in replic management logic

fix: minor bugfix in replic management logic

pkg/controller/attach_detach.go

@@ -209,7 +209,8 @@ func (r *ReconcileAttachDetach) triggerAttach(ctx context.Context, azVolumeAttac
 			if publishCtx = attachResult.PublishContext(); publishCtx != nil {
 				handleSuccess(false)
 			}
-			if attachErr = <-attachResult.ResultChannel(); attachErr != nil {
+			var ok bool
+			if attachErr, ok = <-attachResult.ResultChannel(); !ok || attachErr != nil {
 				handleError()
 			} else {
 				handleSuccess(true)

pkg/controller/azvolume.go

@@ -289,7 +289,7 @@ func (r *ReconcileAzVolume) triggerDelete(ctx context.Context, azVolume *azdiskv
 
 				// start waiting for replica AzVolumeAttachment CRIs to be deleted
 				for i, attachment := range attachments {
-					waiter, err := r.conditionWatcher.NewConditionWaiter(deleteCtx, watcher.AzVolumeAttachmentType, attachment.Name, verifyObjectDeleted)
+					waiter, err := r.conditionWatcher.NewConditionWaiter(deleteCtx, watcher.AzVolumeAttachmentType, attachment.Name, verifyObjectFailedOrDeleted)
 					if err != nil {
 						updateFunc = func(obj client.Object) error {
 							return reportError(obj, err)

pkg/controller/common.go

@@ -995,6 +995,13 @@ func verifyObjectDeleted(obj interface{}, objectDeleted bool) (bool, error) {
 	if obj == nil || objectDeleted {
 		return true, nil
 	}
+	return false, nil
+}
+
+func verifyObjectFailedOrDeleted(obj interface{}, objectDeleted bool) (bool, error) {
+	if obj == nil || objectDeleted {
+		return true, nil
+	}
 
 	// otherwise, the volume detachment has either failed with error or pending
 	azVolumeAttachmentInstance := obj.(*azdiskv1beta2.AzVolumeAttachment)

pkg/controller/replica.go

@@ -78,7 +78,7 @@ func (r *ReconcileReplica) Reconcile(ctx context.Context, request reconcile.Requ
 
 			// If promotion event, create a replacement replica
 			if isAttached(azVolumeAttachment) && azVolumeAttachment.Status.Detail.PreviousRole == azdiskv1beta2.ReplicaRole {
-				r.triggerManageReplica(ctx, azVolumeAttachment.Spec.VolumeName)
+				r.triggerManageReplica(azVolumeAttachment.Spec.VolumeName)
 			}
 		}
 	} else {
@@ -111,7 +111,7 @@ func (r *ReconcileReplica) Reconcile(ctx context.Context, request reconcile.Requ
 					_, _ = waiter.Wait(goCtx)
 
 					// add replica management operation to the queue
-					r.triggerManageReplica(goCtx, azVolumeAttachment.Spec.VolumeName)
+					r.triggerManageReplica(azVolumeAttachment.Spec.VolumeName)
 				}()
 			}
 		} else if azVolumeAttachment.Status.State == azdiskv1beta2.AttachmentFailed {
@@ -125,7 +125,7 @@ func (r *ReconcileReplica) Reconcile(ctx context.Context, request reconcile.Requ
 }
 
 //nolint:contextcheck // context is not inherited by design
-func (r *ReconcileReplica) triggerManageReplica(ctx context.Context, volumeName string) {
+func (r *ReconcileReplica) triggerManageReplica(volumeName string) {
 	manageReplicaCtx, w := workflow.New(context.Background(), workflow.WithDetails(consts.VolumeNameLabel, volumeName))
 	defer w.Finish(nil)
 	r.addToOperationQueue(

pkg/controller/shared_state.go

@@ -177,7 +177,6 @@ func (c *SharedState) addToOperationQueue(ctx context.Context, volumeName string
 	}
 	lockable := v.(*lockableEntry)
 	lockable.Lock()
-
 	isFirst := lockable.entry.(*operationQueue).Len() <= 0
 	_ = lockable.entry.(*operationQueue).PushBack(&replicaOperation{
 		ctx:       ctx,
@@ -199,35 +198,34 @@ func (c *SharedState) addToOperationQueue(ctx context.Context, volumeName string
 	if isFirst {
 		go func() {
 			lockable.Lock()
+			defer lockable.Unlock()
 			for {
 				operationQueue := lockable.entry.(*operationQueue)
 				// pop the first operation
 				front := operationQueue.Front()
 				operation := front.Value.(*replicaOperation)
-				lockable.Unlock()
 
 				// only run the operation if the operation requester is not enlisted in blacklist
 				if !operationQueue.gcExclusionList.has(operation.requester) {
-					if err := operation.operationFunc(operation.ctx); err != nil {
+					lockable.Unlock()
+					err := operation.operationFunc(operation.ctx)
+					lockable.Lock()
+					if err != nil {
 						if shouldRequeueReplicaOperation(operation.isReplicaGarbageCollection, err) {
 							// if failed, push it to the end of the queue
-							lockable.Lock()
 							if operationQueue.isActive {
 								operationQueue.PushBack(operation)
 							}
-							lockable.Unlock()
 						}
 					}
 				}
 
-				lockable.Lock()
 				operationQueue.remove(front)
 				// there is no entry remaining, exit the loop
 				if operationQueue.Front() == nil {
 					break
 				}
 			}
-			lockable.Unlock()
 		}()
 	}
 }
@@ -241,8 +239,8 @@ func (c *SharedState) deleteOperationQueue(volumeName string) {
 	// clear the queue in case, there still is an entry in queue
 	lockable := v.(*lockableEntry)
 	lockable.Lock()
+	defer lockable.Unlock()
 	lockable.entry.(*operationQueue).Init()
-	lockable.Unlock()
 }
 
 func (c *SharedState) closeOperationQueue(volumeName string) func() {
@@ -265,8 +263,8 @@ func (c *SharedState) addToGcExclusionList(volumeName string, target operationRe
 	}
 	lockable := v.(*lockableEntry)
 	lockable.Lock()
+	defer lockable.Unlock()
 	lockable.entry.(*operationQueue).gcExclusionList.add(target)
-	lockable.Unlock()
 }
 
 func (c *SharedState) removeFromExclusionList(volumeName string, target operationRequester) {
@@ -276,8 +274,8 @@ func (c *SharedState) removeFromExclusionList(volumeName string, target operatio
 	}
 	lockable := v.(*lockableEntry)
 	lockable.Lock()
+	defer lockable.Unlock()
 	delete(lockable.entry.(*operationQueue).gcExclusionList, target)
-	lockable.Unlock()
 }
 
 func (c *SharedState) dequeueGarbageCollection(volumeName string) {
@@ -287,6 +285,7 @@ func (c *SharedState) dequeueGarbageCollection(volumeName string) {
 	}
 	lockable := v.(*lockableEntry)
 	lockable.Lock()
+	defer lockable.Unlock()
 	queue := lockable.entry.(*operationQueue)
 	// look for garbage collection operation in the queue and remove from queue
 	var next *list.Element
@@ -296,7 +295,6 @@ func (c *SharedState) dequeueGarbageCollection(volumeName string) {
 			queue.remove(cur)
 		}
 	}
-	lockable.Unlock()
 }
 
 func (c *SharedState) getVolumesFromPod(ctx context.Context, podName string) ([]string, error) {