karmada-io
/
karmada
mirror of https://github.com/karmada-io/karmada.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

refactor static weight strategy 

Signed-off-by: Garrybest <garrybest@foxmail.com>
This commit is contained in:
Garrybest 2022-11-18 17:33:41 +08:00
parent 9fc2c4cd6d
commit 8d9780f838
5 changed files with 380 additions and 254 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
pkg/scheduler/core/assignment.go
View File

@ -48,14 +48,28 @@ func assignByAggregatedStrategy(state *assignState) ([]workv1alpha2.TargetCluste
return divideReplicasByResource(state.candidates, state.object, policyv1alpha1.ReplicaDivisionPreferenceAggregated)
}
// assignByStaticWeightStrategy assigns replicas by StaticWeightStrategy.
/*
* assignByStaticWeightStrategy assigns a total number of replicas to the selected clusters by the weight list.
* For example, we want to assign replicas to two clusters named A and B.
* | Total | Weight(A:B) | Assignment(A:B) |
* | 9 | 1:2 | 3:6 |
* | 9 | 1:3 | 2:7 | Approximate assignment
* Note:
* 1. If any selected cluster which not present on the weight list will be ignored(different with '0' replica).
* 2. In case of not enough replica for specific cluster which will get '0' replica.
*/
func assignByStaticWeightStrategy(state *assignState) ([]workv1alpha2.TargetCluster, error) {
// If ReplicaDivisionPreference is set to "Weighted" and WeightPreference is not set,
// scheduler will weight all clusters averagely.
if state.strategy.WeightPreference == nil {
state.strategy.WeightPreference = getDefaultWeightPreference(state.candidates)
}
return divideReplicasByStaticWeight(state.candidates, state.strategy.WeightPreference.StaticWeightList, state.object.Replicas)
weightList := getStaticWeightInfoList(state.candidates, state.strategy.WeightPreference.StaticWeightList)
acc := newDispenser(state.object.Replicas, nil)
acc.takeByWeight(weightList)
return acc.result, nil
}
// assignByDynamicWeightStrategy assigns replicas by assignByDynamicWeightStrategy.

218
pkg/scheduler/core/assignment_test.go Normal file
View File

