kubernetes 存储流程

PV 与 PVC

PVC (PersistentVolumeClaim)，定名空间（namespace）级别的资源，由 用户 or StatefulSet 控制器（根据VolumeClaimTemplate） 创建。PVC 类似于 Pod，Pod 消耗 Node 资源，PVC 消耗 PV 资源。Pod 可以请求特定级别的资源（CPU 和内存），而 PVC 可以请求特定存储卷的大小及访问模式（Access Mode
PV（PersistentVolume）是集群中的一块存储资源，可以是 NFS、iSCSI、Ceph、GlusterFS 等存储卷，PV 由集群管理员创建，然后由开辟者使用 PVC 来申请 PV，PVC 是对 PV 的申请，类似于 Pod 对 Node 的申请。
静态创建存储卷

也就是我们手动创建一个pv和pvc，然后将pv和pvc绑定，然后pod使用pvc，这样就可以使用pv了。
创建一个 nfs 的 pv 以及 对应的 pvc

1. apiVersion: v1
2. kind: PersistentVolume
3. metadata:
4.   name: nfs-pv
5. spec:
6.   capacity:
7.     storage: 10Gi
8.   accessModes:
9.     - ReadWriteOnce
10.   persistentVolumeReclaimPolicy: Delete
11.   nfs:
12.     server: 192.168.203.110
13.     path: /data/nfs

复制代码

1. apiVersion: v1
2. kind: PersistentVolumeClaim
3. metadata:
4.   name: nfs-pvc
5. spec:
6.   accessModes:
7.   - ReadWriteOnce
8.   resources:
9.     requests:
10.       storage: 10Gi

复制代码

查看 pvc

1. $ kubectl get pvc
2. NAME      STATUS   VOLUME   CAPACITY   ACCESS MODES   STORAGECLASS   AGE
3. nfs-pvc   Bound    nfs-pv   10Gi       RWO                           101s

复制代码

创建一个 pod 使用 pvc

1. apiVersion: v1
2. kind: Pod
3. metadata:
4.   name: test-nfs
5. spec:
6.   containers:
7.   - image: ubuntu:22.04
8.     name: ubuntu
9.     command:
10.     - /bin/sh
11.     - -c
12.     - sleep 10000
13.     volumeMounts:
14.     - mountPath: /data
15.       name: nfs-volume
16.   volumes:
17.   - name: nfs-volume
18.     persistentVolumeClaim:
19.       claimName: nfs-pvc

复制代码

1. ❯ kubectl exec -it test-nfs -- cat /data/nfs
2. 192.168.203.110:/data/nfs

复制代码

pvc pv 绑定流程

