【kubernetes/k8s源碼分析】 k8s csi plugin attach mount源碼分析

    接着kubelet pod 挂載volume源碼分析文章https://blog.csdn.net/zhonglinzhang/article/details/89923875

    volumeAttacher, newAttacherErr := attachableVolumePlugin.NewAttacher()，如果使用CSI插件則調用如下：

func (p *csiPlugin) NewAttacher() (volume.Attacher, error) {
	k8s := p.host.GetKubeClient()
	if k8s == nil {
		klog.Error(log("unable to get kubernetes client from host"))
		return nil, errors.New("unable to get Kubernetes client")
	}

	return &csiAttacher{
		plugin:        p,
		k8s:           k8s,
		waitSleepTime: 1 * time.Second,
	}, nil
}           

        csiAttacher結構實作了Attach方法

type csiAttacher struct {
	plugin        *csiPlugin
	k8s           kubernetes.Interface
	waitSleepTime time.Duration

	csiClient csiClient
}           

    建立挂盤操作，協調挂盤處理流程

reconcile

      --> operationExecutor.VerifyControllerAttachedVolume

      --> operationExecutor.MountVolume

               --> GenerateMountVolumeFunc

                      --> volumePlugin.NewMounter 第3章節講解

                      --> attachableVolumePlugin.NewAttacher()

                      --> volumeAttacher.WaitForAttach 第1章節講解

                      --> volumeDeviceMounter.MountDevice

                      --> volumeMounter.SetUp(fsGroup) 第4章節講解

  node節點資訊

  volumesAttached:

  - devicePath: csi-57bf2260979246d9fc2d571fc7b886854ddc82035e3376fe25c5185b46639104

    name: kubernetes.io/csi/rbd.csi.ceph.com^csi-rbd-vol-9f15531e-76e6-11e9-8f1d-a640193e1ea4

  volumesInUse:

  - kubernetes.io/csi/rbd.csi.ceph.com^csi-rbd-vol-9f15531e-76e6-11e9-8f1d-a640193e1ea4

1. WaitForAttach

func (c *csiAttacher) WaitForAttach(spec *volume.Spec, _ string, pod *v1.Pod, timeout time.Duration) (string, error) {           

    1.1 getCSISourceFromSpce函數 

       getCSISourceFromSpce根據pv得到如下資訊

  csi:

    driver: rbd.csi.ceph.com

    fsType: ext4

    nodePublishSecretRef:

      name: csi-rbd-secret

      namespace: default

    volumeAttributes:

      adminid: admin

      imageFeatures: layering

      imageFormat: "2"

      monitors: 192.168.74.57:6789

      pool: rbd

      storage.kubernetes.io/csiProvisionerIdentity: 1557904506281-8081-

    volumeHandle: csi-rbd-vol-9f15531e-76e6-11e9-8f1d-a640193e1ea4

    1.2 最終邏輯函數，看看幹什麼操作了

func (c *csiAttacher) waitForVolumeAttachmentInternal(volumeHandle, attachID string, timer *time.Timer, timeout time.Duration) (string, error) {
	klog.V(4).Info(log("probing VolumeAttachment [id=%v]", attachID))
	attach, err := c.k8s.StorageV1beta1().VolumeAttachments().Get(attachID, meta.GetOptions{})
	if err != nil {
		klog.Error(log("attacher.WaitForAttach failed for volume [%s] (will continue to try): %v", volumeHandle, err))
		return "", fmt.Errorf("volume %v has GET error for volume attachment %v: %v", volumeHandle, attachID, err)
	}           

2. waitForVolumeAttachmentInternal

      等待插件進行attach操作，接着參看kubernetes csi external-attacher 源碼分析 https://blog.csdn.net/zhonglinzhang/article/details/89643001

      external-attacher watch到volumeAttachments資源，則處理，最終發送GRPC請求ControllerPublishVolumeRequest

      2.1 隻要status為attached則算成功

func verifyAttachmentStatus(attachment *storage.VolumeAttachment, volumeHandle string) (bool, error) {
	// if being deleted, fail fast
	if attachment.GetDeletionTimestamp() != nil {
		klog.Error(log("VolumeAttachment [%s] has deletion timestamp, will not continue to wait for attachment", attachment.Name))
		return false, errors.New("volume attachment is being deleted")
	}
	// attachment OK
	if attachment.Status.Attached {
		return true, nil
	}
	// driver reports attach error
	attachErr := attachment.Status.AttachError
	if attachErr != nil {
		klog.Error(log("attachment for %v failed: %v", volumeHandle, attachErr.Message))
		return false, errors.New(attachErr.Message)
	}
	return false, nil
}           

3. NewMounter函數

    路徑 pkg/volume/csi/csi_mounter.go，csiMountMgr結構體

mounter := &csiMountMgr{
	plugin:       p,
	k8s:          k8s,
	spec:         spec,
	pod:          pod,
	podUID:       pod.UID,
	driverName:   csiDriverName(pvSource.Driver),
	volumeID:     pvSource.VolumeHandle,
	specVolumeID: spec.Name(),
	readOnly:     readOnly,
}
mounter.csiClientGetter.driverName = csiDriverName(pvSource.Driver)           

4. SetUpAt函數

     不知道為沙起個這麼鬼名字函數

    4.1 與CSI插件建立socket連接配接

csiSource, err := getCSISourceFromSpec(c.spec)
if err != nil {
	klog.Error(log("mounter.SetupAt failed to get CSI persistent source: %v", err))
	return err
}
klog.Infof("zzlin SetUpAt csiSource: %v", csiSource.String())

csi, err := c.csiClientGetter.Get()
if err != nil {
	klog.Error(log("mounter.SetUpAt failed to get CSI client: %v", err))
	return err
}
ctx, cancel := context.WithTimeout(context.Background(), csiTimeout)
defer cancel()           

    4.2 發送GRPC請求NodePublishVolume

fsType := csiSource.FSType
err = csi.NodePublishVolume(
	ctx,
	c.volumeID,
	c.readOnly,
	deviceMountPath,
	dir,
	accessMode,
	c.publishContext,
	attribs,
	nodePublishSecrets,
	fsType,
	c.spec.PersistentVolume.Spec.MountOptions,
)