CSI (Container Storage Interface) 原理深度解析

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

service Identity {
    // 返回驱动名称和版本
    rpc GetPluginInfo(GetPluginInfoRequest) returns (GetPluginInfoResponse) {}

    // 返回驱动支持的能力
    rpc GetPluginCapabilities(GetPluginCapabilitiesRequest) returns (GetPluginCapabilitiesResponse) {}

    // 健康检查
    rpc Probe(ProbeRequest) returns (ProbeResponse) {}
}

 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

service Controller {
    // 创建卷
    rpc CreateVolume(CreateVolumeRequest) returns (CreateVolumeResponse) {}

    // 删除卷
    rpc DeleteVolume(DeleteVolumeRequest) returns (DeleteVolumeResponse) {}

    // 将卷附加到节点（控制面操作）
    rpc ControllerPublishVolume(ControllerPublishVolumeRequest) returns (ControllerPublishVolumeResponse) {}

    // 从节点分离卷
    rpc ControllerUnpublishVolume(ControllerUnpublishVolumeRequest) returns (ControllerUnpublishVolumeResponse) {}

    // 验证卷能力
    rpc ValidateVolumeCapabilities(ValidateVolumeCapabilitiesRequest) returns (ValidateVolumeCapabilitiesResponse) {}

    // 列出卷
    rpc ListVolumes(ListVolumesRequest) returns (ListVolumesResponse) {}

    // 获取存储容量
    rpc GetCapacity(GetCapacityRequest) returns (GetCapacityResponse) {}

    // 获取控制器能力
    rpc ControllerGetCapabilities(ControllerGetCapabilitiesRequest) returns (ControllerGetCapabilitiesResponse) {}

    // 创建快照
    rpc CreateSnapshot(CreateSnapshotRequest) returns (CreateSnapshotResponse) {}

    // 删除快照
    rpc DeleteSnapshot(DeleteSnapshotRequest) returns (DeleteSnapshotResponse) {}

    // 扩容卷
    rpc ControllerExpandVolume(ControllerExpandVolumeRequest) returns (ControllerExpandVolumeResponse) {}
}

 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

service Node {
    // 将卷挂载到全局目录（Stage）
    rpc NodeStageVolume(NodeStageVolumeRequest) returns (NodeStageVolumeResponse) {}

    // 从全局目录卸载卷
    rpc NodeUnstageVolume(NodeUnstageVolumeRequest) returns (NodeUnstageVolumeResponse) {}

    // 将卷挂载到 Pod 目录（Publish）
    rpc NodePublishVolume(NodePublishVolumeRequest) returns (NodePublishVolumeResponse) {}

    // 从 Pod 目录卸载卷
    rpc NodeUnpublishVolume(NodeUnpublishVolumeRequest) returns (NodeUnpublishVolumeResponse) {}

    // 获取卷统计信息
    rpc NodeGetVolumeStats(NodeGetVolumeStatsRequest) returns (NodeGetVolumeStatsResponse) {}

    // 节点上扩容卷
    rpc NodeExpandVolume(NodeExpandVolumeRequest) returns (NodeExpandVolumeResponse) {}

    // 获取节点能力
    rpc NodeGetCapabilities(NodeGetCapabilitiesRequest) returns (NodeGetCapabilitiesResponse) {}

    // 获取节点信息
    rpc NodeGetInfo(NodeGetInfoRequest) returns (NodeGetInfoResponse) {}
}

  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

// 驱动主入口
func main() {
    // 创建驱动实例
    driver := NewDriver(
        driverName,     // 如: "csi.example.com"
        nodeID,         // 节点标识
        endpoint,       // Unix socket 路径
    )

    // 启动 gRPC 服务
    driver.Run()
}

// Identity Server 实现
type IdentityServer struct {
    name    string
    version string
}

func (is *IdentityServer) GetPluginInfo(ctx context.Context, req *csi.GetPluginInfoRequest) (*csi.GetPluginInfoResponse, error) {
    return &csi.GetPluginInfoResponse{
        Name:          is.name,
        VendorVersion: is.version,
    }, nil
}

