1
2
3
4
5
6
7
8
9
10
11
12
13
14
// NewAttachDetachController returns a new instance of AttachDetachController.
func NewAttachDetachController(
	kubeClient clientset.Interface,
	csiClient csiclient.Interface,
	podInformer coreinformers.PodInformer,
	nodeInformer coreinformers.NodeInformer,
	pvcInformer coreinformers.PersistentVolumeClaimInformer,
	pvInformer coreinformers.PersistentVolumeInformer,
	cloud cloudprovider.Interface,
	plugins []volume.VolumePlugin,
	prober volume.DynamicPluginProber,
	disableReconciliationSync bool,
	reconcilerSyncDuration time.Duration,
	timerConfig TimerConfig) (AttachDetachController, error) {

1
2
3
4
5
6
7
8
eventBroadcaster := record.NewBroadcaster()
eventBroadcaster.StartLogging(klog.Infof)
eventBroadcaster.StartRecordingToSink(&v1core.EventSinkImpl{Interface: kubeClient.CoreV1().Events("")})
recorder := eventBroadcaster.NewRecorder(scheme.Scheme, v1.EventSource{Component: "attachdetach-controller"})
blkutil := volumepathhandler.NewBlockVolumePathHandler()

adc.desiredStateOfWorld = cache.NewDesiredStateOfWorld(&adc.volumePluginMgr)
adc.actualStateOfWorld = cache.NewActualStateOfWorld(&adc.volumePluginMgr)

1
2
3
4
5
6
7
adc.attacherDetacher =
	operationexecutor.NewOperationExecutor(operationexecutor.NewOperationGenerator(
		kubeClient,
		&adc.volumePluginMgr,
		recorder,
		false, // flag for experimental binary check for volume mount
		blkutil))

1
2
3
4
5
6
7
8
9
10
11
// Default these to values in options
adc.reconciler = reconciler.NewReconciler(
	timerConfig.ReconcilerLoopPeriod,
	timerConfig.ReconcilerMaxWaitForUnmountDuration,
	reconcilerSyncDuration,
	disableReconciliationSync,
	adc.desiredStateOfWorld,
	adc.actualStateOfWorld,
	adc.attacherDetacher,
	adc.nodeStatusUpdater,
	recorder)

1
2
3
4
5
6
7
8
adc.desiredStateOfWorldPopulator = populator.NewDesiredStateOfWorldPopulator(
	timerConfig.DesiredStateOfWorldPopulatorLoopSleepPeriod,
	timerConfig.DesiredStateOfWorldPopulatorListPodsRetryDuration,
	podInformer.Lister(),
	adc.desiredStateOfWorld,
	&adc.volumePluginMgr,
	pvcInformer.Lister(),
	pvInformer.Lister())

1
2
3
4
5
podInformer.Informer().AddEventHandler(kcache.ResourceEventHandlerFuncs{
	AddFunc:    adc.podAdd,
	UpdateFunc: adc.podUpdate,
	DeleteFunc: adc.podDelete,
})

1
2
3
4
5
nodeInformer.Informer().AddEventHandler(kcache.ResourceEventHandlerFuncs{
	AddFunc:    adc.nodeAdd,
	UpdateFunc: adc.nodeUpdate,
	DeleteFunc: adc.nodeDelete,
})

1
2
3
4
5
6
7
8
pvcInformer.Informer().AddEventHandler(kcache.ResourceEventHandlerFuncs{
	AddFunc: func(obj interface{}) {
		adc.enqueuePVC(obj)
	},
	UpdateFunc: func(old, new interface{}) {
		adc.enqueuePVC(new)
	},
})

1
2
3
4
5
6
7
8
9
10
func (adc *attachDetachController) Run(stopCh <-chan struct{}) {
	defer runtime.HandleCrash()
	defer adc.pvcQueue.ShutDown()

	klog.Infof("Starting attach detach controller")
	defer klog.Infof("Shutting down attach detach controller")

	if !controller.WaitForCacheSync("attach detach", stopCh, adc.podsSynced, adc.nodesSynced, adc.pvcsSynced, adc.pvsSynced) {
		return
	}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
func (adc *attachDetachController) populateActualStateOfWorld() error {
	klog.V(5).Infof("Populating ActualStateOfworld")
	nodes, err := adc.nodeLister.List(labels.Everything())
	if err != nil {
		return err
	}

	for _, node := range nodes {
		nodeName := types.NodeName(node.Name)
		for _, attachedVolume := range node.Status.VolumesAttached {
			uniqueName := attachedVolume.Name
			// The nil VolumeSpec is safe only in the case the volume is not in use by any pod.
			// In such a case it should be detached in the first reconciliation cycle and the
			// volume spec is not needed to detach a volume. If the volume is used by a pod, it
			// its spec can be: this would happen during in the populateDesiredStateOfWorld which
			// scans the pods and updates their volumes in the ActualStateOfWorld too.
			err = adc.actualStateOfWorld.MarkVolumeAsAttached(uniqueName, nil /* VolumeSpec */, nodeName, attachedVolume.DevicePath)
			if err != nil {
				klog.Errorf("Failed to mark the volume as attached: %v", err)
				continue
			}
			adc.processVolumesInUse(nodeName, node.Status.VolumesInUse)
			adc.addNodeToDswp(node, types.NodeName(node.Name))
		}
	}
	return nil
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
func (asw *actualStateOfWorld) AddVolumeNode(
	uniqueName v1.UniqueVolumeName, volumeSpec *volume.Spec, nodeName types.NodeName, devicePath string) (v1.UniqueVolumeName, error) {
	asw.Lock()
	defer asw.Unlock()

	var volumeName v1.UniqueVolumeName
	if volumeSpec != nil {
		attachableVolumePlugin, err := asw.volumePluginMgr.FindAttachablePluginBySpec(volumeSpec)
		if err != nil || attachableVolumePlugin == nil {
			return "", fmt.Errorf(
				"failed to get AttachablePlugin from volumeSpec for volume %q err=%v",
				volumeSpec.Name(),
				err)
		}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
volumeObj, volumeExists := asw.attachedVolumes[volumeName]
if !volumeExists {
	volumeObj = attachedVolume{
		volumeName:      volumeName,
		spec:            volumeSpec,
		nodesAttachedTo: make(map[types.NodeName]nodeAttachedTo),
		devicePath:      devicePath,
	}
} else {
	// If volume object already exists, it indicates that the information would be out of date.
	// Update the fields for volume object except the nodes attached to the volumes.
	volumeObj.devicePath = devicePath
	volumeObj.spec = volumeSpec
	klog.V(2).Infof("Volume %q is already added to attachedVolume list to node %q, update device path %q",
		volumeName,
		nodeName,
		devicePath)
}
asw.attachedVolumes[volumeName] = volumeObj

1
2
3
4
5
6
7
8
9
10
11
12
13
14
_, nodeExists := volumeObj.nodesAttachedTo[nodeName]
if !nodeExists {
	// Create object if it doesn't exist.
	volumeObj.nodesAttachedTo[nodeName] = nodeAttachedTo{
		nodeName:              nodeName,
		mountedByNode:         true, // Assume mounted, until proven otherwise
		mountedByNodeSetCount: 0,
		detachRequestedTime:   time.Time{},
	}
} else {
	klog.V(5).Infof("Volume %q is already added to attachedVolume list to the node %q",
		volumeName,
		nodeName)
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
// Add the volumeName to the node's volumesToReportAsAttached list
// This is an internal function and caller should acquire and release the lock
func (asw *actualStateOfWorld) addVolumeToReportAsAttached(
	volumeName v1.UniqueVolumeName, nodeName types.NodeName) {
	// In case the volume/node entry is no longer in attachedVolume list, skip the rest
	if _, _, err := asw.getNodeAndVolume(volumeName, nodeName); err != nil {
		klog.V(4).Infof("Volume %q is no longer attached to node %q", volumeName, nodeName)
		return
	}
	nodeToUpdate, nodeToUpdateExists := asw.nodesToUpdateStatusFor[nodeName]
	if !nodeToUpdateExists {
		// Create object if it doesn't exist
		nodeToUpdate = nodeToUpdateStatusFor{
			nodeName:                  nodeName,
			statusUpdateNeeded:        true,
			volumesToReportAsAttached: make(map[v1.UniqueVolumeName]v1.UniqueVolumeName),
		}
		asw.nodesToUpdateStatusFor[nodeName] = nodeToUpdate
		klog.V(4).Infof("Add new node %q to nodesToUpdateStatusFor", nodeName)
	}
	_, nodeToUpdateVolumeExists :=
		nodeToUpdate.volumesToReportAsAttached[volumeName]
	if !nodeToUpdateVolumeExists {
		nodeToUpdate.statusUpdateNeeded = true
		nodeToUpdate.volumesToReportAsAttached[volumeName] = volumeName
		asw.nodesToUpdateStatusFor[nodeName] = nodeToUpdate
		klog.V(4).Infof("Report volume %q as attached to node %q", volumeName, nodeName)
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
// processVolumesInUse processes the list of volumes marked as "in-use"
// according to the specified Node's Status.VolumesInUse and updates the
// corresponding volume in the actual state of the world to indicate that it is
// mounted.
func (adc *attachDetachController) processVolumesInUse(
	nodeName types.NodeName, volumesInUse []v1.UniqueVolumeName) {
	klog.V(4).Infof("processVolumesInUse for node %q", nodeName)
	for _, attachedVolume := range adc.actualStateOfWorld.GetAttachedVolumesForNode(nodeName) {
		mounted := false
		for _, volumeInUse := range volumesInUse {
			if attachedVolume.VolumeName == volumeInUse {
				mounted = true
				break
			}
		}
		err := adc.actualStateOfWorld.SetVolumeMountedByNode(attachedVolume.VolumeName, nodeName, mounted)
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
func (adc *attachDetachController) addNodeToDswp(node *v1.Node, nodeName types.NodeName) {
	if _, exists := node.Annotations[volumeutil.ControllerManagedAttachAnnotation]; exists {
		keepTerminatedPodVolumes := false

		if t, ok := node.Annotations[volumeutil.KeepTerminatedPodVolumesAnnotation]; ok {
			keepTerminatedPodVolumes = (t == "true")
		}

		// Node specifies annotation indicating it should be managed by attach
		// detach controller. Add it to desired state of world.
		adc.desiredStateOfWorld.AddNode(nodeName, keepTerminatedPodVolumes)
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
func (adc *attachDetachController) podAdd(obj interface{}) {
	pod, ok := obj.(*v1.Pod)
	if pod == nil || !ok {
		return
	}
	if pod.Spec.NodeName == "" {
		// Ignore pods without NodeName, indicating they are not scheduled.
		return
	}

	volumeActionFlag := util.DetermineVolumeAction(
		pod,
		adc.desiredStateOfWorld,
		true /* default volume action */)

	util.ProcessPodVolumes(pod, volumeActionFlag, /* addVolumes */
		adc.desiredStateOfWorld, &adc.volumePluginMgr, adc.pvcLister, adc.pvLister)
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
nodeName := types.NodeName(pod.Spec.NodeName)
if nodeName == "" {
	klog.V(10).Infof(
		"Skipping processing of pod %q/%q: it is not scheduled to a node.",
		pod.Namespace,
		pod.Name)
	return
} else if !desiredStateOfWorld.NodeExists(nodeName) {
	// If the node the pod is scheduled to does not exist in the desired
	// state of the world data structure, that indicates the node is not
	// yet managed by the controller. Therefore, ignore the pod.
	klog.V(4).Infof(
		"Skipping processing of pod %q/%q: it is scheduled to node %q which is not managed by the controller.",
		pod.Namespace,
		pod.Name,
		nodeName)
	return
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
func (rc *reconciler) Run(stopCh <-chan struct{}) {
	wait.Until(rc.reconciliationLoopFunc(), rc.loopPeriod, stopCh)
}

// reconciliationLoopFunc this can be disabled via cli option disableReconciliation.
// It periodically checks whether the attached volumes from actual state
// are still attached to the node and update the status if they are not.
func (rc *reconciler) reconciliationLoopFunc() func() {
	return func() {

		rc.reconcile()

		if rc.disableReconciliationSync {
			klog.V(5).Info("Skipping reconciling attached volumes still attached since it is disabled via the command line.")
		} else if rc.syncDuration < time.Second {
			klog.V(5).Info("Skipping reconciling attached volumes still attached since it is set to less than one second via the command line.")
		} else if time.Since(rc.timeOfLastSync) > rc.syncDuration {
			klog.V(5).Info("Starting reconciling attached volumes still attached")
			rc.sync()
		}
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
func (dswp *desiredStateOfWorldPopulator) Run(stopCh <-chan struct{}) {
	wait.Until(dswp.populatorLoopFunc(), dswp.loopSleepDuration, stopCh)
}

func (dswp *desiredStateOfWorldPopulator) populatorLoopFunc() func() {
	return func() {
		dswp.findAndRemoveDeletedPods()

		// findAndAddActivePods is called periodically, independently of the main
		// populator loop.
		if time.Since(dswp.timeOfLastListPods) < dswp.listPodsRetryDuration {
			klog.V(5).Infof(
				"Skipping findAndAddActivePods(). Not permitted until %v (listPodsRetryDuration %v).",
				dswp.timeOfLastListPods.Add(dswp.listPodsRetryDuration),
				dswp.listPodsRetryDuration)

			return
		}
		dswp.findAndAddActivePods()
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
informerPod, err := dswp.podLister.Pods(namespace).Get(name)
switch {
case errors.IsNotFound(err):
	// if we can't find the pod, we need to delete it below
case err != nil:
	klog.Errorf("podLister Get failed for pod %q (UID %q) with %v", dswPodKey, dswPodUID, err)
	continue
default:
	volumeActionFlag := util.DetermineVolumeAction(
		informerPod,
		dswp.desiredStateOfWorld,
		true /* default volume action */)

	if volumeActionFlag {
		informerPodUID := volutil.GetUniquePodName(informerPod)
		// Check whether the unique identifier of the pod from dsw matches the one retrieved from pod informer
		if informerPodUID == dswPodUID {
			klog.V(10).Infof("Verified pod %q (UID %q) from dsw exists in pod informer.", dswPodKey, dswPodUID)
			continue
		}
	}
}

// the pod from dsw does not exist in pod informer, or it does not match the unique identifier retrieved
// from the informer, delete it from dsw
klog.V(1).Infof("Removing pod %q (UID %q) from dsw because it does not exist in pod informer.", dswPodKey, dswPodUID)
dswp.desiredStateOfWorld.DeletePod(dswPodUID, dswPodToAdd.VolumeName, dswPodToAdd.NodeName)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
func (dswp *desiredStateOfWorldPopulator) findAndAddActivePods() {
	pods, err := dswp.podLister.List(labels.Everything())
	if err != nil {
		klog.Errorf("podLister List failed: %v", err)
		return
	}
	dswp.timeOfLastListPods = time.Now()

	for _, pod := range pods {
		if volutil.IsPodTerminated(pod, pod.Status) {
			// Do not add volumes for terminated pods
			continue
		}
		util.ProcessPodVolumes(pod, true,
			dswp.desiredStateOfWorld, dswp.volumePluginMgr, dswp.pvcLister, dswp.pvLister)

	}

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
// pvcWorker processes items from pvcQueue
func (adc *attachDetachController) pvcWorker() {
	for adc.processNextItem() {
	}
}

func (adc *attachDetachController) processNextItem() bool {
	keyObj, shutdown := adc.pvcQueue.Get()
	if shutdown {
		return false
	}
	defer adc.pvcQueue.Done(keyObj)

	if err := adc.syncPVCByKey(keyObj.(string)); err != nil {
		// Rather than wait for a full resync, re-add the key to the
		// queue to be processed.
		adc.pvcQueue.AddRateLimited(keyObj)
		runtime.HandleError(fmt.Errorf("Failed to sync pvc %q, will retry again: %v", keyObj.(string), err))
		return true
	}

	// Finally, if no error occurs we Forget this item so it does not
	// get queued again until another change happens.
	adc.pvcQueue.Forget(keyObj)
	return true
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
func (adc *attachDetachController) syncPVCByKey(key string) error {
	namespace, name, err := kcache.SplitMetaNamespaceKey(key)
	pvc, err := adc.pvcLister.PersistentVolumeClaims(namespace).Get(name)


	if pvc.Status.Phase != v1.ClaimBound || pvc.Spec.VolumeName == "" {
		// Skip unbound PVCs.
		return nil
	}

	objs, err := adc.podIndexer.ByIndex(pvcKeyIndex, key)

	for _, obj := range objs {
		volumeActionFlag := util.DetermineVolumeAction(
			pod,
			adc.desiredStateOfWorld,
			true /* default volume action */)

		util.ProcessPodVolumes(pod, volumeActionFlag, /* addVolumes */
			adc.desiredStateOfWorld, &adc.volumePluginMgr, adc.pvcLister, adc.pvLister)
	}
	return nil
}

1
2
3
4
5
6
7
8
func (rc *reconciler) reconcile() {
	// Detaches are triggered before attaches so that volumes referenced by
	// pods that are rescheduled to a different node are detached first.

	// Ensure volumes that should be detached are detached.
	for _, attachedVolume := range rc.actualStateOfWorld.GetAttachedVolumes() {
		if !rc.desiredStateOfWorld.VolumeExists(
			attachedVolume.VolumeName, attachedVolume.NodeName) {

1
2
3
4
5
6
7
8
9
10
11
12
13
// Set the detach request time
elapsedTime, err := rc.actualStateOfWorld.SetDetachRequestTime(attachedVolume.VolumeName, attachedVolume.NodeName)
if err != nil {
	klog.Errorf("Cannot trigger detach because it fails to set detach request time with error %v", err)
	continue
}
// Check whether timeout has reached the maximum waiting time
timeout := elapsedTime > rc.maxWaitForUnmountDuration
// Check whether volume is still mounted. Skip detach if it is still mounted unless timeout
if attachedVolume.MountedByNode && !timeout {
	klog.V(12).Infof(attachedVolume.GenerateMsgDetailed("Cannot detach volume because it is still mounted", ""))
	continue
}

1
2
3
4
5
6
7
8
9
// Before triggering volume detach, mark volume as detached and update the node status
// If it fails to update node status, skip detach volume
err = rc.actualStateOfWorld.RemoveVolumeFromReportAsAttached(attachedVolume.VolumeName, attachedVolume.NodeName)
if err != nil {
	klog.V(5).Infof("RemoveVolumeFromReportAsAttached failed while removing volume %q from node %q with: %v",
		attachedVolume.VolumeName,
		attachedVolume.NodeName,
		err)
}

1
2
3
4
5
// Trigger detach volume which requires verifing safe to detach step
// If timeout is true, skip verifySafeToDetach check
klog.V(5).Infof(attachedVolume.GenerateMsgDetailed("Starting attacherDetacher.DetachVolume", ""))
verifySafeToDetach := !timeout
err = rc.attacherDetacher.DetachVolume(attachedVolume.AttachedVolume, verifySafeToDetach, rc.actualStateOfWorld)

1
2
3
4
5
6
7
8
9
10
11
func (rc *reconciler) attachDesiredVolumes() {
	// Ensure volumes that should be attached are attached.
	for _, volumeToAttach := range rc.desiredStateOfWorld.GetVolumesToAttach() {
		if rc.actualStateOfWorld.VolumeNodeExists(volumeToAttach.VolumeName, volumeToAttach.NodeName) {
			// Volume/Node exists, touch it to reset detachRequestedTime
			if klog.V(5) {
				klog.Infof(volumeToAttach.GenerateMsgDetailed("Volume attached--touching", ""))
			}
			rc.actualStateOfWorld.ResetDetachRequestTime(volumeToAttach.VolumeName, volumeToAttach.NodeName)
			continue
		}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
if rc.isMultiAttachForbidden(volumeToAttach.VolumeSpec) {
	nodes := rc.actualStateOfWorld.GetNodesForVolume(volumeToAttach.VolumeName)
	if len(nodes) > 0 {
		if !volumeToAttach.MultiAttachErrorReported {
			rc.reportMultiAttachError(volumeToAttach, nodes)
			rc.desiredStateOfWorld.SetMultiAttachError(volumeToAttach.VolumeName, volumeToAttach.NodeName)
		}
		continue
	}
}

// check if this volume is allowed to be attached to multiple PODs/nodes, if yes, return false
		for _, accessMode := range volumeSpec.PersistentVolume.Spec.AccessModes {
			if accessMode == v1.ReadWriteMany || accessMode == v1.ReadOnlyMany {
				return false
			}
		}

1
2
3
4
5
6
7
8
9
err := rc.attacherDetacher.AttachVolume(volumeToAttach.VolumeToAttach, rc.actualStateOfWorld)
if err == nil {
	klog.Infof(volumeToAttach.GenerateMsgDetailed("attacherDetacher.AttachVolume started", ""))
}
if err != nil && !exponentialbackoff.IsExponentialBackoff(err) {
	// Ignore exponentialbackoff.IsExponentialBackoff errors, they are expected.
	// Log all other errors.
	klog.Errorf(volumeToAttach.GenerateErrorDetailed("attacherDetacher.AttachVolume failed to start", err).Error())
}

1
2
3
func (p *csiPlugin) NewAttacher() (volume.Attacher, error) {
	return p.newAttacherDetacher()
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
func (c *csiAttacher) Attach(spec *volume.Spec, nodeName types.NodeName) (string, error) {
	if spec == nil {
		klog.Error(log("attacher.Attach missing volume.Spec"))
		return "", errors.New("missing spec")
	}

	pvSrc, err := getPVSourceFromSpec(spec)
	if err != nil {
		klog.Error(log("attacher.Attach failed to get CSIPersistentVolumeSource: %v", err))
		return "", err
	}

	node := string(nodeName)
	attachID := getAttachmentName(pvSrc.VolumeHandle, pvSrc.Driver, node)