k8s-编写CSI插件（3）

1、概述

在 Kubernetes 中，存储插件的开发主要有以下几种方式：

CSI插件：Container Storage Interface (CSI) 是 Kubernetes 的标准插件接口，是全新的插件方案，插件和驱动调用通过grpc协议，功能丰富，支持存储卷动态提供、快速、动态扩容等等。例如，可以连 in-tree 的插件做平滑迁移，当系统中运行有对应类型的 CSI 驱动时，其实使用方式虽然还是 in-tree 的方式，但 Kubernetes 已在后台默默为你替换成了CSI 插件方案，Volume 的 Attach/Detah、Mount/Unmount 等操作都会调用相应的 CSI 驱动完成。本文以 CSI 插件为例子学习。后续会详细介绍一下 CSI的实现原理和编写自己的你CSI驱动。
FlexVolume插件：FlexVolume 是 Kubernetes 的早期存储插件接口之一，它提供了一个简单的接口，但局限性却很大，用于将存储驱动程序接入到 Kubernetes 中。通过实现 FlexVolume 接口，可以将各种存储系统接入到 Kubernetes 集群中，包括 NFS、GlusterFS、Ceph 等等。
in-tree插件：in-tree 存储插件是 Kubernetes 的早期存储插件接口之一，它将存储驱动程序嵌入到 Kubernetes 主体代码库中。in-tree 插件可以实现对本地存储、NFS、iSCSI 等存储系统的支持。不过，由于 in-tree 插件需要嵌入到 Kubernetes 主体代码库中，因此对于插件开发者而言，维护成本较高，并且需要适应 Kubernetes 主体代码库的版本变化。

说明：CSI 插件是 Kubernetes 中推荐使用的存储插件接口，它提供了一种标准化的接口，更完善、更友好的编程插件方式，能够将各种存储系统集成到 Kubernetes 中。而 FlexVolume 插件和 in-tree 插件则是早期的存储插件接口，由于它们的维护成本较高，因此在新的 Kubernetes 版本中已经不再被推荐使用。

2、CSI 插件实现原理

CSI 的设计思想,把插件的职责从之前讲的 “两阶段处理”,扩展成了 Provision、Attach 和 Mount 三个阶段。趁着可以复习持久化 Volume 的两个阶段 k8s-持久化存储PV与PVC。

CSI 插件设计还将 kubernetes 里面的部分存储管理功能剥离出来，做成独立的外部组件（External Components）分为 Driver Register、External Provisioner 和 External Attacher。

2.1、CSI 处理的三个阶段

（1）Provision 阶段实现是指与外部存储供应商协凋卷 CreateVolume 和 DeleteVolume。假如外部存储供应商为 Google Cloud ，那么此阶段应该完成在 Google Cloud 存储商创建/删除一个指定大小的块存储设备。

（2）Attach 阶段是指将外部存储供应商提供好的卷存储设备挂载到本地或从本地卸载，其实就是实现 ControllerPublishVolume 和 ControllerUnpublishVolume。假如以外部存储供应商为Google Cloud存储为例，在 Provisioning 阶段创建好的卷的块设备，在此阶段应该实现将其挂载到服务器本地或从本地卸载，在必要的情况下还需要进行格式化等操作，但会在 Mount 进行格式化操作。

（3）Mount 阶段实现会当一个目标 Pod 在某个 Node 节点上调度时，kubelet 会根据前两个阶段返回的结果来创建这个 Pod。以外部存储供应商为Google Cloud云存储为例，此阶段将会把已经 Attaching 的本地块设备以目录形式挂载到 Pod 中或者从 Pod 中卸载这个块设备。

更加简单的描述：

Provision 等价于 “创建磁盘”；
Attach 等价于 “挂载磁盘到虚拟机”；
Mount 等价于 “将该磁盘格式化后，挂载在 Volume 的宿主机目录上”；

2.2、External Components

Provisioner、

（1）External Provisioner 组件：负责的是 Provision 阶段，它会去 Watch APIServer 里面的 PVC 对象，当有 PVC 被创建时，它会去调用 CSI Controller 的 CreateVolume 方法创建 PV 出来。

