kubernetes ceph-csi分析目录导航

基于tag v2.1.1

https://github.com/kubernetes-csi/external-attacher/releases/tag/v2.1.1

external-attacher

external-attacher属于external plugin中的一个。下面我们先来回顾一下external plugin以及csi系统结构。

external plugin

external plugin包括了external-provisioner、external-attacher、external-resizer、external-snapshotter等，external plugin辅助csi plugin组件，共同完成了存储相关操作。external plugin负责watch pvc、volumeAttachment等对象，然后调用volume plugin来完成存储的相关操作。如external-provisioner watch pvc对象，然后调用csi plugin来创建存储，最后创建pv对象；external-attacher watch volumeAttachment对象，然后调用csi plugin来做attach/dettach操作，并修改volumeAttachment对象与pv对象；external-resizer watch pvc对象，然后调用csi plugin来做存储的扩容操作等。

csi系统结构

external-attacher作用分析

根据CSI plugin是否支持ControllerPublish/ControllerUnpublish操作，external-attacher的作用分为如下两种：

（1）当CSI plugin不支持ControllerPublish/ControllerUnpublish操作时，AD controller（或kubelet的volume manager）创建VolumeAttachment对象后，external-attacher仅参与VolumeAttachment对象的修改，将attached属性值patch为true；而external-attacher对pv对象无任何同步处理操作。

（2）当CSI plugin支持ControllerPublish/ControllerUnpublish操作时，external-attacher调用csi plugin（ControllerPublishVolume）进行存储的attach操作，然后更改VolumeAttachment对象，将attached属性值patch为true，并patch pv对象，增加该external-attacher相关的finalizer；对于pv对象，external-attacher负责处理pv对象的finalizer，patch pv对象，去除该external-attacher相关的finalizer（该external-attacher执行attach操作时添加的finalizer）。

源码分析

external-attacher的源码分析将分为两部分：

（1）main方法以及启动参数分析；

（2）核心处理逻辑分析。

前面一篇博客已经对external-attacher的main方法以及启动参数做了分析，这篇博客将对external-attacher的核心处理逻辑进行源码分析。

Run

核心逻辑为跑goroutine不停的调用ctrl.syncVA与ctrl.syncPV对VolumeAttachment对象以及PV对象进行同步处理。

//external-attcher/pkg/controller/controller.go

// Run starts CSI attacher and listens on channel events

func (ctrl *CSIAttachController) Run(workers int, stopCh <-chan struct{}) {

	defer ctrl.vaQueue.ShutDown()

	defer ctrl.pvQueue.ShutDown()

	klog.Infof("Starting CSI attacher")

	defer klog.Infof("Shutting CSI attacher")

	if !cache.WaitForCacheSync(stopCh, ctrl.vaListerSynced, ctrl.pvListerSynced) {

		klog.Errorf("Cannot sync caches")

		return

	}

	for i := 0; i < workers; i++ {

		go wait.Until(ctrl.syncVA, 0, stopCh)

		go wait.Until(ctrl.syncPV, 0, stopCh)

	}

	if ctrl.shouldReconcileVolumeAttachment {

		go wait.Until(func() {

			err := ctrl.handler.ReconcileVA()

			if err != nil {

				klog.Errorf("Failed to reconcile volume attachments: %v", err)

			}

		}, ctrl.reconcileSync, stopCh)

	}

	<-stopCh

}

1.syncVA

syncVA负责VolumeAttachment对象的处理，核心逻辑为：

（1）判断VolumeAttachment对象的.Spec.Attacher属性值，判断是否由本attacher组件来负责VolumeAttachment对象的同步处理；

（2）调用ctrl.handler.SyncNewOrUpdatedVolumeAttachment(va)。

//external-attcher/pkg/controller/controller.go

func (ctrl *CSIAttachController) syncVA() {

	key, quit := ctrl.vaQueue.Get()

	if quit {

		return

	}

	defer ctrl.vaQueue.Done(key)

	vaName := key.(string)

	klog.V(4).Infof("Started VA processing %q", vaName)

	// get VolumeAttachment to process

	va, err := ctrl.vaLister.Get(vaName)

	if err != nil {

		if apierrs.IsNotFound(err) {

			// VolumeAttachment was deleted in the meantime, ignore.

			klog.V(3).Infof("VA %q deleted, ignoring", vaName)

			return

		}

		klog.Errorf("Error getting VolumeAttachment %q: %v", vaName, err)

		ctrl.vaQueue.AddRateLimited(vaName)

		return

	}

	if va.Spec.Attacher != ctrl.attacherName {

		klog.V(4).Infof("Skipping VolumeAttachment %s for attacher %s", va.Name, va.Spec.Attacher)

		return

	}

	ctrl.handler.SyncNewOrUpdatedVolumeAttachment(va)

}

