external-attacher源码分析(2)-核心处理逻辑分析
阅读原文时间:2021年07月10日阅读:1

kubernetes ceph-csi分析目录导航

基于tag v2.1.1

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

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。