（2）External Attacher 组件：负责的是 Attach 阶段，它监听了 APIServer 里 VolumeAttachment 对象的变化，一旦有变化，它会去调用 CSI Controller 的 ControllerPublish 方法完成它所对应的 Volume 的 Attach 阶段。

（3）Driver Registrar 组件，负责将插件注册到 kubelet 里面。

关于描述，建议查阅文档

2.3 、CSI gRPC Server

CSI 的三大阶段实际上更细粒度的划分到 CSI Sidecar Containers 中，其实开发 CSI 实际上是面向 CSI Sidecar Containers 编程。针对于 CSI Sidecar Containers 主要实现 CSI 插件以 gRPC 的方式对外提供的三个 gRPC Service，分别叫作：CSI Identity Service、CSI Controller Service 和 CSI Node Service，这三个服务也是需要我们去编写代码来实现的 CSI 插件。

（1）Identity Service

CSI Identity 服务，主要负责对外暴露这个插件本身的信息，所定义的 proto 文件如下：

service Identity {
// return the version and name of the plugin
rpc GetPluginInfo(GetPluginInfoRequest) returns (GetPluginInfoResponse) {}// 报告插件是否具有为控制器接口提供服务的能力
rpc GetPluginCapabilities(GetPluginCapabilitiesRequest) returns (GetPluginCapabilitiesResponse) {}// 由 CO 调用只是为了检查插件是否正在运行
rpc Probe (ProbeRequest) returns (ProbeResponse) {}}

IdentityServer 定义的接口如下：

// IdentityServer is the server API for Identity service.type IdentityServer interface {GetPluginInfo(context.Context, *GetPluginInfoRequest) (*GetPluginInfoResponse, error)GetPluginCapabilities(context.Context, *GetPluginCapabilitiesRequest) (*GetPluginCapabilitiesResponse, error)Probe(context.Context, *ProbeRequest) (*ProbeResponse, error)}

（2）Controller Service

CSI Controller 服务，定义的是对 CSI Volume（对应 Kubernetes 里的 PV）的管理接口，比如：创建和删除 CSI Volume、对 CSI Volume 进行 Attach/Dettach（在 CSI 里，这个操作被叫作 Publish/Unpublish），以及对 CSI Volume 进行 Snapshot 等。

CSI Controller 服务里定义的这些操作大部分的核心逻辑应该在 ControllerServer 中实现，比如创建/销毁 Volume，创建/销毁 Snapshot 等。在实际开发中，自己编写的 CSI 都会实现 CreateVolume 和 DeleteVolume，至于其他方法根据业务需求以及外部存储供应商实际情况来决定是否进行实现。

接口定义如下所示：

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

（3）CSI Node Service

在 Mount 阶段 kubelet 会通过 node-driver-registrar 容器调用这三个方法：NodePublishVolume、NodeUnpublishVolume、NodeGetCapabilities 。

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

proto 文件源码位置：https://github.com/container-storage-interface/spec/blob/master/csi.proto
go文件源码：https://github.com/container-storage-interface/spec/blob/master/lib/go/csi/csi.pb.go

3、编写一个 CSI 插件

3.1 介绍

了解 CSI 插件机制原理及相关概念后，接下来实战一个自己的 CSI 插件。CSI 插件的代码结构比较清晰的，可以参考 csi-driver-host-path 、csi-digitalocean ，正如 csi-driver-host-path 代码结构如下：

外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传

可以看到，IdentityServer 、ControllerServer 、NodeServer 三个服务都定义在了 pkg/hostpath 目录下。

为了能够让 Kubernetes 访问到 CSI 的三个服务，需要定义一个标准的 gRPC Server，把编写好的 gRPC Server 注册给 CSI，然后它就可以响应来自 External Components 的 CSI 请求了。 csi-driver-host-path 项目是写在了 server.go 文件夹里头的：


func (s *nonBlockingGRPCServer) serve(ep string, ids csi.IdentityServer, cs csi.ControllerServer, ns csi.NodeServer) {listener, cleanup, err := endpoint.Listen(ep)...server := grpc.NewServer(opts...)s.server = servers.cleanup = cleanupif ids != nil {csi.RegisterIdentityServer(server, ids)}if cs != nil {csi.RegisterControllerServer(server, cs)}if ns != nil {csi.RegisterNodeServer(server, ns)}...
server.Serve(listener)}