1.1 SyncNewOrUpdatedVolumeAttachment

ctrl.handler.SyncNewOrUpdatedVolumeAttachment()包含两个实现，将根据CSI plugin是否支持ControllerPublish/ControllerUnpublish操作来调用不同的实现。（调用不同实现的判断逻辑在external-attacher的main方法里，前面分析external-attacher的main方法时已经分析过了，忘记的可以回去看下）

（1）trivialHandler.SyncNewOrUpdatedVolumeAttachment：当CSI plugin不支持ControllerPublish/ControllerUnpublish操作时，AD controller（或kubelet的volume manager）创建VolumeAttachment对象后，external-attacher仅参与VolumeAttachment对象的修改，将attached属性值patch为true。

（2）csiHandler.SyncNewOrUpdatedVolumeAttachment：当CSI plugin支持ControllerPublish/ControllerUnpublish操作时，external-attacher调用csi plugin（ControllerPublishVolume）进行存储的attach操作，然后更改VolumeAttachment对象，将attached属性值patch为true。

1.1.1 trivialHandler.SyncNewOrUpdatedVolumeAttachment

先来看trivialHandler的实现。

当VolumeAttachment的attached属性值为false时，调用markAsAttached做进一步处理。

//external-attcher/pkg/controller/trivial_handler.go

func (h *trivialHandler) SyncNewOrUpdatedVolumeAttachment(va *storage.VolumeAttachment) {

	klog.V(4).Infof("Trivial sync[%s] started", va.Name)

	if !va.Status.Attached {

		// mark as attached

		if _, err := markAsAttached(h.client, va, nil); err != nil {

			klog.Warningf("Error saving VolumeAttachment %s as attached: %s", va.Name, err)

			h.vaQueue.AddRateLimited(va.Name)

			return

		}

		klog.V(2).Infof("Marked VolumeAttachment %s as attached", va.Name)

	}

	h.vaQueue.Forget(va.Name)

}

markAsAttached

markAsAttached主要是将VolumeAttachment的attached属性值patch为true。

//external-attcher/pkg/controller/util.go

func markAsAttached(client kubernetes.Interface, va *storage.VolumeAttachment, metadata map[string]string) (*storage.VolumeAttachment, error) {

	klog.V(4).Infof("Marking as attached %q", va.Name)

	clone := va.DeepCopy()

	clone.Status.Attached = true

	clone.Status.AttachmentMetadata = metadata

	clone.Status.AttachError = nil

	patch, err := createMergePatch(va, clone)

	if err != nil {

		return va, err

	}

	newVA, err := client.StorageV1beta1().VolumeAttachments().Patch(va.Name, types.MergePatchType, patch)

	if err != nil {

		return va, err

	}

	klog.V(4).Infof("Marked as attached %q", va.Name)

	return newVA, nil

}

1.1.2 csiHandler.SyncNewOrUpdatedVolumeAttachment

先来看csiHandler的实现。主要逻辑如下：

（1）当VolumeAttachment对象的.DeletionTimestamp字段为空，调用h.syncAttach（调用了csi plugin的ControllerPublishVolume方法来做存储的attach操作，并调用markAsAttached将VolumeAttachment的attached属性值patch为true）；

（2）当VolumeAttachment对象的.DeletionTimestamp字段不为空，调用h.syncDetach（调用了csi plugin的ControllerUnpublishVolume方法来做存储的dettach操作，并调用markAsDetached将VolumeAttachment的attached属性值patch为false）。

// pkg/controller/csi_handler.go

func (h *csiHandler) SyncNewOrUpdatedVolumeAttachment(va *storage.VolumeAttachment) {

	klog.V(4).Infof("CSIHandler: processing VA %q", va.Name)

	var err error

	if va.DeletionTimestamp == nil {

		err = h.syncAttach(va)

	} else {

		err = h.syncDetach(va)

	}

	if err != nil {

		// Re-queue with exponential backoff

		klog.V(2).Infof("Error processing %q: %s", va.Name, err)

		h.vaQueue.AddRateLimited(va.Name)

		return

	}

	// The operation has finished successfully, reset exponential backoff

	h.vaQueue.Forget(va.Name)

	klog.V(4).Infof("CSIHandler: finished processing %q", va.Name)

}

