ext-attacher | 南风未起｜云原生/Go技术分享博客

# 1.简介

# 1.1.作用

external-attacher负责监听volumeAttachment对象，交互CSI Plugin进行attach/detach及修改VA/PV对象，将卷附着或分离到节点。
注意

CSI Plugin不支持Publish/UnPublish操作，external-attacher仅修改volumeAttachment对象状态

# 1.2.入口

main函数负责实例化handler及attachController，启动attachController以激活worker及周期协程处理VA/PV状态及附着分离任务。

func main() {
	...

	factory := informers.NewSharedInformerFactory(clientset, *resync)
	...

	// Connect to CSI.
	csiConn := connection.Connect(*csiAddress, metricsManager, OnConnectionLoss(ExitOnConnectionLoss()))
	...
	
    // 间隔1s探测直至ready
	rpc.ProbeForever(csiConn, *timeout)
	...

	// Find driver name.
	csiAttacher, err := rpc.GetDriverName(ctx, csiConn)
	...

    // controller-service能力
	supportsService, err := supportsPluginControllerService(ctx, csiConn)
	...

    // 不支持controller-service
	if !supportsService {
        // 实例化trivialHandler(仅修改VA状态)
		handler = controller.NewTrivialHandler(clientset)
	} else {
        // attach/readyonly能力
		supportsAttach, supportsReadOnly, supportsListVolumesPublishedNodes, supportsSingleNodeMultiWriter, err = supportsControllerCapabilities(ctx, csiConn)
		...

        // 支持attach/detach
		if supportsAttach {
            // lister
			pvLister := factory.Core().V1().PersistentVolumes().Lister()
			vaLister := factory.Storage().V1().VolumeAttachments().Lister()
			csiNodeLister := factory.Storage().V1().CSINodes().Lister()
			
            // attacher client
            volAttacher := attacher.NewAttacher(csiConn)
            // volume缓存
			CSIVolumeLister := attacher.NewVolumeLister(csiConn)
            // 实例化CSIHandler
			handler = controller.NewCSIHandler(...)
		} else {
            // 实例化trivialHandler(仅修改VA状态)
			handler = controller.NewTrivialHandler(clientset)
		}
	}
	...

    // 初始化attacher controller
	ctrl := controller.NewCSIAttachController(...)

    // 启动函数
	run := func(ctx context.Context) {
		stopCh := ctx.Done()
		factory.Start(stopCh)
		ctrl.Run(int(*workerThreads), stopCh)
	}

    // 非leader选举执行
	if !*enableLeaderElection {
		run(context.TODO())
    // leader选举执行
	} else {
		...

		le := leaderelection.NewLeaderElection(leClientset, lockName, run)
        ...

		le.Run()
        ...
	}
}

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
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75

注意

trivialHandler是不支持attach/detach的实现，仅更新volumeAttachment状态为true

# 2.controller

# 2.1.初始化

NewCSIAttachController()会实例化csiAttachController，构造vaInformer/pvInformer监听以处理va/pv状态变化及附着。

// NewCSIAttachController returns a new *CSIAttachController
func NewCSIAttachController(...) *CSIAttachController {
	...

	ctrl := &CSIAttachController{
		client:                          client,
		attacherName:                    attacherName,
		handler:                         handler,
		...
		vaQueue:                         workqueue.NewNamedRateLimitingQueue(vaRateLimiter, "csi-attacher-va"),
		pvQueue:                         workqueue.NewNamedRateLimitingQueue(paRateLimiter, "csi-attacher-pv"),
		shouldReconcileVolumeAttachment: shouldReconcileVolumeAttachment,
		reconcileSync:                   1min,
		translator:                      csitrans.New(),
	}

    // valumeAttachment监听
	volumeAttachmentInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc:    ctrl.vaAdded,
		UpdateFunc: ctrl.vaUpdated,
		DeleteFunc: ctrl.vaDeleted,
	})
	ctrl.vaLister = volumeAttachmentInformer.Lister()
	...

	pvInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc:    ctrl.pvAdded,
		UpdateFunc: ctrl.pvUpdated,
		//DeleteFunc: ctrl.pvDeleted, TODO: do we need this?
	})
	ctrl.pvLister = pvInformer.Lister()
	...
    
    // csiHandler初始化
	ctrl.handler.Init(ctrl.vaQueue, ctrl.pvQueue)

	return ctrl
}

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
30
31
32
33
34
35
36
37
38