func (is *IdentityServer) GetPluginCapabilities(ctx context.Context, req *csi.GetPluginCapabilitiesRequest) (*csi.GetPluginCapabilitiesResponse, error) {
    return &csi.GetPluginCapabilitiesResponse{
        Capabilities: []*csi.PluginCapability{
            {
                Type: &csi.PluginCapability_Service_{
                    Service: &csi.PluginCapability_Service{
                        Type: csi.PluginCapability_Service_CONTROLLER_SERVICE,
                    },
                },
            },
        },
    }, nil
}

// Controller Server 实现
type ControllerServer struct {
    // 存储后端客户端
    client StorageClient
}

func (cs *ControllerServer) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequest) (*csi.CreateVolumeResponse, error) {
    // 1. 验证请求参数
    if req.GetName() == "" {
        return nil, status.Error(codes.InvalidArgument, "Volume name required")
    }

    // 2. 调用存储后端创建卷
    vol, err := cs.client.CreateVolume(req.GetName(), req.GetCapacityRange().GetRequiredBytes())
    if err != nil {
        return nil, status.Error(codes.Internal, err.Error())
    }

    // 3. 返回卷信息
    return &csi.CreateVolumeResponse{
        Volume: &csi.Volume{
            VolumeId:      vol.ID,
            CapacityBytes: vol.Size,
            VolumeContext: map[string]string{
                "storage-pool": vol.Pool,
            },
        },
    }, nil
}

// Node Server 实现
type NodeServer struct {
    nodeID string
    mounter mount.Interface
}

func (ns *NodeServer) NodeStageVolume(ctx context.Context, req *csi.NodeStageVolumeRequest) (*csi.NodeStageVolumeResponse, error) {
    // 1. 获取设备路径
    devicePath := req.GetPublishContext()["devicePath"]
    stagingPath := req.GetStagingTargetPath()
    fsType := req.GetVolumeCapability().GetMount().GetFsType()

    // 2. 格式化设备（如果需要）
    if err := ns.formatDevice(devicePath, fsType); err != nil {
        return nil, err
    }

    // 3. 挂载到 staging 目录
    if err := ns.mounter.Mount(devicePath, stagingPath, fsType, nil); err != nil {
        return nil, err
    }

    return &csi.NodeStageVolumeResponse{}, nil
}

