登录注册

kubernetes pv-controller 解析

欢马劈雪

工程师 (已认证)

  • 原创分享签约作者

发表于教程文章
504

一 简介

pv controller是 kcm 的组件之一，它负责处理集群中的pvc/pv对象，对pvc/pv 对象进行状态转换。

二 pvController 初始化

初始化代码在 pkg/controller/volume/persistentvolume/pv_controller_base.go 文件中，NewController 主要做了如下几件事情

  • 初始化 eventRecorder
  • 初始化 PersistentVolumeController 对象，
  • 调用 VolumePluginMgr.InitPlugins() 方法 初始化存储插件，代码存在于 pkg/volume/plugins.go 文件中

  • 开始创建 informer 监听集群内的资源，初始化了如下 informer

    • PersistentVolumeInformer
    • PersistentVolumeClaimInformer
    • StorageClassInformer
    • PodInformer
    • NodeInformer
  • 将 PV & PVC 的 event 分别放入 volumeQueue & claimQueue
  • 为了不每次都迭代 pods ，自定义一个通过 pvc 键索引 pod 的索引器
  • 初始化 intree 存储 -> csi 迁移相关功能的 manager

NewController代码在cmd/kube-controller-manager代码里面被调用，初始化成功之后紧接着调用go Run()方法运行 pvController

三 开始运行

// 开始运行 pvController 
func (ctrl *PersistentVolumeController) Run(stopCh <-chan struct{}) {
  defer utilruntime.HandleCrash()
  defer ctrl.claimQueue.ShutDown()
  defer ctrl.volumeQueue.ShutDown()

  klog.Infof("Starting persistent volume controller")
  defer klog.Infof("Shutting down persistent volume controller")

  if !cache.WaitForNamedCacheSync("persistent volume", stopCh, ctrl.volumeListerSynced, ctrl.claimListerSynced, ctrl.classListerSynced, ctrl.podListerSynced, ctrl.NodeListerSynced) {
    return
  }

  ctrl.initializeCaches(ctrl.volumeLister, ctrl.claimLister)

  go wait.Until(ctrl.resync, ctrl.resyncPeriod, stopCh)
  go wait.Until(ctrl.volumeWorker, time.Second, stopCh)
  go wait.Until(ctrl.claimWorker, time.Second, stopCh)

  metrics.Register(ctrl.volumes.store, ctrl.claims, &ctrl.volumePluginMgr)

  <-stopCh
}

同步缓存之后开始周期性执行 ctrl.resync,ctrl.volumeWorker , ctrl.claimWorker , 我们看下 initalizeCaches 方法

func (ctrl *PersistentVolumeController) initializeCaches(volumeLister corelisters.PersistentVolumeLister, claimLister corelisters.PersistentVolumeClaimLister) {
  // 这里不访问 apiserver，是从本地缓存拿出的对象，这些对象不可以被外部函数修改
  volumeList, err := volumeLister.List(labels.Everything())
  if err != nil {
    klog.Errorf("PersistentVolumeController can't initialize caches: %v", err)
    return
  }
  for _, volume := range volumeList {
    // 我们不能改变 volume 对象，所以这里我们copy一份新对象，对新对象进行操作
    volumeClone := volume.DeepCopy()
    if _, err = ctrl.storeVolumeUpdate(volumeClone); err != nil {
      klog.Errorf("error updating volume cache: %v", err)
    }
  }

  claimList, err := claimLister.List(labels.Everything())
  if err != nil {
    klog.Errorf("PersistentVolumeController can't initialize caches: %v", err)
    return
  }
  for _, claim := range claimList {
    if _, err = ctrl.storeClaimUpdate(claim.DeepCopy()); err != nil {
      klog.Errorf("error updating claim cache: %v", err)
    }
  }
  klog.V(4).Infof("controller initialized")
}

type persistentVolumeOrderedIndex struct {
  store cache.Indexer
}

该方法将 cache.listener 里面的缓存转存在 persistentVolumeOrderedIndex 中，它是按 AccessModes 索引并按存储容量排序的 persistentVolume 的缓存。