syncAttach

syncAttach不展开分析了，感兴趣的可以自己深入分析，从h.csiAttach调用入手。

主要逻辑为：

（1）调用h.csiAttach做attach操作（patch pv对象，增加该external-attacher相关的Finalizer，并最终调用了csi plugin的ControllerPublishVolume方法来做存储的attach操作）；

（2）调用markAsAttached将VolumeAttachment的attached属性值patch为true。

// pkg/controller/csi_handler.go

func (h *csiHandler) syncAttach(va *storage.VolumeAttachment) error {

	if !h.consumeForceSync(va.Name) && va.Status.Attached {

		// Volume is attached and no force sync, there is nothing to be done.

		klog.V(4).Infof("%q is already attached", va.Name)

		return nil

	}

	// Attach and report any error

	klog.V(2).Infof("Attaching %q", va.Name)

	va, metadata, err := h.csiAttach(va)

	if err != nil {

		var saveErr error

		va, saveErr = h.saveAttachError(va, err)

		if saveErr != nil {

			// Just log it, propagate the attach error.

			klog.V(2).Infof("Failed to save attach error to %q: %s", va.Name, saveErr.Error())

		}

		// Add context to the error for logging

		err := fmt.Errorf("failed to attach: %s", err)

		return err

	}

	klog.V(2).Infof("Attached %q", va.Name)

	// Mark as attached

	if _, err := markAsAttached(h.client, va, metadata); err != nil {

		return fmt.Errorf("failed to mark as attached: %s", err)

	}

	klog.V(4).Infof("Fully attached %q", va.Name)

	return nil

}

syncDetach

同样的，syncDetach不展开分析，这里直接给出结果，最终调用了csi plugin的ControllerUnpublishVolume方法来做存储的dettach操作，并调用markAsDetached将VolumeAttachment的attached属性值patch为false，感兴趣的可以自己深入分析，从h.csiDetach调用入手。

// pkg/controller/csi_handler.go

func (h *csiHandler) syncDetach(va *storage.VolumeAttachment) error {

	klog.V(4).Infof("Starting detach operation for %q", va.Name)

	if !h.consumeForceSync(va.Name) && !h.hasVAFinalizer(va) {

		klog.V(4).Infof("%q is already detached", va.Name)

		return nil

	}

	// Detach and report any error

	klog.V(2).Infof("Detaching %q", va.Name)

	va, err := h.csiDetach(va)

	if err != nil {

		var saveErr error

		va, saveErr = h.saveDetachError(va, err)

		if saveErr != nil {

			// Just log it, propagate the detach error.

			klog.V(2).Infof("Failed to save detach error to %q: %s", va.Name, saveErr.Error())

		}

		// Add context to the error for logging

		err := fmt.Errorf("failed to detach: %s", err)

		return err

	}

	klog.V(4).Infof("Fully detached %q", va.Name)

	return nil

}

2.syncPV

syncPV负责PV对象的处理，核心逻辑为调用ctrl.handler.SyncNewOrUpdatedPersistentVolume(pv)来对pv对象做进一步处理。

//external-attcher/pkg/controller/controller.go

// 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()

	if quit {

		return

	}

	defer ctrl.pvQueue.Done(key)

	pvName := key.(string)

	klog.V(4).Infof("Started PV processing %q", pvName)

	// get PV to process

	pv, err := ctrl.pvLister.Get(pvName)

	if err != nil {

		if apierrs.IsNotFound(err) {

			// PV was deleted in the meantime, ignore.

			klog.V(3).Infof("PV %q deleted, ignoring", pvName)

			return

		}

		klog.Errorf("Error getting PersistentVolume %q: %v", pvName, err)

		ctrl.pvQueue.AddRateLimited(pvName)

		return

	}

	ctrl.handler.SyncNewOrUpdatedPersistentVolume(pv)

}

2.1 SyncNewOrUpdatedPersistentVolume

