Kubernetes调度器原理解析

return &schedulerapi.Plugins{
    QueueSort: &schedulerapi.PluginSet{
      Enabled: []schedulerapi.Plugin{
        {Name: queuesort.Name},
      },
    },
    PreFilter: &schedulerapi.PluginSet{
      Enabled: []schedulerapi.Plugin{
        {Name: noderesources.FitName},
        {Name: nodeports.Name},
        {Name: interpodaffinity.Name},
      },
    },
    Filter: &schedulerapi.PluginSet{
      Enabled: []schedulerapi.Plugin{
        {Name: nodeunschedulable.Name},
        {Name: noderesources.FitName},
        {Name: nodename.Name},
        {Name: nodeports.Name},
        {Name: nodeaffinity.Name},
        {Name: volumerestrictions.Name},
        {Name: tainttoleration.Name},
        {Name: nodevolumelimits.EBSName},
        {Name: nodevolumelimits.GCEPDName},
        {Name: nodevolumelimits.CSIName},
        {Name: nodevolumelimits.AzureDiskName},
        {Name: volumebinding.Name},
        {Name: volumezone.Name},
        {Name: interpodaffinity.Name},
      },
    },
    PreScore: &schedulerapi.PluginSet{
      Enabled: []schedulerapi.Plugin{
        {Name: interpodaffinity.Name},
        {Name: defaultpodtopologyspread.Name},
        {Name: tainttoleration.Name},
      },
    },
    Score: &schedulerapi.PluginSet{
      Enabled: []schedulerapi.Plugin{
        {Name: noderesources.BalancedAllocationName, Weight: 1},
        {Name: imagelocality.Name, Weight: 1},
        {Name: interpodaffinity.Name, Weight: 1},
        {Name: noderesources.LeastAllocatedName, Weight: 1},
        {Name: nodeaffinity.Name, Weight: 1},
        {Name: nodepreferavoidpods.Name, Weight: 10000},
        {Name: defaultpodtopologyspread.Name, Weight: 1},
        {Name: tainttoleration.Name, Weight: 1},
      },
    },
    Bind: &schedulerapi.PluginSet{
      Enabled: []schedulerapi.Plugin{
        {Name: defaultbinder.Name},
      },
    },
  }
}

func (sched *Scheduler) scheduleOne(ctx context.Context) {
  podInfo := sched.NextPod()
......
  scheduleResult, err := sched.Algorithm.Schedule(schedulingCycleCtx, prof, state, pod)
  if err != nil {
    if fitError, ok := err.(*core.FitError); ok {
      if sched.DisablePreemption {
        klog.V(3).Infof("Pod priority feature is not enabled or preemption is disabled by scheduler configuration." +
          " No preemption is performed.")
      } else {
        preemptionStartTime := time.Now()
        sched.preempt(schedulingCycleCtx, prof, state, pod, fitError)
        metrics.PreemptionAttempts.Inc()
      }
      metrics.PodScheduleFailures.Inc()
    } else if err == core.ErrNoNodesAvailable {
      // No nodes available is counted as unschedulable rather than an error.
      metrics.PodScheduleFailures.Inc()
    } else {
      klog.Errorf("error selecting node for pod: %v", err)
      metrics.PodScheduleErrors.Inc()
    }
    sched.recordSchedulingFailure(prof, podInfo.DeepCopy(), err, v1.PodReasonUnschedulable, err.Error())
    return
  }
......
}

func (g*genericScheduler) Schedule(ctx context.Context, fwk framework.Framework, state *framework.CycleState, pod *v1.Pod) (result ScheduleResult, err error) {
  defer trace.LogIfLong(100 * time.Millisecond)
  if err := g.snapshot(); err != nil {
    return result, err
  }
  trace.Step("Snapshotting scheduler cache and node infos done")
  if g.nodeInfoSnapshot.NumNodes() == 0 {
    return result, ErrNoNodesAvailable
  }
  feasibleNodes, diagnosis, err := g.findNodesThatFitPod(ctx, fwk, state, pod)
  if err != nil {
    return result, err
  }
  trace.Step("Computing predicates done")
  if len(feasibleNodes) == 0 {
    return result, &framework.FitError{
      Pod:         pod,
      NumAllNodes: g.nodeInfoSnapshot.NumNodes(),
      Diagnosis:   diagnosis,
    }
  }
  // When only one node after predicate, just use it.
  if len(feasibleNodes) == 1 {
    return ScheduleResult{
      SuggestedHost:  feasibleNodes[0].Name,
      EvaluatedNodes: 1 + len(diagnosis.NodeToStatusMap),
      FeasibleNodes:  1,
    }, nil
  }
  priorityList, err := g.prioritizeNodes(ctx, fwk, state, pod, feasibleNodes)
  if err != nil {
    return result, err
  }
  host, err := g.selectHost(priorityList)
  trace.Step("Prioritizing done")
  return ScheduleResult{
    SuggestedHost:  host,
    EvaluatedNodes: len(feasibleNodes) + len(diagnosis.NodeToStatusMap),
    FeasibleNodes:  len(feasibleNodes),
  }, err
}