1 resync

func (ctrl *PersistentVolumeController) resync() {
  klog.V(4).Infof("resyncing PV controller")

  pvcs, err := ctrl.claimLister.List(labels.NewSelector())
  if err != nil {
    klog.Warningf("cannot list claims: %s", err)
    return
  }
  for _, pvc := range pvcs {
    ctrl.enqueueWork(ctrl.claimQueue, pvc)
  }

  pvs, err := ctrl.volumeLister.List(labels.NewSelector())
  if err != nil {
    klog.Warningf("cannot list persistent volumes: %s", err)
    return
  }
  for _, pv := range pvs {
    ctrl.enqueueWork(ctrl.volumeQueue, pv)
  }
}

这里将集群内所有的 pvc/pv 统一都放到对应的 claimQueue & volumeQueue 里面重新处理。 这个resyncPeriod 等于一个random time.Duration * config.time(在 kcm 启动时设置)。

2 volumeWorker

一个无限循环， 不断的处理从 volumeQueue 里面获取到的 PersistentVolume

workFunc := func() bool {
    keyObj, quit := ctrl.volumeQueue.Get()
    if quit {
      return true
    }
    defer ctrl.volumeQueue.Done(keyObj)
    key := keyObj.(string)
    klog.V(5).Infof("volumeWorker[%s]", key)

    _, name, err := cache.SplitMetaNamespaceKey(key)
    if err != nil {
      klog.V(4).Infof("error getting name of volume %q to get volume from informer: %v", key, err)
      return false
    }
    volume, err := ctrl.volumeLister.Get(name)
    if err == nil {
      // The volume still exists in informer cache, the event must have
      // been add/update/sync
      ctrl.updateVolume(volume)
      return false
    }
    if !errors.IsNotFound(err) {
      klog.V(2).Infof("error getting volume %q from informer: %v", key, err)
      return false
    }

    // The volume is not in informer cache, the event must have been
    // "delete"
    volumeObj, found, err := ctrl.volumes.store.GetByKey(key)
    if err != nil {
      klog.V(2).Infof("error getting volume %q from cache: %v", key, err)
      return false
    }
    if !found {
      // The controller has already processed the delete event and
      // deleted the volume from its cache
      klog.V(2).Infof("deletion of volume %q was already processed", key)
      return false
    }
    volume, ok := volumeObj.(*v1.PersistentVolume)
    if !ok {
      klog.Errorf("expected volume, got %+v", volumeObj)
      return false
    }
    ctrl.deleteVolume(volume)
    return false
  }

我们主要关注 ctrl.updateVolume(volume) 方法

updateVolume

updateVolume 方法是对于集群内的 events 实际 handler 方法，它里面主要调用了 ctrl.syncVolume 方法来处理