3.2 CSI 插件，需要实现的三个服务

（1）CSI Identity 服务

CSI Identity 服务，主要负责对外暴露这个插件本身的信息


package driver  import (  `context`  "github.com/container-storage-interface/spec/lib/go/csi""github.com/sirupsen/logrus"
)  // GetPluginInfo 返回插件名字和版本
func (d *Driver) GetPluginInfo(ctx context.Context, req *csi.GetPluginInfoRequest) (*csi.GetPluginInfoResponse, error) {  logrus.Infof("GetPluginInfo: called with args: %+v", *req)  return &csi.GetPluginInfoResponse{  // K8s 通过 DriverName 这个值来找到在 StorageClass 里声明要使用的 CSI 插件的Name:          d.config.DriverName,VendorVersion: d.config.VendorVersion,  }, nil  
}  // GetPluginCapabilities 返回插件所支持的功能  
func (d *Driver) GetPluginCapabilities(ctx context.Context, req *csi.GetPluginCapabilitiesRequest) (*csi.GetPluginCapabilitiesResponse, error) {  logrus.Infof("GetPluginCapabilities: called with args: %+v", *req)  caps := []*csi.PluginCapability{  {  Type: &csi.PluginCapability_Service_{  Service: &csi.PluginCapability_Service{  Type: csi.PluginCapability_Service_CONTROLLER_SERVICE,  },  },  },  {  Type: &csi.PluginCapability_Service_{  Service: &csi.PluginCapability_Service{  Type: csi.PluginCapability_Service_VOLUME_ACCESSIBILITY_CONSTRAINTS,  },  },  },  }  return &csi.GetPluginCapabilitiesResponse{Capabilities: caps}, nil  
}  // K8S 调用它来检查这个 CSI 插件是否正常工作（健康检测 ）。
func (d *Driver) Probe(ctx context.Context, req *csi.ProbeRequest) (*csi.ProbeResponse, error) {  logrus.Infof("Probe: called with args %+v", req)  return &csi.ProbeResponse{}, nil  
}

（2）CSI Controller 服务

这个服务主要任务是负责 “Provision ”和“Attach ” 阶段。

Provision 阶段对应的接口，是 CreateVolume 和 DeleteVolume，它们的调用者是 External Provisoner。代码如下所示：

package driver  import (  `context`  `github.com/container-storage-interface/spec/lib/go/csi`     `github.com/sirupsen/logrus`
)  var (  // controllerCaps 代表Controller Plugin支持的功能// 这里只实现Volume的创建/删除，附加/卸载  controllerCaps = []csi.ControllerServiceCapability_RPC_Type{  csi.ControllerServiceCapability_RPC_CREATE_DELETE_VOLUME,  csi.ControllerServiceCapability_RPC_PUBLISH_UNPUBLISH_VOLUME,  }  
)  // CreateVolume 创建  
func (d *Driver) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequest) (resp *csi.CreateVolumeResponse, finalErr error) {  logrus.Infof("CreateVolume: called with args %+v", *req)  // 对于nfs来说，这里就是创建一个存储卷，其它类型块存储如Ceph RDB 也类似vol, err := hp.createVolume(volumeID, req.GetName(), capacity, requestedAccessType, ...)glog.V(4).Infof("created volume %s at path %s", vol.VolID, vol.VolPath)vmReq := &csi.Volume{  VolumeId:      "w-123",  CapacityBytes: 10 * (1 << 30),  VolumeContext: req.GetParameters(),  }  return &csi.CreateVolumeResponse{Volume: vmReq}, nil  
}  // DeleteVolume 删除  
func (d *Driver) DeleteVolume(ctx context.Context, req *csi.DeleteVolumeRequest) (*csi.DeleteVolumeResponse, error) {  logrus.Infof("DeleteVolume: called with args %+v", *req)  return &csi.DeleteVolumeResponse{}, nil  
}

