Kubernetes 动态 PV 实现原理

apiVersion: v1
kind: PersistentVolumeClaim
metadata:
  name: myclaim
spec:
  accessModes:
    - ReadWriteOnce
  volumeMode: Filesystem
  resources:
    requests:
      storage: 8Gi
  storageClassName: slow

$ kubectl get CSIDriver
NAME                            ATTACHREQUIRED   PODINFOONMOUNT   STORAGECAPACITY   TOKENREQUESTS   REQUIRESREPUBLISH   MODES        AGE
rook-ceph.cephfs.csi.ceph.com   true             false            false             <unset>         false               Persistent   77d
rook-ceph.rbd.csi.ceph.com      true             false            false             <unset>         false               Persistent   77d

// https://github.com/kubernetes-csi/external-provisioner/blob/v3.1.0/cmd/csi-provisioner/csi-provisioner.go
func main() {
    // a lot of code here
    provisionController = controller.NewProvisionController(
        clientset,
        provisionerName,
        csiProvisioner,
        provisionerOptions...,
    )

    csiClaimController := ctrl.NewCloningProtectionController(
        clientset,
        claimLister,
        claimInformer,
        claimQueue,
        controllerCapabilities,
    )
    // a lot of code here

    run := func(ctx context.Context) {
        // a lot of code here
        provisionController.Run(ctx)
    }
    // a lot of code here
}

// 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,
    options ...func(*ProvisionController) error,
) *ProvisionController {
    // a lot of code here
    claimHandler := cache.ResourceEventHandlerFuncs{
        AddFunc:    func(obj interface{}) { controller.enqueueClaim(obj) },
        UpdateFunc: func(oldObj, newObj interface{}) { controller.enqueueClaim(newObj) },
        DeleteFunc: func(obj interface{}) {
            // NOOP. The claim is either in claimsInProgress and in the queue, so it will be processed as usual
            // or it's not in claimsInProgress and then we don't care
        },
    }

    if controller.claimInformer != nil {
        controller.claimInformer.AddEventHandlerWithResyncPeriod(claimHandler, controller.resyncPeriod)
    } else {
        controller.claimInformer = informer.Core().V1().PersistentVolumeClaims().Informer()
        controller.claimInformer.AddEventHandler(claimHandler)
    }
    // a lot of code here
}

// syncClaim checks if the claim should have a volume provisioned for it and
// provisions one if so. Returns an error if the claim is to be requeued.
func (ctrl *ProvisionController) syncClaim(ctx context.Context, obj interface{}) error {
    claim, ok := obj.(*v1.PersistentVolumeClaim)
    if !ok {
        return fmt.Errorf("expected claim but got %+v", obj)
    }
    should, err := ctrl.shouldProvision(ctx, claim)
    if err != nil {
        ctrl.updateProvisionStats(claim, err, time.Time{})
        return err
    } else if should {
        startTime := time.Now()

        status, err := ctrl.provisionClaimOperation(ctx, claim) // here
        // a lot of code here
    }
    return nil
}

// provisionClaimOperation attempts to provision a volume for the given claim.
// Returns nil error only when the volume was provisioned (in which case it also returns ProvisioningFinished),
// a normal error when the volume was not provisioned and provisioning should be retried (requeue the claim),
// or the special errStopProvision when provisioning was impossible and no further attempts to provision should be tried.
func (ctrl *ProvisionController) provisionClaimOperation(ctx context.Context, claim *v1.PersistentVolumeClaim) (ProvisioningState, error) {
    // Most code here is identical to that found in controller.go of kube's PV controller...
    claimClass := util.GetPersistentVolumeClaimClass(claim)

    //  A previous doProvisionClaim may just have finished while we were waiting for
    //  the locks. Check that PV (with deterministic name) hasn't been provisioned
    //  yet.
    pvName := ctrl.getProvisionedVolumeNameForClaim(claim)
    _, exists, err := ctrl.volumes.GetByKey(pvName)
    if err == nil && exists {
        // Volume has been already provisioned, nothing to do.
        klog.Info(logOperation(operation, "persistentvolume %q already exists, skipping", pvName))
        return ProvisioningFinished, errStopProvision
    }

    // a lot of code here

    // For any issues getting fields from StorageClass (including reclaimPolicy & mountOptions),
    // retry the claim because the storageClass can be fixed/(re)created independently of the claim
    class, err := ctrl.getStorageClass(claimClass)
    if err != nil {
        klog.Error(logOperation(operation, "error getting claim's StorageClass's fields: %v", err))
        return ProvisioningFinished, err
    }

    // a lot of code here
    options := ProvisionOptions{
        StorageClass: class,
        PVName:       pvName,
        PVC:          claim,
        SelectedNode: selectedNode,
    }

    ctrl.eventRecorder.Event(claim, v1.EventTypeNormal, "Provisioning", fmt.Sprintf("External provisioner is provisioning volume for claim %q", claimToClaimKey(claim)))

    volume, result, err := ctrl.provisioner.Provision(ctx, options)

    // a lot of code here
}