func (ctrl *PersistentVolumeController) syncVolume(ctx context.Context, volume *v1.PersistentVolume) error {
  klog.V(4).Infof("synchronizing PersistentVolume[%s]: %s", volume.Name, getVolumeStatusForLogging(volume))

    ...
  // [Unit test set 4]
  if volume.Spec.ClaimRef == nil {
    // Volume is unused
    klog.V(4).Infof("synchronizing PersistentVolume[%s]: volume is unused", volume.Name)
    if _, err := ctrl.updateVolumePhase(volume, v1.VolumeAvailable, ""); err != nil {
      // Nothing was saved; we will fall back into the same
      // condition in the next call to this method
      return err
    }
    return nil
  } else /* pv.Spec.ClaimRef != nil */ {
    // Volume is bound to a claim.
    if volume.Spec.ClaimRef.UID == "" {
      // The PV is reserved for a PVC; that PVC has not yet been
      // bound to this PV; the PVC sync will handle it.
      klog.V(4).Infof("synchronizing PersistentVolume[%s]: volume is pre-bound to claim %s", volume.Name, claimrefToClaimKey(volume.Spec.ClaimRef))
      if _, err := ctrl.updateVolumePhase(volume, v1.VolumeAvailable, ""); err != nil {
        // Nothing was saved; we will fall back into the same
        // condition in the next call to this method
        return err
      }
      return nil
    }
    klog.V(4).Infof("synchronizing PersistentVolume[%s]: volume is bound to claim %s", volume.Name, claimrefToClaimKey(volume.Spec.ClaimRef))
    // Get the PVC by _name_
    var claim *v1.PersistentVolumeClaim
    claimName := claimrefToClaimKey(volume.Spec.ClaimRef)
    obj, found, err := ctrl.claims.GetByKey(claimName)
    if err != nil {
      return err
    }
    if !found {
      if volume.Status.Phase != v1.VolumeReleased && volume.Status.Phase != v1.VolumeFailed {
        obj, err = ctrl.claimLister.PersistentVolumeClaims(volume.Spec.ClaimRef.Namespace).Get(volume.Spec.ClaimRef.Name)
        if err != nil && !apierrors.IsNotFound(err) {
          return err
        }
        found = !apierrors.IsNotFound(err)
        if !found {
          obj, err = ctrl.kubeClient.CoreV1().PersistentVolumeClaims(volume.Spec.ClaimRef.Namespace).Get(context.TODO(), volume.Spec.ClaimRef.Name, metav1.GetOptions{})
          if err != nil && !apierrors.IsNotFound(err) {
            return err
          }
          found = !apierrors.IsNotFound(err)
        }
      }
    }
    if !found {
      klog.V(4).Infof("synchronizing PersistentVolume[%s]: claim %s not found", volume.Name, claimrefToClaimKey(volume.Spec.ClaimRef))
      // Fall through with claim = nil
    } else {
      var ok bool
      claim, ok = obj.(*v1.PersistentVolumeClaim)
      if !ok {
        return fmt.Errorf("cannot convert object from volume cache to volume %q!?: %#v", claim.Spec.VolumeName, obj)
      }
      klog.V(4).Infof("synchronizing PersistentVolume[%s]: claim %s found: %s", volume.Name, claimrefToClaimKey(volume.Spec.ClaimRef), getClaimStatusForLogging(claim))
    }
    if claim != nil && claim.UID != volume.Spec.ClaimRef.UID {
      klog.V(4).Infof("Maybe cached claim: %s is not the newest one, we should fetch it from apiserver", claimrefToClaimKey(volume.Spec.ClaimRef))

      claim, err = ctrl.kubeClient.CoreV1().PersistentVolumeClaims(volume.Spec.ClaimRef.Namespace).Get(context.TODO(), volume.Spec.ClaimRef.Name, metav1.GetOptions{})
      if err != nil && !apierrors.IsNotFound(err) {
        return err
      } else if claim != nil {
        // Treat the volume as bound to a missing claim.
        if claim.UID != volume.Spec.ClaimRef.UID {
          klog.V(4).Infof("synchronizing PersistentVolume[%s]: claim %s has a newer UID than pv.ClaimRef, the old one must have been deleted", volume.Name, claimrefToClaimKey(volume.Spec.ClaimRef))
          claim = nil
        } else {
          klog.V(4).Infof("synchronizing PersistentVolume[%s]: claim %s has a same UID with pv.ClaimRef", volume.Name, claimrefToClaimKey(volume.Spec.ClaimRef))
        }
      }
    }

    if claim == nil {
      if volume.Status.Phase != v1.VolumeReleased && volume.Status.Phase != v1.VolumeFailed {
        // Also, log this only once:
        klog.V(2).Infof("volume %q is released and reclaim policy %q will be executed", volume.Name, volume.Spec.PersistentVolumeReclaimPolicy)
        if volume, err = ctrl.updateVolumePhase(volume, v1.VolumeReleased, ""); err != nil {
          // Nothing was saved; we will fall back into the same condition
          // in the next call to this method
          return err
        }
      }
      if err = ctrl.reclaimVolume(volume); err != nil {
        // Release failed, we will fall back into the same condition
        // in the next call to this method
        return err
      }
      if volume.Spec.PersistentVolumeReclaimPolicy == v1.PersistentVolumeReclaimRetain {
        // volume is being retained, it references a claim that does not exist now.
        klog.V(4).Infof("PersistentVolume[%s] references a claim %q (%s) that is not found", volume.Name, claimrefToClaimKey(volume.Spec.ClaimRef), volume.Spec.ClaimRef.UID)
      }
      return nil
    } else if claim.Spec.VolumeName == "" {
      if pvutil.CheckVolumeModeMismatches(&claim.Spec, &volume.Spec) {
        volumeMsg := fmt.Sprintf("Cannot bind PersistentVolume to requested PersistentVolumeClaim %q due to incompatible volumeMode.", claim.Name)
        ctrl.eventRecorder.Event(volume, v1.EventTypeWarning, events.VolumeMismatch, volumeMsg)
        claimMsg := fmt.Sprintf("Cannot bind PersistentVolume %q to requested PersistentVolumeClaim due to incompatible volumeMode.", volume.Name)
        ctrl.eventRecorder.Event(claim, v1.EventTypeWarning, events.VolumeMismatch, claimMsg)
        // Skipping syncClaim
        return nil
      }

      if metav1.HasAnnotation(volume.ObjectMeta, pvutil.AnnBoundByController) {
        // The binding is not completed; let PVC sync handle it
        klog.V(4).Infof("synchronizing PersistentVolume[%s]: volume not bound yet, waiting for syncClaim to fix it", volume.Name)
      } else {
        // Dangling PV; try to re-establish the link in the PVC sync
        klog.V(4).Infof("synchronizing PersistentVolume[%s]: volume was bound and got unbound (by user?), waiting for syncClaim to fix it", volume.Name)
      }
      ctrl.claimQueue.Add(claimToClaimKey(claim))
      return nil
    } else if claim.Spec.VolumeName == volume.Name {
      // Volume is bound to a claim properly, update status if necessary
      klog.V(4).Infof("synchronizing PersistentVolume[%s]: all is bound", volume.Name)
      if _, err = ctrl.updateVolumePhase(volume, v1.VolumeBound, ""); err != nil {
        // Nothing was saved; we will fall back into the same
        // condition in the next call to this method
        return err
      }
      return nil
    } else {
      // Volume is bound to a claim, but the claim is bound elsewhere
      if metav1.HasAnnotation(volume.ObjectMeta, pvutil.AnnDynamicallyProvisioned) && volume.Spec.PersistentVolumeReclaimPolicy == v1.PersistentVolumeReclaimDelete {
        if volume.Status.Phase != v1.VolumeReleased && volume.Status.Phase != v1.VolumeFailed {
          // Also, log this only once:
          klog.V(2).Infof("dynamically volume %q is released and it will be deleted", volume.Name)
          if volume, err = ctrl.updateVolumePhase(volume, v1.VolumeReleased, ""); err != nil {
            // Nothing was saved; we will fall back into the same condition
            // in the next call to this method
            return err
          }
        }
        if err = ctrl.reclaimVolume(volume); err != nil {
          return err
        }
        return nil
      } else {
        if metav1.HasAnnotation(volume.ObjectMeta, pvutil.AnnBoundByController) {
          klog.V(4).Infof("synchronizing PersistentVolume[%s]: volume is bound by controller to a claim that is bound to another volume, unbinding", volume.Name)
          if err = ctrl.unbindVolume(volume); err != nil {
            return err
          }
          return nil
        } else {
          // The PV must have been created with this ptr; leave it alone.
          klog.V(4).Infof("synchronizing PersistentVolume[%s]: volume is bound by user to a claim that is bound to another volume, waiting for the claim to get unbound", volume.Name)
          if err = ctrl.unbindVolume(volume); err != nil {
            return err
          }
          return nil
        }
      }
    }
  }
}

