CSI以及本地盘的相关实现记录

接口

CSI有众多接口，按逻辑共分为三类，开发者需要按照需求实现对应的接口。外部组件(node-driver-registrar、external-provisioner 等)会通过unix socket+gpc调用这些接口。

在CSI代码中经常可以看到这么几句话，就是注册对应的接口。

csi.RegisterIdentityServer(srv, identity)
csi.RegisterControllerServer(srv, ctl)
csi.RegisterNodeServer(srv, node)

之所以分这么多类，是因为这些类是不同的外部组件分别调用的。实际上，你开发的CSI pod 完全可以全部注册这些类，然后CSI pod + External-Provisioner 部署在一起时，该CSI pod就会被调用controller类的函数，没人调用其他类的函数，同理可得其他。
或者，你也可以将你的CSI pod分为3个pod分别注册3种类，那么 controller类 的CSI pod 只能和 External-Provisioner在一起，CSI pod node 只能和…类似微服务化，实际比较少见。

identity类

// identity需要实现入如下接口
type IdentityServer interface {
    
    GetPluginInfo(context.Context, *GetPluginInfoRequest) (*GetPluginInfoResponse, error)
    GetPluginCapabilities(context.Context, *GetPluginCapabilitiesRequest) (*GetPluginCapabilitiesResponse, error)
    Probe(context.Context, *ProbeRequest) (*ProbeResponse, error)
}

GetPluginInfo: 获取信息，例如drivername，就是常见的 disk.xx.com
GetPluginCapabilities: 返回能力，PluginCapability_Service_CONTROLLER_SERVICE 和 PluginCapability_Service_VOLUME_ACCESSIBILITY_CONSTRAINTS，前者表示自己提供controller能力，后者表示自己支持topology，这个在实现本地盘时很关键，必须支持。

controller类

// ControllerServer is the server API for Controller service.
type ControllerServer interface {
    
    CreateVolume(context.Context, *CreateVolumeRequest) (*CreateVolumeResponse, error)
    DeleteVolume(context.Context, *DeleteVolumeRequest) (*DeleteVolumeResponse, error)
    ControllerPublishVolume(context.Context, *ControllerPublishVolumeRequest) (*ControllerPublishVolumeResponse, error)
    ControllerUnpublishVolume(context.Context, *ControllerUnpublishVolumeRequest) (*ControllerUnpublishVolumeResponse, error)
    ValidateVolumeCapabilities(context.Context, *ValidateVolumeCapabilitiesRequest) (*ValidateVolumeCapabilitiesResponse, error)
    ListVolumes(context.Context, *ListVolumesRequest) (*ListVolumesResponse, error)
    GetCapacity(context.Context, *GetCapacityRequest) (*GetCapacityResponse, error)
    ControllerGetCapabilities(context.Context, *ControllerGetCapabilitiesRequest) (*ControllerGetCapabilitiesResponse, error)
    CreateSnapshot(context.Context, *CreateSnapshotRequest) (*CreateSnapshotResponse, error)
    DeleteSnapshot(context.Context, *DeleteSnapshotRequest) (*DeleteSnapshotResponse, error)
    ListSnapshots(context.Context, *ListSnapshotsRequest) (*ListSnapshotsResponse, error)
    ControllerExpandVolume(context.Context, *ControllerExpandVolumeRequest) (*ControllerExpandVolumeResponse, error)
    ControllerGetVolume(context.Context, *ControllerGetVolumeRequest) (*ControllerGetVolumeResponse, error)
}

下面除非特殊指明，都是external-provisioner触发调用的
CreateVolume: 创建盘，如果是云盘，那么就是调用RPC去外部创建一块云盘；如果是本地盘，那么可能就是要在某个目录下创建一个目录之类
DeleteVolume: 删除盘
ControllerPublishVolume: 挂载，如果是云盘，那么就是类似的mount -t nfs之类的，将远程盘挂载到本地的目录；如果是本地盘，其实没有什么作用。由external-attacher 调用
ControllerUnpublishVolume: 卸载，由external-attacher 调用
ValidateVolumeCapabilities: 通常来说是否允许volume被多个node挂载，例如某个数据库的主备要求使用同一块volume才会用到
ListVolumes: 顾名思义
GetCapacity: 返回盘剩余大小。非常关键的一个接口，当且仅当 External-Provisioner 以及csidriver 支持 storagecapacity时才会生效
https://github.com/kubernetes-csi/external-provisioner#capacity-support
ControllerGetCapabilities: 表示controller支持哪些接口，实际上外部组件先调用这个获取能力，然后再决定是否调用CreateVolume ControllerPublishVolume等
CreateSnapshot: 快照 external-snapshot 会调用该接口
DeleteSnapshot: 快照 external-snapshot 会调用该接口
ListSnapshots: 快照 external-snapshot 会调用该接口
ControllerExpandVolume: 扩容，注意CSI目前还不支持缩容。externa-resizer 会调用该接口
ControllerGetVolume: 顾名思义

node类