注意

PV监听相对特殊，只有删除、可迁移但未迁移状态下才会入队处理PV

# 2.2.run

ctrl.Run()会启动多个协程执行ctrl.syncVA和ctrl.syncPV同步处理VA/PV状态，执行ctrl.handler.ReconcileVA执行卷附着或分离。

// Run starts CSI attacher and listens on channel events
func (ctrl *CSIAttachController) Run(workers int, stopCh <-chan struct{}) {
	...

    // VA+PVC同步完成
	if !cache.WaitForCacheSync(stopCh, ctrl.vaListerSynced, ctrl.pvListerSynced) {
		klog.Errorf("Cannot sync caches")
		return
	}
    
    // 启动10个worker
	for i := 0; i < workers; i++ {
		go wait.Until(ctrl.syncVA, 0, stopCh)
		go wait.Until(ctrl.syncPV, 0, stopCh)
	}

    // CSIDriver支持Attach/Detach
	if ctrl.shouldReconcileVolumeAttachment {
        // 间隔1min触发一次
		go wait.Until(func() {
			// VA协调
            ctrl.handler.ReconcileVA()
			...
		}, ctrl.reconcileSync, stopCh)
	}

	<-stopCh
}

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

注意

ReconcileVA负责attach/detach，syncVA仅同步状态

# 3.VA处理

# 3.1.sync

ctrl.syncVA()会处理volumeAttachment对象，检查

// syncVA deals with one key off the queue.  It returns false when it's time to quit.
func (ctrl *CSIAttachController) syncVA() {
	key, quit := ctrl.vaQueue.Get()
	...
    
	defer ctrl.vaQueue.Done(key)

	vaName := key.(string)

	// 获取volumeAttachment对象
	va, err := ctrl.vaLister.Get(vaName)
	if err != nil {
		if apierrs.IsNotFound(err) {
			return
		}
		ctrl.vaQueue.AddRateLimited(vaName)
		return
	}
    
    // driver不匹配
	if va.Spec.Attacher != ctrl.attacherName {
		return
	}
    
    // 处理VA
	ctrl.handler.SyncNewOrUpdatedVolumeAttachment(va)
}

// trivialHandler
func (h *trivialHandler) SyncNewOrUpdatedVolumeAttachment(va *storage.VolumeAttachment) {
	// 未attach
	if !va.Status.Attached {
		// 修改VA为attached
		if _, err := markAsAttached(h.client, va, nil); err != nil {
			h.vaQueue.AddRateLimited(va.Name)
			return
		}
	}
    
	h.vaQueue.Forget(va.Name)
}

// csiHandler
func (h *csiHandler) SyncNewOrUpdatedVolumeAttachment(va *storage.VolumeAttachment) {
	...
    
    // 未删除
	if va.DeletionTimestamp == nil {
        // 执行attach
		err = h.syncAttach(va)
	} else {
        // 执行detach
		err = h.syncDetach(va)
	}
    
	if err != nil {
		// Re-queue with exponential backoff
		h.vaQueue.AddRateLimited(va.Name)
		return
	}
    
	// The operation has finished successfully, reset exponential backoff
	h.vaQueue.Forget(va.Name)
}

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
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64

注意

trivialHandler未交互CSI执行attach/detach，仅修改volumeAttachment状态，所以后续分析的都是csiHandler