1、当 pv 的 Spec.ClaimRef 的值为空的时候，说明当前 pv 未被使用，调用 ctrl.updateVolumePhase 使得 pv 进入 Available 状态

2、当 pv 的 Spec.ClaimRef 的值不为空的时候， 说明当前 pv 已绑定一个pvc

  • 当Spec.ClaimRef.UID 为空的时候，说明 pvc 还未绑定 pv， 调用ctrl.updateVolumePhase 使得 pv 进入 Available 状态， 方法返回，等待 pvc syncClaim 方法处理
  • 使用 Spec.ClaimRef 相关的 pvc 信息获取 pv_controller缓存的pvc

  • 如果 pvc 没有找到

    • 有可能是集群压力过大缓存没有更新，则进一步从 informercache 中找，如果 informercache里面还是没有的话则进一步从apiserver中去找
    • 这里如果发现 非 Released & 非 Failed 的pv 经过上述步骤仍然找不到 pvc 的话，说明 pvc 被删除。在最新的kubernetes 版本中会检查reclaimPoilcy，对 pv的状态进行处理
  • 找到 pvc 之后

1）如果 pvc 的 uid 和 Spec.ClaimRef.UID 不一致，这样一般是 pv 指向的 pvc 被删了，然后立即创建了一个同名的pvc， 而缓存还没有更新，这时我们需要doublecheck一下，若 double check 之后依旧不存在，则判断是pv绑定了一个不存在的pvc， 将pvc置为空，执行上述pvc 没有找到的逻辑