@ -0,0 +1,218 @@
package core
import (
"testing"
clusterv1alpha1 "github.com/karmada-io/karmada/pkg/apis/cluster/v1alpha1"
policyv1alpha1 "github.com/karmada-io/karmada/pkg/apis/policy/v1alpha1"
workv1alpha2 "github.com/karmada-io/karmada/pkg/apis/work/v1alpha2"
"github.com/karmada-io/karmada/test/helper"
)
func Test_assignByStaticWeightStrategy(t *testing.T) {
tests := []struct {
name string
clusters []*clusterv1alpha1.Cluster
weightPreference *policyv1alpha1.ClusterPreferences
replicas int32
want []workv1alpha2.TargetCluster
wantErr bool
}{
{
name: "replica 12, weight 3:2:1",
clusters: []*clusterv1alpha1.Cluster{
helper.NewCluster(ClusterMember1),
helper.NewCluster(ClusterMember2),
helper.NewCluster(ClusterMember3),
},
weightPreference: &policyv1alpha1.ClusterPreferences{
StaticWeightList: []policyv1alpha1.StaticClusterWeight{
{
TargetCluster: policyv1alpha1.ClusterAffinity{
ClusterNames: []string{ClusterMember1},
},
Weight: 3,
},
{
TargetCluster: policyv1alpha1.ClusterAffinity{
ClusterNames: []string{ClusterMember2},
},
Weight: 2,
},
{
TargetCluster: policyv1alpha1.ClusterAffinity{
ClusterNames: []string{ClusterMember3},
},
Weight: 1,
},
},
},
replicas: 12,
want: []workv1alpha2.TargetCluster{
{
Name: ClusterMember1,
Replicas: 6,
},
{
Name: ClusterMember2,
Replicas: 4,
},
{
Name: ClusterMember3,
Replicas: 2,
},
},
wantErr: false,
},
{
name: "replica 12, default weight",
clusters: []*clusterv1alpha1.Cluster{
helper.NewCluster(ClusterMember1),
helper.NewCluster(ClusterMember2),
helper.NewCluster(ClusterMember3),
},
replicas: 12,
want: []workv1alpha2.TargetCluster{
{
Name: ClusterMember1,
Replicas: 4,
},
{
Name: ClusterMember2,
Replicas: 4,
},
{
Name: ClusterMember3,
Replicas: 4,
},
},
wantErr: false,
},
{
name: "replica 14, weight 3:2:1",
clusters: []*clusterv1alpha1.Cluster{
helper.NewCluster(ClusterMember1),
helper.NewCluster(ClusterMember2),
helper.NewCluster(ClusterMember3),
},
weightPreference: &policyv1alpha1.ClusterPreferences{
StaticWeightList: []policyv1alpha1.StaticClusterWeight{
{
TargetCluster: policyv1alpha1.ClusterAffinity{
ClusterNames: []string{ClusterMember1},
},
Weight: 3,
},
{
TargetCluster: policyv1alpha1.ClusterAffinity{
ClusterNames: []string{ClusterMember2},
},
Weight: 2,
},
{
TargetCluster: policyv1alpha1.ClusterAffinity{
ClusterNames: []string{ClusterMember3},
},
Weight: 1,
},
},
},
replicas: 14,
want: []workv1alpha2.TargetCluster{
{
Name: ClusterMember1,
Replicas: 8,
},
{
Name: ClusterMember2,
Replicas: 4,
},
{
Name: ClusterMember3,
Replicas: 2,
},
},
wantErr: false,
},
{
name: "insufficient replica assignment should get 0 replica",
clusters: []*clusterv1alpha1.Cluster{
helper.NewCluster(ClusterMember1),
helper.NewCluster(ClusterMember2),
},
weightPreference: &policyv1alpha1.ClusterPreferences{
StaticWeightList: []policyv1alpha1.StaticClusterWeight{
{
TargetCluster: policyv1alpha1.ClusterAffinity{
ClusterNames: []string{ClusterMember1},
},
Weight: 1,
},
{
TargetCluster: policyv1alpha1.ClusterAffinity{
ClusterNames: []string{ClusterMember2},
},
Weight: 1,
},
},
},
replicas: 0,
want: []workv1alpha2.TargetCluster{
{
Name: ClusterMember1,
Replicas: 0,
},
{
Name: ClusterMember2,
Replicas: 0,
},
},
wantErr: false,
},
{
name: "selected cluster without weight should be ignored",
clusters: []*clusterv1alpha1.Cluster{
helper.NewCluster(ClusterMember1),
helper.NewCluster(ClusterMember2),
},
weightPreference: &policyv1alpha1.ClusterPreferences{
StaticWeightList: []policyv1alpha1.StaticClusterWeight{
{
TargetCluster: policyv1alpha1.ClusterAffinity{
ClusterNames: []string{ClusterMember1},
},
Weight: 1,
},
},
},
replicas: 2,
want: []workv1alpha2.TargetCluster{
{
Name: ClusterMember1,
Replicas: 2,
},
},
wantErr: false,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
got, err := assignByStaticWeightStrategy(&assignState{
candidates: tt.clusters,
strategy: &policyv1alpha1.ReplicaSchedulingStrategy{
ReplicaSchedulingType: policyv1alpha1.ReplicaSchedulingTypeDivided,
ReplicaDivisionPreference: policyv1alpha1.ReplicaDivisionPreferenceWeighted,
WeightPreference: tt.weightPreference,
},
object: &workv1alpha2.ResourceBindingSpec{Replicas: tt.replicas},
})
if (err != nil) != tt.wantErr {
t.Errorf("divideReplicasByStaticWeight() error = %v, wantErr %v", err, tt.wantErr)
return
}
if !helper.IsScheduleResultEqual(got, tt.want) {
t.Errorf("divideReplicasByStaticWeight() got = %v, want %v", got, tt.want)
}
})
}
}

131
pkg/scheduler/core/division_algorithm.go
View File