1. func (ctrl *PersistentVolumeController) syncUnboundClaim(ctx context.Context, claim *v1.PersistentVolumeClaim) error {
2.         logger := klog.FromContext(ctx)
3.         if claim.Spec.VolumeName == "" {
4.                 // 是不是延迟绑定 也就是 VolumeBindingMode 为 WaitForFirstConsumer
5.                 delayBinding, err := storagehelpers.IsDelayBindingMode(claim, ctrl.classLister)
6.                 if err != nil {
7.                         return err
8.                 }
9.     // 通过 pvc 找到最合适的 pv
10.                 volume, err := ctrl.volumes.findBestMatchForClaim(claim, delayBinding)
11.                 if err != nil {
12.                         logger.V(2).Info("Synchronizing unbound PersistentVolumeClaim, Error finding PV for claim", "PVC", klog.KObj(claim), "err", err)
13.                         return fmt.Errorf("error finding PV for claim %q: %w", claimToClaimKey(claim), err)
14.                 }
15.                 if volume == nil {
16.                         //// No PV found for this claim
17.                 } else /* pv != nil */ {
18.                        
19.                         claimKey := claimToClaimKey(claim)
20.                         logger.V(4).Info("Synchronizing unbound PersistentVolumeClaim, volume found", "PVC", klog.KObj(claim), "volumeName", volume.Name, "volumeStatus", getVolumeStatusForLogging(volume))
21.       // 绑定 pv 和 pvc
22.       // 这里会处理 pvc 的 spec.volumeName status 和 pv 的 status
23.                         if err = ctrl.bind(ctx, volume, claim); err != nil {
24.                                 return err
25.                         }
26.                         return nil
27.                 }
28.         } else /* pvc.Spec.VolumeName != nil */ {
29.                 /*
30.       ......
31.     */
32.         }
33. }
35. // 选择
36. func FindMatchingVolume(
37.         claim *v1.PersistentVolumeClaim,
38.         volumes []*v1.PersistentVolume,
39.         node *v1.Node,
40.         excludedVolumes map[string]*v1.PersistentVolume,
41.         delayBinding bool) (*v1.PersistentVolume, error) {
43.         var smallestVolume *v1.PersistentVolume
44.         var smallestVolumeQty resource.Quantity
45.         requestedQty := claim.Spec.Resources.Requests[v1.ResourceName(v1.ResourceStorage)]
46.         requestedClass := GetPersistentVolumeClaimClass(claim)
48.         var selector labels.Selector
49.         if claim.Spec.Selector != nil {
50.                 internalSelector, err := metav1.LabelSelectorAsSelector(claim.Spec.Selector)
51.                 if err != nil {
52.                         return nil, fmt.Errorf("error creating internal label selector for claim: %v: %v", claimToClaimKey(claim), err)
53.                 }
54.                 selector = internalSelector
55.         }
57.         // Go through all available volumes with two goals:
58.         // - find a volume that is either pre-bound by user or dynamically
59.         //   provisioned for this claim. Because of this we need to loop through
60.         //   all volumes.
61.         // - find the smallest matching one if there is no volume pre-bound to
62.         //   the claim.
63.         for _, volume := range volumes {
64.                 if _, ok := excludedVolumes[volume.Name]; ok {
65.                         // Skip volumes in the excluded list
66.                         continue
67.                 }
68.                 if volume.Spec.ClaimRef != nil && !IsVolumeBoundToClaim(volume, claim) {
69.                         continue
70.                 }
72.                 volumeQty := volume.Spec.Capacity[v1.ResourceStorage]
73.                 if volumeQty.Cmp(requestedQty) < 0 {
74.                         continue
75.                 }
76.                 // filter out mismatching volumeModes
77.                 if CheckVolumeModeMismatches(&claim.Spec, &volume.Spec) {
78.                         continue
79.                 }
81.                 // check if PV's DeletionTimeStamp is set, if so, skip this volume.
82.                 if volume.ObjectMeta.DeletionTimestamp != nil {
83.                         continue
84.                 }
86.                 nodeAffinityValid := true
87.                 if node != nil {
88.                         // Scheduler path, check that the PV NodeAffinity
89.                         // is satisfied by the node
90.                         // CheckNodeAffinity is the most expensive call in this loop.
91.                         // We should check cheaper conditions first or consider optimizing this function.
92.                         err := CheckNodeAffinity(volume, node.Labels)
93.                         if err != nil {
94.                                 nodeAffinityValid = false
95.                         }
96.                 }
98.                 if IsVolumeBoundToClaim(volume, claim) {
99.                         // If PV node affinity is invalid, return no match.
100.                         // This means the prebound PV (and therefore PVC)
101.                         // is not suitable for this node.
102.                         if !nodeAffinityValid {
103.                                 return nil, nil
104.                         }
106.                         return volume, nil
107.                 }
109.                 if node == nil && delayBinding {
110.                         // PV controller does not bind this claim.
111.                         // Scheduler will handle binding unbound volumes
112.                         // Scheduler path will have node != nil
113.                         continue
114.                 }
116.                 // filter out:
117.                 // - volumes in non-available phase
118.                 // - volumes whose labels don't match the claim's selector, if specified
119.                 // - volumes in Class that is not requested
120.                 // - volumes whose NodeAffinity does not match the node
121.                 if volume.Status.Phase != v1.VolumeAvailable {
122.                         // We ignore volumes in non-available phase, because volumes that
123.                         // satisfies matching criteria will be updated to available, binding
124.                         // them now has high chance of encountering unnecessary failures
125.                         // due to API conflicts.
126.                         continue
127.                 } else if selector != nil && !selector.Matches(labels.Set(volume.Labels)) {
128.                         continue
129.                 }
130.                 if GetPersistentVolumeClass(volume) != requestedClass {
131.                         continue
132.                 }
133.                 if !nodeAffinityValid {
134.                         continue
135.                 }
137.                 if node != nil {
138.                         // Scheduler path
139.                         // Check that the access modes match
140.                         if !CheckAccessModes(claim, volume) {
141.                                 continue
142.                         }
143.                 }
145.                 if smallestVolume == nil || smallestVolumeQty.Cmp(volumeQty) > 0 {
146.                         smallestVolume = volume
147.                         smallestVolumeQty = volumeQty
148.                 }
149.         }
151.         if smallestVolume != nil {
152.                 // Found a matching volume
153.                 return smallestVolume, nil
154.         }
156.         return nil, nil
157. }