# 3.2.attach

h.syncAttach()会准备请求数据，向PV补充finalizer，执行h.csiAttach()向节点附着volume，进而修改volumeAttachment状态。

func (h *csiHandler) syncAttach(va *storage.VolumeAttachment) error {
    // 非强制同步+Attached
	if !h.consumeForceSync(va.Name) && va.Status.Attached {
		return nil
	}

	// 执行Attach
	va, metadata, err := h.csiAttach(va)
	if err != nil {
        // 设置err及更新VA对象
		va, saveErr = h.saveAttachError(va, err)
		...
        
		return err
	}

	// VA标记为Attached
	markAsAttached(h.client, va, metadata)
    ...
    
	return nil
}

func (h *csiHandler) csiAttach(va *storage.VolumeAttachment) (...) {
	...
    
    // VA关联PV
	if va.Spec.Source.PersistentVolumeName != nil {
		...
        
        // 获取PV对象
		pv, err := h.pvLister.Get(*va.Spec.Source.PersistentVolumeName)
		...
        
		// 正在删除的PV不允许Attach
		if pv.DeletionTimestamp != nil {
			return va, nil, fmt.Errorf("PersistentVolume %q is marked for deletion", pv.Name)
		}
        
        // 设置attach finalizer及更新PV
		pv, err = h.addPVFinalizer(pv)
		...

        // 可迁移的Driver
		if h.translator.IsPVMigratable(pv) {
            // in-tree PV转为csi PV
			pv, err = h.translator.TranslateInTreePVToCSI(pv)
			...
			migratable = true
		}

		// 获取pv.spec.csi
		csiSource, err = getCSISource(&pv.Spec)
		...

		pvSpec = &pv.Spec
    // VS内联volume
	} else if va.Spec.Source.InlineVolumeSpec != nil {
		if va.Spec.Source.InlineVolumeSpec.CSI == nil {
            return va, nil, errors.New("inline volume spec contains nil CSI source")
		}
        
        csiSource = va.Spec.Source.InlineVolumeSpec.CSI
		pvSpec = va.Spec.Source.InlineVolumeSpec
	} else {
		return va, nil, errors.New("neither InlineCSIVolumeSource nor PersistentVolumeName specified in VA")
	}

    // volume属性
	attributes, err := GetVolumeAttributes(csiSource)
	...
    
    // volume名称及readyOnly  
	volumeHandle, readOnly, err := GetVolumeHandle(csiSource)
	...
    
    // Driver不支持readyOnly
	if !h.supportsPublishReadOnly {
		readOnly = false
	}

    // 基于PV生成volume能力
	volumeCapabilities, err := GetVolumeCapabilities(pvSpec, h.supportsSingleNodeMultiWriter, h.defaultFSType)
	...

    // 获取attach凭证
	secrets, err := h.getCredentialsFromPV(csiSource)
	...

    // 由CSINode获取nodeID
	nodeID, err := h.getNodeID(h.attacherName, va.Spec.NodeName, nil)
	...

	// 设置attach finalizer
	va, finalizerAdded := h.prepareVAFinalizer(va)
    // 设置nodeID Anno
	va, nodeIDAdded := h.prepareVANodeID(va, nodeID)

    // 更新VA
	if finalizerAdded || nodeIDAdded {
		va = h.patchVA(originalVA, va)
        ...
	}
	...
    
	// 执行Attach
	publishInfo := h.attacher.Attach(ctx, volumeHandle, readOnly, nodeID, capabilities, attributes, secrets)
    ...

	return va, publishInfo, nil
}

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
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111

注意

h.syncAttach主要负责请求准备，h.attacher.Attach会调用csiClient.ControllerPublishVolume进行附着

# 3.3.detach

h.syncDetach()会调用csiClient.ControllerUnpublishVolume()完成detach动作，同时将volumeAttachment附着状态释放。

