Kubernetes 调度器与绑定

// bind the pod to its host asynchronously (we can do this b/c of the assumption step above).
go func() {
    bindingCycleCtx, cancel := context.WithCancel(ctx)
    defer cancel()
}()

waitOnPermitStatus := fwk.WaitOnPermit(bindingCycleCtx, assumedPod)
if !waitOnPermitStatus.IsSuccess() {
    var reason string
    if waitOnPermitStatus.IsUnschedulable() {
        metrics.PodUnschedulable(fwk.ProfileName(), metrics.SinceInSeconds(start))
        reason = v1.PodReasonUnschedulable
    } else {
        metrics.PodScheduleError(fwk.ProfileName(), metrics.SinceInSeconds(start))
        reason = SchedulerError
    }
    // trigger un-reserve plugins to clean up state associated with the reserved Pod
    fwk.RunReservePluginsUnreserve(bindingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost)
    if forgetErr := sched.SchedulerCache.ForgetPod(assumedPod); forgetErr != nil {
        klog.ErrorS(forgetErr, "scheduler cache ForgetPod failed")
    } else {
        // "Forget"ing an assumed Pod in binding cycle should be treated as a PodDelete event,
        // as the assumed Pod had occupied a certain amount of resources in scheduler cache.
        // TODO(#103853): de-duplicate the logic.
        // Avoid moving the assumed Pod itself as it's always Unschedulable.
        // It's intentional to "defer" this operation; otherwise MoveAllToActiveOrBackoffQueue() would
        // update `q.moveRequest` and thus move the assumed pod to backoffQ anyways.
        defer sched.SchedulingQueue.MoveAllToActiveOrBackoffQueue(internalqueue.AssignedPodDelete, func(pod *v1.Pod) bool {
            return assumedPod.UID != pod.UID
        })
    }
    sched.recordSchedulingFailure(fwk, assumedPodInfo, waitOnPermitStatus.AsError(), reason, clearNominatedNode)
    return
}

// Run "prebind" plugins.
preBindStatus := fwk.RunPreBindPlugins(bindingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost)

func (f *frameworkImpl) RunPreBindPlugins(ctx context.Context, state *framework.CycleState, pod *v1.Pod, nodeName string) (status *framework.Status) {
    // a lot of code here
    for _, pl := range f.preBindPlugins {
        status = f.runPreBindPlugin(ctx, pl, state, pod, nodeName)
        if !status.IsSuccess() {
            err := status.AsError()
            klog.ErrorS(err, "Failed running PreBind plugin", "plugin", pl.Name(), "pod", klog.KObj(pod))
            return framework.AsStatus(fmt.Errorf("running PreBind plugin %q: %w", pl.Name(), err))
        }
    }
    return nil
}

func (f *frameworkImpl) runPreBindPlugin(ctx context.Context, pl framework.PreBindPlugin, state *framework.CycleState, pod *v1.Pod, nodeName string) *framework.Status {
    // a lot of code here
    status := pl.PreBind(ctx, state, pod, nodeName)
    f.metricsRecorder.observePluginDurationAsync(preBind, pl.Name(), status, metrics.SinceInSeconds(startTime))
    return status
}

func NewInTreeRegistry() runtime.Registry {
    // a lot of code here
    return runtime.Registry{
        // a lot of code here
        volumebinding.Name:                   runtime.FactoryAdapter(fts, volumebinding.New),
    }
}

type VolumeBinding struct {
    Binder    SchedulerVolumeBinder
    PVCLister corelisters.PersistentVolumeClaimLister
    scorer    volumeCapacityScorer
    fts       feature.Features
}

func (pl *VolumeBinding) PreBind(ctx context.Context, cs *framework.CycleState, pod *v1.Pod, nodeName string) *framework.Status {
    s, err := getStateData(cs)
    if err != nil {
        return framework.AsStatus(err)
    }
    if s.allBound {
        // no need to bind volumes
        return nil
    }
    // we don't need to hold the lock as only one node will be pre-bound for the given pod
    podVolumes, ok := s.podVolumesByNode[nodeName]
    if !ok {
        return framework.AsStatus(fmt.Errorf("no pod volumes found for node %q", nodeName))
    }
    klog.V(5).InfoS("Trying to bind volumes for pod", "pod", klog.KObj(pod))
    err = pl.Binder.BindPodVolumes(pod, podVolumes)
    if err != nil {
        klog.V(1).InfoS("Failed to bind volumes for pod", "pod", klog.KObj(pod), "err", err)
        return framework.AsStatus(err)
    }
    klog.V(5).InfoS("Success binding volumes for pod", "pod", klog.KObj(pod))
    return nil
}