复制代码

kubelet 绑定

1. if err := os.MkdirAll(dir, 0750); err != nil {
2.                 return err
3. }
4. source := fmt.Sprintf("%s:%s", nfsMounter.server, nfsMounter.exportPath)
5. options := []string{}
6. if nfsMounter.readOnly {
7.   options = append(options, "ro")
8. }
9. mountOptions := util.JoinMountOptions(nfsMounter.mountOptions, options)
10. err = nfsMounter.mounter.MountSensitiveWithoutSystemd(source, dir, "nfs", mountOptions, nil)

复制代码

kubelet 就会在调用 sudo mount -t nfs ... 命令把 nfs 绑定到主机上 绑定的目录大概为 /var/lib/kubelet/pods/[POD-ID]/volumes/
StorageClass

StorageClass 是 Kubernetes 中用来界说存储卷的范例的资源对象，StorageClass 用来界说存储卷的范例，比如 NFS、iSCSI、Ceph、GlusterFS 等存储卷。StorageClass 是集群级别的资源，由集群管理员创建，用户可以使用 StorageClass 来动态创建 PV。
动态创建存储卷

动态创建存储卷相比静态创建存储卷，少了集群管理员的干预，流程如下图所示：

创建一个 StorageClass pvc pod

1. apiVersion: storage.k8s.io/v1
2. kind: StorageClass
3. metadata:
4.   name: local-storage
5. provisioner: rancher.io/local-path
6. reclaimPolicy: Delete
7. volumeBindingMode: WaitForFirstConsumer
8. ---
9. apiVersion: v1
10. kind: PersistentVolumeClaim
11. metadata:
12.   name: my-local-pvc
13. spec:
14.   accessModes:
15.     - ReadWriteOnce
16.   resources:
17.     requests:
18.       storage: 128Mi
19.   storageClassName: local-storage
20. ---
21. apiVersion: v1
22. kind: Pod
23. metadata:
24.   name: test
25. spec:
26.   containers:
27.   - image: ubuntu:22.04
28.     name: ubuntu
29.     command:
30.     - /bin/sh
31.     - -c
32.     - sleep 10000
33.     volumeMounts:
34.     - mountPath: /data
35.       name: my-local-pvc
36.   volumes:
37.   - name: my-local-pvc
38.     persistentVolumeClaim:
39.       claimName: my-local-pvc

复制代码

查看 pv

1. ❯ kubectl get pv
2. NAME                                       CAPACITY   ACCESS MODES   RECLAIM POLICY   STATUS   CLAIM                  STORAGECLASS    REASON   AGE
3. pvc-9d257d8a-29a8-4abf-a1e2-c7e4953fc0ca   128Mi      RWO            Delete           Bound    default/my-local-pvc   local-storage            85s

复制代码

StorageClass 创建 pv 流程

