csi-provisioner源码分析
本文个人博客地址:https://www.huweihuang.com/kubernetes-notes/develop/csi-provisioner.html
本文主要分析
csi-provisioner
的源码,关于开发一个Dynamic Provisioner
,具体可参考nfs-client-provisioner的源码分析
1. Dynamic Provisioner
1.1. Provisioner Interface
开发Dynamic Provisioner
需要实现Provisioner接口,该接口有两个方法,分别是:
- Provision:创建存储资源,并且返回一个PV对象。
- Delete:移除对应的存储资源,但并没有删除PV对象。
1.2. 开发provisioner的步骤
- 写一个
provisioner
实现Provisioner
接口(包含Provision
和Delete
的方法)。 - 通过该
provisioner
构建ProvisionController
。 - 执行
ProvisionController
的Run
方法。
2. CSI Provisioner
CSI Provisioner的源码可参考:https://github.com/kubernetes-csi/external-provisioner。
2.1. Main 函数
2.1.1. 读取环境变量
源码如下:
var (provisioner = flag.String("provisioner", "", "Name of the provisioner. The provisioner will only provision volumes for claims that request a StorageClass with a provisioner field set equal to this name.")master = flag.String("master", "", "Master URL to build a client config from. Either this or kubeconfig needs to be set if the provisioner is being run out of cluster.")kubeconfig = flag.String("kubeconfig", "", "Absolute path to the kubeconfig file. Either this or master needs to be set if the provisioner is being run out of cluster.")csiEndpoint = flag.String("csi-address", "/run/csi/socket", "The gRPC endpoint for Target CSI Volume")connectionTimeout = flag.Duration("connection-timeout", 10*time.Second, "Timeout for waiting for CSI driver socket.")volumeNamePrefix = flag.String("volume-name-prefix", "pvc", "Prefix to apply to the name of a created volume")volumeNameUUIDLength = flag.Int("volume-name-uuid-length", -1, "Truncates generated UUID of a created volume to this length. Defaults behavior is to NOT truncate.")showVersion = flag.Bool("version", false, "Show version.")provisionController *controller.ProvisionControllerversion = "unknown"
)func init() {var config *rest.Configvar err errorflag.Parse()flag.Set("logtostderr", "true")if *showVersion {fmt.Println(os.Args[0], version)os.Exit(0)}glog.Infof("Version: %s", version)...
}
通过init函数
解析相关参数,其实provisioner
指明为PVC提供PV的provisioner的名字,需要和StorageClass
对象中的provisioner
字段一致。
2.1.2. 获取clientset对象
源码如下:
// get the KUBECONFIG from env if specified (useful for local/debug cluster)
kubeconfigEnv := os.Getenv("KUBECONFIG")
if kubeconfigEnv != "" {glog.Infof("Found KUBECONFIG environment variable set, using that..")kubeconfig = &kubeconfigEnv
}
if *master != "" || *kubeconfig != "" {glog.Infof("Either master or kubeconfig specified. building kube config from that..")config, err = clientcmd.BuildConfigFromFlags(*master, *kubeconfig)
} else {glog.Infof("Building kube configs for running in cluster...")config, err = rest.InClusterConfig()
}
if err != nil {glog.Fatalf("Failed to create config: %v", err)
}
clientset, err := kubernetes.NewForConfig(config)
if err != nil {glog.Fatalf("Failed to create client: %v", err)
}// snapclientset.NewForConfig creates a new Clientset for VolumesnapshotV1alpha1Client
snapClient, err := snapclientset.NewForConfig(config)
if err != nil {glog.Fatalf("Failed to create snapshot client: %v", err)
}
csiAPIClient, err := csiclientset.NewForConfig(config)
if err != nil {glog.Fatalf("Failed to create CSI API client: %v", err)
}
通过读取对应的k8s的配置,创建clientset
对象,用来执行k8s对应的API,其中主要包括对PV和PVC等对象的创建删除等操作。
2.1.3. k8s版本校验
// The controller needs to know what the server version is because out-of-tree
// provisioners aren't officially supported until 1.5
serverVersion, err := clientset.Discovery().ServerVersion()
if err != nil {glog.Fatalf("Error getting server version: %v", err)
}
获取了k8s的版本信息,因为provisioners的功能在k8s 1.5及以上版本才支持。
2.1.4. 连接 csi socket
// Generate a unique ID for this provisioner
timeStamp := time.Now().UnixNano() / int64(time.Millisecond)
identity := strconv.FormatInt(timeStamp, 10) + "-" + strconv.Itoa(rand.Intn(10000)) + "-" + *provisioner// Provisioner will stay in Init until driver opens csi socket, once it's done
// controller will exit this loop and proceed normally.
socketDown := true
grpcClient := &grpc.ClientConn{}
for socketDown {grpcClient, err = ctrl.Connect(*csiEndpoint, *connectionTimeout)if err == nil {socketDown = falsecontinue}time.Sleep(10 * time.Second)
}
在Provisioner
会停留在初始化状态,直到csi socket
连接成功才正常运行。如果连接失败,会暂停10秒
后重试,其中涉及以下2个参数:
- csiEndpoint:CSI Volume的gRPC地址,默认通过为
/run/csi/socket
。 - connectionTimeout:连接CSI driver socket的超时时间,默认为10秒。
2.1.5. 构造csi-Provisioner对象
// Create the provisioner: it implements the Provisioner interface expected by
// the controller
csiProvisioner := ctrl.NewCSIProvisioner(clientset, csiAPIClient, *csiEndpoint, *connectionTimeout, identity, *volumeNamePrefix, *volumeNameUUIDLength, grpcClient, snapClient)
provisionController = controller.NewProvisionController(clientset,*provisioner,csiProvisioner,serverVersion.GitVersion,
)
通过参数clientset
,csiAPIClient
, csiEndpoint
, connectionTimeout
, identity
, volumeNamePrefix
, volumeNameUUIDLength
,grpcClient
, snapClient
构造csi-Provisioner对象。
通过csiProvisioner
构造ProvisionController
对象。
2.1.6. 运行ProvisionController
func main() {provisionController.Run(wait.NeverStop)
}
ProvisionController
实现了具体的PV和PVC的相关逻辑,Run
方法以常驻进程的方式运行。
2.2. Provision和Delete方法
2.2.1. Provision方法
csiProvisioner
的Provision
方法具体源码参考:https://github.com/kubernetes-csi/external-provisioner/blob/master/pkg/controller/controller.go#L336
Provision
方法用来创建存储资源,并且返回一个PV
对象。其中入参是VolumeOptions
,用来指定PV
对象的相关属性。
1、构造PV相关属性
pvName, err := makeVolumeName(p.volumeNamePrefix, fmt.Sprintf("%s", options.PVC.ObjectMeta.UID), p.volumeNameUUIDLength)
if err != nil {return nil, err
}
2、构造CSIPersistentVolumeSource相关属性
driverState, err := checkDriverState(p.grpcClient, p.timeout, needSnapshotSupport)
if err != nil {return nil, err
}...
// Resolve controller publish, node stage, node publish secret references
controllerPublishSecretRef, err := getSecretReference(controllerPublishSecretNameKey, controllerPublishSecretNamespaceKey, options.Parameters, pvName, options.PVC)
if err != nil {return nil, err
}
nodeStageSecretRef, err := getSecretReference(nodeStageSecretNameKey, nodeStageSecretNamespaceKey, options.Parameters, pvName, options.PVC)
if err != nil {return nil, err
}
nodePublishSecretRef, err := getSecretReference(nodePublishSecretNameKey, nodePublishSecretNamespaceKey, options.Parameters, pvName, options.PVC)
if err != nil {return nil, err
}...
volumeAttributes := map[string]string{provisionerIDKey: p.identity}
for k, v := range rep.Volume.Attributes {volumeAttributes[k] = v
}...
fsType := ""
for k, v := range options.Parameters {switch strings.ToLower(k) {case "fstype":fsType = v}
}
if len(fsType) == 0 {fsType = defaultFSType
}
3、创建CSI CreateVolumeRequest
// Create a CSI CreateVolumeRequest and Response
req := csi.CreateVolumeRequest{Name: pvName,Parameters: options.Parameters,VolumeCapabilities: volumeCaps,CapacityRange: &csi.CapacityRange{RequiredBytes: int64(volSizeBytes),},
}
...
glog.V(5).Infof("CreateVolumeRequest %+v", req)rep := &csi.CreateVolumeResponse{}
...
opts := wait.Backoff{Duration: backoffDuration, Factor: backoffFactor, Steps: backoffSteps}
err = wait.ExponentialBackoff(opts, func() (bool, error) {ctx, cancel := context.WithTimeout(context.Background(), p.timeout)defer cancel()rep, err = p.csiClient.CreateVolume(ctx, &req)if err == nil {// CreateVolume has finished successfullyreturn true, nil}if status, ok := status.FromError(err); ok {if status.Code() == codes.DeadlineExceeded {// CreateVolume timed out, give it another chance to completeglog.Warningf("CreateVolume timeout: %s has expired, operation will be retried", p.timeout.String())return false, nil}}// CreateVolume failed , no reason to retry, bailing from ExponentialBackoffreturn false, err
})if err != nil {return nil, err
}if rep.Volume != nil {glog.V(3).Infof("create volume rep: %+v", *rep.Volume)
}respCap := rep.GetVolume().GetCapacityBytes()
if respCap < volSizeBytes {capErr := fmt.Errorf("created volume capacity %v less than requested capacity %v", respCap, volSizeBytes)delReq := &csi.DeleteVolumeRequest{VolumeId: rep.GetVolume().GetId(),}delReq.ControllerDeleteSecrets = provisionerCredentialsctx, cancel := context.WithTimeout(context.Background(), p.timeout)defer cancel()_, err := p.csiClient.DeleteVolume(ctx, delReq)if err != nil {capErr = fmt.Errorf("%v. Cleanup of volume %s failed, volume is orphaned: %v", capErr, pvName, err)}return nil, capErr
}
Provison
方法核心功能是调用p.csiClient.CreateVolume(ctx, &req)
。
4、构造PV对象
pv := &v1.PersistentVolume{ObjectMeta: metav1.ObjectMeta{Name: pvName,},Spec: v1.PersistentVolumeSpec{PersistentVolumeReclaimPolicy: options.PersistentVolumeReclaimPolicy,AccessModes: options.PVC.Spec.AccessModes,Capacity: v1.ResourceList{v1.ResourceName(v1.ResourceStorage): bytesToGiQuantity(respCap),},// TODO wait for CSI VolumeSource APIPersistentVolumeSource: v1.PersistentVolumeSource{CSI: &v1.CSIPersistentVolumeSource{Driver: driverState.driverName,VolumeHandle: p.volumeIdToHandle(rep.Volume.Id),FSType: fsType,VolumeAttributes: volumeAttributes,ControllerPublishSecretRef: controllerPublishSecretRef,NodeStageSecretRef: nodeStageSecretRef,NodePublishSecretRef: nodePublishSecretRef,},},},
}if driverState.capabilities.Has(PluginCapability_ACCESSIBILITY_CONSTRAINTS) {pv.Spec.NodeAffinity = GenerateVolumeNodeAffinity(rep.Volume.AccessibleTopology)
}glog.Infof("successfully created PV %+v", pv.Spec.PersistentVolumeSource)return pv, nil
Provision
方法只是通过VolumeOptions
参数来构建PV
对象,并没有执行具体PV
的创建或删除的操作。
不同类型的Provisioner
的,一般是PersistentVolumeSource
类型和参数不同,例如csi-provisioner
对应的PersistentVolumeSource
为CSI
,并且需要传入CSI
相关的参数:
Driver
VolumeHandle
FSType
VolumeAttributes
ControllerPublishSecretRef
NodeStageSecretRef
NodePublishSecretRef
2.2.2. Delete方法
csiProvisioner
的delete
方法具体源码参考:https://github.com/kubernetes-csi/external-provisioner/blob/master/pkg/controller/controller.go#L606
func (p *csiProvisioner) Delete(volume *v1.PersistentVolume) error {if volume == nil || volume.Spec.CSI == nil {return fmt.Errorf("invalid CSI PV")}volumeId := p.volumeHandleToId(volume.Spec.CSI.VolumeHandle)_, err := checkDriverState(p.grpcClient, p.timeout, false)if err != nil {return err}req := csi.DeleteVolumeRequest{VolumeId: volumeId,}// get secrets if StorageClass specifies itstorageClassName := volume.Spec.StorageClassNameif len(storageClassName) != 0 {if storageClass, err := p.client.StorageV1().StorageClasses().Get(storageClassName, metav1.GetOptions{}); err == nil {// Resolve provision secret credentials.// No PVC is provided when resolving provision/delete secret names, since the PVC may or may not exist at delete time.provisionerSecretRef, err := getSecretReference(provisionerSecretNameKey, provisionerSecretNamespaceKey, storageClass.Parameters, volume.Name, nil)if err != nil {return err}credentials, err := getCredentials(p.client, provisionerSecretRef)if err != nil {return err}req.ControllerDeleteSecrets = credentials}}ctx, cancel := context.WithTimeout(context.Background(), p.timeout)defer cancel()_, err = p.csiClient.DeleteVolume(ctx, &req)return err
}
Delete
方法主要是调用了p.csiClient.DeleteVolume(ctx, &req)
方法。
2.3. 总结
csi provisioner
实现了Provisioner
接口,其中包含Provison
和Delete
两个方法:
Provision
:调用csiClient.CreateVolume
方法,同时构造并返回PV对象。Delete
:调用csiClient.DeleteVolume
方法。
csi provisioner
的核心方法都调用了csi-client
相关方法。
3. csi-client
csi client
的相关代码参考:https://github.com/container-storage-interface/spec/blob/master/lib/go/csi/v0/csi.pb.go
3.1. 构造csi-client
3.1.1. 构造grpcClient
// Provisioner will stay in Init until driver opens csi socket, once it's done
// controller will exit this loop and proceed normally.
socketDown := true
grpcClient := &grpc.ClientConn{}
for socketDown {grpcClient, err = ctrl.Connect(*csiEndpoint, *connectionTimeout)if err == nil {socketDown = falsecontinue}time.Sleep(10 * time.Second)
}
通过连接csi socket
,连接成功才构造可用的grpcClient
。
3.1.2. 构造csi-client
通过grpcClient
构造csi-client
。
// Create the provisioner: it implements the Provisioner interface expected by
// the controller
csiProvisioner := ctrl.NewCSIProvisioner(clientset, csiAPIClient, *csiEndpoint, *connectionTimeout, identity, *volumeNamePrefix, *volumeNameUUIDLength, grpcClient, snapClient)
NewCSIProvisioner
// NewCSIProvisioner creates new CSI provisioner
func NewCSIProvisioner(client kubernetes.Interface,csiAPIClient csiclientset.Interface,csiEndpoint string,connectionTimeout time.Duration,identity string,volumeNamePrefix string,volumeNameUUIDLength int,grpcClient *grpc.ClientConn,snapshotClient snapclientset.Interface) controller.Provisioner {csiClient := csi.NewControllerClient(grpcClient)provisioner := &csiProvisioner{client: client,grpcClient: grpcClient,csiClient: csiClient,csiAPIClient: csiAPIClient,snapshotClient: snapshotClient,timeout: connectionTimeout,identity: identity,volumeNamePrefix: volumeNamePrefix,volumeNameUUIDLength: volumeNameUUIDLength,}return provisioner
}
NewControllerClient
csiClient := csi.NewControllerClient(grpcClient)
...
type controllerClient struct {cc *grpc.ClientConn
}func NewControllerClient(cc *grpc.ClientConn) ControllerClient {return &controllerClient{cc}
}
3.2. csiClient.CreateVolume
csi provisoner
中调用csiClient.CreateVolume
代码如下:
opts := wait.Backoff{Duration: backoffDuration, Factor: backoffFactor, Steps: backoffSteps}
err = wait.ExponentialBackoff(opts, func() (bool, error) {ctx, cancel := context.WithTimeout(context.Background(), p.timeout)defer cancel()rep, err = p.csiClient.CreateVolume(ctx, &req)if err == nil {// CreateVolume has finished successfullyreturn true, nil}if status, ok := status.FromError(err); ok {if status.Code() == codes.DeadlineExceeded {// CreateVolume timed out, give it another chance to completeglog.Warningf("CreateVolume timeout: %s has expired, operation will be retried", p.timeout.String())return false, nil}}// CreateVolume failed , no reason to retry, bailing from ExponentialBackoffreturn false, err
})
CreateVolumeRequest的构造:
// Create a CSI CreateVolumeRequest and Response
req := csi.CreateVolumeRequest{Name: pvName,Parameters: options.Parameters,VolumeCapabilities: volumeCaps,CapacityRange: &csi.CapacityRange{RequiredBytes: int64(volSizeBytes),},
}
...
req.VolumeContentSource = volumeContentSource
...
req.AccessibilityRequirements = requirements
...
req.ControllerCreateSecrets = provisionerCredentials
具体的Create
实现方法如下:
其中
csiClient
是个接口类型
具体代码参考controllerClient.CreateVolume
func (c *controllerClient) CreateVolume(ctx context.Context, in *CreateVolumeRequest, opts ...grpc.CallOption) (*CreateVolumeResponse, error) {out := new(CreateVolumeResponse)err := grpc.Invoke(ctx, "/csi.v0.Controller/CreateVolume", in, out, c.cc, opts...)if err != nil {return nil, err}return out, nil
}
3.3. csiClient.DeleteVolume
csi provisoner
中调用csiClient.DeleteVolume
代码如下:
func (p *csiProvisioner) Delete(volume *v1.PersistentVolume) error {...req := csi.DeleteVolumeRequest{VolumeId: volumeId,}// get secrets if StorageClass specifies it...ctx, cancel := context.WithTimeout(context.Background(), p.timeout)defer cancel()_, err = p.csiClient.DeleteVolume(ctx, &req)return err
}
DeleteVolumeRequest的构造:
req := csi.DeleteVolumeRequest{VolumeId: volumeId,
}
...
req.ControllerDeleteSecrets = credentials
将构造的DeleteVolumeRequest
传给DeleteVolume
方法。
具体的Delete
实现方法如下:
具体代码参考:controllerClient.DeleteVolume
func (c *controllerClient) DeleteVolume(ctx context.Context, in *DeleteVolumeRequest, opts ...grpc.CallOption) (*DeleteVolumeResponse, error) {out := new(DeleteVolumeResponse)err := grpc.Invoke(ctx, "/csi.v0.Controller/DeleteVolume", in, out, c.cc, opts...)if err != nil {return nil, err}return out, nil
}
4. ProvisionController.Run
自定义的provisioner
实现了Provisoner接口
的Provision
和Delete
方法,这两个方法主要对后端存储做创建和删除操作,并没有对PV对象进行创建和删除操作。
func main() {provisionController.Run(wait.NeverStop)
}
这块代码逻辑可参考:nfs-client-provisioner 源码分析
- https://github.com/kubernetes-csi/external-provisioner
- https://github.com/container-storage-interface/spec
- https://github.com/kubernetes/community/blob/master/contributors/design-proposals/storage/container-storage-interface.md
- https://github.com/container-storage-interface/spec/blob/master/spec.md
csi-provisioner源码分析相关推荐
- kube-scheduler 磁盘调度源码分析
作者 | leadersnowy 来源 | CSDN博客 kube-scheduler介绍 首先我们知道,kube-scheduler的根本工作任务是根据各种调度算法将Pod调度到最合适的工作节点上 ...
- SVM算法及OpenCV源码分析
关于SVM原理,请参看: 系统学习机器学习之SVM(一) 系统学习机器学习之SVM(二) 系统学习机器学习之SVM(三)--Liblinear,LibSVM使用整理,总结 系统学习机器学习之SVM(四 ...
- kubelet源码分析(一)之 NewKubeletCommand
本文个人博客地址:https://www.huweihuang.com/kubernetes-notes/code-analysis/kubelet/NewKubeletCommand.html ku ...
- 【Golang源码分析】Go Web常用程序包gorilla/mux的使用与源码简析
目录[阅读时间:约10分钟] 一.概述 二.对比: gorilla/mux与net/http DefaultServeMux 三.简单使用 四.源码简析 1.NewRouter函数 2.HandleF ...
- SpringBoot-web开发(四): SpringMVC的拓展、接管(源码分析)
[SpringBoot-web系列]前文: SpringBoot-web开发(一): 静态资源的导入(源码分析) SpringBoot-web开发(二): 页面和图标定制(源码分析) SpringBo ...
- SpringBoot-web开发(二): 页面和图标定制(源码分析)
[SpringBoot-web系列]前文: SpringBoot-web开发(一): 静态资源的导入(源码分析) 目录 一.首页 1. 源码分析 2. 访问首页测试 二.动态页面 1. 动态资源目录t ...
- SpringBoot-web开发(一): 静态资源的导入(源码分析)
目录 方式一:通过WebJars 1. 什么是webjars? 2. webjars的使用 3. webjars结构 4. 解析源码 5. 测试访问 方式二:放入静态资源目录 1. 源码分析 2. 测 ...
- Yolov3Yolov4网络结构与源码分析
Yolov3&Yolov4网络结构与源码分析 从2018年Yolov3年提出的两年后,在原作者声名放弃更新Yolo算法后,俄罗斯的Alexey大神扛起了Yolov4的大旗. 文章目录 论文汇总 ...
- ViewGroup的Touch事件分发(源码分析)
Android中Touch事件的分发又分为View和ViewGroup的事件分发,View的touch事件分发相对比较简单,可参考 View的Touch事件分发(一.初步了解) View的Touch事 ...
最新文章
- JAVA代码实现下载单个文件,和下载打包文件
- 华数传媒牵手阿里云计算共建媒体云平台
- java左手画圆右手画方_左手画圆,右手画方作文
- 将计算机设置成交换机主机名,CISCO2950交换机的配置(设置密码、IP地址、主机名)...
- 折半查找的思想及源码_结构与算法(04):排序规则与查找算法
- libevent源码学习-----统一事件源及信号绑定函数
- 11行代码AC——习题2-4 子序列的和(subsequence)——解题报告
- 动词变名词的变化规则_动词过去式变化规则 - 点评英语网
- iOS 通知中心(NSNotificationCenter)
- 面试了二十多个人,终于定下来一个
- 浏览器文件分段断点上传简单示例(python 篇)
- matlab 矩阵动态,matlab – 如何以块为单位动态重塑矩阵?
- (查看dll函数名和参数)通过VS自带的dumpbin查看dll包含的函数
- IE浏览器访问出现Automation 服务器不能创建对象
- java多级继承_java代码继承------多层继承
- 使用 Autel MaxiFlash Elite 进行 GM J2534 编程
- Qlikview---变量
- DisGeNET的初步认识
- MySQL 递归 sql语句 WITH表达式实现
- 【论文阅读笔记】Multi-Task Feature Learning for Knowledge Graph Enhanced