Attach 阶段”对应的接口是 ControllerPublishVolume 和 ControllerUnpublishVolume，它们的调用者是 External Attacher。

Attach 阶段主要的工作是调用供应商提供存储 API，将前面创建好的存储卷，挂载到指定的虚拟机上。


// ControllerPublishVolume 附加  
func (d *Driver) ControllerPublishVolume(ctx context.Context, req *csi.ControllerPublishVolumeRequest) (*csi.ControllerPublishVolumeResponse, error) {  logrus.Infof("ControllerPublishVolume: called with args %+v", *req)  pvInfo := map[string]string{"DevicePathKey": "/dev/sdb"}  return &csi.ControllerPublishVolumeResponse{PublishContext: pvInfo}, nil  
}  // ControllerUnpublishVolume 卸载  
func (d *Driver) ControllerUnpublishVolume(ctx context.Context, req *csi.ControllerUnpublishVolumeRequest) (*csi.ControllerUnpublishVolumeResponse, error) {  logrus.Infof("ControllerUnpublishVolume: called with args %+v", *req)  return &csi.ControllerUnpublishVolumeResponse{}, nil  
}

（3）CSI Node 服务

CSI Node 服务所对应的是 “Mount 阶段”，而调用 CSI Node 服务来完成 “Mount 阶段”的是 “kubelet 的 VolumeManagerReconciler 控制循环”。

在 “Mount 阶段” 中，它是手续需要格式化这个设备，然后才能把它挂载到 Volume 对应的宿主机目录上。

在 kubelet 的 VolumeManagerReconciler 控制循环中，这两步操作分别叫作 MountDevice 和 SetUp，SetUp 操作则会调用 CSI Node 服务的 NodePublishVolume 接口，而 MountDevice 操作，就是直接调用了 CSI Node 服务里的 NodeStageVolume 接口。

nodeServer go文件中两个重要接口：NodeStageVolume 和 NodePublishVolume

// 格式化 Volume 在宿主机上对应的存储设备，然后挂载到一个临时目录（Staging 目录）上。
func (hp *hostPath) NodeStageVolume(ctx context.Context, req *csi.NodeStageVolumeRequest) (*csi.NodeStageVolumeResponse, error) {  // Check arguments  if len(req.GetVolumeId()) == 0 {  return nil, status.Error(codes.InvalidArgument, "Volume ID missing in request")  }  stagingTargetPath := req.GetStagingTargetPath()  if stagingTargetPath == "" {  return nil, status.Error(codes.InvalidArgument, "Target path missing in request")  }  if req.GetVolumeCapability() == nil {  return nil, status.Error(codes.InvalidArgument, "Volume Capability missing in request")  }  // Lock before acting on global state. A production-quality  // driver might use more fine-grained locking.   hp.mutex.Lock()  defer hp.mutex.Unlock()  vol, err := hp.state.GetVolumeByID(req.VolumeId)  if err != nil {  return nil, err  }  if hp.config.EnableAttach && !vol.Attached {  return nil, status.Errorf(codes.Internal, "ControllerPublishVolume must be called on volume '%s' before staging on node",  vol.VolID)  }  if vol.Staged.Has(stagingTargetPath) {  glog.V(4).Infof("Volume %q is already staged at %q, nothing to do.", req.VolumeId, stagingTargetPath)  return &csi.NodeStageVolumeResponse{}, nil  }  if !vol.Staged.Empty() {  return nil, status.Errorf(codes.FailedPrecondition, "volume %q is already staged at %v", req.VolumeId, vol.Staged)  }  vol.Staged.Add(stagingTargetPath)  if err := hp.state.UpdateVolume(vol); err != nil {  return nil, err  }  return &csi.NodeStageVolumeResponse{}, nil  
}

总结

1、以上主要是分析 nfs 的 CSI 插件为例，大致了解编写 CSI 插件的流程。CSI 开发其实是针对 Kubernetes CSI Sidecar Containers 的 gRPC 开发，根据 CSI 规范实现驱动程序，开发者可以根据需求，完成三大阶段中对应三大 gRPC Server 相应方法即可。需要实现以下接口：