【kubernetes/k8s源码分析】 external nfs storage 源码分析

一时失言乱红尘 2022-01-26 06:51 224阅读 0赞

[https://github.com/kubernetes-incubator/external-storage/tree/master/nfs-client][https_github.com_kubernetes-incubator_external-storage_tree_master_nfs-client]

### storageclass.yaml ###

>     apiVersion: storage.k8s.io/v1
>     kind: StorageClass
>     metadata:
>       name: managed-nfs-storage
>     provisioner: fuseim.pri/ifs # or choose another name, must match deployment's env PROVISIONER_NAME'
>     parameters:
>       archiveOnDelete: "false"

# 1. [Main函数][Main] #

从pod中获取环境变量，包括 NFS\_SERVER，NFS\_PATH，PROVISIONER\_NAME

func main() {
    	flag.Parse()
    	flag.Set("logtostderr", "true")
    
    	server := os.Getenv("NFS_SERVER")
    	if server == "" {
    		glog.Fatal("NFS_SERVER not set")
    	}
    	path := os.Getenv("NFS_PATH")
    	if path == "" {
    		glog.Fatal("NFS_PATH not set")
    	}
    	provisionerName := os.Getenv(provisionerNameKey)
    	if provisionerName == "" {
    		glog.Fatalf("environment variable %s is not set! Please set it.", provisionerNameKey)
    	}

###     1.2 使用 incluster 创建k8s client ###

主要操作的资源为 PV 和 PVC

// Create an InClusterConfig and use it to create a client for the controller
    // to use to communicate with Kubernetes
    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)
    }
    
    // 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)
    }

# 2. [NewProvisionController函数][Main] #

实例化 ProvisionController 结构，主要是为 PVC 提供 PV 操作

nfsProvisioner 实现了 Provision 和 Delete 方法