func (b *volumeBinder) BindPodVolumes(assumedPod *v1.Pod, podVolumes *PodVolumes) (err error) {
    // a lot of code here
    bindings := podVolumes.StaticBindings
    claimsToProvision := podVolumes.DynamicProvisions

    // Start API operations
    err = b.bindAPIUpdate(assumedPod, bindings, claimsToProvision)
    if err != nil {
        return err
    }

    err = wait.Poll(time.Second, b.bindTimeout, func() (bool, error) {
        b, err := b.checkBindings(assumedPod, bindings, claimsToProvision)
        return b, err
    })
    if err != nil {
        return fmt.Errorf("binding volumes: %w", err)
    }
    return nil
}

// Bind the volume to the claim if it is not bound yet
if volume.Spec.ClaimRef == nil ||
    volume.Spec.ClaimRef.Name != claim.Name ||
    volume.Spec.ClaimRef.Namespace != claim.Namespace ||
    volume.Spec.ClaimRef.UID != claim.UID {

    claimRef, err := reference.GetReference(scheme.Scheme, claim)
    if err != nil {
        return nil, false, fmt.Errorf("unexpected error getting claim reference: %w", err)
    }
    volumeClone.Spec.ClaimRef = claimRef
    dirty = true
}

err := sched.bind(bindingCycleCtx, fwk, assumedPod, scheduleResult.SuggestedHost, state)

func (f *frameworkImpl) RunBindPlugins(ctx context.Context, state *framework.CycleState, pod *v1.Pod, nodeName string) (status *framework.Status) {
    // a lot of code here
    for _, bp := range f.bindPlugins {
        status = f.runBindPlugin(ctx, bp, state, pod, nodeName)
        if status != nil && status.Code() == framework.Skip {
            continue
        }
        if !status.IsSuccess() {
            err := status.AsError()
            klog.ErrorS(err, "Failed running Bind plugin", "plugin", bp.Name(), "pod", klog.KObj(pod))
            return framework.AsStatus(fmt.Errorf("running Bind plugin %q: %w", bp.Name(), err))
        }
        return status
    }
    return status
}

func (f *frameworkImpl) runBindPlugin(ctx context.Context, bp framework.BindPlugin, state *framework.CycleState, pod *v1.Pod, nodeName string) *framework.Status {
    // a lot of code here
    status := bp.Bind(ctx, state, pod, nodeName)
    f.metricsRecorder.observePluginDurationAsync(bind, bp.Name(), status, metrics.SinceInSeconds(startTime))
    return status
}

func (b DefaultBinder) Bind(ctx context.Context, state *framework.CycleState, p *v1.Pod, nodeName string) *framework.Status {
    klog.V(3).InfoS("Attempting to bind pod to node", "pod", klog.KObj(p), "node", nodeName)
    binding := &v1.Binding{
        ObjectMeta: metav1.ObjectMeta{Namespace: p.Namespace, Name: p.Name, UID: p.UID},
        Target:     v1.ObjectReference{Kind: "Node", Name: nodeName},
    }
    err := b.handle.ClientSet().CoreV1().Pods(binding.Namespace).Bind(ctx, binding, metav1.CreateOptions{})
    if err != nil {
        return framework.AsStatus(err)
    }
    return nil
}

// Run "postbind" plugins.
fwk.RunPostBindPlugins(bindingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost)

// RunPostBindPlugins runs the set of configured postbind plugins.
func (f *frameworkImpl) RunPostBindPlugins(ctx context.Context, state *framework.CycleState, pod *v1.Pod, nodeName string) {
    //a lot of code here
    for _, pl := range f.postBindPlugins {
        f.runPostBindPlugin(ctx, pl, state, pod, nodeName)
    }
}

func (f *frameworkImpl) runPostBindPlugin(ctx context.Context, pl framework.PostBindPlugin, state *framework.CycleState, pod *v1.Pod, nodeName string) {
    // a lot of code here
    startTime := time.Now()
    pl.PostBind(ctx, state, pod, nodeName)
    f.metricsRecorder.observePluginDurationAsync(postBind, pl.Name(), nil, metrics.SinceInSeconds(startTime))
}