2）如果pvc 的 volumeName 为空

  • 检查 pvc的 volumeMode 和 pv 的 volumeMode是否一致，不一致报 event 出来

  • 如果发现有这个 pv 有 AnnBoundByController = "http://pv.kubernetes.io/bound-by-controller" 这个annotation 说明 pvc/pv 流程正在绑定中
  • 将 pvc 放到 claimQueue 里面， 让 claimWorker 进行处理

3）如果 pvc.Spec.volumeName == pv.volumeName 的时候，直接将 pv 设置为 bound 状态

4）如果 pvc.Spec.volumeName != pv.volumeName 的时候

  • 如果是 pv 是动态创建的情况下，并且 pv 的 ReclaimPolicy 是 delete 的情况下， 说明 pvc 已经绑定了其他pv， 将 pv 置为 released 的状态， 等待deleters 删除

  • 如果 pv 不是动态创建的情况下，将 pv 的 ClaimRef 字段置为空，将其 unbound 掉

3 claimWorker

一个无限循环，不断的处理从 claimQueue 里面获取到的 PersistentVolumeClaim

    workFunc := func() bool {
    keyObj, quit := ctrl.claimQueue.Get()
    if quit {
      return true
    }
    defer ctrl.claimQueue.Done(keyObj)
    key := keyObj.(string)
    klog.V(5).Infof("claimWorker[%s]", key)

    namespace, name, err := cache.SplitMetaNamespaceKey(key)
    if err != nil {
      klog.V(4).Infof("error getting namespace & name of claim %q to get claim from informer: %v", key, err)
      return false
    }
    claim, err := ctrl.claimLister.PersistentVolumeClaims(namespace).Get(name)
    if err == nil {
      // The claim still exists in informer cache, the event must have
      // been add/update/sync
      ctrl.updateClaim(claim)
      return false
    }
    if !errors.IsNotFound(err) {
      klog.V(2).Infof("error getting claim %q from informer: %v", key, err)
      return false
    }

    // The claim is not in informer cache, the event must have been "delete"
    claimObj, found, err := ctrl.claims.GetByKey(key)
    if err != nil {
      klog.V(2).Infof("error getting claim %q from   cache: %v", key, err)
      return false
    }
    if !found {
      // The controller has already processed the delete event and
      // deleted the claim from its cache
      klog.V(2).Infof("deletion of claim %q was already processed", key)
      return false
    }
    claim, ok := claimObj.(*v1.PersistentVolumeClaim)
    if !ok {
      klog.Errorf("expected claim, got %+v", claimObj)
      return false
    }
    ctrl.deleteClaim(claim)
    return false
    }