跟上面分析的syncVA中的SyncNewOrUpdatedVolumeAttachment一样，ctrl.handler.SyncNewOrUpdatedPersistentVolume()也包含两个实现，将根据CSI plugin是否支持ControllerPublish/ControllerUnpublish操作来调用不同的实现。（调用不同实现的判断逻辑在external-attacher的main方法里，前面分析external-attacher的main方法时已经分析过了，忘记的可以回去看下）

（1）trivialHandler.SyncNewOrUpdatedPersistentVolume：当CSI plugin不支持ControllerPublish/ControllerUnpublish操作时，external-attacher不对pv对象做任何同步处理操做。

（2）csiHandler.SyncNewOrUpdatedPersistentVolume：当CSI plugin支持ControllerPublish/ControllerUnpublish操作时，external-attacher负责处理pv对象的finalizer，patch pv对象，去除该external-attacher相关的finalizer（该external-attacher执行attach操作时添加的finalizer）。

2.1.1 trivialHandler.SyncNewOrUpdatedPersistentVolume

trivialHandler.SyncNewOrUpdatedPersistentVolume方法直接返回，可以看出对于pv对象，不做处理。

//external-attcher/pkg/controller/trivial_handler.go

func (h *trivialHandler) SyncNewOrUpdatedPersistentVolume(pv *v1.PersistentVolume) {

	return

}

与ceph-csi搭配使用的external-attacher相关日志

I0907 03:30:18.426009       1 main.go:166] CSI driver does not support ControllerPublishUnpublish, using trivial handler

I0907 08:13:38.148672       1 controller.go:198] Started VA processing "csi-706ace7b44a2f618280cba51951595f98893b34a38015e3c4ef7d1160be7f67b"

I0907 08:13:38.148698       1 trivial_handler.go:53] Trivial sync[csi-706ace7b44a2f618280cba51951595f98893b34a38015e3c4ef7d1160be7f67b] started

I0907 08:13:38.148707       1 util.go:35] Marking as attached "csi-706ace7b44a2f618280cba51951595f98893b34a38015e3c4ef7d1160be7f67b"

I0907 08:13:38.156486       1 util.go:48] Marked as attached "csi-706ace7b44a2f618280cba51951595f98893b34a38015e3c4ef7d1160be7f67b"

I0907 08:13:38.156510       1 trivial_handler.go:61] Marked VolumeAttachment csi-706ace7b44a2f618280cba51951595f98893b34a38015e3c4ef7d1160be7f67b as attached

2.1.2 csiHandler.SyncNewOrUpdatedPersistentVolume

csiHandler.SyncNewOrUpdatedPersistentVolume主要是处理pv对象的finalizer，patch pv对象，去除该external-attacher相关的finalizer（该external-attacher执行attach操作时添加的finalizer）。

主要逻辑：

（1）判断pv对象的DeletionTimestamp是否为空，为空则直接返回；

（2）检查pv对象是否含有该external-attacher相关的finalizer，没有则直接返回；

（3）查询volumeAttachment对象列表，遍历查询是否有va对象记录着该pv，有则直接返回；

（4）去除pv对象中该external-attacher相关的finalizer；

（5）patch pv对象。

// pkg/controller/csi_handler.go