1. // 还是 syncUnboundClaim 中
2. // volume 为空，说明没有找到合适的 pv 那么去检查 如果 pvc 的 storageClassName 不为空，那么就会去找到对应的 storageClass
3. if volume == nil {
5.                         switch {
6.                         case delayBinding && !storagehelpers.IsDelayBindingProvisioning(claim):
7.         // ......
8.                         case storagehelpers.GetPersistentVolumeClaimClass(claim) != "":
9.                                 // 如果 pvc 的 storageClassName 不为空，那么就会去找到对应的 storageClass
10.                                 if err = ctrl.provisionClaim(ctx, claim); err != nil {
11.                                         return err
12.                                 }
13.                                 return nil
14.                         default:
15.                         }
16.                         return nil
17. }
19. func (ctrl *PersistentVolumeController) provisionClaim(ctx context.Context, claim *v1.PersistentVolumeClaim) error {
20.         plugin, storageClass, err := ctrl.findProvisionablePlugin(claim)
21.         ctrl.scheduleOperation(logger, opName, func() error {
22.                 var err error
23.                 if plugin == nil {
24.       // 如果是外部的 provisioner 这里我们就安装了 rancher.io/local-path 这个插件
25.       // 所以这里会调用 provisionClaimOperationExternal
26.                         _, err = ctrl.provisionClaimOperationExternal(ctx, claim, storageClass)
27.                 } else {
28.       // 内部的 provisioner 直接处理
29.                         _, err = ctrl.provisionClaimOperation(ctx, claim, plugin, storageClass)
30.                 }
31.                 return err
32.         })
33.         return nil
34. }
36. // 如果是外部的 provisioner 会在 pvc 的 annotations 加入 volume.beta.kubernetes.io/storage-provisioner: rancher.io/local-path 和 volume.kubernetes.io/storage-provisioner: rancher.io/local-path
37. func (ctrl *PersistentVolumeController) setClaimProvisioner(ctx context.Context, claim *v1.PersistentVolumeClaim, provisionerName string) (*v1.PersistentVolumeClaim, error) {
38.         if val, ok := claim.Annotations[storagehelpers.AnnStorageProvisioner]; ok && val == provisionerName {
39.                 // annotation is already set, nothing to do
40.                 return claim, nil
41.         }
43.         // The volume from method args can be pointing to watcher cache. We must not
44.         // modify these, therefore create a copy.
45.         claimClone := claim.DeepCopy()
46.         // TODO: remove the beta storage provisioner anno after the deprecation period
47.         logger := klog.FromContext(ctx)
48.         metav1.SetMetaDataAnnotation(&claimClone.ObjectMeta, storagehelpers.AnnBetaStorageProvisioner, provisionerName)
49.         metav1.SetMetaDataAnnotation(&claimClone.ObjectMeta, storagehelpers.AnnStorageProvisioner, provisionerName)
50.         updateMigrationAnnotations(logger, ctrl.csiMigratedPluginManager, ctrl.translator, claimClone.Annotations, true)
51.         newClaim, err := ctrl.kubeClient.CoreV1().PersistentVolumeClaims(claim.Namespace).Update(ctx, claimClone, metav1.UpdateOptions{})
52.         if err != nil {
53.                 return newClaim, err
54.         }
55.         _, err = ctrl.storeClaimUpdate(logger, newClaim)
56.         if err != nil {
57.                 return newClaim, err
58.         }
59.         return newClaim, nil
60. }

复制代码

kubernetes external provisioner

kubernetes external provisioner 是一个独立的历程，用来动态创建 PV，它通过监听 StorageClass 的事件，当 StorageClass 的 ReclaimPolicy 为 Retain 时，会创建 PV。
在这里我新建一个 关于 nfs 的 external provisioner