我们主要关注 ctrl.updateClaim(claim) 方法, 与上面同样，它里面主要调用了 ctrl.syncClaim 方法来处理， 在 syncClaim 里面根据 pvc 的状态分别调用了 ctrl.syncUnboundClaim & ctrl.syncBoundClaim 方法来处理

syncUnboundClaim

func (ctrl *PersistentVolumeController) syncUnboundClaim(ctx context.Context, claim *v1.PersistentVolumeClaim) error {
  if claim.Spec.VolumeName == "" {
    // User did not care which PV they get.
    delayBinding, err := pvutil.IsDelayBindingMode(claim, ctrl.classLister)
    if err != nil {
      return err
    }

    // [Unit test set 1]
    volume, err := ctrl.volumes.findBestMatchForClaim(claim, delayBinding)
    if err != nil {
      klog.V(2).Infof("synchronizing unbound PersistentVolumeClaim[%s]: Error finding PV for claim: %v", claimToClaimKey(claim), err)
      return fmt.Errorf("error finding PV for claim %q: %w", claimToClaimKey(claim), err)
    }
    if volume == nil {
      klog.V(4).Infof("synchronizing unbound PersistentVolumeClaim[%s]: no volume found", claimToClaimKey(claim))
      switch {
      case delayBinding && !pvutil.IsDelayBindingProvisioning(claim):
        if err = ctrl.emitEventForUnboundDelayBindingClaim(claim); err != nil {
          return err
        }
      case storagehelpers.GetPersistentVolumeClaimClass(claim) != "":
        if err = ctrl.provisionClaim(ctx, claim); err != nil {
          return err
        }
        return nil
      default:
        ctrl.eventRecorder.Event(claim, v1.EventTypeNormal, events.FailedBinding, "no persistent volumes available for this claim and no storage class is set")
      }

      // Mark the claim as Pending and try to find a match in the next
      // periodic syncClaim
      if _, err = ctrl.updateClaimStatus(claim, v1.ClaimPending, nil); err != nil {
        return err
      }
      return nil
    } else /* pv != nil */ {
      // Found a PV for this claim
      // OBSERVATION: pvc is "Pending", pv is "Available"
      claimKey := claimToClaimKey(claim)
      klog.V(4).Infof("synchronizing unbound PersistentVolumeClaim[%s]: volume %q found: %s", claimKey, volume.Name, getVolumeStatusForLogging(volume))
      if err = ctrl.bind(volume, claim); err != nil {
        metrics.RecordMetric(claimKey, &ctrl.operationTimestamps, err)
        return err
      }
      metrics.RecordMetric(claimKey, &ctrl.operationTimestamps, nil)
      return nil
    }
  } else /* pvc.Spec.VolumeName != nil */ {
    klog.V(4).Infof("synchronizing unbound PersistentVolumeClaim[%s]: volume %q requested", claimToClaimKey(claim), claim.Spec.VolumeName)
    obj, found, err := ctrl.volumes.store.GetByKey(claim.Spec.VolumeName)
    if err != nil {
      return err
    }
    if !found {
      klog.V(4).Infof("synchronizing unbound PersistentVolumeClaim[%s]: volume %q requested and not found, will try again next time", claimToClaimKey(claim), claim.Spec.VolumeName)
      if _, err = ctrl.updateClaimStatus(claim, v1.ClaimPending, nil); err != nil {
        return err
      }
      return nil
    } else {
      volume, ok := obj.(*v1.PersistentVolume)
      if !ok {
        return fmt.Errorf("cannot convert object from volume cache to volume %q!?: %+v", claim.Spec.VolumeName, obj)
      }
      klog.V(4).Infof("synchronizing unbound PersistentVolumeClaim[%s]: volume %q requested and found: %s", claimToClaimKey(claim), claim.Spec.VolumeName, getVolumeStatusForLogging(volume))
      if volume.Spec.ClaimRef == nil {
        klog.V(4).Infof("synchronizing unbound PersistentVolumeClaim[%s]: volume is unbound, binding", claimToClaimKey(claim))
        if err = checkVolumeSatisfyClaim(volume, claim); err != nil {
          klog.V(4).Infof("Can't bind the claim to volume %q: %v", volume.Name, err)
          // send an event
          msg := fmt.Sprintf("Cannot bind to requested volume %q: %s", volume.Name, err)
          ctrl.eventRecorder.Event(claim, v1.EventTypeWarning, events.VolumeMismatch, msg)
          // volume does not satisfy the requirements of the claim
          if _, err = ctrl.updateClaimStatus(claim, v1.ClaimPending, nil); err != nil {
            return err
          }
        } else if err = ctrl.bind(volume, claim); err != nil {
          // On any error saving the volume or the claim, subsequent
          // syncClaim will finish the binding.
          return err
        }
        // OBSERVATION: pvc is "Bound", pv is "Bound"
        return nil
      } else if pvutil.IsVolumeBoundToClaim(volume, claim) {
        // User asked for a PV that is claimed by this PVC
        // OBSERVATION: pvc is "Pending", pv is "Bound"
        klog.V(4).Infof("synchronizing unbound PersistentVolumeClaim[%s]: volume already bound, finishing the binding", claimToClaimKey(claim))

        // Finish the volume binding by adding claim UID.
        if err = ctrl.bind(volume, claim); err != nil {
          return err
        }
        // OBSERVATION: pvc is "Bound", pv is "Bound"
        return nil
      } else {
        // User asked for a PV that is claimed by someone else
        // OBSERVATION: pvc is "Pending", pv is "Bound"
        if !metav1.HasAnnotation(claim.ObjectMeta, pvutil.AnnBoundByController) {
          klog.V(4).Infof("synchronizing unbound PersistentVolumeClaim[%s]: volume already bound to different claim by user, will retry later", claimToClaimKey(claim))
          claimMsg := fmt.Sprintf("volume %q already bound to a different claim.", volume.Name)
          ctrl.eventRecorder.Event(claim, v1.EventTypeWarning, events.FailedBinding, claimMsg)
          // User asked for a specific PV, retry later
          if _, err = ctrl.updateClaimStatus(claim, v1.ClaimPending, nil); err != nil {
            return err
          }
          return nil
        } else {
          klog.V(4).Infof("synchronizing unbound PersistentVolumeClaim[%s]: volume already bound to different claim %q by controller, THIS SHOULD NEVER HAPPEN", claimToClaimKey(claim), claimrefToClaimKey(volume.Spec.ClaimRef))
          claimMsg := fmt.Sprintf("volume %q already bound to a different claim.", volume.Name)
          ctrl.eventRecorder.Event(claim, v1.EventTypeWarning, events.FailedBinding, claimMsg)

          return fmt.Errorf("invalid binding of claim %q to volume %q: volume already claimed by %q", claimToClaimKey(claim), claim.Spec.VolumeName, claimrefToClaimKey(volume.Spec.ClaimRef))
        }
      }
    }
  }
}

梳理下整体流程

  • 如果当前 pvc 的 volumeName 为空

    • 判断当前pvc 是否是延迟绑定的
    • 调用 volume, err := ctrl.volumes.findBestMatchForClaim(claim, delayBinding) 找出对应的 pv

  • 如果找到 volume 的话

    • 调用 ctrl.bind(volume, claim) 方法进行绑定

  • 如果没有找到 volume 的话

    • 如果是延迟绑定， 并且还未触发（pod 未引用）则 emit event 到 pvc 上
    • 如果 pvc 绑定了 sc, 调用 ctrl.provisionClaim(ctx, claim) 方法
  1. 分析 pvc yaml， 找到 provisioner driver
  2. 启动一个 goroutine
  3. 调用 ctrl.provisionClaimOperation(ctx, claim, plugin, storageClass) 进行创建工作

provisionClaimOperation

展开阅读全文

本文系作者在时代Java发表，未经许可，不得转载。

如有侵权，请联系nowjava@qq.com删除。

编辑于
举报

关注时代Java

教程文章

文章订阅

关注时代Java

Copyright © 2014-2024NowJava. All Rights Reserved. 时代Java 版权所有 京ICP备14059627号-7

关注时代Java