const (
  // FitName is the name of the plugin used in the plugin registry and configurations.
  FitName = "NodeResourcesFit"
  // preFilterStateKey is the key in CycleState to NodeResourcesFit pre-computed data.
  // Using the name of the plugin will likely help us avoid collisions with other plugins.
  preFilterStateKey = "PreFilter" + FitName
)
// Fit is a plugin that checks if a node has sufficient resources.
type Fit struct {
  ignoredResources sets.String
}
// Name returns name of the plugin. It is used in logs, etc.
func (f *Fit) Name() string {
  return FitName
}
func computePodResourceRequest(pod *v1.Pod) *preFilterState {
  result := &preFilterState{}
  for _, container := range pod.Spec.Containers {
    result.Add(container.Resources.Requests)
  }
  // take max_resource(sum_pod, any_init_container)
  for _, container := range pod.Spec.InitContainers {
    result.SetMaxResource(container.Resources.Requests)
  }
  // If Overhead is being utilized, add to the total requests for the pod
  if pod.Spec.Overhead != nil && utilfeature.DefaultFeatureGate.Enabled(features.PodOverhead) {
    result.Add(pod.Spec.Overhead)
  }
  return result
}
// PreFilter invoked at the prefilter extension point.
func (f *Fit) PreFilter(ctx context.Context, cycleState *framework.CycleState, pod *v1.Pod) *framework.Status {
  cycleState.Write(preFilterStateKey, computePodResourceRequest(pod))
  return nil
}
func (f *Fit) Filter(ctx context.Context, cycleState *framework.CycleState, pod *v1.Pod, nodeInfo *schedulernodeinfo.NodeInfo) *framework.Status {
  s, err := getPreFilterState(cycleState)
  if err != nil {
    return framework.NewStatus(framework.Error, err.Error())
  }
  insufficientResources := fitsRequest(s, nodeInfo, f.ignoredResources)
  if len(insufficientResources) != 0 {
    // We will keep all failure reasons.
    failureReasons := make([]string, 0, len(insufficientResources))
    for _, r := range insufficientResources {
      failureReasons = append(failureReasons, r.Reason)
    }
    return framework.NewStatus(framework.Unschedulable, failureReasons...)
  }
  return nil
}

// Option configures a framework.Registry.
type Option func(framework.Registry) error
// NewSchedulerCommand creates a *cobra.Command object with default parameters and registryOptions
func NewSchedulerCommand(registryOptions ...Option) *cobra.Command {
...
}

// WithPlugin creates an Option based on plugin name and factory. Please don't remove this function: it is used to register out-of-tree plugins,
// hence there are no references to it from the kubernetes scheduler code base.
func WithPlugin(name string, factory framework.PluginFactory) Option {
  return func(registry framework.Registry) error {
    return registry.Register(name, factory)
  }

func main() {
  rand.Seed(time.Now().UnixNano())
  command := app.NewSchedulerCommand(app.WithPlugin(plugins.Name, plugins.New))
  logs.InitLogs()
  defer logs.FlushLogs()
  if err := command.Execute(); err != nil {
    os.Exit(1)
  }
}

profiles:
- pluginConfig:
- args:
args1: "example1"
name: sample-plugin
schedulerName: sample-scheduler
plugins:
preFilter:
enabled:
- name: "sample-plugin"
filter:
enabled:
- name: "sample-plugin"

func (g *genericScheduler) findNodesThatFitPod(ctx context.Context, prof *profile.Profile, state *framework.CycleState, pod *v1.Pod) ([]*v1.Node, framework.NodeToStatusMap, error) {
  filteredNodesStatuses := make(framework.NodeToStatusMap)
  filtered, err := g.findNodesThatPassFilters(ctx, prof, state, pod, filteredNodesStatuses)
  if err != nil {
    return nil, nil, err
  }
  filtered, err = g.findNodesThatPassExtenders(pod, filtered, filteredNodesStatuses)
  if err != nil {
    return nil, nil, err
  }
  return filtered, filteredNodesStatuses, nil
}

apiVersion: kubescheduler.config.k8s.io/v1alpha1
kind: KubeSchedulerConfiguration
clientConnection:
kubeconfig: /home/xyz2277/.kube/config
leaderElection:
leaderElect: false
leaseDuration: 25s
renewDeadline: 24s
algorithmSource:
policy:
file:
path: policy-config.json
policy-config.json:
{
"kind": "Policy",
"apiVersion": "v1",
"predicates": [
{
"name": "PodFitsHostPorts"
},
{
"name": "PodFitsResources"
},
{
"name": "NoDiskConflict"
},
{
"name": "MatchNodeSelector"
},
{
"name": "HostName"
}
],
"extenders": [
{
"urlPrefix": "http://<ip>:<port>/scheduler",
"apiVersion": "v1beta1",
"filterVerb": "predicates",
"prioritizeVerb": "prioritizes",
"bindVerb":   "bind",
"weight": 1,
"enableHttps": false,
"nodeCacheCapable": false,
"managedResources": [
{
"name": "ecloud.10086.cn/vcuda-core",
"ignoredByScheduler": false
}
]
}
],
"hardPodAffinitySymmetricWeight": 10,
"alwaysCheckAllPredicates": false
}