1. package main
3. import (
4.         "context"
5.         "fmt"
6.         "path/filepath"
8.         "github.com/golang/glog"
9.         v1 "k8s.io/api/core/v1"
10.         metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
11.         "k8s.io/client-go/kubernetes"
12.         "k8s.io/client-go/tools/clientcmd"
13.         "k8s.io/client-go/util/homedir"
14.         "sigs.k8s.io/controller-runtime/pkg/log"
15.         "sigs.k8s.io/sig-storage-lib-external-provisioner/v10/controller"
16. )
18. const provisionerName = "provisioner.test.com/nfs"
20. var _ controller.Provisioner = &nfsProvisioner{}
22. type nfsProvisioner struct {
23.         client kubernetes.Interface
24. }
26. func (p *nfsProvisioner) Provision(ctx context.Context, options controller.ProvisionOptions) (*v1.PersistentVolume, controller.ProvisioningState, error) {
27.         if options.PVC.Spec.Selector != nil {
28.                 return nil, controller.ProvisioningFinished, fmt.Errorf("claim Selector is not supported")
29.         }
30.         glog.V(4).Infof("nfs provisioner: VolumeOptions %v", options)
32.         pv := &v1.PersistentVolume{
33.                 ObjectMeta: metav1.ObjectMeta{
34.                         Name: options.PVName,
35.                 },
36.                 Spec: v1.PersistentVolumeSpec{
37.                         PersistentVolumeReclaimPolicy: *options.StorageClass.ReclaimPolicy,
38.                         AccessModes:                   options.PVC.Spec.AccessModes,
39.                         MountOptions:                  options.StorageClass.MountOptions,
40.                         Capacity: v1.ResourceList{
41.                                 v1.ResourceName(v1.ResourceStorage): options.PVC.Spec.Resources.Requests[v1.ResourceName(v1.ResourceStorage)],
42.                         },
43.                         PersistentVolumeSource: v1.PersistentVolumeSource{
44.                                 NFS: &v1.NFSVolumeSource{
45.                                         Server:   options.StorageClass.Parameters["server"],
46.                                         Path:     options.StorageClass.Parameters["path"],
47.                                         ReadOnly: options.StorageClass.Parameters["readOnly"] == "true",
48.                                 },
49.                         },
50.                 },
51.         }
53.         return pv, controller.ProvisioningFinished, nil
54. }
56. func (p *nfsProvisioner) Delete(ctx context.Context, volume *v1.PersistentVolume) error {
57.         // 因为是 nfs 没有产生实际的资源，所以这里不需要删除
58.         // 如果在 provisioner 中创建了资源，那么这里需要删除
59.         // 一般是调用 csi 创建/删除资源
60.         return nil
61. }
63. func main() {
64.         l := log.FromContext(context.Background())
65.         config, err := clientcmd.BuildConfigFromFlags("", filepath.Join(homedir.HomeDir(), ".kube", "config"))
66.         if err != nil {
67.                 glog.Fatalf("Failed to create kubeconfig: %v", err)
68.         }
69.         clientset, err := kubernetes.NewForConfig(config)
70.         if err != nil {
71.                 glog.Fatalf("Failed to create client: %v", err)
72.         }
74.         clientNFSProvisioner := &nfsProvisioner{
75.                 client: clientset,
76.         }
78.         pc := controller.NewProvisionController(l,
79.                 clientset,
80.                 provisionerName,
81.                 clientNFSProvisioner,
82.                 controller.LeaderElection(true),
83.         )
84.         glog.Info("Starting provision controller")
85.         pc.Run(context.Background())
86. }

复制代码

创建一个 nfs 的 storageClass

1. apiVersion: storage.k8s.io/v1
2. kind: StorageClass
3. metadata:
4.   name: my-nfs
5. provisioner: provisioner.test.com/nfs
6. reclaimPolicy: Delete
7. volumeBindingMode: Immediate
8. parameters:
9.   server: "192.168.203.110"
10.   path: /data/nfs
11.   readOnly: "false"

复制代码

1. apiVersion: v1
2. kind: PersistentVolumeClaim
3. metadata:
4.   name: my-nfs-pvc
5. spec:
6.   storageClassName: my-nfs
7.   accessModes:
8.   - ReadWriteOnce
9.   resources:
10.     requests:
11.       storage: 1Gi

