从nfs-subdir-external-provisioner

简单总结

创建pvc后

对于设置了storageclassname的

provisioner watch到后匹配此storageclass对应provisioner是自身则创建pv

对于没设置storageclassname的

DefaultStorageClass这个准入控制器会设置此pvc的storageclassname为集群内设置的default storageclass对应的name
provisioner watch到后匹配此storageclass对应provisioner是自身则创建pv

provisioner相关源码

nfs-subdir-external-provisioner入口
cmd/nfs-subdir-external-provisioner/provisioner.go中

func main() {
...
    pc := controller.NewProvisionController(clientset,
        provisionerName,
        clientNFSProvisioner,
        serverVersion.GitVersion,
        controller.LeaderElection(leaderElection),
    )
    pc.Run(context.Background())
}

这边watch和判断pvc是否属于自己处理是sig-storage-lib-external-provisioner这个提供的库在处理
相对sig-storage-lib-external-provisioner
controller/controller.go中

func NewProvisionController(
    client kubernetes.Interface,
    provisionerName string,
    provisioner Provisioner,
    kubeVersion string,
    options ...func(*ProvisionController) error,
) *ProvisionController {
...
    volumeHandler := cache.ResourceEventHandlerFuncs{
        AddFunc:    func(obj interface{}) { controller.enqueueVolume(obj) },
        UpdateFunc: func(oldObj, newObj interface{}) { controller.enqueueVolume(newObj) },
        DeleteFunc: func(obj interface{}) { controller.forgetVolume(obj) },
    }
...
}

func (ctrl *ProvisionController) enqueueVolume(obj interface{}) {
    var key string
    var err error
    if key, err = cache.DeletionHandlingMetaNamespaceKeyFunc(obj); err != nil {
        utilruntime.HandleError(err)
        return
    }
    // Re-Adding is harmless but try to add it to the queue only if it is not
    // already there, because if it is already there we *must* be retrying it
    if ctrl.volumeQueue.NumRequeues(key) == 0 {
        ctrl.volumeQueue.Add(key)
    }
}

func (ctrl *ProvisionController) Run(ctx context.Context) {
...
    for i := 0; i < ctrl.threadiness; i++ {
            go wait.Until(func() { ctrl.runClaimWorker(ctx) }, time.Second, ctx.Done())
            go wait.Until(func() { ctrl.runVolumeWorker(ctx) }, time.Second, ctx.Done())
        }
...
}


func (ctrl *ProvisionController) runClaimWorker(ctx context.Context) {
    for ctrl.processNextClaimWorkItem(ctx) {
    }
}

func (ctrl *ProvisionController) processNextClaimWorkItem(ctx context.Context) bool {
...
if err := ctrl.syncClaimHandler(ctx, key); err != nil {
...
}
...
}

func (ctrl *ProvisionController) syncClaimHandler(ctx context.Context, key string) error {
...
return ctrl.syncClaim(ctx, claimObj)
}

func (ctrl *ProvisionController) syncClaim(ctx context.Context, obj interface{}) error {
should, err := ctrl.shouldProvision(ctx, claim)
...
status, err := ctrl.provisionClaimOperation(ctx, claim)
...
}


判断是否属于自己处理
func (ctrl *ProvisionController) shouldProvision(ctx context.Context, claim *v1.PersistentVolumeClaim) (bool, error) {
...
        class, err := ctrl.getStorageClass(claimClass)
        if err != nil {
            glog.Errorf("Error getting claim %q's StorageClass's fields: %v", claimToClaimKey(claim), err)
            return false, err
        }
        if class.Provisioner != ctrl.provisionerName {
            return false, nil
        }
...
}

调用Provision方法

func (ctrl *ProvisionController) provisionClaimOperation(ctx context.Context, claim *v1.PersistentVolumeClaim) (ProvisioningState, error) {
...
    volume, result, err := ctrl.provisioner.Provision(ctx, options)
...
}

最终进入的是nfs-subdir-external-provisioner提供的Provision方法,返回pv
相对nfs-subdir-external-provisioner
cmd/nfs-subdir-external-provisioner/provisioner.go中

func (p *nfsProvisioner) Provision(ctx context.Context, options controller.ProvisionOptions) (*v1.PersistentVolume, controller.ProvisioningState, error) {
...
    pv := &v1.PersistentVolume{
        ObjectMeta: metav1.ObjectMeta{
            Name: options.PVName,
        },
        Spec: v1.PersistentVolumeSpec{
            PersistentVolumeReclaimPolicy: *options.StorageClass.ReclaimPolicy,
            AccessModes:                   options.PVC.Spec.AccessModes,
            MountOptions:                  options.StorageClass.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,
                },
            },
        },
    }
    return pv, controller.ProvisioningFinished, nil
}

DefaultStorageClass相关代码

相对kubernetes
plugin/pkg/admission/storage/storageclass/setdefault/admission.go中

const (
    // PluginName is the name of this admission controller plugin
    PluginName = "DefaultStorageClass"
)

// Register registers a plugin
func Register(plugins *admission.Plugins) {
    plugins.Register(PluginName, func(config io.Reader) (admission.Interface, error) {
        plugin := newPlugin()
        return plugin, nil
    })
}

func (a *claimDefaulterPlugin) Admit(ctx context.Context, attr admission.Attributes, o admission.ObjectInterfaces) error {
...
设置了storageclassname,则跳过
    if helper.PersistentVolumeClaimHasClass(pvc) {
        // The user asked for a class.
        return nil
    }

...
未设置storageclassname,则设置默认的storageclass对应的name
    def, err := getDefaultClass(a.lister)
    if err != nil {
        return admission.NewForbidden(attr, err)
    }
    if def == nil {
        // No default class selected, do nothing about the PVC.
        return nil
    }

    pvc.Spec.StorageClassName = &def.Name
...
}

补充

所有的k8s内置的准入控制器
相对kubernetes
pkg/kubeapiserver/options/admission.go中
···
func NewAdmissionOptions() *AdmissionOptions {
...
RegisterAllAdmissionPlugins(options.Plugins)
...
return &AdmissionOptions{
GenericAdmission: options,
}
}
···
pkg/kubeapiserver/options/plugins.go中

func RegisterAllAdmissionPlugins(plugins *admission.Plugins) {
    admit.Register(plugins) // DEPRECATED as no real meaning
    alwayspullimages.Register(plugins)
    antiaffinity.Register(plugins)
    defaulttolerationseconds.Register(plugins)
    defaultingressclass.Register(plugins)
    denyserviceexternalips.Register(plugins)
    deny.Register(plugins) // DEPRECATED as no real meaning
    eventratelimit.Register(plugins)
    extendedresourcetoleration.Register(plugins)
    gc.Register(plugins)
    imagepolicy.Register(plugins)
    limitranger.Register(plugins)
    autoprovision.Register(plugins)
    exists.Register(plugins)
    noderestriction.Register(plugins)
    nodetaint.Register(plugins)
    label.Register(plugins) // DEPRECATED, future PVs should not rely on labels for zone topology
    podnodeselector.Register(plugins)
    podtolerationrestriction.Register(plugins)
    runtimeclass.Register(plugins)
    resourcequota.Register(plugins)
    podsecurity.Register(plugins)
    podsecuritypolicy.Register(plugins)
    podpriority.Register(plugins)
    scdeny.Register(plugins)
    serviceaccount.Register(plugins)
    setdefault.Register(plugins)
    resize.Register(plugins)
    storageobjectinuseprotection.Register(plugins)
    certapproval.Register(plugins)
    certsigning.Register(plugins)
    certsubjectrestriction.Register(plugins)
}