func (h *csiHandler) syncDetach(va *storage.VolumeAttachment) error {
	// !forceSync+attach finalizer不存在
	if !h.consumeForceSync(va.Name) && !h.hasVAFinalizer(va) {
		return nil
	}

	// 执行detach
	va, err := h.csiDetach(va)
	if err != nil {
		// 设置err及更新VA对象
		va, saveErr = h.saveDetachError(va, err)
		...

		return err
	}

	return nil
}

func (h *csiHandler) csiDetach(va *storage.VolumeAttachment) (*storage.VolumeAttachment, error) {
	...
    
    // VA关联PV
	if va.Spec.Source.PersistentVolumeName != nil {
		...
        
        // 获取PV对象
		pv, err := h.pvLister.Get(*va.Spec.Source.PersistentVolumeName)
		...
        
        // 可迁移Driver
		if h.translator.IsPVMigratable(pv) {
            // in-tree PV转为csi PV
			pv, err = h.translator.TranslateInTreePVToCSI(pv)
			...
			migratable = true
		}
        
        // 获取pv.spec.csi
		csiSource, err = getCSISource(&pv.Spec)
		...
    // VA内联volume
	} else if va.Spec.Source.InlineVolumeSpec != nil {
		if va.Spec.Source.InlineVolumeSpec.CSI == nil {
            return va, errors.New("inline volume spec contains nil CSI source")
		}
        
        csiSource = va.Spec.Source.InlineVolumeSpec.CSI
	} else {
		return va, errors.New("neither InlineCSIVolumeSource nor PersistentVolumeName specified in VA source")
	}

    // 获取volume名称
	volumeHandle, _, err := GetVolumeHandle(csiSource)
	...
    
    // 获取detach凭证
	secrets, err := h.getCredentialsFromPV(csiSource)
	...

    // 由CSINode获取nodeID
	nodeID, err := h.getNodeID(h.attacherName, va.Spec.NodeName, va)
	...

    // 执行detach
	err = h.attacher.Detach(ctx, volumeHandle, nodeID, secrets)
	...

    // 释放attached状态
	markAsDetached(h.client, va)
    ...

	return va, nil
}

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
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74

注意

h.attacher.Detach会执行到csiClient.ControllerUnpublishVolume

# 3.4.reconcile

ctrl.handler.ReconcileVA()会周期获取CSIDriver的volume attach状态，基于volumeAttachment对象进行对账及触发强制处理。

func (a *CSIVolumeLister) ListVolumes(ctx context.Context) (map[string]([]string), error) {
	...
    
	for {
        // 获取driver侧的volume
		rsp, err := a.client.ListVolumes(ctx, &csi.ListVolumesRequest{StartingToken: tok})
		...
        
        // 缓存volume
		for _, e := range rsp.Entries {
			p[e.GetVolume().VolumeId] = e.Status.PublishedNodeIds
		}
        
        // 检查结束条件
		tok = rsp.NextToken
		if len(tok) == 0 {
			break
		}
	}

	return p, nil
}


// lists volumes from CSIDriver and reconciles attachment status with the corresponding VolumeAttachment object.
func (h *csiHandler) ReconcileVA() error {
	...

	// Loop over all volume attachment objects
	vas, err := h.vaLister.List(labels.Everything())
	...

    // 获取volume attach真实状态
	published, err := h.CSIVolumeLister.ListVolumes(ctx)
	...

	for _, va := range vas {
        // 由CSINode/VA Anno获取nodeID
		nodeID, err := h.getNodeID(h.attacherName, va.Spec.NodeName, va)
		...
        
        // 获取关联的pv.spec(PV/InlineVolume)
		pvSpec, err := h.getProcessedPVSpec(va)
		...

        // 获取pv.spec.csi
		source, err := getCSISource(pvSpec)
		...

        // 获取volume名称
		volumeHandle, _, err := GetVolumeHandle(source)
		...
        
        // va attach状态
		attachedStatus := va.Status.Attached
		...
        
        // 迁移的driver
		if isMig {
            // 基于in-tree plugin重新生成volume名称
			volumeHandle, err = h.translator.RepairVolumeHandle(source.Driver, volumeHandle, nodeID)
			...
		}
		...
        
		// 匹配attached Volume
		for _, gotNodeID := range published[volumeHandle] {
			if gotNodeID == nodeID {
				found = true
				break
			}
		}

		// VA attach+driver未找到
		if attachedStatus != found {
			// 标记强制执行
			h.setForceSync(va.Name)
            // VA重新入队
			h.vaQueue.Add(va.Name)
		}
	}
	return nil
}

