6188855463

Committed 14 Sep 2023 05:24PM UTC coverage: 69.242% (+0.3%) from 68.933%

Build # 6188855463

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #1781 from NedAnd1/event-persistence

[V2] feat: persist replica attachment failures

Run Details

118 of 118 new or added lines in 3 files covered. (100.0%)

7393 of 10677 relevant lines covered (69.24%)

7.81 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

73.16

/pkg/controller/shared_state.go

/*
Copyright 2021 The Kubernetes Authors.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package controller

import (
        "container/list"
        "context"
        "errors"
        "fmt"
        "math"
        "reflect"
        "sort"
        "strings"
        "sync"
        "sync/atomic"
        "time"

        "google.golang.org/grpc/codes"
        "google.golang.org/grpc/status"
        v1 "k8s.io/api/core/v1"
        crdClientset "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset"
        apiErrors "k8s.io/apimachinery/pkg/api/errors"
        metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
        "k8s.io/apimachinery/pkg/labels"
        "k8s.io/apimachinery/pkg/runtime"
        "k8s.io/apimachinery/pkg/selection"
        "k8s.io/apimachinery/pkg/types"
        "k8s.io/apimachinery/pkg/util/wait"
        utilfeature "k8s.io/apiserver/pkg/util/feature"
        "k8s.io/client-go/kubernetes"
        cache "k8s.io/client-go/tools/cache"
        "k8s.io/client-go/tools/record"
        "k8s.io/client-go/util/retry"
        csitranslator "k8s.io/csi-translation-lib/plugins"
        "k8s.io/klog/v2"
        "k8s.io/kubernetes/pkg/features"
        azdiskv1beta2 "sigs.k8s.io/azuredisk-csi-driver/pkg/apis/azuredisk/v1beta2"
        azdisk "sigs.k8s.io/azuredisk-csi-driver/pkg/apis/client/clientset/versioned"
        consts "sigs.k8s.io/azuredisk-csi-driver/pkg/azureconstants"
        "sigs.k8s.io/azuredisk-csi-driver/pkg/azureutils"
        "sigs.k8s.io/azuredisk-csi-driver/pkg/watcher"
        "sigs.k8s.io/azuredisk-csi-driver/pkg/workflow"
        "sigs.k8s.io/controller-runtime/pkg/client"
)

type DriverLifecycle interface {
        GetDiskClientSet() azdisk.Interface
        GetConditionWatcher() *watcher.ConditionWatcher
        IsDriverUninstall() bool
}

type SharedState struct {
        recoveryComplete              uint32
        config                        *azdiskv1beta2.AzDiskDriverConfiguration
        topologyKey                   string
        podToClaimsMap                sync.Map
        podToInlineMap                sync.Map
        claimToPodsMap                sync.Map
        volumeToClaimMap              sync.Map
        claimToVolumeMap              sync.Map
        azVolumeAttachmentToVaMap     sync.Map
        pvToVolumeMap                 sync.Map
        podLocks                      sync.Map
        visitedVolumes                sync.Map
        volumeOperationQueues         sync.Map
        cleanUpMap                    sync.Map
        priorityReplicaRequestsQueue  *VolumeReplicaRequestsPriorityQueue
        processingReplicaRequestQueue int32
        eventRecorder                 record.EventRecorder
        cachedClient                  client.Client
        azClient                      azdisk.Interface
        kubeClient                    kubernetes.Interface
        crdClient                     crdClientset.Interface
        conditionWatcher              *watcher.ConditionWatcher
        azureDiskCSITranslator        csitranslator.InTreePlugin
        availableAttachmentsMap       sync.Map
        driverLifecycle               DriverLifecycle
        eventsToPersistQueue          chan ReplicaAttachmentFailureInfo
        eventsToUnpersistQueue        chan string
}

type ReplicaAttachmentFailureInfo struct {
        volumeName string
        message    string
        pods       []runtime.Object
        timestamp  time.Time
}

const (
        // persistent event and their refreshes should overlap by this amount of time
        eventOverlapDuration = 30 * time.Second
        // allow persistent events to overlap by up to this additional amount of time for batched reporting etc.
        eventOverlapVariance = 30 * time.Millisecond
)

func NewSharedState(config *azdiskv1beta2.AzDiskDriverConfiguration, topologyKey string, eventRecorder record.EventRecorder, cachedClient client.Client, crdClient crdClientset.Interface, kubeClient kubernetes.Interface, driverLifecycle DriverLifecycle) *SharedState {
        newSharedState := &SharedState{
                config:                 config,
                topologyKey:            topologyKey,
                eventRecorder:          eventRecorder,
                cachedClient:           cachedClient,
                crdClient:              crdClient,
                azClient:               driverLifecycle.GetDiskClientSet(),
                kubeClient:             kubeClient,
                conditionWatcher:       driverLifecycle.GetConditionWatcher(),
                azureDiskCSITranslator: csitranslator.NewAzureDiskCSITranslator(),
                driverLifecycle:        driverLifecycle,
        }
        newSharedState.createReplicaRequestsQueue()
        newSharedState.createEventQueues()

        return newSharedState
}

func (c *SharedState) isRecoveryComplete() bool {
        return atomic.LoadUint32(&c.recoveryComplete) == 1
}

func (c *SharedState) MarkRecoveryComplete() {
        atomic.StoreUint32(&c.recoveryComplete, 1)
}

func (c *SharedState) DeleteAPIVersion(ctx context.Context, deleteVersion string) error {
        w, _ := workflow.GetWorkflowFromContext(ctx)
        crdNames := []string{consts.AzDriverNodeCRDName, consts.AzVolumeCRDName, consts.AzVolumeAttachmentCRDName}
        for _, crdName := range crdNames {
                err := retry.RetryOnConflict(retry.DefaultBackoff,
                        func() error {
                                crd, err := c.crdClient.ApiextensionsV1().CustomResourceDefinitions().Get(ctx, crdName, metav1.GetOptions{})
                                if err != nil {
                                        if apiErrors.IsNotFound(err) {
                                                return err
                                        }
                                        return nil
                                }

                                updated := crd.DeepCopy()
                                var storedVersions []string
                                // remove version from status stored versions
                                for _, version := range updated.Status.StoredVersions {
                                        if version == deleteVersion {
                                                continue
                                        }
                                        storedVersions = append(storedVersions, version)
                                }
                                updated.Status.StoredVersions = storedVersions
                                _, err = c.crdClient.ApiextensionsV1().CustomResourceDefinitions().UpdateStatus(ctx, updated, metav1.UpdateOptions{})
                                if err != nil {
                                        // log the error and continue
                                        return err
                                }
                                return nil
                        })

                if err != nil {
                        w.Logger().Errorf(err, "failed to delete %s api version from CRD (%s)", deleteVersion, crdName)
                }

                // Uncomment when the all deployments have rolled over to v1beta1.
                // updated = crd.DeepCopy()
                // // remove version from spec versions
                // var specVersions []crdv1.CustomResourceDefinitionVersion
                // for _, version := range updated.Spec.Versions {
                //         if version.Name == deleteVersion {
                //                 continue
                //         }
                //         specVersions = append(specVersions, version)
                // }
                // updated.Spec.Versions = specVersions

                // // update the crd
                // crd, err = c.crdClient.ApiextensionsV1().CustomResourceDefinitions().Update(ctx, updated, metav1.UpdateOptions{})
                // if err != nil {
                //         // log the error and continue
                //         w.Logger().Errorf(err, "failed to remove %s spec version from CRD (%s)", deleteVersion, crd.Name)
                //         continue
                // }
        }
        return nil
}

func (c *SharedState) createOperationQueue(volumeName string) {
        _, _ = c.volumeOperationQueues.LoadOrStore(volumeName, newLockableEntry(newOperationQueue()))
}

func (c *SharedState) addToOperationQueue(ctx context.Context, volumeName string, requester operationRequester, operationFunc func(context.Context) error, isReplicaGarbageCollection bool) {
        // It is expected for caller to provide parent workflow via context.
        // The child workflow will be created below and be fed to the queued operation for necessary workflow information.
        ctx, w := workflow.New(ctx, workflow.WithDetails(consts.VolumeNameLabel, volumeName))

        v, ok := c.volumeOperationQueues.Load(volumeName)
        if !ok {
                return
        }
        lockable := v.(*lockableEntry)
        lockable.Lock()
        isFirst := lockable.entry.(*operationQueue).Len() <= 0
        _ = lockable.entry.(*operationQueue).PushBack(&replicaOperation{
                ctx:       ctx,
                requester: requester,
                operationFunc: func(ctx context.Context) (err error) {
                        defer func() {
                                if !shouldRequeueReplicaOperation(isReplicaGarbageCollection, err) {
                                        w.Finish(err)
                                }
                        }()
                        err = operationFunc(ctx)
                        return
                },
                isReplicaGarbageCollection: isReplicaGarbageCollection,
        })
        lockable.Unlock()

        // If this is the first operation, start the goroutine
        if isFirst {
                go func() {
                        lockable.Lock()
                        defer lockable.Unlock()
                        operationQueue := lockable.entry.(*operationQueue)

                        for {
                                // Get the first operation exiting the loop if the queue is empty.
                                front := operationQueue.Front()
                                if front == nil {
                                        break
                                }

                                operation := front.Value.(*replicaOperation)

                                // Only run the operation if the operation requester is not enlisted in blacklist
                                if !operationQueue.gcExclusionList.has(operation.requester) {

                                        // Release the lock while executing the operation to avoid deadlocks.
                                        lockable.Unlock()
                                        err := operation.operationFunc(operation.ctx)
                                        lockable.Lock()

                                        if shouldRequeueReplicaOperation(operation.isReplicaGarbageCollection, err) {
                                                // If operation failed, push it to the end of the queue if the queue is
                                                // still active.
                                                if operationQueue.isActive {
                                                        operationQueue.PushBack(operation)
                                                }
                                        }
                                }

                                // Remove the operation from the queue.
                                operationQueue.safeRemove(front)
                        }
                }()
        }
}

func (c *SharedState) deleteOperationQueue(volumeName string) {
        v, ok := c.volumeOperationQueues.LoadAndDelete(volumeName)
        // if operation queue has already been deleted, return
        if !ok {
                return
        }
        // clear the queue in case, there still is an entry in queue
        lockable := v.(*lockableEntry)
        lockable.Lock()
        defer lockable.Unlock()
        lockable.entry.(*operationQueue).Init()
}

func (c *SharedState) closeOperationQueue(volumeName string) func() {
        v, ok := c.volumeOperationQueues.Load(volumeName)
        if !ok {
                return nil
        }
        lockable := v.(*lockableEntry)

        lockable.Lock()
        lockable.entry.(*operationQueue).isActive = false
        lockable.entry.(*operationQueue).Init()
        return lockable.Unlock
}

func (c *SharedState) addToGcExclusionList(volumeName string, target operationRequester) {
        v, ok := c.volumeOperationQueues.Load(volumeName)
        if !ok {
                return
        }
        lockable := v.(*lockableEntry)
        lockable.Lock()
        defer lockable.Unlock()
        lockable.entry.(*operationQueue).gcExclusionList.add(target)
}

func (c *SharedState) removeFromExclusionList(volumeName string, target operationRequester) {
        v, ok := c.volumeOperationQueues.Load(volumeName)
        if !ok {
                return
        }
        lockable := v.(*lockableEntry)
        lockable.Lock()
        defer lockable.Unlock()
        delete(lockable.entry.(*operationQueue).gcExclusionList, target)
}

func (c *SharedState) dequeueGarbageCollection(volumeName string) {
        v, ok := c.volumeOperationQueues.Load(volumeName)
        if !ok {
                return
        }
        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
        for cur := queue.Front(); cur != nil; cur = next {
                next = cur.Next()
                if cur.Value.(*replicaOperation).isReplicaGarbageCollection {
                        queue.safeRemove(cur)
                }
        }
}

func (c *SharedState) getVolumesFromPod(ctx context.Context, podName string) ([]string, error) {
        w, _ := workflow.GetWorkflowFromContext(ctx)

        var claims []string
        w.Logger().V(5).Infof("Getting requested volumes for pod (%s).", podName)
        value, ok := c.podToClaimsMap.Load(podName)
        if !ok {
                return nil, status.Errorf(codes.NotFound, "unable to find an entry for pod (%s) in podToClaims map", podName)
        }
        claims, ok = value.([]string)
        if !ok {
                return nil, status.Errorf(codes.Internal, "wrong output type: expected []string")
        }

        volumes := []string{}
        for _, claim := range claims {
                value, ok := c.claimToVolumeMap.Load(claim)
                if !ok {
                        // the pvc entry is not an azure resource
                        w.Logger().V(5).Infof("Requested volume %s for pod %s is not an azure resource", value, podName)
                        continue
                }
                volume, ok := value.(string)
                if !ok {
                        return nil, status.Errorf(codes.Internal, "wrong output type: expected string")
                }
                volumes = append(volumes, volume)
                w.Logger().V(5).Infof("Requested volumes for pod %s are now the following: Volumes: %v, Len: %d", podName, volumes, len(volumes))
        }
        return volumes, nil
}

func (c *SharedState) getPodsFromVolume(ctx context.Context, client client.Client, volumeName string) ([]v1.Pod, error) {
        w, _ := workflow.GetWorkflowFromContext(ctx)
        pods, err := c.getPodNamesFromVolume(volumeName)
        if err != nil {
                return nil, err
        }
        podObjs := []v1.Pod{}
        for _, pod := range pods {
                namespace, name, err := parseQualifiedName(pod)
                if err != nil {
                        w.Logger().Errorf(err, "cannot get podObj for pod (%s)", pod)
                        continue
                }
                var podObj v1.Pod
                if err := client.Get(ctx, types.NamespacedName{Namespace: namespace, Name: name}, &podObj); err != nil {
                        return nil, err
                }
                podObjs = append(podObjs, podObj)
        }
        return podObjs, nil
}

func (c *SharedState) getPodNamesFromVolume(volumeName string) ([]string, error) {
        v, ok := c.volumeToClaimMap.Load(volumeName)
        if !ok {
                return nil, status.Errorf(codes.NotFound, "no bound persistent volume claim was found for AzVolume (%s)", volumeName)
        }
        claimName, ok := v.(string)
        if !ok {
                return nil, status.Errorf(codes.Internal, "volumeToClaimMap should should hold string")
        }

        value, ok := c.claimToPodsMap.Load(claimName)
        if !ok {
                return nil, status.Errorf(codes.NotFound, "no pods found for PVC (%s)", claimName)
        }
        lockable, ok := value.(*lockableEntry)
        if !ok {
                return nil, status.Errorf(codes.Internal, "claimToPodsMap should hold lockable entry")
        }

        lockable.RLock()
        defer lockable.RUnlock()

        podMap, ok := lockable.entry.(set)
        if !ok {
                return nil, status.Errorf(codes.Internal, "claimToPodsMap entry should hold a set")
        }

        pods := make([]string, len(podMap))
        i := 0
        for v := range podMap {
                pod := v.(string)
                pods[i] = pod
                i++
        }

        return pods, nil
}

func (c *SharedState) getVolumesForPodObjs(ctx context.Context, pods []v1.Pod) ([]string, error) {
        volumes := []string{}
        for _, pod := range pods {
                podVolumes, err := c.getVolumesFromPod(ctx, getQualifiedName(pod.Namespace, pod.Name))
                if err != nil {
                        return nil, err
                }
                volumes = append(volumes, podVolumes...)
        }
        return volumes, nil
}

func (c *SharedState) addPod(ctx context.Context, pod *v1.Pod, updateOption updateWithLock) error {
        var err error
        w, _ := workflow.GetWorkflowFromContext(ctx)
        podKey := getQualifiedName(pod.Namespace, pod.Name)
        v, _ := c.podLocks.LoadOrStore(podKey, &sync.Mutex{})

        w.Logger().V(5).Infof("Adding pod %s to shared map with keyName %s.", pod.Name, podKey)
        podLock := v.(*sync.Mutex)
        if updateOption == acquireLock {
                podLock.Lock()
                defer podLock.Unlock()
        }
        w.Logger().V(5).Infof("Pod spec of pod %s is: %+v. With volumes: %+v", pod.Name, pod.Spec, pod.Spec.Volumes)

        // If the claims already exist for the podKey, add them to a set
        value, _ := c.podToClaimsMap.LoadOrStore(podKey, []string{})
        claims := value.([]string)
        claimSet := set{}
        for _, claim := range claims {
                claimSet.add(claim)
        }

        for _, volume := range pod.Spec.Volumes {
                // TODO: investigate if we need special support for CSI ephemeral volume or generic ephemeral volume
                // if csiMigration is enabled and there is an inline volume, create AzVolume CRI for the inline volume.
                if utilfeature.DefaultFeatureGate.Enabled(features.CSIMigration) &&
                        utilfeature.DefaultFeatureGate.Enabled(features.CSIMigrationAzureDisk) &&
                        volume.AzureDisk != nil {
                        // inline volume: create AzVolume resource
                        var pv *v1.PersistentVolume
                        if pv, err = c.azureDiskCSITranslator.TranslateInTreeInlineVolumeToCSI(&volume, pod.Namespace); err != nil {
                                w.Logger().V(5).Errorf(err, "failed to translate inline volume to csi")
                                continue
                        } else if pv == nil {
                                w.Logger().V(5).Errorf(status.Errorf(codes.Internal, "unexpected failure in translating inline volume to csi"), "nil pv returned")
                                continue
                        }
                        w.Logger().V(5).Infof("Creating AzVolume instance for inline volume %s.", volume.AzureDisk.DiskName)
                        if err := c.createAzVolumeFromPv(ctx, *pv, map[string]string{consts.InlineVolumeAnnotation: volume.AzureDisk.DataDiskURI}); err != nil {
                                return err
                        }
                        v, exists := c.podToInlineMap.Load(podKey)
                        var inlines []string
                        if exists {
                                inlines = v.([]string)
                        }
                        inlines = append(inlines, volume.AzureDisk.DiskName)
                        c.podToInlineMap.Store(podKey, inlines)
                }
                if volume.PersistentVolumeClaim == nil {
                        w.Logger().V(5).Infof("Pod %s: Skipping Volume %s. No persistent volume exists.", pod.Name, volume)
                        continue
                }
                namespacedClaimName := getQualifiedName(pod.Namespace, volume.PersistentVolumeClaim.ClaimName)
                if _, ok := c.claimToVolumeMap.Load(namespacedClaimName); !ok {
                        // Log message if the Pod status is Running
                        if pod.Status.Phase == v1.PodRunning {
                                w.Logger().V(5).Infof("Skipping Pod %s. Volume %s not csi. Driver: %+v", pod.Name, volume.Name, volume.CSI)
                        }
                        continue
                }
                w.Logger().V(5).Infof("Pod %s. Volume %v is csi.", pod.Name, volume)
                claimSet.add(namespacedClaimName)
                v, _ := c.claimToPodsMap.LoadOrStore(namespacedClaimName, newLockableEntry(set{}))

                lockable := v.(*lockableEntry)
                lockable.Lock()
                pods := lockable.entry.(set)
                if !pods.has(podKey) {
                        pods.add(podKey)
                }
                // No need to restore the amended set to claimToPodsMap because set is a reference type
                lockable.Unlock()

                w.Logger().V(5).Infof("Storing pod %s and claim %s to claimToPodsMap map.", pod.Name, namespacedClaimName)
        }
        w.Logger().V(5).Infof("Storing pod %s and claim %s to podToClaimsMap map.", pod.Name, claims)

        allClaims := []string{}
        for key := range claimSet {
                allClaims = append(allClaims, key.(string))
        }
        c.podToClaimsMap.Store(podKey, allClaims)
        return nil
}

func (c *SharedState) deletePod(ctx context.Context, podKey string) error {
        w, _ := workflow.GetWorkflowFromContext(ctx)
        value, exists := c.podLocks.LoadAndDelete(podKey)
        if !exists {
                return nil
        }
        podLock := value.(*sync.Mutex)
        podLock.Lock()
        defer podLock.Unlock()

        value, exists = c.podToInlineMap.LoadAndDelete(podKey)
        if exists {
                inlines := value.([]string)

                for _, inline := range inlines {
                        _, err := c.cleanUpAzVolumeAttachmentByVolume(ctx, inline, pod, azureutils.AllRoles, cleanUpAttachment, deleteAndWait)
                        if err != nil && !apiErrors.IsNotFound(err) {
                                w.Logger().Errorf(err, "failed to list AzVolumeAttachments (%s) for inline (%s): %v", inline, inline, err)
                                return err
                        }
                        if err := c.azClient.DiskV1beta2().AzVolumes(c.config.ObjectNamespace).Delete(ctx, inline, metav1.DeleteOptions{}); err != nil && !apiErrors.IsNotFound(err) {
                                w.Logger().Errorf(err, "failed to delete AzVolume (%s) for inline (%s): %v", inline, inline, err)
                                return err
                        }
                }
        }

        value, exists = c.podToClaimsMap.LoadAndDelete(podKey)
        if !exists {
                return nil
        }
        claims := value.([]string)

        for _, claim := range claims {
                value, ok := c.claimToPodsMap.Load(claim)
                if !ok {
                        w.Logger().Errorf(nil, "No pods found for PVC (%s)", claim)
                }

                // Scope the duration that we hold the lockable lock using a function.
                func() {
                        lockable, ok := value.(*lockableEntry)
                        if !ok {
                                w.Logger().Error(nil, "claimToPodsMap should hold lockable entry")
                                return
                        }

                        lockable.Lock()
                        defer lockable.Unlock()

                        podSet, ok := lockable.entry.(set)
                        if !ok {
                                w.Logger().Error(nil, "claimToPodsMap entry should hold a set")
                        }

                        podSet.remove(podKey)
                        if len(podSet) == 0 {
                                c.claimToPodsMap.Delete(claim)
                        }
                }()
        }
        return nil
}

func (c *SharedState) addVolumeAndClaim(azVolumeName, pvName, pvClaimName string) {
        c.pvToVolumeMap.Store(pvName, azVolumeName)
        c.volumeToClaimMap.Store(azVolumeName, pvClaimName)
        c.claimToVolumeMap.Store(pvClaimName, azVolumeName)
}

func (c *SharedState) deletePV(pvName string) error {
        var err error

        ctx := context.Background()
        ctx, w := workflow.New(ctx, workflow.WithDetails())
        defer func() { w.Finish(err) }()
        defer func() {
                if apiErrors.IsNotFound(err) {
                        err = nil
                }
        }()

        var azVolume azdiskv1beta2.AzVolume
        if val, ok := c.pvToVolumeMap.Load(pvName); ok {
                volumeName := val.(string)
                err = c.cachedClient.Get(ctx, types.NamespacedName{Namespace: c.config.ObjectNamespace, Name: volumeName}, &azVolume)
                if err != nil {
                        if !apiErrors.IsNotFound(err) {
                                return err
                        }
                        return nil
                }
        } else {
                // if no volume name can be found for PV, try fetching azVolume using labels
                var azVolumeList azdiskv1beta2.AzVolumeList
                req, err := azureutils.CreateLabelRequirements(consts.PvNameLabel, selection.Equals, pvName)
                if err != nil {
                        return err
                }
                err = c.cachedClient.List(ctx, &azVolumeList, &client.ListOptions{LabelSelector: labels.NewSelector().Add(*req)})
                if err != nil && !apiErrors.IsNotFound(err) {
                        return err
                } else if apiErrors.IsNotFound(err) || len(azVolumeList.Items) == 0 {
                        return nil
                }
                azVolume = azVolumeList.Items[0]
        }

        // deletion timestamp is set and AzVolume reconcliler will handle the delete request.
        // The volume itself will not be deleted.
        w.AddDetailToLogger(workflow.GetObjectDetails(&azVolume)...)

        if !isPreProvisioned(&azVolume) {
                return nil
        }

        err = c.cachedClient.Delete(ctx, &azVolume)
        if err != nil {
                if apiErrors.IsNotFound(err) {
                        return nil
                }
                return err
        }

        waitCh := make(chan goSignal)
        go func() {
                goCtx, w := workflow.New(ctx)
                defer func() { w.Finish(err) }()
                waitCh <- goSignal{}

                waiter := c.conditionWatcher.NewConditionWaiter(ctx, watcher.AzVolumeType, azVolume.Name, verifyObjectFailedOrDeleted)

                for {
                        // if AzVolume was successfully deleted
                        obj, err := waiter.Wait(goCtx)
                        if err == nil {
                                // remove the entry from pv to volume map, once AzVolumeCRI is deleted
                                c.pvToVolumeMap.Delete(pvName)
                                return
                        }

                        azVolume := obj.(*azdiskv1beta2.AzVolume)

                        if azVolume.Status.State == azdiskv1beta2.VolumeDeletionFailed || azVolume.Status.Error != nil {
                                updateFunc := func(obj client.Object) error {
                                        azVolume := obj.(*azdiskv1beta2.AzVolume)
                                        azVolume.Status.Error = nil
                                        azVolume.Status.State = azdiskv1beta2.VolumeCreated
                                        return nil
                                }
                                _, _ = azureutils.UpdateCRIWithRetry(goCtx, nil, c.cachedClient, c.azClient, azVolume, updateFunc, consts.ForcedUpdateMaxNetRetry, azureutils.UpdateCRIStatus)
                        }
                }
        }()
        <-waitCh

        return nil
}

func (c *SharedState) deleteVolumeAndClaim(azVolumeName string) {
        v, ok := c.volumeToClaimMap.LoadAndDelete(azVolumeName)
        if ok {
                pvClaimName := v.(string)
                c.claimToVolumeMap.Delete(pvClaimName)
        }
}

func (c *SharedState) markVolumeVisited(azVolumeName string) {
        c.visitedVolumes.Store(azVolumeName, struct{}{})
}

func (c *SharedState) unmarkVolumeVisited(azVolumeName string) {
        c.visitedVolumes.Delete(azVolumeName)
}

func (c *SharedState) isVolumeVisited(azVolumeName string) bool {
        _, visited := c.visitedVolumes.Load(azVolumeName)
        return visited
}

func (c *SharedState) getRankedNodesForReplicaAttachments(ctx context.Context, volumes []string, podObjs []v1.Pod) ([]string, error) {
        var err error
        ctx, w := workflow.New(ctx)
        defer func() { w.Finish(err) }()

        w.Logger().V(5).Info("Getting ranked list of nodes for creating AzVolumeAttachments")

        nodeList := &v1.NodeList{}
        if err := c.cachedClient.List(ctx, nodeList); err != nil {
                return nil, err
        }

        var selectedNodeObjs []v1.Node
        selectedNodeObjs, err = c.selectNodesPerTopology(ctx, nodeList.Items, podObjs, volumes)
        if err != nil {
                w.Logger().Errorf(err, "failed to select nodes for volumes (%+v)", volumes)
                return nil, err
        }

        selectedNodes := make([]string, len(selectedNodeObjs))
        for i, selectedNodeObj := range selectedNodeObjs {
                selectedNodes[i] = selectedNodeObj.Name
        }

        w.Logger().V(5).Infof("Selected nodes (%+v) for replica AzVolumeAttachments for volumes (%+v)", selectedNodes, volumes)
        return selectedNodes, nil
}

func (c *SharedState) filterNodes(ctx context.Context, nodes []v1.Node, pods []v1.Pod, volumes []string) ([]v1.Node, error) {
        var err error
        ctx, w := workflow.New(ctx)
        defer func() { w.Finish(err) }()

        pvs := make([]*v1.PersistentVolume, len(volumes))
        for i, volume := range volumes {
                var azVolume *azdiskv1beta2.AzVolume
                azVolume, err = azureutils.GetAzVolume(ctx, c.cachedClient, c.azClient, volume, c.config.ObjectNamespace, true)
                if err != nil {
                        w.Logger().V(5).Errorf(err, "AzVolume for volume %s is not found.", volume)
                        return nil, err
                }

                var pv v1.PersistentVolume
                if err = c.cachedClient.Get(ctx, types.NamespacedName{Name: azVolume.Spec.PersistentVolume}, &pv); err != nil {
                        return nil, err
                }
                pvs[i] = &pv
        }

        var filterPlugins = []filterPlugin{
                &interPodAffinityFilter{},
                &interPodAntiAffinityFilter{},
                &podTolerationFilter{},
                &podNodeAffinityFilter{},
                &podNodeSelectorFilter{},
                &volumeNodeSelectorFilter{},
        }

        filteredNodes := nodes
        for _, filterPlugin := range filterPlugins {
                filterPlugin.setup(pods, pvs, c)
                if updatedFilteredNodes, err := filterPlugin.filter(ctx, filteredNodes); err != nil {
                        w.Logger().Errorf(err, "failed to filter node with filter plugin (%s). Ignoring filtered results.", filterPlugin.name())
                } else {
                        filteredNodes = updatedFilteredNodes
                        nodeStrs := make([]string, len(filteredNodes))
                        for i, filteredNode := range filteredNodes {
                                nodeStrs[i] = filteredNode.Name
                        }
                        w.Logger().V(10).Infof("Filtered node list from filter plugin (%s): %+v", filterPlugin.name(), nodeStrs)
                }
        }

        return filteredNodes, nil
}

func (c *SharedState) prioritizeNodes(ctx context.Context, pods []v1.Pod, volumes []string, nodes []v1.Node) []v1.Node {
        ctx, w := workflow.New(ctx)
        defer w.Finish(nil)

        nodeScores := map[string]int{}
        for _, node := range nodes {
                nodeScores[node.Name] = 0
        }

        var nodeScorerPlugins = []nodeScorerPlugin{
                &scoreByNodeCapacity{},
                &scoreByReplicaCount{},
                &scoreByInterPodAffinity{},
                &scoreByInterPodAntiAffinity{},
                &scoreByPodNodeAffinity{},
        }

        for _, nodeScorerPlugin := range nodeScorerPlugins {
                nodeScorerPlugin.setup(nodes, pods, volumes, c)
                if updatedNodeScores, err := nodeScorerPlugin.score(ctx, nodeScores); err != nil {
                        w.Logger().Errorf(err, "failed to score nodes by node scorer (%s)", nodeScorerPlugin.name())
                } else {
                        // update node scores if scorer plugin returned success
                        nodeScores = updatedNodeScores
                }
                var nodeScoreResult string
                for nodeName, score := range nodeScores {
                        nodeScoreResult += fmt.Sprintf("<%s: %d> ", nodeName, score)
                }
                w.Logger().V(10).Infof("node score after node score plugin (%s): %s", nodeScorerPlugin.name(), nodeScoreResult)
        }

        // normalize score
        numFiltered := 0
        for _, node := range nodes {
                if _, exists := nodeScores[node.Name]; !exists {
                        nodeScores[node.Name] = -1
                        numFiltered++
                }
        }

        sort.Slice(nodes[:], func(i, j int) bool {
                return nodeScores[nodes[i].Name] > nodeScores[nodes[j].Name]
        })

        return nodes[:len(nodes)-numFiltered]
}

func (c *SharedState) filterAndSortNodes(ctx context.Context, nodes []v1.Node, pods []v1.Pod, volumes []string) ([]v1.Node, error) {
        var err error
        ctx, w := workflow.New(ctx)
        defer func() { w.Finish(err) }()

        var filteredNodes []v1.Node
        filteredNodes, err = c.filterNodes(ctx, nodes, pods, volumes)
        if err != nil {
                w.Logger().Errorf(err, "failed to filter nodes for volumes (%+v): %v", volumes, err)
                return nil, err
        }
        sortedNodes := c.prioritizeNodes(ctx, pods, volumes, filteredNodes)
        return sortedNodes, nil
}

func (c *SharedState) selectNodesPerTopology(ctx context.Context, nodes []v1.Node, pods []v1.Pod, volumes []string) ([]v1.Node, error) {
        var err error
        ctx, w := workflow.New(ctx)
        defer func() { w.Finish(err) }()

        selectedNodes := []v1.Node{}
        numReplicas := 0

        // disperse node topology if possible
        compatibleZonesSet := set{}
        var primaryNode string
        for i, volume := range volumes {
                var azVolume *azdiskv1beta2.AzVolume
                azVolume, err = azureutils.GetAzVolume(ctx, c.cachedClient, c.azClient, volume, c.config.ObjectNamespace, true)
                if err != nil {
                        err = status.Errorf(codes.Aborted, "failed to get AzVolume CRI (%s)", volume)
                        return nil, err
                }

                numReplicas = max(numReplicas, azVolume.Spec.MaxMountReplicaCount)
                w.Logger().V(5).Infof("Number of requested replicas for Azvolume (%s) is: %d. Max replica count is: %d.",
                        volume, numReplicas, azVolume.Spec.MaxMountReplicaCount)

                var pv v1.PersistentVolume
                if err = c.cachedClient.Get(ctx, types.NamespacedName{Name: azVolume.Spec.PersistentVolume}, &pv); err != nil {
                        return nil, err
                }

                if pv.Spec.NodeAffinity == nil || pv.Spec.NodeAffinity.Required == nil {
                        continue
                }

                // Find the intersection of the zones for all the volumes
                topologyKey := c.topologyKey
                if i == 0 {
                        compatibleZonesSet = getSupportedZones(pv.Spec.NodeAffinity.Required.NodeSelectorTerms, topologyKey)
                } else {
                        listOfZones := getSupportedZones(pv.Spec.NodeAffinity.Required.NodeSelectorTerms, topologyKey)
                        for key := range compatibleZonesSet {
                                if !listOfZones.has(key) {
                                        compatibleZonesSet.remove(key)
                                }
                        }
                }

                // find primary node if not already found
                if primaryNode == "" {
                        if primaryAttachment, err := azureutils.GetAzVolumeAttachmentsForVolume(ctx, c.cachedClient, volume, azureutils.PrimaryOnly); err != nil || len(primaryAttachment) == 0 {
                                continue
                        } else {
                                primaryNode = primaryAttachment[0].Spec.NodeName
                        }
                }
        }

        var compatibleZones []string
        if len(compatibleZonesSet) > 0 {
                for key := range compatibleZonesSet {
                        compatibleZones = append(compatibleZones, key.(string))
                }
        }

        if len(compatibleZones) == 0 {
                selectedNodes, err = c.filterAndSortNodes(ctx, nodes, pods, volumes)
                if err != nil {
                        err = status.Errorf(codes.Aborted, "failed to select nodes for volumes (%+v): %v", volumes, err)
                        return nil, err
                }
        } else {
                w.Logger().V(5).Infof("The list of zones to select nodes from is: %s", strings.Join(compatibleZones, ","))

                var primaryNodeZone string
                if primaryNode != "" {
                        nodeObj := &v1.Node{}
                        err = c.cachedClient.Get(ctx, types.NamespacedName{Name: primaryNode}, nodeObj)
                        if err != nil {
                                w.Logger().Errorf(err, "failed to retrieve the primary node")
                        }

                        var ok bool
                        if primaryNodeZone, ok = nodeObj.Labels[consts.WellKnownTopologyKey]; ok {
                                w.Logger().V(5).Infof("failed to find zone annotations for primary node")
                        }
                }

                nodeSelector := labels.NewSelector()
                zoneRequirement, _ := labels.NewRequirement(consts.WellKnownTopologyKey, selection.In, compatibleZones)
                nodeSelector = nodeSelector.Add(*zoneRequirement)

                compatibleNodes := &v1.NodeList{}
                if err = c.cachedClient.List(ctx, compatibleNodes, &client.ListOptions{LabelSelector: nodeSelector}); err != nil {
                        err = status.Errorf(codes.Aborted, "failed to retrieve node list: %v", err)
                        return nodes, err
                }

                // Create a zone to node map
                zoneToNodeMap := map[string][]v1.Node{}
                for _, node := range compatibleNodes.Items {
                        zoneName := node.Labels[consts.WellKnownTopologyKey]
                        zoneToNodeMap[zoneName] = append(zoneToNodeMap[zoneName], node)
                }

                // Get prioritized nodes per zone
                nodesPerZone := [][]v1.Node{}
                primaryZoneNodes := []v1.Node{}
                totalCount := 0
                for zone, nodeList := range zoneToNodeMap {
                        var sortedNodes []v1.Node
                        sortedNodes, err = c.filterAndSortNodes(ctx, nodeList, pods, volumes)
                        if err != nil {
                                err = status.Errorf(codes.Aborted, "failed to select nodes for volumes (%+v): %v", volumes, err)
                                return nil, err
                        }

                        totalCount += len(sortedNodes)
                        if zone == primaryNodeZone {
                                primaryZoneNodes = sortedNodes
                                continue
                        }
                        nodesPerZone = append(nodesPerZone, sortedNodes)
                }
                // Append the nodes from the zone of the primary node at last
                if len(primaryZoneNodes) > 0 {
                        nodesPerZone = append(nodesPerZone, primaryZoneNodes)
                }
                // Select the nodes from each of the zones one by one and append to the list
                i, j, countSoFar := 0, 0, 0
                for len(selectedNodes) < numReplicas && countSoFar < totalCount {
                        if len(nodesPerZone[i]) > j {
                                selectedNodes = append(selectedNodes, nodesPerZone[i][j])
                                countSoFar++
                        }
                        if i < len(nodesPerZone)-1 {
                                i++
                        } else {
                                i = 0
                                j++
                        }
                }
        }

        return selectedNodes, nil
}

func (c *SharedState) getNodesWithReplica(ctx context.Context, volumeName string) ([]string, error) {
        w, _ := workflow.GetWorkflowFromContext(ctx)
        w.Logger().V(5).Infof("Getting nodes with replica AzVolumeAttachments for volume %s.", volumeName)
        azVolumeAttachments, err := azureutils.GetAzVolumeAttachmentsForVolume(ctx, c.cachedClient, volumeName, azureutils.ReplicaOnly)
        if err != nil {
                w.Logger().V(5).Errorf(err, "failed to get AzVolumeAttachments for volume %s.", volumeName)
                return nil, err
        }

        nodes := []string{}
        for _, azVolumeAttachment := range azVolumeAttachments {
                if deleteRequested, _ := objectDeletionRequested(&azVolumeAttachment); !deleteRequested {
                        nodes = append(nodes, azVolumeAttachment.Spec.NodeName)
                }
        }
        w.Logger().V(5).Infof("Nodes with replica AzVolumeAttachments for volume %s are: %v, Len: %d", volumeName, nodes, len(nodes))
        return nodes, nil
}

func (c *SharedState) createReplicaAzVolumeAttachment(ctx context.Context, volumeID, node string, volumeContext map[string]string) error {
        var err error
        ctx, w := workflow.New(ctx, workflow.WithDetails(consts.NodeNameLabel, node))
        defer func() { w.Finish(err) }()

        var diskName string
        diskName, err = azureutils.GetDiskName(volumeID)
        if err != nil {
                err = status.Errorf(codes.Internal, "failed to extract volume name from volumeID (%s)", volumeID)
                return err
        }
        w.AddDetailToLogger(consts.VolumeNameLabel, diskName)

        w.Logger().V(5).Info("Creating replica AzVolumeAttachments")
        if volumeContext == nil {
                volumeContext = make(map[string]string)
        }
        // creating azvolumeattachment
        volumeName := strings.ToLower(diskName)
        replicaName := azureutils.GetAzVolumeAttachmentName(volumeName, node)
        azVolumeAttachment := azdiskv1beta2.AzVolumeAttachment{
                ObjectMeta: metav1.ObjectMeta{
                        Name:      replicaName,
                        Namespace: c.config.ObjectNamespace,
                        Labels: map[string]string{
                                consts.NodeNameLabel:   node,
                                consts.VolumeNameLabel: volumeName,
                                consts.RoleLabel:       string(azdiskv1beta2.ReplicaRole),
                        },
                        Annotations: map[string]string{consts.VolumeAttachRequestAnnotation: "controller"},
                        Finalizers:  []string{consts.AzVolumeAttachmentFinalizer},
                },
                Spec: azdiskv1beta2.AzVolumeAttachmentSpec{
                        NodeName:      node,
                        VolumeID:      volumeID,
                        VolumeName:    volumeName,
                        RequestedRole: azdiskv1beta2.ReplicaRole,
                        VolumeContext: volumeContext,
                },
        }
        w.AnnotateObject(&azVolumeAttachment)
        azureutils.AnnotateAPIVersion(&azVolumeAttachment)

        _, err = c.azClient.DiskV1beta2().AzVolumeAttachments(c.config.ObjectNamespace).Create(ctx, &azVolumeAttachment, metav1.CreateOptions{})
        if err != nil {
                err = status.Errorf(codes.Internal, "failed to create replica AzVolumeAttachment %s.", replicaName)
                return err
        }
        return nil
}

func (c *SharedState) cleanUpAzVolumeAttachmentByVolume(ctx context.Context, azVolumeName string, caller operationRequester, role azureutils.AttachmentRoleMode, cleanupMode attachmentCleanUpMode, attachmentDeleteMode deleteMode) ([]azdiskv1beta2.AzVolumeAttachment, error) {
        var err error
        ctx, w := workflow.New(ctx, workflow.WithDetails(consts.VolumeNameLabel, azVolumeName))
        defer func() { w.Finish(err) }()

        w.Logger().Infof("AzVolumeAttachment clean up requested by %s for AzVolume (%s)", caller, azVolumeName)

        var attachments []azdiskv1beta2.AzVolumeAttachment
        attachments, err = azureutils.GetAzVolumeAttachmentsForVolume(ctx, c.cachedClient, azVolumeName, role)
        if err != nil {
                if apiErrors.IsNotFound(err) {
                        err = nil
                        return nil, nil
                }
                err = status.Errorf(codes.Aborted, "failed to get AzVolumeAttachments: %v", err)
                return nil, err
        }

        if err = c.cleanUpAzVolumeAttachments(ctx, attachments, cleanupMode, caller); err != nil {
                return attachments, err
        }
        c.unmarkVolumeVisited(azVolumeName)

        if attachmentDeleteMode == deleteAndWait {
                attachmentsCount := len(attachments)
                errorMessageCh := make(chan string, attachmentsCount)

                // start waiting for replica AzVolumeAttachment CRIs to be deleted
                for _, attachment := range attachments {
                        // wait async and report error to go channel
                        go func(ctx context.Context, attachment azdiskv1beta2.AzVolumeAttachment) {
                                waiter := c.conditionWatcher.NewConditionWaiter(ctx, watcher.AzVolumeAttachmentType, attachment.Name, verifyObjectFailedOrDeleted)
                                defer waiter.Close()

                                _, derr := waiter.Wait(ctx)
                                if derr != nil {
                                        errorMessageCh <- fmt.Sprintf("%s: %v", attachment.Name, derr)
                                } else {
                                        errorMessageCh <- ""
                                }
                        }(ctx, attachment)
                }

                // if errors have been found with the conditionWatcher calls, format the error msg and report via CRI
                var errMsgs []string
                for i := 0; i < attachmentsCount; i++ {
                        v, ok := <-errorMessageCh
                        if ok && v != "" {
                                errMsgs = append(errMsgs, v)
                        }
                }
                close(errorMessageCh)
                if len(errMsgs) > 0 {
                        err = status.Errorf(codes.Internal, strings.Join(errMsgs, ", "))
                }
        }

        return attachments, err
}

func (c *SharedState) cleanUpAzVolumeAttachmentByNode(ctx context.Context, azDriverNodeName string, caller operationRequester, role azureutils.AttachmentRoleMode, cleanupMode attachmentCleanUpMode, attachmentDeleteMode deleteMode) ([]azdiskv1beta2.AzVolumeAttachment, error) {
        var err error
        ctx, w := workflow.New(ctx, workflow.WithDetails(consts.NodeNameLabel, azDriverNodeName))
        defer func() { w.Finish(err) }()
        w.Logger().Infof("AzVolumeAttachment clean up requested by %s for AzDriverNode (%s)", caller, azDriverNodeName)

        var nodeRequirement *labels.Requirement
        nodeRequirement, err = azureutils.CreateLabelRequirements(consts.NodeNameLabel, selection.Equals, azDriverNodeName)
        if err != nil {
                return nil, err
        }
        labelSelector := labels.NewSelector().Add(*nodeRequirement)

        var attachments *azdiskv1beta2.AzVolumeAttachmentList
        attachments, err = c.azClient.DiskV1beta2().AzVolumeAttachments(c.config.ObjectNamespace).List(ctx, metav1.ListOptions{LabelSelector: labelSelector.String()})
        if err != nil {
                if apiErrors.IsNotFound(err) {
                        err = nil
                        return nil, nil
                }
                err = status.Errorf(codes.Aborted, "failed to get AzVolumeAttachments: %v", err)
                return nil, err
        }

        cleanUpMap := map[string][]azdiskv1beta2.AzVolumeAttachment{}
        for _, attachment := range attachments.Items {
                if shouldCleanUp(attachment, role) {
                        cleanUpMap[attachment.Spec.VolumeName] = append(cleanUpMap[attachment.Spec.VolumeName], attachment)
                }
        }

        for volumeName, cleanUps := range cleanUpMap {
                volumeName := volumeName
                c.addToOperationQueue(ctx,
                        volumeName,
                        caller,
                        func(ctx context.Context) error {
                                return c.cleanUpAzVolumeAttachments(ctx, cleanUps, cleanupMode, caller)
                        },
                        false)
        }
        return attachments.Items, nil
}

func (c *SharedState) cleanUpAzVolumeAttachments(ctx context.Context, attachments []azdiskv1beta2.AzVolumeAttachment, cleanUpMode attachmentCleanUpMode, caller operationRequester) error {
        var err error

        for _, attachment := range attachments {
                patched := attachment.DeepCopy()

                if attachment.Spec.RequestedRole == azdiskv1beta2.PrimaryRole {
                        if cleanUpMode == cleanUpAttachment && !volumeDetachRequested(patched) {
                                markDetachRequest(patched, caller)
                        } else if deleteRequested, _ := objectDeletionRequested(&attachment); !deleteRequested {
                                // if primary azvolumeattachments are being cleaned up for driver uninstall, issue a DELETE call and continue
                                // note that this DELETE request will remove AzVolumeAttachment CRI without detaching the volume from node
                                if err = c.cachedClient.Delete(ctx, patched); err != nil {
                                        return err
                                }
                        }
                } else {
                        // replica mount should always be detached in cleanup regardless to the cleanup mode
                        if !volumeDetachRequested(patched) {
                                markDetachRequest(patched, caller)
                        }

                        // append cleanup annotation to prevent replica recreations except for when the clean up was triggered by node controller due to node failure.
                        if caller != node && !metav1.HasAnnotation(patched.ObjectMeta, consts.CleanUpAnnotation) {
                                markCleanUp(patched, caller)
                        }
                }

                if !reflect.DeepEqual(attachment.Status, patched.Status) {
                        if err = c.cachedClient.Status().Patch(ctx, patched, client.MergeFrom(&attachment)); err != nil && apiErrors.IsNotFound(err) {
                                err = status.Errorf(codes.Internal, "failed to patch AzVolumeAttachment (%s)", attachment.Name)
                                return err
                        }
                }
        }
        return nil
}

func (c *SharedState) createReplicaRequestsQueue() {
        c.priorityReplicaRequestsQueue = &VolumeReplicaRequestsPriorityQueue{}
        c.priorityReplicaRequestsQueue.queue = cache.NewHeap(
                func(obj interface{}) (string, error) {
                        return obj.(*ReplicaRequest).VolumeName, nil
                },
                func(left, right interface{}) bool {
                        return left.(*ReplicaRequest).Priority > right.(*ReplicaRequest).Priority
                })
}

// Removes replica requests from the priority queue and adds to operation queue.
func (c *SharedState) tryCreateFailedReplicas(ctx context.Context, requester operationRequester) {
        if atomic.SwapInt32(&c.processingReplicaRequestQueue, 1) == 0 {
                ctx, w := workflow.New(ctx)
                defer w.Finish(nil)
                requests := c.priorityReplicaRequestsQueue.DrainQueue()
                for i := 0; i < len(requests); i++ {
                        replicaRequest := requests[i]
                        c.addToOperationQueue(ctx,
                                replicaRequest.VolumeName,
                                requester,
                                func(ctx context.Context) error {
                                        return c.manageReplicas(ctx, replicaRequest.VolumeName)
                                },
                                false,
                        )
                }
                atomic.StoreInt32(&c.processingReplicaRequestQueue, 0)
        }
}

func (c *SharedState) garbageCollectReplicas(ctx context.Context, volumeName string, requester operationRequester) {
        c.addToOperationQueue(
                ctx,
                volumeName,
                replica,
                func(ctx context.Context) error {
                        if _, err := c.cleanUpAzVolumeAttachmentByVolume(ctx, volumeName, requester, azureutils.ReplicaOnly, cleanUpAttachment, deleteOnly); err != nil {
                                return err
                        }
                        c.addToGcExclusionList(volumeName, requester)
                        c.removeGarbageCollection(volumeName)
                        c.unmarkVolumeVisited(volumeName)
                        return nil
                },
                true,
        )
}

func (c *SharedState) removeGarbageCollection(volumeName string) {
        v, ok := c.cleanUpMap.LoadAndDelete(volumeName)
        if ok {
                cancelFunc := v.(context.CancelFunc)
                cancelFunc()
        }
        // if there is any garbage collection enqueued in operation queue, remove it
        c.dequeueGarbageCollection(volumeName)
}

func (c *SharedState) manageReplicas(ctx context.Context, volumeName string) error {
        var err error
        ctx, w := workflow.New(ctx)
        defer func() { w.Finish(err) }()

        var azVolume *azdiskv1beta2.AzVolume
        azVolume, err = azureutils.GetAzVolume(ctx, c.cachedClient, c.azClient, volumeName, c.config.ObjectNamespace, true)

        // in case the volume attachment succeeds or terminally errors out
        c.unpersistAttachmentFailure(volumeName) // attempt un-persisting a previous failure (if the event refresher is active)

        if apiErrors.IsNotFound(err) {
                w.Logger().V(5).Info("Volume no longer exists. Aborting manage replica operation")
                return nil
        } else if err != nil {
                w.Logger().Error(err, "failed to get AzVolume")
                return err
        }

        // replica management should not be executed or retried if AzVolume is scheduled for a deletion or not created.
        deleteRequested, _ := objectDeletionRequested(azVolume)
        if !isCreated(azVolume) || deleteRequested {
                w.Logger().Errorf(errors.New("no valid azVolume"), "azVolume (%s) is scheduled for deletion or has no underlying volume object", azVolume.Name)
                return nil
        }

        currentReplicaCount, err := c.countValidReplicasForVolume(ctx, volumeName)
        if err != nil {
                return err
        }

        desiredReplicaCount := azVolume.Spec.MaxMountReplicaCount
        w.Logger().Infof("Control number of replicas for volume (%s): desired=%d, current:%d", azVolume.Spec.VolumeName, desiredReplicaCount, currentReplicaCount)

        if desiredReplicaCount > currentReplicaCount {
                w.Logger().Infof("Need %d more replicas for volume (%s)", desiredReplicaCount-currentReplicaCount, azVolume.Spec.VolumeName)
                if azVolume.Status.Detail == nil || azVolume.Status.State == azdiskv1beta2.VolumeDeleting || azVolume.Status.State == azdiskv1beta2.VolumeDeleted {
                        // underlying volume does not exist, so volume attachment cannot be made
                        return nil
                }
                if err = c.createReplicas(ctx, desiredReplicaCount-currentReplicaCount, azVolume.Name, azVolume.Status.Detail.VolumeID, azVolume.Spec.Parameters); err != nil {
                        w.Logger().Errorf(err, "failed to create %d replicas for volume (%s): %v", desiredReplicaCount-currentReplicaCount, azVolume.Spec.VolumeName, err)
                        return err
                }
        }
        return nil
}

// Count the number of replica attachments that aren't scheduled for deletion for a given volume
func (c *SharedState) countValidReplicasForVolume(ctx context.Context, volumeName string) (int, error) {
        w, _ := workflow.GetWorkflowFromContext(ctx)
        validReplicaCount := 0

        azVolumeAttachments, err := azureutils.GetAzVolumeAttachmentsForVolume(ctx, c.cachedClient, volumeName, azureutils.ReplicaOnly)
        if err != nil {
                w.Logger().Errorf(err, "failed to list replica AzVolumeAttachments")
                return validReplicaCount, err
        }

        for _, azVolumeAttachment := range azVolumeAttachments {
                if deleteRequested, _ := objectDeletionRequested(&azVolumeAttachment); !deleteRequested {
                        validReplicaCount++
                }
        }
        return validReplicaCount, nil
}

func (c *SharedState) createReplicas(ctx context.Context, remainingReplicas int, volumeName, volumeID string, volumeContext map[string]string) error {
        var err error
        ctx, w := workflow.New(ctx)
        defer func() { w.Finish(err) }()

        // if volume is scheduled for clean up, skip replica creation
        if _, cleanUpScheduled := c.cleanUpMap.Load(volumeName); cleanUpScheduled {
                return nil
        }

        // get pods linked to the volume
        var pods []v1.Pod
        pods, err = c.getPodsFromVolume(ctx, c.cachedClient, volumeName)
        if err != nil {
                return err
        }

        // acquire per-pod lock to be released upon creation of replica AzVolumeAttachment CRIs
        for _, pod := range pods {
                podKey := getQualifiedName(pod.Namespace, pod.Name)
                v, _ := c.podLocks.LoadOrStore(podKey, &sync.Mutex{})
                podLock := v.(*sync.Mutex)
                podLock.Lock()
                defer podLock.Unlock()
        }

        var nodes []string
        nodes, err = c.getNodesForReplica(ctx, volumeName, pods)
        if err != nil {
                w.Logger().Errorf(err, "failed to get a list of nodes for replica attachment")
                return err
        }

        requiredReplicas := remainingReplicas
        for _, node := range nodes {
                if err = c.createReplicaAzVolumeAttachment(ctx, volumeID, node, volumeContext); err != nil {
                        w.Logger().Errorf(err, "failed to create replica AzVolumeAttachment for volume %s", volumeName)
                        // continue to try attachment with next node
                        continue
                }
                remainingReplicas--
                if remainingReplicas <= 0 {
                        // no more remainingReplicas, don't need to create replica AzVolumeAttachment
                        break
                }
        }

        if remainingReplicas > 0 {
                //no failed replica attachments, but there are still more replicas to reach MaxShares
                request := ReplicaRequest{VolumeName: volumeName, Priority: remainingReplicas}
                c.priorityReplicaRequestsQueue.Push(ctx, &request)
                message := fmt.Sprintf("Not enough suitable nodes to attach %d of %d replica mount(s) for volume %s", remainingReplicas, requiredReplicas, volumeName)
                podCopies := make([]runtime.Object, len(pods))
                for i, pod := range pods {
                        podCopies[i] = pod.DeepCopyObject()
                }
                timestamp := time.Now()
                for _, podCopy := range podCopies {
                        c.eventRecorder.Eventf(podCopy, v1.EventTypeWarning, consts.ReplicaAttachmentFailedEvent, message)
                }
                message += timestamp.UTC().Format(" [0102 15:04:05") // append original timestamp
                c.persistAttachmentFailure(volumeName, message, podCopies, timestamp)
        }
        return nil
}

func (c *SharedState) persistAttachmentFailure(volumeName string, message string, podCopies []runtime.Object, timestamp time.Time) {
        c.eventsToPersistQueue <- ReplicaAttachmentFailureInfo{volumeName, message, podCopies, timestamp}
}

// Only requests an event un-persist if the event refresher is active
func (c *SharedState) unpersistAttachmentFailure(volumeName string) {
        c.eventsToUnpersistQueue <- volumeName
}

func (c *SharedState) createEventQueues() {
        c.eventsToPersistQueue = make(chan ReplicaAttachmentFailureInfo, c.config.ControllerConfig.WorkerThreads)
        c.eventsToUnpersistQueue = make(chan string, c.config.ControllerConfig.WorkerThreads*2)
        go c._eventRefresherRoutine()
}

func (c *SharedState) _eventRefresherRoutine() {
        type eventInfo struct {
                message   string
                timestamp time.Time
                objects   []runtime.Object
        }

        var eventTTL = time.Duration(c.config.ControllerConfig.EventTTLInSec) * time.Second
        eventMap := map[string]*circularLinkedListNode[eventInfo]{}

        var events circularLinkedList[eventInfo]
        var lastTime time.Time
        var delay time.Duration            // how long the alarm was last set to wait for
        expLatency := eventOverlapVariance // we request the alarm to wake us up earlier by this amount, to negate latency due to other operations and timer imprecision
        alarm := time.NewTimer(math.MaxInt64)
        if !alarm.Stop() {
                <-alarm.C
        }

        for {
                select {
                case newFailureInfo := <-c.eventsToPersistQueue:
                        // add the new failure
                        newEvent := &circularLinkedListNode[eventInfo]{
                                curr: eventInfo{
                                        message:   newFailureInfo.message,
                                        objects:   newFailureInfo.pods,
                                        timestamp: newFailureInfo.timestamp.Add(eventTTL - eventOverlapDuration),
                                },
                        }

                        if events.isEmpty() {
                                lastTime = time.Now()
                                delay = newEvent.curr.timestamp.Sub(lastTime) - expLatency
                                alarm.Reset(delay)
                        }

                        events.add(newEvent)

                        eventMap[newFailureInfo.volumeName].tryRemove()
                        eventMap[newFailureInfo.volumeName] = newEvent
                case newVolumeName := <-c.eventsToUnpersistQueue:
                        oldEvent := eventMap[newVolumeName]
                        if oldEvent != nil {
                                delete(eventMap, newVolumeName)
                                if oldEvent.next == oldEvent {
                                        // no more events to refresh!
                                        events.clear()
                                } else {
                                        oldEvent.remove()
                                }
                        }
                case <-alarm.C:
                        currTime := time.Now()
                        currLatency := expLatency + currTime.Sub(lastTime) - delay
                        expLatency = (expLatency*3 + currLatency) / 4 // update the expected latency with a 25% expontially-weighted moving average
                        lastTime = currTime
                        for {
                                for _, object := range events.curr.objects { // can we keep this as is? or do we need to check / keep track of whether each pod is still relevant
                                        c.eventRecorder.Event(object, v1.EventTypeWarning, consts.ReplicaAttachmentFailedEvent, events.curr.message)
                                }
                                events.curr.timestamp = events.curr.timestamp.Add(eventTTL)
                                events.next()
                                delay = time.Until(events.curr.timestamp) - expLatency
                                if delay >= eventOverlapVariance {
                                        break
                                }
                                // events which are really close together will be processed at the same, preventing the overuse of sleep()
                                // event refresher's responsiviness may be more variable, so we'll increase its channel buffers as needed
                        }
                        alarm.Reset(delay)
                }
        }
}

func (c *SharedState) getNodesForReplica(ctx context.Context, volumeName string, pods []v1.Pod) ([]string, error) {
        var err error
        ctx, w := workflow.New(ctx)
        defer func() { w.Finish(err) }()

        if len(pods) == 0 {
                pods, err = c.getPodsFromVolume(ctx, c.cachedClient, volumeName)
                if err != nil {
                        return nil, err
                }
        }

        var volumes []string
        volumes, err = c.getVolumesForPodObjs(ctx, pods)
        if err != nil {
                return nil, err
        }

        var nodes []string
        nodes, err = c.getRankedNodesForReplicaAttachments(ctx, volumes, pods)
        if err != nil {
                return nil, err
        }

        var replicaNodes []string
        replicaNodes, err = c.getNodesWithReplica(ctx, volumeName)
        if err != nil {
                return nil, err
        }

        skipSet := map[string]bool{}
        for _, replicaNode := range replicaNodes {
                skipSet[replicaNode] = true
        }

        filtered := []string{}
        for _, node := range nodes {
                if skipSet[node] {
                        continue
                }
                // if the node has no capacity for disk attachment, we should skip it
                remainingCapacity, nodeExists := c.availableAttachmentsMap.Load(node)
                if !nodeExists || remainingCapacity == nil || remainingCapacity.(*atomic.Int32).Load() <= int32(0) {
                        w.Logger().V(5).Infof("skip node(%s) because it has no capacity for disk attachment", node)
                        continue
                }
                filtered = append(filtered, node)
        }

        return filtered, nil
}

func (c *SharedState) createAzVolumeFromPv(ctx context.Context, pv v1.PersistentVolume, annotations map[string]string) error {
        var err error
        ctx, w := workflow.New(ctx)
        defer func() { w.Finish(err) }()

        var desiredAzVolume *azdiskv1beta2.AzVolume
        requiredBytes, _ := pv.Spec.Capacity.Storage().AsInt64()
        volumeCapability := c.getVolumeCapabilityFromPv(&pv)

        // translate intree pv to csi pv to convert them into AzVolume resource
        if utilfeature.DefaultFeatureGate.Enabled(features.CSIMigration) &&
                utilfeature.DefaultFeatureGate.Enabled(features.CSIMigrationAzureDisk) &&
                pv.Spec.AzureDisk != nil {
                var transPV *v1.PersistentVolume
                // if an error occurs while translating, it's unrecoverable, so return no error
                if transPV, err = c.translateInTreePVToCSI(&pv); err != nil {
                        return err
                }
                pv = *transPV
        }

        // skip if PV is not managed by azuredisk driver
        if pv.Spec.CSI == nil || pv.Spec.CSI.Driver != c.config.DriverName {
                return nil
        }

        // create AzVolume CRI for CSI Volume Source
        desiredAzVolume, err = c.createAzVolumeFromCSISource(pv.Spec.CSI)
        if err != nil {
                return err
        }

        if pv.Spec.NodeAffinity != nil && pv.Spec.NodeAffinity.Required != nil {
                desiredAzVolume.Status.Detail.AccessibleTopology = azureutils.GetTopologyFromNodeSelector(*pv.Spec.NodeAffinity.Required, c.topologyKey)
        }
        if azureutils.IsMultiNodePersistentVolume(pv) {
                desiredAzVolume.Spec.MaxMountReplicaCount = 0
        }

        // if it's an inline volume, no pv label or pvc label should be added
        if !azureutils.MapContains(annotations, consts.InlineVolumeAnnotation) {
                desiredAzVolume.Labels = azureutils.AddToMap(desiredAzVolume.Labels, consts.PvNameLabel, pv.Name)

                if pv.Spec.ClaimRef != nil {
                        desiredAzVolume.Labels = azureutils.AddToMap(desiredAzVolume.Labels, consts.PvcNameLabel, pv.Spec.ClaimRef.Name)
                        desiredAzVolume.Labels = azureutils.AddToMap(desiredAzVolume.Labels, consts.PvcNamespaceLabel, pv.Spec.ClaimRef.Namespace)
                }
        }

        desiredAzVolume.Spec.VolumeCapability = volumeCapability
        desiredAzVolume.Spec.PersistentVolume = pv.Name
        desiredAzVolume.Spec.CapacityRange = &azdiskv1beta2.CapacityRange{RequiredBytes: requiredBytes}

        desiredAzVolume.Status.Detail.CapacityBytes = requiredBytes

        for k, v := range annotations {
                desiredAzVolume.Status.Annotations = azureutils.AddToMap(desiredAzVolume.Status.Annotations, k, v)
        }

        w.AddDetailToLogger(consts.PvNameKey, pv.Name, consts.VolumeNameLabel, desiredAzVolume.Name)

        if err = c.createAzVolume(ctx, desiredAzVolume); err != nil {
                err = status.Errorf(codes.Internal, "failed to create AzVolume (%s) for PV (%s): %v", desiredAzVolume.Name, pv.Name, err)
                return err
        }
        return nil
}

func (c *SharedState) getVolumeCapabilityFromPv(pv *v1.PersistentVolume) []azdiskv1beta2.VolumeCapability {
        volCaps := []azdiskv1beta2.VolumeCapability{}

        for _, accessMode := range pv.Spec.AccessModes {
                volCap := azdiskv1beta2.VolumeCapability{}
                // default to Mount
                if pv.Spec.VolumeMode != nil && *pv.Spec.VolumeMode == v1.PersistentVolumeBlock {
                        volCap.AccessType = azdiskv1beta2.VolumeCapabilityAccessBlock
                }
                switch accessMode {
                case v1.ReadWriteOnce:
                        volCap.AccessMode = azdiskv1beta2.VolumeCapabilityAccessModeSingleNodeSingleWriter
                case v1.ReadWriteMany:
                        volCap.AccessMode = azdiskv1beta2.VolumeCapabilityAccessModeMultiNodeMultiWriter
                case v1.ReadOnlyMany:
                        volCap.AccessMode = azdiskv1beta2.VolumeCapabilityAccessModeMultiNodeReaderOnly
                default:
                        volCap.AccessMode = azdiskv1beta2.VolumeCapabilityAccessModeUnknown
                }
                volCaps = append(volCaps, volCap)
        }
        return volCaps
}

func (c *SharedState) createAzVolumeFromCSISource(source *v1.CSIPersistentVolumeSource) (*azdiskv1beta2.AzVolume, error) {
        diskName, err := azureutils.GetDiskName(source.VolumeHandle)
        if err != nil {
                return nil, fmt.Errorf("failed to extract diskName from volume handle (%s): %v", source.VolumeHandle, err)
        }

        _, maxMountReplicaCount := azureutils.GetMaxSharesAndMaxMountReplicaCount(source.VolumeAttributes, false)

        diskParameters, _ := azureutils.ParseDiskParameters(source.VolumeAttributes, azureutils.IgnoreUnknown)
        volumeParams := diskParameters.VolumeContext

        azVolumeName := strings.ToLower(diskName)

        azVolume := azdiskv1beta2.AzVolume{
                ObjectMeta: metav1.ObjectMeta{
                        Name:       azVolumeName,
                        Finalizers: []string{consts.AzVolumeFinalizer},
                },
                Spec: azdiskv1beta2.AzVolumeSpec{
                        MaxMountReplicaCount: maxMountReplicaCount,
                        Parameters:           volumeParams,
                        VolumeName:           diskName,
                },
                Status: azdiskv1beta2.AzVolumeStatus{
                        Detail: &azdiskv1beta2.AzVolumeStatusDetail{
                                VolumeID:      source.VolumeHandle,
                                VolumeContext: source.VolumeAttributes,
                        },
                        State: azdiskv1beta2.VolumeCreated,
                },
        }
        azureutils.AnnotateAPIVersion(&azVolume)

        return &azVolume, nil
}

func (c *SharedState) createAzVolume(ctx context.Context, desiredAzVolume *azdiskv1beta2.AzVolume) error {
        w, _ := workflow.GetWorkflowFromContext(ctx)

        var err error
        var azVolume *azdiskv1beta2.AzVolume
        var updated *azdiskv1beta2.AzVolume

        azVolume, err = c.azClient.DiskV1beta2().AzVolumes(c.config.ObjectNamespace).Get(ctx, desiredAzVolume.Name, metav1.GetOptions{})
        if err != nil {
                if apiErrors.IsNotFound(err) {
                        azVolume, err = c.azClient.DiskV1beta2().AzVolumes(c.config.ObjectNamespace).Create(ctx, desiredAzVolume, metav1.CreateOptions{})
                        if err != nil {
                                return err
                        }
                        updated = azVolume.DeepCopy()
                        updated.Status = desiredAzVolume.Status
                } else {
                        return err
                }
        }

        if apiVersion, ok := azureutils.GetFromMap(azVolume.Annotations, consts.APIVersion); !ok || apiVersion != azdiskv1beta2.APIVersion {
                w.Logger().Infof("Found AzVolume (%s) with older api version. Converting to apiVersion(%s)", azVolume.Name, azdiskv1beta2.APIVersion)

                azVolume.Spec.PersistentVolume = desiredAzVolume.Spec.PersistentVolume

                for k, v := range desiredAzVolume.Labels {
                        azVolume.Labels = azureutils.AddToMap(azVolume.Labels, k, v)
                }

                // for now, we don't empty the meta annotations after migrating them to status annotation for safety.
                // note that this will leave some remnant garbage entries in meta annotations
                var statusAnnotation []string
                for k, v := range azVolume.Annotations {
                        statusAnnotation = append(statusAnnotation, k, v)
                }

                for k, v := range desiredAzVolume.Annotations {
                        azVolume.Annotations = azureutils.AddToMap(azVolume.Annotations, k, v)
                }

                azVolume, err = c.azClient.DiskV1beta2().AzVolumes(c.config.ObjectNamespace).Update(ctx, azVolume, metav1.UpdateOptions{})
                if err != nil {
                        return err
                }
                updated = azVolume.DeepCopy()
                updated.Status.Annotations = azureutils.AddToMap(updated.Status.Annotations, statusAnnotation...)
        }

        if updated != nil {
                if _, err = azureutils.UpdateCRIWithRetry(ctx, nil, c.cachedClient, c.azClient, azVolume, func(obj client.Object) error {
                        azvolume := obj.(*azdiskv1beta2.AzVolume)
                        azvolume.Status = updated.Status
                        return nil
                }, consts.NormalUpdateMaxNetRetry, azureutils.UpdateCRIStatus); err != nil {
                        return err
                }
        }

        // if AzVolume CRI successfully recreated, also recreate the operation queue for the volume
        c.createOperationQueue(desiredAzVolume.Name)
        return nil
}

func (c *SharedState) translateInTreePVToCSI(pv *v1.PersistentVolume) (*v1.PersistentVolume, error) {
        var err error
        // translate intree pv to csi pv to convert them into AzVolume resource
        if utilfeature.DefaultFeatureGate.Enabled(features.CSIMigration) &&
                utilfeature.DefaultFeatureGate.Enabled(features.CSIMigrationAzureDisk) &&
                pv.Spec.AzureDisk != nil {
                // if an error occurs while translating, it's unrecoverable, so return no error
                if pv, err = c.azureDiskCSITranslator.TranslateInTreePVToCSI(pv); err != nil {
                } else if pv == nil {
                        err = status.Errorf(codes.Internal, "unexpected failure in translating inline volume to csi")
                }

        }
        return pv, err
}

// waitForVolumeAttachmentNAme waits for the VolumeAttachment name to be updated in the azVolumeAttachmentVaMap by the volumeattachment controller
func (c *SharedState) waitForVolumeAttachmentName(ctx context.Context, azVolumeAttachment *azdiskv1beta2.AzVolumeAttachment) (string, error) {
        var vaName string
        err := wait.PollImmediateUntilWithContext(ctx, consts.DefaultPollingRate, func(ctx context.Context) (bool, error) {
                val, exists := c.azVolumeAttachmentToVaMap.Load(azVolumeAttachment.Name)
                if exists {
                        vaName = val.(string)
                }
                return exists, nil
        })
        return vaName, err
}

// Returns set of node names that qualify pod affinity term and set of node names with qualifying replica attachments.
func (c *SharedState) getQualifiedNodesForPodAffinityTerm(ctx context.Context, nodes []v1.Node, podNamespace string, affinityTerm v1.PodAffinityTerm) (podNodes, replicaNodes set) {
        var err error
        w, _ := workflow.GetWorkflowFromContext(ctx)
        candidateNodes := set{}
        for _, node := range nodes {
                candidateNodes.add(node.Name)
        }
        podNodes = set{}
        replicaNodes = set{}

        var podSelector labels.Selector
        podSelector, err = metav1.LabelSelectorAsSelector(affinityTerm.LabelSelector)
        // if failed to convert pod affinity label selector to selector, log error and skip
        if err != nil {
                w.Logger().Errorf(err, "failed to convert pod affinity (%v) to selector", affinityTerm.LabelSelector)
        }

        nsList := &v1.NamespaceList{}
        if affinityTerm.NamespaceSelector != nil {
                nsSelector, err := metav1.LabelSelectorAsSelector(affinityTerm.NamespaceSelector)
                // if failed to convert pod affinity label selector to selector, log error and skip
                if err != nil {
                        w.Logger().Errorf(err, "failed to convert pod affinity (%v) to selector", affinityTerm.LabelSelector)
                } else {
                        if err = c.cachedClient.List(ctx, nsList, &client.ListOptions{LabelSelector: nsSelector}); err != nil {
                                w.Logger().Errorf(err, "failed to list namespaces with selector (%v)", nsSelector)
                                return
                        }

                }
        }

        namespaces := affinityTerm.Namespaces
        for _, ns := range nsList.Items {
                namespaces = append(namespaces, ns.Name)
        }

        pods := []v1.Pod{}
        if len(namespaces) > 0 {
                for _, namespace := range namespaces {
                        podList := &v1.PodList{}
                        if err = c.cachedClient.List(ctx, podList, &client.ListOptions{LabelSelector: podSelector, Namespace: namespace}); err != nil {
                                w.Logger().Errorf(err, "failed to retrieve pod list: %v", err)
                                pods = append(pods, podList.Items...)
                        }
                }
        } else {
                podList := &v1.PodList{}
                if err = c.cachedClient.List(ctx, podList, &client.ListOptions{LabelSelector: podSelector, Namespace: podNamespace}); err != nil {
                        w.Logger().Errorf(err, "failed to retrieve pod list: %v", err)
                }
                pods = podList.Items
        }

        // get replica nodes for pods that satisfy pod label selector
        replicaNodes = c.getReplicaNodesForPods(ctx, pods)
        for replicaNode := range replicaNodes {
                if !candidateNodes.has(replicaNode) {
                        replicaNodes.remove(replicaNode)
                }
        }

        // get nodes with pod that share the same topology as pods satisfying pod label selector
        for _, pod := range pods {
                podNodes.add(pod.Spec.NodeName)
        }

        var podNodeObjs []v1.Node
        for node := range podNodes {
                var nodeObj v1.Node
                if err = c.cachedClient.Get(ctx, types.NamespacedName{Name: node.(string)}, &nodeObj); err != nil {
                        w.Logger().Errorf(err, "failed to get node (%s)", node.(string))
                        continue
                }
                podNodeObjs = append(podNodeObjs, nodeObj)
        }

        topologyLabel := c.getNodesTopologySelector(ctx, podNodeObjs, affinityTerm.TopologyKey)
        for _, node := range nodes {
                if topologyLabel != nil && topologyLabel.Matches(labels.Set(node.Labels)) {
                        podNodes.add(node.Name)
                }
        }
        return
}

// Returns set of node names where replica mounts of given pod can be found
func (c *SharedState) getReplicaNodesForPods(ctx context.Context, pods []v1.Pod) (replicaNodes set) {
        // add nodes, to which replica attachments of matching pods' volumes are attached, to replicaNodes
        replicaNodes = set{}
        if volumes, err := c.getVolumesForPodObjs(ctx, pods); err == nil {
                for _, volume := range volumes {
                        attachments, err := azureutils.GetAzVolumeAttachmentsForVolume(ctx, c.cachedClient, volume, azureutils.ReplicaOnly)
                        if err != nil {
                                continue
                        }
                        for _, attachment := range attachments {
                                if deleteRequested, _ := objectDeletionRequested(&attachment); !deleteRequested {
                                        node := attachment.Spec.NodeName
                                        replicaNodes.add(node)
                                }
                        }
                }
        }

        return replicaNodes
}

// Returns a label selector corresponding to a list of nodes and a topology key (aka label key)
func (c *SharedState) getNodesTopologySelector(ctx context.Context, nodes []v1.Node, topologyKey string) labels.Selector {
        w, _ := workflow.GetWorkflowFromContext(ctx)
        if len(nodes) == 0 {
                return nil
        }

        topologyValues := set{}
        for _, node := range nodes {
                nodeLabels := node.GetLabels()
                if topologyValue, exists := nodeLabels[topologyKey]; exists {
                        topologyValues.add(topologyValue)
                } else {
                        w.Logger().V(5).Infof("node (%s) doesn't have label value for topologyKey (%s)", node.Name, topologyKey)
                }
        }

        topologySelector := labels.NewSelector()
        topologyRequirement, err := azureutils.CreateLabelRequirements(topologyKey, selection.In, topologyValues.toStringSlice()...)
        // if failed to create label requirement, log error and return empty selector
        if err != nil {
                w.Logger().Errorf(err, "failed to create label requirement for topologyKey (%s)", topologyKey)
        } else {
                topologySelector = topologySelector.Add(*topologyRequirement)
        }
        return topologySelector
}

// addNodeToAvailableAttachmentsMap returns true if the node is added to or already in the availableAttachmentsMap, and false otherwise.
func (c *SharedState) addNodeToAvailableAttachmentsMap(ctx context.Context, nodeName string, nodeLables map[string]string) bool {
        if _, ok := c.availableAttachmentsMap.Load(nodeName); !ok {
                capacity, err := azureutils.GetNodeRemainingDiskCountActual(ctx, c.cachedClient, nodeName)
                if err != nil {
                        klog.Errorf("Failed to get node(%s) remaining disk count with error: %v", nodeName, err)
                        // store the maximum capacity if an entry for the node doesn't exist.
                        capacity, err = azureutils.GetNodeMaxDiskCountWithLabels(nodeLables)
                        if err != nil {
                                klog.Errorf("Failed to add node(%s) in availableAttachmentsMap, because get capacity of available attachments is failed with error: %v", nodeName, err)
                                return false
                        }
                }
                var count atomic.Int32
                count.Store(int32(capacity))
                klog.Infof("Added node(%s) to availableAttachmentsMap with capacity: %d", nodeName, capacity)
                c.availableAttachmentsMap.LoadOrStore(nodeName, &count)
        }
        return true
}

func (c *SharedState) deleteNodeFromAvailableAttachmentsMap(ctx context.Context, node string) {
        klog.Infof("Deleted node(%s) from availableAttachmentsMap", node)
        c.availableAttachmentsMap.Delete(node)
}

func (c *SharedState) decrementNodeCapacity(ctx context.Context, node string) bool {
        remainingCapacity, nodeExists := c.availableAttachmentsMap.Load(node)
        if nodeExists && remainingCapacity != nil {
                for {
                        currentCapacity := remainingCapacity.(*atomic.Int32).Load()
                        if currentCapacity == int32(0) {
                                klog.Errorf("Failed to decrement disk capacity for node(%s) because no remaining capacity", node)
                                return false
                        }
                        if remainingCapacity.(*atomic.Int32).CompareAndSwap(currentCapacity, currentCapacity-1) {
                                return true
                        }
                }
        }

        klog.Errorf("Failed to decrement disk capacity because node(%s) not found", node)
        return false
}

func (c *SharedState) incrementNodeCapacity(ctx context.Context, node string) bool {
        remainingCapacity, nodeExists := c.availableAttachmentsMap.Load(node)
        if nodeExists && remainingCapacity != nil {
                remainingCapacity.(*atomic.Int32).Add(1)
                return true
        }

        klog.Errorf("Failed to increment disk capacity because node(%s) not found", node)
        return false
}

kubernetes-sigs / azuredisk-csi-driver / 6188855463

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous