前言

Kubernetes集群中，如果没有存储，所有pod中应用产生的数据都是临时的，pod挂掉，被rc重新拉起之后，以前产生的数据就丢掉了，这对有些场景是不可接受的，此时，外部存储就显得尤为重要。 

这里重点介绍两个API资源：

PersistentVolume(PV)：集群中的一块网络存储，是集群中的资源，可类比集群中的Node资源；

PersistentVolumeClaim(PVC) : 用户对存储的需求，可类比pod，pod消费node资源，PVC就消费PV资源。

当然还有StorageClass等概念，这里不做详细说明（稳定后，后期文章专门介绍）。K8s存储管理主要分布在两个组件中（这里不包括api）：kube-controller-manager和 kubelet。由于涉及的点比较多，我们分成几篇文章来介绍，本篇主要分析PersistentVolume。
代码基于社区，commit id： 65ddace3ed8e7c25546d12912c8dfdcd06ffe1e0

用例

Kubernetes支持的外部存储非常的多，如：AWSElasticBlockStore，AzureFile，AzureDisk，CephFS，Cinder，FlexVolume，GCEPersistentDisk，Glusterfs，HostPath，iSCSI，NFS，RBD，VsphereVolume等。
简单起见，以HostPath存储的方式，举例说明。

创建PV（hostpath方式存储，目录 /tmp/data） Yaml文件： kind: PersistentVolume
apiVersion: v1
metadata:
name: task-pv-volume
labels:
type: local
spec:
capacity:
storage: 10Gi
accessModes:
– ReadWriteOnce
hostPath:
path: “/tmp/data”

创建命令： kubectl create -f http://k8s.io/docs/tasks/configure-pod-container/task-pv-volume.yaml
persistentvolume “task-pv-volume” created

查看结果： kubectl get pv task-pv-volume
NAME CAPACITY ACCESSMODES RECLAIMPOLICY STATUS CLAIM REASON AGE
task-pv-volume 10Gi RWO Retain Available 17s

创建PVC Yaml文件：
kind: PersistentVolumeClaim
apiVersion: v1
metadata:
name: task-pv-claim
spec:
accessModes:
– ReadWriteOnce
resources:
requests:
storage: 3Gi

创建命令： kubectl create -f http://k8s.io/docs/tasks/configure-pod-container/task-pv-claim.yaml
persistentvolumeclaim “task-pv-claim” created

查看结果：（已经绑定上面创建的PV） kubectl get pvc task-pv-claim
NAME STATUS VOLUME CAPACITY ACCESSMODES AGE
task-pv-claim Bound task-pv-volume 10Gi RWO 5s
PVC配置中没有指定volume name，PVController会从所有的volume中，找到合适的，和PVC进行绑定。
再查看上面创建的PV：
kubectl get pv tasck-pv-cvolume
NAME CAPACITY ACCESSMODES RECLAIMPOLICY STATUS CLAIM REASON AGE
task-pv-volume 10Gi RWO Retain Bound default/task-pv-claim 8m

创建Pod，使用上面创建的PV： Yaml文件：
kind: Pod
apiVersion: v1
metadata:
name: task-pv-pod
spec:
volumes:
– name: task-pv-storage
persistentVolumeClaim:
claimName: task-pv-claim
containers:
– name: task-pv-container
image: nginx
ports:
– containerPort: 80
name: “http-server”
volumeMounts:
– mountPath: “/usr/share/nginx/html”
name: task-pv-storage 注：pod和PVC要在同一个namespace中。

创建命令： kubectl create -f http://k8s.io/docs/tasks/configure-pod-container/task-pv-pod.yaml
pod “task-pv-pod” created

查看pod： kubectl get pod task-pv-pod
NAME READY STATUS RESTARTS AGE
task-pv-pod 0/1 ContainerCreating 0 19s
kubectl get pod task-pv-pod
NAME READY STATUS RESTARTS AGE
task-pv-pod 1/1 Running 0 1m

进入pod，创建文件: kubectl exec -it task-pv-pod — /bin/bash
root@task-pv-pod:~# cd /usr/share/nginx/html/
root@task-pv-pod:/usr/share/nginx/html# echo “hello world” >pv.log
root@task-pv-pod:/usr/share/nginx/html# ls
pv.log
然后退出pod(容器)，看下host是否有此文件：
root@task-pv-pod:/usr/share/nginx/html# exit
exit
nickren@nickren-thinkpad-t450:/tmp/data$ cd /tmp/data/
nickren@nickren-thinkpad-t450:/tmp/data$ ls
pv.log
nickren@nickren-thinkpad-t450:/tmp/data$ cat pv.log
hello world
由此可见，pod中的信息，已经存在了host里面。

代码

PV（PVC）即 PersistentVolume（PersistentVolumeClaim），在k8s中，有专门的controller管理他们。
下面分析创建Controller以及Controller管理PV（PVC）的逻辑

1、创建并启动PersistentVolumeController