复制代码

1. apiVersion: v1
2. kind: Pod
3. metadata:
4.   name: test-nfs
5. spec:
6.   containers:
7.   - image: ubuntu:22.04
8.     name: ubuntu
9.     command:
10.     - /bin/sh
11.     - -c
12.     - sleep 10000
13.     volumeMounts:
14.     - mountPath: /data
15.       name: my-nfs-pvc
16.   volumes:
17.   - name: my-nfs-pvc
18.     persistentVolumeClaim:
19.       claimName: my-nfs-pvc

复制代码

1. ❯ kubectl exec -it test-nfs -- cat /data/nfs
2. 192.168.203.110:/data/nfs

复制代码

CSI 流程

持久化存储流程图如下：

Provisioner

当摆设 csi-controller 时 ，会启动一个伴生容器，项目地点为 https://github.com/kubernetes-csi/external-provisioner  这个项目是一个 csi 的 provisioner
它会监控属于自己的pvc，当有新的pvc创建时，会调用 csi 的 createVolume 方法，创建一个 volume，然后创建一个 pv。当 pvc 删除时，会调用 csi 的 deleteVolume 方法，然后删除 volume 和 pv。

Attacher

external-attacher 也是 csi-controller 的伴生容器，项目地点为 https://github.com/kubernetes-csi/external-attacher 这个项目是一个 csi 的 attacher, 它会监控 AttachDetachController 资源，当有新的资源创建时，会调用 csi 的 controllerPublishVolume 方法，挂载 volume 到 node 上。当资源删除时，会调用 csi 的 controllerUnpublishVolume 方法，卸载 volume。

Snapshotter

external-snapshotter 也是 csi-controller 的伴生容器，项目地点为 https://github.com/kubernetes-csi/external-snapshotter 这个项目是一个 csi 的 snapshotter, 它会监控 VolumeSnapshot 资源，当有新的资源创建时，会调用 csi 的 createSnapshot 方法，创建一个快照。当资源删除时，会调用 csi 的 deleteSnapshot 方法，删除快照。
csi-node

csi-node 是一个 kubelet 的插件，所以它需要每个节点上都运行，当 pod 创建时，并且 VolumeAttachment 的 .spec.Attached 时，kubelet 会调用 csi 的 NodeStageVolume 函数，之后插件（csiAttacher）调用内部 in-tree CSI 插件（csiMountMgr）的 SetUp 函数，该函数内部会调用 csi 的 NodePublishVolume 函数，挂载 volume 到 pod 上。当 pod 删除时，kubelet 观察到包含 CSI 存储卷的 Pod 被删除，于是调用内部 in-tree CSI 插件（csiMountMgr）的 TearDown 函数，该函数内部会通过 unix domain socket 调用外部 CSI 插件的 NodeUnpublishVolume 函数。kubelet 调用内部 in-tree CSI 插件（csiAttacher）的 UnmountDevice 函数，该函数内部会通过 unix domain socket 调用外部 CSI 插件的 NodeUnstageVolume 函数。

csi-node-driver-registrar

这个是 csi-node 的伴生容器，项目地点为 https://github.com/kubernetes-csi/node-driver-registrar,
它的主要作用是向 kubelet 注册 csi 插件，kubelet 会调用 csi 插件的 Probe 方法，假如返回成功，kubelet 会调用 csi 插件的 NodeGetInfo 方法，获取节点信息。
csi-livenessprobe

这个是 csi-node 的伴生容器，项目地点为 https://github.com/kubernetes-csi/livenessprobe, 它的主要作用是给 kubernetes 的 livenessprobe 提供一个接口，用来检查 csi 插件是否正常运行。它在 /healthz 时，会调用 csi 的 Probe 方法，假如返回成功，返回 200，否则返回 500。
Reference

• https://developer.aliyun.com/article/754434
  
• https://developer.aliyun.com/article/783464
  

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。