// NodeServer is the server API for Node service.
type NodeServer interface {
    
    NodeStageVolume(context.Context, *NodeStageVolumeRequest) (*NodeStageVolumeResponse, error)
    NodeUnstageVolume(context.Context, *NodeUnstageVolumeRequest) (*NodeUnstageVolumeResponse, error)
    NodePublishVolume(context.Context, *NodePublishVolumeRequest) (*NodePublishVolumeResponse, error)
    NodeUnpublishVolume(context.Context, *NodeUnpublishVolumeRequest) (*NodeUnpublishVolumeResponse, error)
    NodeGetVolumeStats(context.Context, *NodeGetVolumeStatsRequest) (*NodeGetVolumeStatsResponse, error)
    NodeExpandVolume(context.Context, *NodeExpandVolumeRequest) (*NodeExpandVolumeResponse, error)
    NodeGetCapabilities(context.Context, *NodeGetCapabilitiesRequest) (*NodeGetCapabilitiesResponse, error)
    NodeGetInfo(context.Context, *NodeGetInfoRequest) (*NodeGetInfoResponse, error)
}

上述函数 都被kubelet调用，kubelet如何知道这些接口和地址？这涉及Node-Driver-Registrar 它将这些接口告诉kubelet。

NodeStageVolume: 将指定的volume挂载到指定的global path上。实际上，你要做的就是执行bind mount操作，之所以存在global path，目的就是为了方便多个人挂载这个global path，做到readwritemany效果
NodeUnstageVolume: 如上
NodePublishVolume: 将指定的global path挂载到指定的pod path之下，这个pod path实际就是容器的path，容器起来时，又会将这个pod path bind mount到自己进程空间。这个函数要做的只是执行 bind mount操作
NodeUnpublishVolume: 如上
NodeGetVolumeStats:
NodeExpandVolume: ControllerExpandVolume的信息中若指定NodeExpansionRequired，则会被调用
NodeGetCapabilities: 和 ControllerGetCapabilities 类似，告诉kubelet自己支持哪些接口
NodeGetInfo: 告诉 kubelet 自己的nodeid、最大volume个数以及拓扑信息。拓扑信息会被打在node的label上，nodeid会出现在node的Annotations上，例如{csi.volume.kubernetes.io/nodeid: {disk.xxx.com: $nodeid}}

协调逻辑

以StatefulSet + volumeClaimTemplates + WaitForFirstConsumer 创建为例 ，当StatefulSet的yaml提交到apiserver后
1、CO 创建PVC
2、external-provisioner watch到这个PVC的创建，如果pvc中的volume.beta.kubernetes.io/storage-provisioner 是自己，则获得操作权
3、external-provisioner 判断 如果 PVC的类型是 WaitForFirstConsumer，则调度器必然已经创建好pod，并且pvc中的volume.kubernetes.io/selected-node写上了pod在那个node上
4、external-provisioner 通过unix socket调用本地的CSI pod的CreateVolume函数
5、external-provisioner 创建PV
6、CO等待PV创建完成后，会创建 VolumeAttachment 对象
7、external-attacherwatch 到VolumeAttachment 对象 后，通过unix socket调用本地的CSI pod的ControllerPublish函数，完成后设置.Status.Attached为true。如果csidriver设置了attachRequired为false的话则不走attach流程。
8、kubelet watch到CSI类型的PV调度到本地，于是等待VolumeAttachment 对象 .Status.Attached为true，在7执行完成后，则调用自己的MountDevice函数，它有通过unix socket调用NodeStageVolume和NodePublishVolume函数

storagecapacity

scheduler中，Filter->FindPodVolumes->checkVolumeProvisions，scheduler 在筛选pod的时候，会考虑storage的存储容量，这就是storagecapacity的能力，由CSI的controller GetCapacity接口吐出。同时它还有一个值可以告诉scheduler MaximumVolumeSize，有了这个值，多盘管理也非常方便。考虑这么一种情况，2个盘，各剩1G，你上报给scheduler自己容量2G的话，那么当一个2G的volume请求过来，实际上是无法分配的，因为2个盘是非线性，此时告诉scheduler MaximumVolumeSize = 1，可以使得scheduler只调度2个1G volume的POD。

func (pl *VolumeBinding) Filter(ctx context.Context, cs *framework.CycleState, pod *v1.Pod, nodeInfo *framework.NodeInfo) *framework.Status {
    
    node := nodeInfo.Node()
    if node == nil {
    
        return framework.NewStatus(framework.Error, "node not found")
    }

    state, err := getStateData(cs)
    if err != nil {
    
        return framework.AsStatus(err)
    }

    if state.skip {
    
        return nil
    }

    podVolumes, reasons, err := pl.Binder.FindPodVolumes(pod, state.boundClaims, state.claimsToBind, node)
    ......
}

本地盘CSI