func (p *csiProvisioner) Provision(ctx context.Context, options controller.ProvisionOptions) (*v1.PersistentVolume, controller.ProvisioningState, error) {
    // a lot of code here
    result, state, err := p.prepareProvision(ctx, claim, options.StorageClass, options.SelectedNode)
    if result == nil {
        return nil, state, err
    }
    req := result.req
    volSizeBytes := req.CapacityRange.RequiredBytes
    pvName := req.Name
    provisionerCredentials := req.Secrets

    rep, err := p.csiClient.CreateVolume(createCtx, req)
}

// prepareProvision does non-destructive parameter checking and preparations for provisioning a volume.
func (p *csiProvisioner) prepareProvision(ctx context.Context, claim *v1.PersistentVolumeClaim, sc *storagev1.StorageClass, selectedNode *v1.Node) (*prepareProvisionResult, controller.ProvisioningState, error) {
    // a lot of code here

    // Create a CSI CreateVolumeRequest and Response
    req := csi.CreateVolumeRequest{
        Name:               pvName,
        Parameters:         sc.Parameters,
        VolumeCapabilities: volumeCaps,
        CapacityRange: &csi.CapacityRange{
            RequiredBytes: int64(volSizeBytes),
        },
    }
    // a lot of code here
    return &prepareProvisionResult{
        fsType:         fsType,
        migratedVolume: migratedVolume,
        req:            &req,
        csiPVSource:    csiPVSource,
    }, controller.ProvisioningNoChange, nil
}

func (p *csiProvisioner) Provision(ctx context.Context, options controller.ProvisionOptions) (*v1.PersistentVolume, controller.ProvisioningState, error) {
    // a lot of code here
    pv := &v1.PersistentVolume{
        ObjectMeta: metav1.ObjectMeta{
            Name: pvName,
        },
        Spec: v1.PersistentVolumeSpec{
            AccessModes:  options.PVC.Spec.AccessModes,
            MountOptions: options.StorageClass.MountOptions,
            Capacity: v1.ResourceList{
                v1.ResourceName(v1.ResourceStorage): bytesToQuantity(respCap),
            },
            // TODO wait for CSI VolumeSource API
            PersistentVolumeSource: v1.PersistentVolumeSource{
                CSI: result.csiPVSource,
            },
        },
    }

    // a lot of code here
    return pv, controller.ProvisioningFinished, nil
}

// provisionClaimOperation attempts to provision a volume for the given claim.
// Returns nil error only when the volume was provisioned (in which case it also returns ProvisioningFinished),
// a normal error when the volume was not provisioned and provisioning should be retried (requeue the claim),
// or the special errStopProvision when provisioning was impossible and no further attempts to provision should be tried.
func (ctrl *ProvisionController) provisionClaimOperation(ctx context.Context, claim *v1.PersistentVolumeClaim) (ProvisioningState, error) {
    // a lot of code here
    volume, result, err := ctrl.provisioner.Provision(ctx, options)
    // a lot of code here
    if err := ctrl.volumeStore.StoreVolume(claim, volume); err != nil {
        return ProvisioningFinished, err
    }
}

// syncClaim is the main controller method to decide what to do with a claim.
// It's invoked by appropriate cache.Controller callbacks when a claim is
// created, updated or periodically synced. We do not differentiate between
// these events.
// For easier readability, it was split into syncUnboundClaim and syncBoundClaim
// methods.
func (ctrl *PersistentVolumeController) syncClaim(ctx context.Context, claim *v1.PersistentVolumeClaim) error {
    klog.V(4).Infof("synchronizing PersistentVolumeClaim[%s]: %s", claimToClaimKey(claim), getClaimStatusForLogging(claim))

    // Set correct "migrated-to" annotations on PVC and update in API server if
    // necessary
    newClaim, err := ctrl.updateClaimMigrationAnnotations(ctx, claim)
    if err != nil {
        // Nothing was saved; we will fall back into the same
        // condition in the next call to this method
        return err
    }
    claim = newClaim

    if !metav1.HasAnnotation(claim.ObjectMeta, pvutil.AnnBindCompleted) {
        return ctrl.syncUnboundClaim(ctx, claim)
    } else {
        return ctrl.syncBoundClaim(claim)
    }
}

// syncUnboundClaim is the main controller method to decide what to do with an
// unbound claim.
func (ctrl *PersistentVolumeController) syncUnboundClaim(ctx context.Context, claim *v1.PersistentVolumeClaim) error {
    // a lot of code here
    obj, found, err := ctrl.volumes.store.GetByKey(claim.Spec.VolumeName)
    if err != nil {
        return err
    }
    if !found {
        // User asked for a PV that does not exist.
        // OBSERVATION: pvc is "Pending"
        // Retry later.
        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 {
            // a lot of code here
            } 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
        }
    }
}