@ -20,6 +20,82 @@ func (a TargetClustersList) Len() int { return len(a) }
func (a TargetClustersList) Swap(i, j int) { a[i], a[j] = a[j], a[i] }
func (a TargetClustersList) Less(i, j int) bool { return a[i].Replicas > a[j].Replicas }
type dispenser struct {
numReplicas int32
result []workv1alpha2.TargetCluster
}
func newDispenser(numReplicas int32, init []workv1alpha2.TargetCluster) *dispenser {
cp := make([]workv1alpha2.TargetCluster, len(init))
copy(cp, init)
return &dispenser{numReplicas: numReplicas, result: cp}
}
func (a *dispenser) done() bool {
return a.numReplicas == 0 && len(a.result) != 0
}
func (a *dispenser) takeByWeight(w helper.ClusterWeightInfoList) {
if a.done() {
return
}
sum := w.GetWeightSum()
if sum == 0 {
return
}
sort.Sort(w)
result := make([]workv1alpha2.TargetCluster, 0, w.Len())
remain := a.numReplicas
for _, info := range w {
replicas := int32(info.Weight * int64(a.numReplicas) / sum)
result = append(result, workv1alpha2.TargetCluster{
Name: info.ClusterName,
Replicas: replicas,
})
remain -= replicas
}
// TODO(Garrybest): take rest replicas by fraction part
for i := range result {
if remain == 0 {
break
}
result[i].Replicas++
remain--
}
a.numReplicas = remain
a.result = util.MergeTargetClusters(a.result, result)
}
func getStaticWeightInfoList(clusters []*clusterv1alpha1.Cluster, weightList []policyv1alpha1.StaticClusterWeight) helper.ClusterWeightInfoList {
list := make(helper.ClusterWeightInfoList, 0)
for _, cluster := range clusters {
var weight int64
for _, staticWeightRule := range weightList {
if util.ClusterMatches(cluster, staticWeightRule.TargetCluster) {
weight += staticWeightRule.Weight
}
}
if weight > 0 {
list = append(list, helper.ClusterWeightInfo{
ClusterName: cluster.Name,
Weight: weight,
})
}
}
if list.GetWeightSum() == 0 {
for _, cluster := range clusters {
list = append(list, helper.ClusterWeightInfo{
ClusterName: cluster.Name,
Weight: 1,
})
}
}
return list
}
// divideReplicasByDynamicWeight assigns a total number of replicas to the selected clusters by the dynamic weight list.
func divideReplicasByDynamicWeight(clusters []*clusterv1alpha1.Cluster, dynamicWeight policyv1alpha1.DynamicWeightFactor, spec *workv1alpha2.ResourceBindingSpec) ([]workv1alpha2.TargetCluster, error) {
switch dynamicWeight {
@ -63,61 +139,6 @@ func divideReplicasByResource(
}
}
// divideReplicasByStaticWeight assigns a total number of replicas to the selected clusters by the weight list.
// For example, we want to assign replicas to two clusters named A and B.
// | Total | Weight(A:B) | Assignment(A:B) |
// | 9 | 1:2 | 3:6 |
// | 9 | 1:3 | 2:7 | Approximate assignment
// Note:
// 1. If any selected cluster which not present on the weight list will be ignored(different with '0' replica).
// 2. In case of not enough replica for specific cluster which will get '0' replica.
func divideReplicasByStaticWeight(clusters []*clusterv1alpha1.Cluster, weightList []policyv1alpha1.StaticClusterWeight,
replicas int32) ([]workv1alpha2.TargetCluster, error) {
weightSum := int64(0)
matchClusters := make(map[string]int64)
desireReplicaInfos := make(map[string]int64)
for _, cluster := range clusters {
for _, staticWeightRule := range weightList {
if util.ClusterMatches(cluster, staticWeightRule.TargetCluster) {
weightSum += staticWeightRule.Weight
matchClusters[cluster.Name] = staticWeightRule.Weight
break
}
}
}
if weightSum == 0 {
for _, cluster := range clusters {
weightSum++
matchClusters[cluster.Name] = 1
}
}
allocatedReplicas := int32(0)
for clusterName, weight := range matchClusters {
desireReplicaInfos[clusterName] = weight * int64(replicas) / weightSum
allocatedReplicas += int32(desireReplicaInfos[clusterName])
}
clusterWeights := helper.SortClusterByWeight(matchClusters)
var clusterNames []string
for _, clusterWeightInfo := range clusterWeights {
clusterNames = append(clusterNames, clusterWeightInfo.ClusterName)
}
divideRemainingReplicas(int(replicas-allocatedReplicas), desireReplicaInfos, clusterNames)
targetClusters := make([]workv1alpha2.TargetCluster, len(desireReplicaInfos))
i := 0
for key, value := range desireReplicaInfos {
targetClusters[i] = workv1alpha2.TargetCluster{Name: key, Replicas: int32(value)}
i++
}
return targetClusters, nil
}
// divideReplicasByPreference assigns a total number of replicas to the selected clusters by preference according to the resource.
func divideReplicasByPreference(
clusterAvailableReplicas []workv1alpha2.TargetCluster,

258
pkg/scheduler/core/division_algorithm_test.go
View File

@ -11,6 +11,7 @@ import (
policyv1alpha1 "github.com/karmada-io/karmada/pkg/apis/policy/v1alpha1"
workv1alpha2 "github.com/karmada-io/karmada/pkg/apis/work/v1alpha2"
"github.com/karmada-io/karmada/pkg/util"
utilhelper "github.com/karmada-io/karmada/pkg/util/helper"
"github.com/karmada-io/karmada/test/helper"
)
@ -21,222 +22,85 @@ const (
ClusterMember4 = "member4"
)
func Test_divideReplicasByStaticWeight(t *testing.T) {
type args struct {
clusters []*clusterv1alpha1.Cluster
weightList []policyv1alpha1.StaticClusterWeight
replicas int32
}
func Test_dispenser_takeByWeight(t *testing.T) {
tests := []struct {
name string
args args
want []workv1alpha2.TargetCluster
wantErr bool
name string
numReplicas int32
result []workv1alpha2.TargetCluster
weightList utilhelper.ClusterWeightInfoList
desired []workv1alpha2.TargetCluster
done bool
}{
{
name: "replica 12, weight 3:2:1",
args: args{
clusters: []*clusterv1alpha1.Cluster{
helper.NewCluster(ClusterMember1),
helper.NewCluster(ClusterMember2),
helper.NewCluster(ClusterMember3),
},
weightList: []policyv1alpha1.StaticClusterWeight{
{
TargetCluster: policyv1alpha1.ClusterAffinity{
ClusterNames: []string{ClusterMember1},
},
Weight: 3,
},
{
TargetCluster: policyv1alpha1.ClusterAffinity{
ClusterNames: []string{ClusterMember2},
},
Weight: 2,
},
{
TargetCluster: policyv1alpha1.ClusterAffinity{
ClusterNames: []string{ClusterMember3},
},
Weight: 1,
},
},
replicas: 12,
name: "Scale up 6 replicas",
numReplicas: 6,
result: []workv1alpha2.TargetCluster{
{Name: "A", Replicas: 1},
{Name: "B", Replicas: 2},
{Name: "C", Replicas: 3},
},
want: []workv1alpha2.TargetCluster{
{
Name: ClusterMember1,
Replicas: 6,
},
{
Name: ClusterMember2,
Replicas: 4,
},
{
Name: ClusterMember3,
Replicas: 2,
},
weightList: []utilhelper.ClusterWeightInfo{
{ClusterName: "A", Weight: 1},
{ClusterName: "B", Weight: 2},
{ClusterName: "C", Weight: 3},
},
wantErr: false,
desired: []workv1alpha2.TargetCluster{
{Name: "A", Replicas: 2},
{Name: "B", Replicas: 4},
{Name: "C", Replicas: 6},
},
done: true,
},
{
name: "replica 12, default weight",
args: struct {
clusters []*clusterv1alpha1.Cluster
weightList []policyv1alpha1.StaticClusterWeight
replicas int32
}{
clusters: []*clusterv1alpha1.Cluster{
helper.NewCluster(ClusterMember1),
helper.NewCluster(ClusterMember2),
helper.NewCluster(ClusterMember3),
},
replicas: 12,
name: "Scale up 3 replicas",
numReplicas: 3,
result: []workv1alpha2.TargetCluster{
{Name: "A", Replicas: 1},
{Name: "B", Replicas: 2},
{Name: "C", Replicas: 3},
},
want: []workv1alpha2.TargetCluster{
{
Name: ClusterMember1,
Replicas: 4,
},
{
Name: ClusterMember2,
Replicas: 4,
},
{
Name: ClusterMember3,
Replicas: 4,
},
weightList: []utilhelper.ClusterWeightInfo{
{ClusterName: "A", Weight: 1},
{ClusterName: "B", Weight: 2},
{ClusterName: "C", Weight: 3},
},
wantErr: false,
desired: []workv1alpha2.TargetCluster{
{Name: "A", Replicas: 1},
{Name: "B", Replicas: 3},
{Name: "C", Replicas: 5},
},
done: true,
},
{
name: "replica 14, weight 3:2:1",
args: struct {
clusters []*clusterv1alpha1.Cluster
weightList []policyv1alpha1.StaticClusterWeight
replicas int32
}{
clusters: []*clusterv1alpha1.Cluster{
helper.NewCluster(ClusterMember1),
helper.NewCluster(ClusterMember2),
helper.NewCluster(ClusterMember3),
},
weightList: []policyv1alpha1.StaticClusterWeight{
{
TargetCluster: policyv1alpha1.ClusterAffinity{
ClusterNames: []string{ClusterMember1},
},
Weight: 3,
},
{
TargetCluster: policyv1alpha1.ClusterAffinity{
ClusterNames: []string{ClusterMember2},
},
Weight: 2,
},
{
TargetCluster: policyv1alpha1.ClusterAffinity{
ClusterNames: []string{ClusterMember3},
},
Weight: 1,
},
},
replicas: 14,
name: "Scale up 2 replicas",
numReplicas: 2,
result: []workv1alpha2.TargetCluster{
{Name: "A", Replicas: 1},
{Name: "B", Replicas: 2},
{Name: "C", Replicas: 3},
},
want: []workv1alpha2.TargetCluster{
{
Name: ClusterMember1,
Replicas: 8,
},
{
Name: ClusterMember2,
Replicas: 4,
},
{
Name: ClusterMember3,
Replicas: 2,
},
weightList: []utilhelper.ClusterWeightInfo{
{ClusterName: "A", Weight: 1},
{ClusterName: "B", Weight: 2},
{ClusterName: "C", Weight: 3},
},
wantErr: false,
},
{
name: "insufficient replica assignment should get 0 replica",
args: struct {
clusters []*clusterv1alpha1.Cluster
weightList []policyv1alpha1.StaticClusterWeight
replicas int32
}{
clusters: []*clusterv1alpha1.Cluster{
helper.NewCluster(ClusterMember1),
helper.NewCluster(ClusterMember2),
},
weightList: []policyv1alpha1.StaticClusterWeight{
{
TargetCluster: policyv1alpha1.ClusterAffinity{
ClusterNames: []string{ClusterMember1},
},
Weight: 1,
},
{
TargetCluster: policyv1alpha1.ClusterAffinity{
ClusterNames: []string{ClusterMember2},
},
Weight: 1,
},
},
replicas: 0,
desired: []workv1alpha2.TargetCluster{
{Name: "A", Replicas: 1},
{Name: "B", Replicas: 2},
{Name: "C", Replicas: 5},
},
want: []workv1alpha2.TargetCluster{
{
Name: ClusterMember1,
Replicas: 0,
},
{
Name: ClusterMember2,
Replicas: 0,
},
},
wantErr: false,
},
{
name: "selected cluster without weight should be ignored",
args: struct {
clusters []*clusterv1alpha1.Cluster
weightList []policyv1alpha1.StaticClusterWeight
replicas int32
}{
clusters: []*clusterv1alpha1.Cluster{
helper.NewCluster(ClusterMember1),
helper.NewCluster(ClusterMember2),
},
weightList: []policyv1alpha1.StaticClusterWeight{
{
TargetCluster: policyv1alpha1.ClusterAffinity{
ClusterNames: []string{ClusterMember1},
},
Weight: 1,
},
},
replicas: 2,
},
want: []workv1alpha2.TargetCluster{
{
Name: ClusterMember1,
Replicas: 2,
},
},
wantErr: false,
done: true,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
got, err := divideReplicasByStaticWeight(tt.args.clusters, tt.args.weightList, tt.args.replicas)
if (err != nil) != tt.wantErr {
t.Errorf("divideReplicasByStaticWeight() error = %v, wantErr %v", err, tt.wantErr)
return
a := newDispenser(tt.numReplicas, tt.result)
a.takeByWeight(tt.weightList)
if a.done() != tt.done {
t.Errorf("expected after takeByWeight: %v, but got: %v", tt.done, a.done())
}
if !helper.IsScheduleResultEqual(got, tt.want) {
t.Errorf("divideReplicasByStaticWeight() got = %v, want %v", got, tt.want)
if !helper.IsScheduleResultEqual(a.result, tt.desired) {
t.Errorf("expected result after takeByWeight: %v, but got: %v", tt.desired, a.result)
}
})
}

9
pkg/util/helper/binding.go
View File

@ -57,6 +57,15 @@ func SortClusterByWeight(m map[string]int64) ClusterWeightInfoList {
return p
}
// GetWeightSum returns the sum of the weight info.
func (p ClusterWeightInfoList) GetWeightSum() int64 {
var res int64
for i := range p {
res += p[i].Weight
}
return res
}
// IsBindingScheduled will check if resourceBinding/clusterResourceBinding is successfully scheduled.
func IsBindingScheduled(status *workv1alpha2.ResourceBindingStatus) bool {
return meta.IsStatusConditionTrue(status.Conditions, workv1alpha2.Scheduled)