func StartControllers(controllers map[string]InitFunc, s *options.CMServer, rootClientBuilder, clientBuilder controller.ControllerClientBuilder, stop <-chan struct{}) error {
。。。。。。
if ctx.IsControllerEnabled(pvBinderControllerName) {
// alphaProvisioner本应该在v1.5版本中去掉的，没做，现在有PR正在做，可以//不用关心
alphaProvisioner, err := NewAlphaVolumeProvisioner(cloud, s.VolumeConfiguration)
if err != nil {
return fmt.Errorf(“an backward-compatible provisioner could not be created: %v, but one was expected. Provisioning will not work. This functionality is considered an early Alpha version.”, err)
}
//构造ControllerParameters
params := persistentvolumecontroller.ControllerParameters{
KubeClient: clientBuilder.ClientOrDie(“persistent-volume-binder”),
SyncPeriod: s.PVClaimBinderSyncPeriod.Duration,
AlphaProvisioner: alphaProvisioner,
VolumePlugins: ProbeControllerVolumePlugins(cloud, s.VolumeConfiguration),
Cloud: cloud,
ClusterName: s.ClusterName,
VolumeInformer: sharedInformers.Core().V1().PersistentVolumes(),
ClaimInformer: sharedInformers.Core().V1().PersistentVolumeClaims(),
ClassInformer: sharedInformers.Storage().V1beta1().StorageClasses(),
EnableDynamicProvisioning: s.VolumeConfiguration.EnableDynamicProvisioning,
}
//这里创建PersistentVolumeController
volumeController := persistentvolumecontroller.NewController(params)
//启动PersistentVolumeController goroutine
go volumeController.Run(stop)
time.Sleep(wait.Jitter(s.ControllerStartInterval.Duration, ControllerStartJitter))
} else {
glog.Warningf(“%q is disabled”, pvBinderControllerName)
}
。。。
}
在kube-controller-manager的StartControllers（）函数中，构造PersistentVolumeController 并运行PersistentVolumeController goroutine。

2、PersistentVolumeController构造方法

func NewController(p ControllerParameters) *PersistentVolumeController {
。。。
//构造PersistentVolumeController
controller := &PersistentVolumeController{
volumes: newPersistentVolumeOrderedIndex(),
claims: cache.NewStore(cache.DeletionHandlingMetaNamespaceKeyFunc),
kubeClient: p.KubeClient,
eventRecorder: eventRecorder,
runningOperations: goroutinemap.NewGoRoutineMap(true /* exponentialBackOffOnError */),
cloud: p.Cloud,
enableDynamicProvisioning: p.EnableDynamicProvisioning,
clusterName: p.ClusterName,
createProvisionedPVRetryCount: createProvisionedPVRetryCount,
createProvisionedPVInterval: createProvisionedPVInterval,
alphaProvisioner: p.AlphaProvisioner,
claimQueue: workqueue.NewNamed(“claims”),
volumeQueue: workqueue.NewNamed(“volumes”),
}
//初始化相应的 VolumePlugin
controller.volumePluginMgr.InitPlugins(p.VolumePlugins, controller)
。。。
//给VolumeInformer 添加相应的event handler
p.VolumeInformer.Informer().AddEventHandlerWithResyncPeriod(
cache.ResourceEventHandlerFuncs{
AddFunc: func(obj interface{}) { controller.enqueueWork(controller.volumeQueue, obj) },
UpdateFunc: func(oldObj, newObj interface{}) { controller.enqueueWork(controller.volumeQueue, newObj) },
DeleteFunc: func(obj interface{}) { controller.enqueueWork(controller.volumeQueue, obj) },
},
p.SyncPeriod,
)
。。。
//给ClaimInformer 添加相应的event handler
p.ClaimInformer.Informer().AddEventHandlerWithResyncPeriod(
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.enqueueWork(controller.claimQueue, obj) },
},
p.SyncPeriod,
)
。。。
return controller
}

3、PersistentVolumeController goroutine运行流程

func (ctrl *PersistentVolumeController) Run(stopCh <-chan struct{}) {
。。。
//从etcd中取数据初始化Controller缓存
ctrl.initializeCaches(ctrl.volumeLister, ctrl.claimLister)
//运行volume具体工作的goroutine
go wait.Until(ctrl.volumeWorker, time.Second, stopCh)
//运行claim具体工作的goroutine
go wait.Until(ctrl.claimWorker, time.Second, stopCh)
<-stopCh
ctrl.claimQueue.ShutDown()
ctrl.volumeQueue.ShutDown()
}

3.1、volumeWorker具体工作 func (ctrl *PersistentVolumeController) volumeWorker() {
workFunc := func() bool {
//从volume队列中取出一个volume object
keyObj, quit := ctrl.volumeQueue.Get()
if quit {
return true
}
defer ctrl.volumeQueue.Done(keyObj)
key := keyObj.(string)
glog.V(5).Infof(“volumeWorker[%s]”, key)
。。。
//检查此volume是否还存在
volume, err := ctrl.volumeLister.Get(name)
if err == nil {
//volume 存在于 informer cache，更新volume
ctrl.updateVolume(volume)
return false
}
if !errors.IsNotFound(err) {
glog.V(2).Infof(“error getting volume %q from informer: %v”, key, err)
return false
}
//volume不存在，删除此volume
。。。
ctrl.deleteVolume(volume)
return false
}
。。。
}

3.2、 claimWorker具体工作 类似volumeWorker工作流程
func (ctrl *PersistentVolumeController) claimWorker() {
workFunc := func() bool {
//从claim队列中取出一个claim object
keyObj, quit := ctrl.claimQueue.Get()
if quit {
return true
}
defer ctrl.claimQueue.Done(keyObj)
key := keyObj.(string)
glog.V(5).Infof(“claimWorker[%s]”, key)
namespace, name, err := cache.SplitMetaNamespaceKey(key)
。。。
//检查此claim是否还存在
claim, err := ctrl.claimLister.PersistentVolumeClaims(namespace).Get(name)
if err == nil {
//claim存在于 informer cache，更新claim
ctrl.updateClaim(claim)
return false
}
if !errors.IsNotFound(err) {
glog.V(2).Infof(“error getting claim %q from informer: %v”, key, err)
return false
}
//claim不存在，删除此claim
ctrl.deleteClaim(claim)
return false
}
。。。
}
K8s中PersistentVolumeController的主体逻辑就是上面所述，具体的update、delete操作由于涉及较多的函数，篇幅所限，不一一列举，可自行阅读代码。

本文转移K8S技术社区-容器技术干货┃K8s存储篇之PV（PVC）