注意

由于trivialHandler实现为空，这里仅分析csiHandler

# 4.PV

# 4.1.sync

ctrl.syncPV()负责PV对象的处理，核心逻辑为调用ctrl.handler.SyncNewOrUpdatedPersistentVolume()修改PV相关finalizer。

// syncPV deals with one key off the queue.  It returns false when it's time to quit.
func (ctrl *CSIAttachController) syncPV() {
	key, quit := ctrl.pvQueue.Get()
	...
    
	defer ctrl.pvQueue.Done(key)

	pvName := key.(string)
	// get PV to process
	pv, err := ctrl.pvLister.Get(pvName)
	if err != nil {
		if apierrs.IsNotFound(err) {
			return
		}
		ctrl.pvQueue.AddRateLimited(pvName)
		return
	}
    
    // PV处理
	ctrl.handler.SyncNewOrUpdatedPersistentVolume(pv)
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

注意

SyncNewOrUpdatedPersistentVolume有两种实现，trivialHandler实现为空，后续仅分析csiHandler实现

# 4.2.reconcile

ctrl.handler.SyncNewOrUpdatedPersistentVolume()负责处理正在删除的PV，移除VA已不存在的PV Attach Finalizer以确保删除。

func (h *csiHandler) SyncNewOrUpdatedPersistentVolume(pv *v1.PersistentVolume) {
	// Sync and remove finalizer on given PV
	if pv.DeletionTimestamp == nil {
		ignore := true

		// 可迁移
		if h.translator.IsPVMigratable(pv) {
			ignore = false
			if ann := pv.Annotations; ann != nil {
                // driver匹配
				if migratedToDriver := ann[annMigratedTo]; migratedToDriver == h.attacherName {
					ignore = true
				}
			}
		}

        // Don't process anything that has no deletion timestamp.
		if ignore {
			h.pvQueue.Forget(pv.Name)
			return
		}
	}
	...
    
	// attach finalizer
	for _, f := range pv.Finalizers {
		if f == finalizer {
			found = true
			break
		}
	}
    
    // 没有,说明未attach过
	if !found {
		// No finalizer -> no action required
		h.pvQueue.Forget(pv.Name)
		return
	}

	// 获取VA
	vas, err := h.vaLister.List(labels.Everything())
	if err != nil {
		h.pvQueue.AddRateLimited(pv.Name)
		return
	}
    
	for _, va := range vas {
		// 关联VA还存在
		if va.Spec.Source.PersistentVolumeName != nil && *va.Spec.Source.PersistentVolumeName == pv.Name {
			// This PV is needed by this VA, don't remove finalizer
			h.pvQueue.Forget(pv.Name)
			return
		}
	}
    
	// No VA found -> remove finalizer
	clone := pv.DeepCopy()
    clone.removeFinalizer(finalizer)
	...

    // 更新PV
	if _, err = h.patchPV(pv, clone); err != nil {
		h.pvQueue.AddRateLimited(pv.Name)
		return
	}

	h.pvQueue.Forget(pv.Name)

	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
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70

注意

SyncNewOrUpdatedPersistentVolume会检查VA清理情况，VA不存在代表detach完成，PV可以安全删除，移除相应finalizer