func (ns *NodeServer) NodePublishVolume(ctx context.Context, req *csi.NodePublishVolumeRequest) (*csi.NodePublishVolumeResponse, error) {
    stagingPath := req.GetStagingTargetPath()
    targetPath := req.GetTargetPath()

    // 使用 bind mount 将 staging 目录挂载到 Pod 目录
    mountOptions := []string{"bind"}
    if req.GetReadonly() {
        mountOptions = append(mountOptions, "ro")
    }

    if err := ns.mounter.Mount(stagingPath, targetPath, "", mountOptions); err != nil {
        return nil, err
    }

    return &csi.NodePublishVolumeResponse{}, 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

# CSI Controller Deployment
apiVersion: apps/v1
kind: Deployment
metadata:
  name: csi-controller
spec:
  replicas: 1
  template:
    spec:
      containers:
        # CSI 驱动容器
        - name: csi-driver
          image: example/csi-driver:v1.0
          args:
            - "--endpoint=unix:///csi/csi.sock"
            - "--nodeid=$(NODE_ID)"
          volumeMounts:
            - name: socket-dir
              mountPath: /csi

        # External Provisioner Sidecar
        - name: csi-provisioner
          image: registry.k8s.io/sig-storage/csi-provisioner:v5.0.1
          args:
            - "--csi-address=/csi/csi.sock"
            - "--feature-gates=Topology=true"
          volumeMounts:
            - name: socket-dir
              mountPath: /csi

        # External Attacher Sidecar
        - name: csi-attacher
          image: registry.k8s.io/sig-storage/csi-attacher:v4.6.1
          args:
            - "--csi-address=/csi/csi.sock"
          volumeMounts:
            - name: socket-dir
              mountPath: /csi

      volumes:
        - name: socket-dir
          emptyDir: {}

---
# CSI Node DaemonSet
apiVersion: apps/v1
kind: DaemonSet
metadata:
  name: csi-node
spec:
  template:
    spec:
      containers:
        # CSI 驱动容器
        - name: csi-driver
          image: example/csi-driver:v1.0
          args:
            - "--endpoint=unix:///csi/csi.sock"
            - "--nodeid=$(NODE_ID)"
          securityContext:
            privileged: true
          volumeMounts:
            - name: socket-dir
              mountPath: /csi
            - name: pods-mount-dir
              mountPath: /var/lib/kubelet/pods
              mountPropagation: Bidirectional
            - name: device-dir
              mountPath: /dev

        # Node Driver Registrar Sidecar
        - name: node-driver-registrar
          image: registry.k8s.io/sig-storage/csi-node-driver-registrar:v2.13.0
          args:
            - "--csi-address=/csi/csi.sock"
            - "--kubelet-registration-path=/var/lib/kubelet/plugins/csi.example.com/csi.sock"
          volumeMounts:
            - name: socket-dir
              mountPath: /csi
            - name: registration-dir
              mountPath: /registration

      volumes:
        - name: socket-dir
          hostPath:
            path: /var/lib/kubelet/plugins/csi.example.com
            type: DirectoryOrCreate
        - name: registration-dir
          hostPath:
            path: /var/lib/kubelet/plugins_registry
            type: Directory
        - name: pods-mount-dir
          hostPath:
            path: /var/lib/kubelet/pods
            type: Directory
        - name: device-dir
          hostPath:
            path: /dev
            type: Directory

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

// RegistrationHandler 处理 CSI 驱动注册
type RegistrationHandler struct {
    csiPlugin *csiPlugin
}

// RegisterPlugin 在驱动注册时调用
func (h *RegistrationHandler) RegisterPlugin(pluginName string, endpoint string, versions []string, pluginClientTimeout *time.Duration) error {
    // 1. 存储驱动信息
    csiDrivers.Set(pluginName, Driver{
        endpoint:                endpoint,
        highestSupportedVersion: highestSupportedVersion,
    })

    // 2. 获取节点信息
    csi, err := newCsiDriverClient(csiDriverName(pluginName))
    nodeID, maxVolumePerNode, accessibleTopology, err := csi.NodeGetInfo(ctx)

    // 3. 更新 CSINode 对象
    err = nim.InstallCSIDriver(pluginName, nodeID, maxVolumePerNode, accessibleTopology)

    return 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

# CSIDriver - 描述 CSI 驱动能力
apiVersion: storage.k8s.io/v1
kind: CSIDriver
metadata:
  name: csi.example.com
spec:
  attachRequired: true          # 是否需要 Attach 操作
  podInfoOnMount: true          # 是否传递 Pod 信息
  fsGroupPolicy: File           # FSGroup 策略
  volumeLifecycleModes:
    - Persistent                # 支持持久卷
    - Ephemeral                 # 支持临时卷
  tokenRequests:                # SA Token 请求
    - audience: "api"
  requiresRepublish: false      # 是否需要重新发布
  seLinuxMount: false           # SELinux 挂载支持

---
# CSINode - 描述节点上的 CSI 驱动
apiVersion: storage.k8s.io/v1
kind: CSINode
metadata:
  name: node-1
spec:
  drivers:
    - name: csi.example.com
      nodeID: node-1-id
      topologyKeys:
        - topology.kubernetes.io/zone
      allocatable:
        count: 100              # 最大卷数量

---
# VolumeAttachment - 表示卷附加状态
apiVersion: storage.k8s.io/v1
kind: VolumeAttachment
metadata:
  name: csi-xxx
spec:
  attacher: csi.example.com
  nodeName: node-1
  source:
    persistentVolumeName: pv-xxx
status:
  attached: true
  attachmentMetadata:
    devicePath: /dev/sdb

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

func (ctrl *PersistentVolumeController) syncUnboundClaim(ctx context.Context, claim *v1.PersistentVolumeClaim) error {
    // 1. 查找匹配的 PV
    volume, err := ctrl.volumes.findBestMatchForClaim(claim, delayBinding)

    if volume == nil {
        // 2. 没有匹配的 PV，检查是否可以动态供应
        if ctrl.shouldProvision(claim) {
            // External Provisioner 会处理
            return nil
        }
    }

    // 3. 绑定 PV 和 PVC
    return ctrl.bind(ctx, volume, claim)
}

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

func (c *csiAttacher) Attach(spec *volume.Spec, nodeName types.NodeName) (string, error) {
    // 1. 创建 VolumeAttachment 对象
    attachment := &storage.VolumeAttachment{
        ObjectMeta: metav1.ObjectMeta{
            Name: attachID,
        },
        Spec: storage.VolumeAttachmentSpec{
            Attacher: pvSrc.Driver,
            NodeName: string(nodeName),
            Source: storage.VolumeAttachmentSource{
                PersistentVolumeName: &pvName,
            },
        },
    }

    _, err := c.k8s.StorageV1().VolumeAttachments().Create(ctx, attachment, metav1.CreateOptions{})

    // 2. 等待 VolumeAttachment 完成
    return c.waitForVolumeAttachment(ctx, attachID, timeout)
}

 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

func (c *csiMountMgr) SetUpAt(dir string, mounterArgs volume.MounterArgs) error {
    // 1. 获取 publishContext（包含设备路径）
    publishContext, err := c.getPublishContext()

    // 2. 调用 NodeStageVolume（如果支持）
    if stageUnstageSet {
        err = csi.NodeStageVolume(ctx,
            c.volumeID,
            publishContext,
            stagingTargetPath,
            fsType,
            accessMode,
            nodeStageSecrets,
            volumeContext,
            mountOptions,
            fsGroup,
        )
    }

    // 3. 调用 NodePublishVolume
    err = csi.NodePublishVolume(ctx,
        c.volumeID,
        c.readOnly,
        stagingTargetPath,
        dir,                    // Pod 挂载目录
        accessMode,
        publishContext,
        volumeContext,
        nodePublishSecrets,
        fsType,
        mountOptions,
        fsGroup,
    )

    return 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

# 1. 查看块设备
lsblk
# NAME   MAJ:MIN RM   SIZE RO TYPE MOUNTPOINT
# sda      8:0    0   100G  0 disk
# └─sda1   8:1    0   100G  0 part /
# sdb      8:16   0    50G  0 disk /var/lib/kubelet/plugins/...

# 2. 格式化设备
mkfs.ext4 /dev/sdb

# 3. 挂载设备
mount /dev/sdb /var/lib/kubelet/plugins/kubernetes.io~csi/driver/vol-xxx/globalmount

# 4. Bind Mount
mount --bind /source /target

# 5. 查看挂载点
findmnt /var/lib/kubelet/pods/xxx/volumes/kubernetes.io~csi/pvc-xxx/mount
# TARGET                                                    SOURCE     FSTYPE OPTIONS
# /var/lib/kubelet/pods/.../mount                          /dev/sdb   ext4   rw,relatime

# 6. 查看挂载传播
cat /proc/self/mountinfo | grep kubelet
# 可以看到 shared, slave, private 等传播类型

1
2
3
4

volumeMounts:
  - name: pods-mount-dir
    mountPath: /var/lib/kubelet/pods
    mountPropagation: Bidirectional  # 关键配置

 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

// CSI 客户端接口 (csi_client.go)
type csiClient interface {
    NodeGetInfo(ctx context.Context) (nodeID string, maxVolumePerNode int64, accessibleTopology map[string]string, err error)
    NodePublishVolume(ctx context.Context, volumeid string, readOnly bool, stagingTargetPath string, targetPath string, accessMode api.PersistentVolumeAccessMode, publishContext map[string]string, volumeContext map[string]string, secrets map[string]string, fsType string, mountOptions []string, fsGroup *int64) error
    NodeUnpublishVolume(ctx context.Context, volID string, targetPath string) error
    NodeStageVolume(ctx context.Context, volID string, publishVolumeInfo map[string]string, stagingTargetPath string, fsType string, accessMode api.PersistentVolumeAccessMode, secrets map[string]string, volumeContext map[string]string, mountOptions []string, fsGroup *int64) error
    NodeUnstageVolume(ctx context.Context, volID, stagingTargetPath string) error
    NodeSupportsStageUnstage(ctx context.Context) (bool, error)
    // ... 更多方法
}

// Volume Mounter (csi_mounter.go)
type csiMountMgr struct {
    csiClientGetter
    k8s                 kubernetes.Interface
    plugin              *csiPlugin
    driverName          csiDriverName
    volumeLifecycleMode storage.VolumeLifecycleMode
    volumeID            string
    specVolumeID        string
    readOnly            bool
    spec                *volume.Spec
    pod                 *api.Pod
    podUID              types.UID
    publishContext      map[string]string
    kubeVolHost         volume.KubeletVolumeHost
}

// Volume Attacher (csi_attacher.go)
type csiAttacher struct {
    plugin       *csiPlugin
    k8s          kubernetes.Interface
    watchTimeout time.Duration
    csiClient    csiClient
}