package scheduling import ( "context" "fmt" "sort" "strings" "time" "github.com/prometheus/client_golang/prometheus" kevents "k8s.io/client-go/tools/events" "k8s.io/klog/v2" clusterclient "open-cluster-management.io/api/client/cluster/clientset/versioned" clusterlisterv1 "open-cluster-management.io/api/client/cluster/listers/cluster/v1" clusterlisterv1alpha1 "open-cluster-management.io/api/client/cluster/listers/cluster/v1alpha1" clusterlisterv1beta1 "open-cluster-management.io/api/client/cluster/listers/cluster/v1beta1" clusterapiv1 "open-cluster-management.io/api/cluster/v1" clusterapiv1beta1 "open-cluster-management.io/api/cluster/v1beta1" "open-cluster-management.io/ocm/pkg/placement/controllers/framework" "open-cluster-management.io/ocm/pkg/placement/controllers/metrics" "open-cluster-management.io/ocm/pkg/placement/plugins" "open-cluster-management.io/ocm/pkg/placement/plugins/addon" "open-cluster-management.io/ocm/pkg/placement/plugins/balance" "open-cluster-management.io/ocm/pkg/placement/plugins/predicate" "open-cluster-management.io/ocm/pkg/placement/plugins/resource" "open-cluster-management.io/ocm/pkg/placement/plugins/steady" "open-cluster-management.io/ocm/pkg/placement/plugins/tainttoleration" ) const ( PrioritizerBalance string = "Balance" PrioritizerSteady string = "Steady" PrioritizerResourceAllocatableCPU string = "ResourceAllocatableCPU" PrioritizerResourceAllocatableMemory string = "ResourceAllocatableMemory" ) // PrioritizerScore defines the score for each cluster type PrioritizerScore map[string]int64 // Scheduler is an interface for scheduler, it returs the scheduler results type Scheduler interface { Schedule( ctx context.Context, placement *clusterapiv1beta1.Placement, clusters []*clusterapiv1.ManagedCluster, ) (ScheduleResult, *framework.Status) } type ScheduleResult interface { // FilterResults returns results for each filter FilterResults() []FilterResult // PrioritizerResults returns results for each prioritizer PrioritizerResults() []PrioritizerResult // PrioritizerScores returns total score for each cluster PrioritizerScores() PrioritizerScore // Decisions returns the decision groups of the schedule Decisions() []*clusterapiv1.ManagedCluster // NumOfUnscheduled returns the number of unscheduled. NumOfUnscheduled() int // RequeueAfter returns the requeue time interval of the placement RequeueAfter() *time.Duration } type FilterResult struct { Name string `json:"name"` FilteredClusters []string `json:"filteredClusters"` } // PrioritizerResult defines the result of one prioritizer, // include name, weight, and score of each cluster. type PrioritizerResult struct { Name string `json:"name"` Weight int32 `json:"weight"` Scores PrioritizerScore `json:"scores"` } // ScheduleResult is the result for a certain schedule. type scheduleResult struct { feasibleClusters []*clusterapiv1.ManagedCluster scheduledDecisions []*clusterapiv1.ManagedCluster unscheduledDecisions int filteredRecords map[string][]*clusterapiv1.ManagedCluster scoreRecords []PrioritizerResult scoreSum PrioritizerScore requeueAfter *time.Duration } type schedulerHandler struct { eventsRecorder kevents.EventRecorder metricsRecorder *metrics.ScheduleMetrics placementDecisionLister clusterlisterv1beta1.PlacementDecisionLister scoreLister clusterlisterv1alpha1.AddOnPlacementScoreLister clusterLister clusterlisterv1.ManagedClusterLister clusterClient clusterclient.Interface } func NewSchedulerHandler( clusterClient clusterclient.Interface, placementDecisionLister clusterlisterv1beta1.PlacementDecisionLister, scoreLister clusterlisterv1alpha1.AddOnPlacementScoreLister, clusterLister clusterlisterv1.ManagedClusterLister, eventsRecorder kevents.EventRecorder, metricsRecorder *metrics.ScheduleMetrics, ) plugins.Handle { return &schedulerHandler{ eventsRecorder: eventsRecorder, metricsRecorder: metricsRecorder, placementDecisionLister: placementDecisionLister, scoreLister: scoreLister, clusterLister: clusterLister, clusterClient: clusterClient, } } func (s *schedulerHandler) EventRecorder() kevents.EventRecorder { return s.eventsRecorder } func (s *schedulerHandler) DecisionLister() clusterlisterv1beta1.PlacementDecisionLister { return s.placementDecisionLister } func (s *schedulerHandler) ScoreLister() clusterlisterv1alpha1.AddOnPlacementScoreLister { return s.scoreLister } func (s *schedulerHandler) ClusterLister() clusterlisterv1.ManagedClusterLister { return s.clusterLister } func (s *schedulerHandler) ClusterClient() clusterclient.Interface { return s.clusterClient } func (s *schedulerHandler) MetricsRecorder() *metrics.ScheduleMetrics { return s.metricsRecorder } // Initialize the default prioritizer weight. // Balane and Steady weight 1, others weight 0. // The default weight can be replaced by each placement's PrioritizerConfigs. var defaultPrioritizerConfig = map[clusterapiv1beta1.ScoreCoordinate]int32{ { Type: clusterapiv1beta1.ScoreCoordinateTypeBuiltIn, BuiltIn: PrioritizerBalance, }: 1, { Type: clusterapiv1beta1.ScoreCoordinateTypeBuiltIn, BuiltIn: PrioritizerSteady, }: 1, } type pluginScheduler struct { handle plugins.Handle filters []plugins.Filter prioritizerWeights map[clusterapiv1beta1.ScoreCoordinate]int32 } func NewPluginScheduler(handle plugins.Handle) *pluginScheduler { return &pluginScheduler{ handle: handle, filters: []plugins.Filter{ predicate.New(handle), tainttoleration.New(handle), }, prioritizerWeights: defaultPrioritizerConfig, } } func (s *pluginScheduler) Schedule( ctx context.Context, placement *clusterapiv1beta1.Placement, clusters []*clusterapiv1.ManagedCluster, ) (ScheduleResult, *framework.Status) { logger := klog.FromContext(ctx) filtered := clusters finalStatus := framework.NewStatus("", framework.Success, "") results := &scheduleResult{ filteredRecords: map[string][]*clusterapiv1.ManagedCluster{}, scoreRecords: []PrioritizerResult{}, } // filter clusters var filterPipline []string for _, f := range s.filters { startTime := time.Now() filterResult, status := f.Filter(ctx, placement, filtered) metrics.PluginDuration.With(prometheus.Labels{ "name": metrics.SchedulingName, "plugin_type": "filter", "plugin_name": f.Name(), }).Observe(s.handle.MetricsRecorder().SinceInSeconds(startTime)) filtered = filterResult.Filtered switch { case status.IsError(): return results, status case status.Code() == framework.Warning: logger.Info("Warning status message", "message", status.Message()) finalStatus = status } filterPipline = append(filterPipline, f.Name()) results.filteredRecords[strings.Join(filterPipline, ",")] = filtered } // Prioritize clusters // 1. Get weight for each prioritizers. // For example, weights is {"Steady": 1, "Balance":1, "AddOn/default/ratio":3}. weights, status := getWeights(s.prioritizerWeights, placement) switch { case status.IsError(): return results, status case status.Code() == framework.Warning: logger.Info("Warning status message", "message", status.Message()) finalStatus = status } // 2. Generate prioritizers for each placement whose weight != 0. prioritizers, status := getPrioritizers(weights, s.handle) switch { case status.IsError(): return results, status case status.Code() == framework.Warning: logger.Info("Warning status message", "message", status.Message()) finalStatus = status } // 3. Calculate clusters scores. scoreSum := PrioritizerScore{} for _, cluster := range filtered { scoreSum[cluster.Name] = 0 } for sc, p := range prioritizers { // Get cluster score. startTime := time.Now() scoreResult, status := p.Score(ctx, placement, filtered) metrics.PluginDuration.With(prometheus.Labels{ "name": metrics.SchedulingName, "plugin_type": "prioritizer", "plugin_name": p.Name(), }).Observe(s.handle.MetricsRecorder().SinceInSeconds(startTime)) score := scoreResult.Scores switch { case status.IsError(): return results, status case status.Code() == framework.Warning: logger.Info("Warning status message", "message", status.Message()) finalStatus = status } // Record prioritizer score and weight weight := weights[sc] results.scoreRecords = append(results.scoreRecords, PrioritizerResult{Name: p.Name(), Weight: weight, Scores: score}) // The final score is a sum of each prioritizer score * weight. // A higher weight indicates that the prioritizer weights more in the cluster selection, // while 0 weight indicate that the prioritizer is disabled. for name, val := range score { scoreSum[name] = scoreSum[name] + val*int64(weight) } } // 4. Sort clusters by score, if score is equal, sort by name sort.SliceStable(filtered, func(i, j int) bool { if scoreSum[filtered[i].Name] == scoreSum[filtered[j].Name] { return filtered[i].Name < filtered[j].Name } else { return scoreSum[filtered[i].Name] > scoreSum[filtered[j].Name] } }) results.feasibleClusters = filtered results.scoreSum = scoreSum // select clusters and generate cluster decisions decisions := selectClusters(placement, filtered) scheduled, unscheduled := len(decisions), 0 if placement.Spec.NumberOfClusters != nil { unscheduled = int(*placement.Spec.NumberOfClusters) - scheduled } results.scheduledDecisions = decisions results.unscheduledDecisions = unscheduled // set placement requeue time for _, f := range s.filters { if r, _ := f.RequeueAfter(ctx, placement); r.RequeueTime != nil { newRequeueAfter := time.Until(*r.RequeueTime) results.requeueAfter = setRequeueAfter(results.requeueAfter, &newRequeueAfter) } } for _, p := range prioritizers { if r, _ := p.RequeueAfter(ctx, placement); r.RequeueTime != nil { newRequeueAfter := time.Until(*r.RequeueTime) results.requeueAfter = setRequeueAfter(results.requeueAfter, &newRequeueAfter) } } return results, finalStatus } // selects clusters based on given cluster slice and number of clusters func selectClusters(placement *clusterapiv1beta1.Placement, clusters []*clusterapiv1.ManagedCluster) []*clusterapiv1.ManagedCluster { numOfDecisions := len(clusters) if placement.Spec.NumberOfClusters != nil { numOfDecisions = int(*placement.Spec.NumberOfClusters) } // truncate the cluster slice if the desired number of decisions is less than // the number of the candidate clusters if numOfDecisions < len(clusters) { clusters = clusters[:numOfDecisions] } return clusters } // setRequeueAfter selects minimal time.Duration as requeue time func setRequeueAfter(requeueAfter, newRequeueAfter *time.Duration) *time.Duration { if newRequeueAfter == nil { return requeueAfter } if requeueAfter == nil || *newRequeueAfter < *requeueAfter { return newRequeueAfter } return requeueAfter } // Get prioritizer weight for the placement. // In Additive and "" mode, will override defaultWeight with what placement has defined and return. // In Exact mode, will return the name and weight defined in placement. func getWeights(defaultWeight map[clusterapiv1beta1.ScoreCoordinate]int32, placement *clusterapiv1beta1.Placement) (map[clusterapiv1beta1.ScoreCoordinate]int32, *framework.Status) { mode := placement.Spec.PrioritizerPolicy.Mode switch { case mode == clusterapiv1beta1.PrioritizerPolicyModeExact: return mergeWeights(nil, placement.Spec.PrioritizerPolicy.Configurations) case mode == clusterapiv1beta1.PrioritizerPolicyModeAdditive || mode == "": return mergeWeights(defaultWeight, placement.Spec.PrioritizerPolicy.Configurations) default: msg := fmt.Sprintf("incorrect prioritizer policy mode: %s", mode) return nil, framework.NewStatus("", framework.Misconfigured, msg) } } func mergeWeights(defaultWeight map[clusterapiv1beta1.ScoreCoordinate]int32, customizedWeight []clusterapiv1beta1.PrioritizerConfig, ) (map[clusterapiv1beta1.ScoreCoordinate]int32, *framework.Status) { weights := make(map[clusterapiv1beta1.ScoreCoordinate]int32) status := framework.NewStatus("", framework.Success, "") // copy the default weight for sc, w := range defaultWeight { weights[sc] = w } // override default weight for _, c := range customizedWeight { if c.ScoreCoordinate != nil { weights[*c.ScoreCoordinate] = c.Weight } else { return nil, framework.NewStatus("", framework.Misconfigured, "scoreCoordinate field is required") } } return weights, status } // Generate prioritizers for the placement. func getPrioritizers(weights map[clusterapiv1beta1.ScoreCoordinate]int32, handle plugins.Handle, ) (map[clusterapiv1beta1.ScoreCoordinate]plugins.Prioritizer, *framework.Status) { result := make(map[clusterapiv1beta1.ScoreCoordinate]plugins.Prioritizer) status := framework.NewStatus("", framework.Success, "") for k, v := range weights { if v == 0 { continue } if k.Type == clusterapiv1beta1.ScoreCoordinateTypeBuiltIn { switch { case k.BuiltIn == PrioritizerBalance: result[k] = balance.New(handle) case k.BuiltIn == PrioritizerSteady: result[k] = steady.New(handle) case k.BuiltIn == PrioritizerResourceAllocatableCPU || k.BuiltIn == PrioritizerResourceAllocatableMemory: result[k] = resource.NewResourcePrioritizerBuilder(handle).WithPrioritizerName(k.BuiltIn).Build() default: msg := fmt.Sprintf("incorrect builtin prioritizer: %s", k.BuiltIn) return nil, framework.NewStatus("", framework.Misconfigured, msg) } } else { if k.AddOn == nil { return nil, framework.NewStatus("", framework.Misconfigured, "addOn should not be empty") } result[k] = addon.NewAddOnPrioritizerBuilder(handle).WithResourceName(k.AddOn.ResourceName).WithScoreName(k.AddOn.ScoreName).Build() } } return result, status } func (r *scheduleResult) FilterResults() []FilterResult { var results []FilterResult // order the FilterResults by key length var filteredRecordsKey []string for name := range r.filteredRecords { filteredRecordsKey = append(filteredRecordsKey, name) } sort.SliceStable(filteredRecordsKey, func(i, j int) bool { return len(filteredRecordsKey[i]) < len(filteredRecordsKey[j]) }) // go through the FilterResults by key length for _, name := range filteredRecordsKey { result := FilterResult{Name: name, FilteredClusters: []string{}} for _, c := range r.filteredRecords[name] { result.FilteredClusters = append(result.FilteredClusters, c.Name) } results = append(results, result) } return results } func (r *scheduleResult) PrioritizerResults() []PrioritizerResult { return r.scoreRecords } func (r *scheduleResult) PrioritizerScores() PrioritizerScore { return r.scoreSum } func (r *scheduleResult) Decisions() []*clusterapiv1.ManagedCluster { return r.scheduledDecisions } func (r *scheduleResult) NumOfUnscheduled() int { return r.unscheduledDecisions } func (r *scheduleResult) RequeueAfter() *time.Duration { return r.requeueAfter }