这里粗略记录下几个关键点
1、storageclass的VolumeBindingMode字段需要是WaitForFirstConsumer，若非如此，volume会提前与pod调度前创建，volume和pod不在同一个node上
2、external-provisioner 需要启用 --node-deployment ，这个样 会根据 selected-node 判断谁负责处理volume，因为当前是 WaitForFirstConsumer模式，scheduler已经通过计算资源选出了node。当然如果不开启node-deployment，你需要自己实现rpc，让实际pod调度的node去处理createvolume请求。
3、external-provisioner和csidriver需要开启 storagecapacity特性，当然，对k8s版本也有要求，1.21以上(1.19需要改apiserver启动参数)https://github.com/kubernetes-csi/external-provisioner#capacity-support。如果没有这个特性，则CSI需要主动更新node.Status.Capacity 上报自定义资源的容量，然后再POD的resource中加上对应资源，从而达到资源调度的情况。
4、CreateVolume如果容量不够（并发创盘），可以返回ResourceExhausted类型的错误码，这样 external-provisioner 会删除PVC的selected-node(详见external-provisioner的rescheduleProvisioning函数)，使得scheduler重新调度POD，重新调度POD时，上面提到会考虑storagecapacity。

5、需要开启topology。external-provisioner 需要启用 --feature-gates=Topology=True，其次 CreateVolume+ NodeGetInfo 的响应中 AccessibleTopology需要带上nodeid。
关于topology，涉及到几方面，PV需要设置 https://kubernetes.io/docs/concepts/storage/persistent-volumes/#node-affinity ，而这部分逻辑由external-provisioner完成。为什么PV需要设置node-affinity，在K8s理念里面，PV是可以跨node访问的，你POD在nodeA是，PV在nodeB上也能访问，关于这一点，CSI相关代码的注释也提到了

type Volume struct {
    
    ......
    // Specifies where (regions, zones, racks, etc.) the provisioned
    // volume is accessible from.
    // A plugin that returns this field MUST also set the
    // VOLUME_ACCESSIBILITY_CONSTRAINTS plugin capability.
    // An SP MAY specify multiple topologies to indicate the volume is
    // accessible from multiple locations.
    // COs MAY use this information along with the topology information
    // returned by NodeGetInfo to ensure that a given volume is accessible
    // from a given node when scheduling workloads.
    // This field is OPTIONAL. If it is not specified, the CO MAY assume
    // the volume is equally accessible from all nodes in the cluster and
    // MAY schedule workloads referencing the volume on any available
    // node.
    //
    // Example 1:
    // accessible_topology = {"region": "R1", "zone": "Z2"}
    // Indicates a volume accessible only from the "region" "R1" and the
    // "zone" "Z2".
    //
    // Example 2:
    // accessible_topology =
    // {"region": "R1", "zone": "Z2"},
    // {"region": "R1", "zone": "Z3"}
    // Indicates a volume accessible from both "zone" "Z2" and "zone" "Z3"
    // in the "region" "R1".
    AccessibleTopology   []*Topology `protobuf:"bytes,5,rep,name=accessible_topology,json=accessibleTopology,proto3" json:"accessible_topology,omitempty"`

上面注释也提到了，如果不设置topology也就是PV没有node-affinity，新建的POD和PVC没问题，但是当POD被delete掉然后重新被调度时，并不会调度到PV所在的node上.

//boundClaims就是待调度的POD所关联的、已经绑定PV的PVC
func (b *volumeBinder) FindPodVolumes(pod *v1.Pod, boundClaims, claimsToBind []*v1.PersistentVolumeClaim, node *v1.Node) (podVolumes *PodVolumes, reasons ConflictReasons, err error) {
    
    ......
    // Check PV node affinity on bound volumes
    if len(boundClaims) > 0 {
    
        //核心是 checkBoundClaims
        boundVolumesSatisfied, boundPVsFound, err = b.checkBoundClaims(boundClaims, node, podName)
        if err != nil {
    
            return
        }
    }

    return
}

func (b *volumeBinder) checkBoundClaims(claims []*v1.PersistentVolumeClaim, node *v1.Node, podName string) (bool, bool, error) {
    
    csiNode, err := b.csiNodeLister.Get(node.Name)
    if err != nil {
    
        // TODO: return the error once CSINode is created by default
        klog.V(4).Infof("Could not get a CSINode object for the node %q: %v", node.Name, err)
    }

    //遍历所有PVC下的PV
    for _, pvc := range claims {
    
        pvName := pvc.Spec.VolumeName

        pv, err := b.pvCache.GetPV(pvName)
        if err != nil {
    
            if _, ok := err.(*errNotFound); ok {
    
                err = nil
            }
            return true, false, err
        }

        pv, err = b.tryTranslatePVToCSI(pv, csiNode)
        if err != nil {
    
            return false, true, err
        }

        //比较 NodeAffinity 和 node.Labels，注意 node.Labels 上是 NodeGetInfo 吐给kubelet，然后kubelet打到label上的
        err = volumeutil.CheckNodeAffinity(pv, node.Labels)
        if err != nil {
    
            klog.V(4).Infof("PersistentVolume %q, Node %q mismatch for Pod %q: %v", pvName, node.Name, podName, err)
            return false, true, nil
        }
        klog.V(5).Infof("PersistentVolume %q, Node %q matches for Pod %q", pvName, node.Name, podName)
    }

    klog.V(4).Infof("All bound volumes for Pod %q match with Node %q", podName, node.Name)
    return true, true, nil
}