// NewProvisionController creates a new provision controller using
    // the given configuration parameters and with private (non-shared) informers.
    func NewProvisionController(
    	client kubernetes.Interface,
    	provisionerName string,
    	provisioner Provisioner,
    	kubeVersion string,
    	options ...func(*ProvisionController) error,
    ) *ProvisionController {
    	id, err := os.Hostname()
    	if err != nil {
    		glog.Fatalf("Error getting hostname: %v", err)
    	}

claims 与 volumes 队列，也就是 PVC 与 PV 队列，该 controller 主要就是做这个工作的

ratelimiter := workqueue.NewMaxOfRateLimiter(
       workqueue.NewItemExponentialFailureRateLimiter(15*time.Second, 1000*time.Second),
       &workqueue.BucketRateLimiter{Limiter: rate.NewLimiter(rate.Limit(10), 100)},
    )
    if !controller.exponentialBackOffOnError {
       ratelimiter = workqueue.NewMaxOfRateLimiter(
          workqueue.NewItemExponentialFailureRateLimiter(15*time.Second, 15*time.Second),
          &workqueue.BucketRateLimiter{Limiter: rate.NewLimiter(rate.Limit(10), 100)},
       )
    }
    controller.claimQueue = workqueue.NewNamedRateLimitingQueue(ratelimiter, "claims")
    controller.volumeQueue = workqueue.NewNamedRateLimitingQueue(ratelimiter, "volumes")

使用 informer 机制，这个主要是 PVC，Add Update Delete 回调函数

// ----------------------
    // PersistentVolumeClaims
    
    claimHandler := cache.ResourceEventHandlerFuncs{
       AddFunc:    func(obj interface{}) { controller.enqueueWork(controller.claimQueue, obj) },
       UpdateFunc: func(oldObj, newObj interface{}) { controller.enqueueWork(controller.claimQueue, newObj) },
       DeleteFunc: func(obj interface{}) { controller.forgetWork(controller.claimQueue, obj) },
    }
    
    if controller.claimInformer != nil {
       controller.claimInformer.AddEventHandlerWithResyncPeriod(claimHandler, controller.resyncPeriod)
    } else {
       controller.claimInformer = informer.Core().V1().PersistentVolumeClaims().Informer()
       controller.claimInformer.AddEventHandler(claimHandler)
    }
    controller.claims = controller.claimInformer.GetStore()

还有对 PV Storageclass 使用 informer 机制，流程一样一样的

# 3. [Provision 函数][Main] #

pv 名字为 $\{namespace\}-$\{pvcName\}-$\{options.PVName\}

pvcNamespace := options.PVC.Namespace
    pvcName := options.PVC.Name
    
    pvName := strings.Join([]string{pvcNamespace, pvcName, options.PVName}, "-")

在 /persistentvolumes 目录下创建目录，修改权限，这个就是 PV 的NFS 路径

fullPath := filepath.Join(mountPath, pvName)
    glog.V(4).Infof("creating path %s", fullPath)
    if err := os.MkdirAll(fullPath, 0777); err != nil {
       return nil, errors.New("unable to create directory to provision new pv: " + err.Error())
    }
    os.Chmod(fullPath, 0777)

返回 PV，path 就是原来的加上刚创建的目录，这个就是给pod 挂载的NFS路径，这就体现了动态创建目录的本质

path := filepath.Join(p.path, pvName)
    
    pv := &v1.PersistentVolume{
    	ObjectMeta: metav1.ObjectMeta{
    		Name: options.PVName,
    	},
    	Spec: v1.PersistentVolumeSpec{
    		PersistentVolumeReclaimPolicy: options.PersistentVolumeReclaimPolicy,
    		AccessModes:                   options.PVC.Spec.AccessModes,
    		MountOptions:                  options.MountOptions,
    		Capacity: v1.ResourceList{
    			v1.ResourceName(v1.ResourceStorage): options.PVC.Spec.Resources.Requests[v1.ResourceName(v1.ResourceStorage)],
    		},
    		PersistentVolumeSource: v1.PersistentVolumeSource{
    			NFS: &v1.NFSVolumeSource{
    				Server:   p.server,
    				Path:     path,
    				ReadOnly: false,
    			},
    		},
    	},
    }

# 4. [Delete 函数][Main] #

取得 path，查看在容器中路径是否存在，不存在就没有该路径，报个警，不管了

path := volume.Spec.PersistentVolumeSource.NFS.Path
    pvName := filepath.Base(path)
    oldPath := filepath.Join(mountPath, pvName)
    if _, err := os.Stat(oldPath); os.IsNotExist(err) {
       glog.Warningf("path %s does not exist, deletion skipped", oldPath)
       return nil
    }
    // Get the storage class for this volume.
    storageClass, err := p.getClassForVolume(volume)

如果 archiveOnDelete 为 false，则简单暴力直接删除该目录

// Determine if the "archiveOnDelete" parameter exists.
    // If it exists and has a false value, delete the directory.
    // Otherwise, archive it.
    archiveOnDelete, exists := storageClass.Parameters["archiveOnDelete"]
    if exists {
       archiveBool, err := strconv.ParseBool(archiveOnDelete)
       if err != nil {
          return err
       }
       if !archiveBool {
          return os.RemoveAll(oldPath)
       }
    }

如果 archiveOnDelete 为 true，则该给名，前缀为 archived

archivePath := filepath.Join(mountPath, "archived-"+pvName)
    glog.V(4).Infof("archiving path %s to %s", oldPath, archivePath)
    return os.Rename(oldPath, archivePath)

## 总结： ##

可以提供动态存储功能，实现很简单，就是在容器里挂载nfs 至 /persistentvolumes，在这里创建目录，让 pod 挂载 nfs-path / 刚创建的目录，每一个 PVC 创建的 PV 指定一个目录

缺点觉得就是只能挂载一个 NFS，不能动态传入 NFS 的 server 和 path，这样就可以万能接入

[https_github.com_kubernetes-incubator_external-storage_tree_master_nfs-client]: https://github.com/kubernetes-incubator/external-storage/tree/master/nfs-client
[Main]: https://github.com/kubernetes-incubator/external-storage/blob/master/nfs-client/cmd/nfs-client-provisioner/provisioner.go#L148