func (h *csiHandler) SyncNewOrUpdatedPersistentVolume(pv *v1.PersistentVolume) {

	klog.V(4).Infof("CSIHandler: processing PV %q", pv.Name)

	// Sync and remove finalizer on given PV

	if pv.DeletionTimestamp == nil {

		// Don't process anything that has no deletion timestamp.

		klog.V(4).Infof("CSIHandler: processing PV %q: no deletion timestamp, ignoring", pv.Name)

		h.pvQueue.Forget(pv.Name)

		return

	}

	// Check if the PV has finalizer

	finalizer := GetFinalizerName(h.attacherName)

	found := false

	for _, f := range pv.Finalizers {

		if f == finalizer {

			found = true

			break

		}

	}

	if !found {

		// No finalizer -> no action required

		klog.V(4).Infof("CSIHandler: processing PV %q: no finalizer, ignoring", pv.Name)

		h.pvQueue.Forget(pv.Name)

		return

	}

	// Check that there is no VA that requires the PV

	vas, err := h.vaLister.List(labels.Everything())

	if err != nil {

		// Failed listing VAs? Try again with exp. backoff

		klog.Errorf("Failed to list VolumeAttachments for PV %q: %s", pv.Name, err.Error())

		h.pvQueue.AddRateLimited(pv.Name)

		return

	}

	for _, va := range vas {

		if va.Spec.Source.PersistentVolumeName != nil && *va.Spec.Source.PersistentVolumeName == pv.Name {

			// This PV is needed by this VA, don't remove finalizer

			klog.V(4).Infof("CSIHandler: processing PV %q: VA %q found", pv.Name, va.Name)

			h.pvQueue.Forget(pv.Name)

			return

		}

	}

	// No VA found -> remove finalizer

	klog.V(4).Infof("CSIHandler: processing PV %q: no VA found, removing finalizer", pv.Name)

	clone := pv.DeepCopy()

	newFinalizers := []string{}

	for _, f := range pv.Finalizers {

		if f == finalizer {

			continue

		}

		newFinalizers = append(newFinalizers, f)

	}

	if len(newFinalizers) == 0 {

		// Canonize empty finalizers for unit test (so we don't need to

		// distinguish nil and [] there)

		newFinalizers = nil

	}

	clone.Finalizers = newFinalizers

	if _, err = h.patchPV(pv, clone); err != nil {

		klog.Errorf("Failed to remove finalizer from PV %q: %s", pv.Name, err.Error())

		h.pvQueue.AddRateLimited(pv.Name)

		return

	}

	klog.V(2).Infof("Removed finalizer from PV %q", pv.Name)

	h.pvQueue.Forget(pv.Name)

	return

}

总结

external-attacher属于external plugin中的一个。

external-attacher作用分析

根据CSI plugin是否支持ControllerPublish/ControllerUnpublish操作，external-attacher的作用分为如下两种：

（1）当CSI plugin不支持ControllerPublish/ControllerUnpublish操作时，AD controller（或kubelet的volume manager）创建VolumeAttachment对象后，external-attacher仅参与VolumeAttachment对象的修改，将attached属性值patch为true；而external-attacher对pv对象无任何同步处理操作。

（2）当CSI plugin支持ControllerPublish/ControllerUnpublish操作时，external-attacher调用csi plugin（ControllerPublishVolume）进行存储的attach操作，然后更改VolumeAttachment对象，将attached属性值patch为true，并patch pv对象，增加该external-attacher相关的Finalizer；对于pv对象，external-attacher负责处理pv对象的finalizer，patch pv对象，去除该external-attacher相关的finalizer（该external-attacher执行attach操作时添加的finalizer）。

external-attacher与ceph-csi rbd结合使用

ceph-csi不支持ControllerPublish/ControllerUnpublish操作，所以external-attacher与ceph-csi rbd结合使用，external-attacher仅参与VolumeAttachment对象的修改，将attached属性值patch为true。

最新文章

1. mysql在linux下的安装
2. Android 监听锁屏、解锁、开屏 操作
3. 《BI项目笔记》历年理化指标分析Cube的建立
4. JSON相关（一）：JSON.parse()和JSON.stringify()
5. centos6.7 安装Docker
6. Unieap3.5-禁用Form表单中的全部标签
7. 字符串和数组中split().toString(),join(),splice(),slice(),substr()和substring()
8. Call Hierarchy(方法调用层次)
9. USB 2.0 A型、B型、Mini和Micro接口定义及封装
10. Activity中setResult(int resultCode, Intent data)与onActivityResult(int requestCode, int resultCode, Intent data)方法的调用
11. cf581A Vasya the Hipster
12. Android 自定义EditText实现粘贴，复制，剪切的监听
13. aix archPlat
14. Android Realm数据库使用指南
15. 解决SQLServer 2008 日志无法收缩，收缩后大小不改变
16. Codeforces Round #162 (Div. 1) B. Good Sequences (dp+分解素数)
17. PA教材提纲 TAW10-2
18. Netty 之 Netty生产级的心跳和重连机制
19. 转载：HBuilder常用快捷键
20. oc中的oop基础及类的基本介绍

热门文章

1. linux服务器市场特性高可用高性能 （集群上体现）安全
2. Docker Swarm（三）Service（服务）分配策略
3. C语言关于指针函数与函数指针个人理解
4. osi七层模型与tcp/ip四层模型的差别
5. https 真的安全吗，可以抓包吗，如何防止抓包吗
6. 路由信息相关 route 网卡
7. MCU，硅片，BOM
8. Jmeter读取python生成的参数
9. 我的N年软件测试感悟
10. VRRP协议的原理与配置