Repository: clusternet/clusternet
Pull request #383 (2022-06-16 08:26:30)

add scheduling loop admission
Signed-off-by: Garrybest <garrybest@foxmail.com>

<!--  Thanks for sending a pull request!

-->

#### What type of PR is this?
kind/enhancement

#### What this PR does / why we need it:

#### Which issue(s) this PR fixes:
<!--
*Automatically closes linked issue when PR is merged.
Usage: `Fixes #<issue number>`, or `Fixes (paste link of issue)`.
-->
Fixes #

#### Special notes for your reviewer:


codecov-commenter: # [Codecov](https://codecov.io/gh/clusternet/clusternet/pull/383?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clusternet) Report
> Merging [#383](https://codecov.io/gh/clusternet/clusternet/pull/383?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clusternet) (98b4f59) into [main](https://codecov.io/gh/clusternet/clusternet/commit/122bc211e8ddb82bf419be112c7741ae46e50df7?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clusternet) (122bc21) will **not change** coverage.
> The diff coverage is `n/a`.

```diff
@@           Coverage Diff           @@
##             main     #383   +/-   ##
=======================================
  Coverage   12.92%   12.92%           
=======================================
  Files          53       53           
  Lines        5331     5331           
=======================================
  Hits          689      689           
  Misses       4594     4594           
  Partials       48       48           
```


| [Impacted Files](https://codecov.io/gh/clusternet/clusternet/pull/383?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clusternet) | Coverage Δ | |
|---|---|---|
| [...ithub.com/clusternet/clusternet/pkg/utils/slice.go](https://codecov.io/gh/clusternet/clusternet/pull/383/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clusternet#diff-c3JjL2dpdGh1Yi5jb20vY2x1c3Rlcm5ldC9jbHVzdGVybmV0L3BrZy91dGlscy9zbGljZS5nbw==) | `57.62% <0.00%> (-5.34%)` | :arrow_down: |
| [...ler/framework/plugins/defaultassigner/algorithm.go](https://codecov.io/gh/clusternet/clusternet/pull/383/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clusternet#diff-c3JjL2dpdGh1Yi5jb20vY2x1c3Rlcm5ldC9jbHVzdGVybmV0L3BrZy9zY2hlZHVsZXIvZnJhbWV3b3JrL3BsdWdpbnMvZGVmYXVsdGFzc2lnbmVyL2FsZ29yaXRobS5nbw==) | `57.73% <0.00%> (+2.82%)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/clusternet/clusternet/pull/383?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clusternet).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clusternet)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/clusternet/clusternet/pull/383?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clusternet). Last update [122bc21...98b4f59](https://codecov.io/gh/clusternet/clusternet/pull/383?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clusternet). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clusternet).


Patch:
diff --git a/pkg/scheduler/framework/plugins/defaultassigner/algorithm.go b/pkg/scheduler/framework/plugins/defaultassigner/algorithm.go
index 3bbb829..ebdc4f3 100644
--- a/pkg/scheduler/framework/plugins/defaultassigner/algorithm.go
+++ b/pkg/scheduler/framework/plugins/defaultassigner/algorithm.go
@@ -43,7 +43,7 @@ func DynamicDivideReplicas(selected *framework.TargetClusters, sub *appsapi.Subs
 // dynamicDivideReplicas divides replicas by the MaxAvailableReplicas
 func dymamicDivideReplicas(desiredReplicas int32, maxAvailableReplicas []int32) []int32 {
 	res := make([]int32, len(maxAvailableReplicas))
-	sumAvailableReplicas := sumArray(maxAvailableReplicas)
+	sumAvailableReplicas := utils.SumArrayInt32(maxAvailableReplicas)
 
 	if desiredReplicas > sumAvailableReplicas {
 		return maxAvailableReplicas
@@ -54,7 +54,7 @@ func dymamicDivideReplicas(desiredReplicas int32, maxAvailableReplicas []int32)
 	}
 
 	// Even out the rest replicas among all candidate clusters.
-	rest := desiredReplicas - sumArray(res)
+	rest := desiredReplicas - utils.SumArrayInt32(res)
 	if rest > 0 {
 		for i := 0; i < int(rest) && i < len(res); i++ {
 			res[i]++
@@ -64,13 +64,6 @@ func dymamicDivideReplicas(desiredReplicas int32, maxAvailableReplicas []int32)
 	return res
 }
 
-func sumArray(array []int32) (sum int32) {
-	for _, v := range array {
-		sum += v
-	}
-	return
-}
-
 // StaticDivideReplicas will fill the target replicas of all feeds based on a weight list and feed finv.
 func StaticDivideReplicas(selected *framework.TargetClusters, sub *appsapi.Subscription, clusters []*clusterapi.ManagedCluster, finv *appsapi.FeedInventory) error {
 	wl := getWeightList(sub.Spec.Subscribers, clusters)
diff --git a/pkg/scheduler/scheduler.go b/pkg/scheduler/scheduler.go
index c454926..7d09430 100644
--- a/pkg/scheduler/scheduler.go
+++ b/pkg/scheduler/scheduler.go
@@ -256,6 +256,10 @@ func (sched *Scheduler) scheduleOne(ctx context.Context) {
 		}
 	}
 
+	if !admit(sub, finv) {
+		return
+	}
+
 	// Synchronously attempt to find a fit for the subscription.
 	start := time.Now()
 
@@ -520,3 +524,38 @@ func truncateMessage(message string) string {
 	suffix := " ..."
 	return message[:max-len(suffix)] + suffix
 }
+
+func admit(sub *appsapi.Subscription, finv *appsapi.FeedInventory) bool {
+	// always schedule replication subscription
+	if sub.Spec.SchedulingStrategy == appsapi.ReplicaSchedulingStrategyType {
+		return true
+	}
+	specHashChanged := utils.HashSubscriptionSpec(&sub.Spec) == sub.Status.SpecHash
+	feedChanged := isFeedChanged(sub, finv)
+	return specHashChanged || feedChanged
+}
+
+func isFeedChanged(sub *appsapi.Subscription, finv *appsapi.FeedInventory) bool {
+	feeds := sub.Spec.Feeds
+	replicas := sub.Status.Replicas
+	if len(feeds) != len(replicas) {
+		return true
+	}
+	for i := range feeds {
+		if _, exist := replicas[utils.GetFeedKey(feeds[i])]; !exist {
+			return true
+		}
+	}
+
+	feedOrders := finv.Spec.Feeds
+	for i := range feedOrders {
+		var desired int32
+		if feedOrders[i].DesiredReplicas != nil {
+			desired = *feedOrders[i].DesiredReplicas
+		}
+		if utils.SumArrayInt32(replicas[utils.GetFeedKey(feedOrders[i].Feed)]) != desired {
+			return true
+		}
+	}
+	return false
+}
diff --git a/pkg/utils/slice.go b/pkg/utils/slice.go
index dd839ca..40dc5ea 100644
--- a/pkg/utils/slice.go
+++ b/pkg/utils/slice.go
@@ -107,3 +107,10 @@ func MinInt32(a, b int32) int32 {
 	}
 	return b
 }
+
+func SumArrayInt32(array []int32) (sum int32) {
+	for _, v := range array {
+		sum += v
+	}
+	return
+}


Test patch:
diff --git a/pkg/scheduler/scheduler_test.go b/pkg/scheduler/scheduler_test.go
new file mode 100644
index 0000000..a5e9d20
--- /dev/null
+++ b/pkg/scheduler/scheduler_test.go
@@ -0,0 +1,178 @@
+package scheduler
+
+import (
+	"testing"
+
+	"k8s.io/utils/pointer"
+
+	appsapi "github.com/clusternet/clusternet/pkg/apis/apps/v1alpha1"
+	"github.com/clusternet/clusternet/pkg/utils"
+)
+
+func Test_isFeedChanged(t *testing.T) {
+	feeds := []appsapi.Feed{
+		{
+			Kind:       "Deployment",
+			APIVersion: "apps/v1",
+			Namespace:  "default",
+			Name:       "deploy",
+		},
+		{
+			Kind:       "StatefulSet",
+			APIVersion: "apps/v1",
+			Namespace:  "default",
+			Name:       "sts",
+		},
+		{
+			Kind:       "ConfigMap",
+			APIVersion: "v1",
+			Namespace:  "default",
+			Name:       "cm",
+		},
+	}
+	type args struct {
+		replicas   map[string][]int32
+		feedOrders []appsapi.FeedOrder
+	}
+	tests := []struct {
+		name string
+		args args
+		want bool
+	}{
+		{
+			name: "same replicas",
+			args: args{
+				replicas: map[string][]int32{
+					utils.GetFeedKey(feeds[0]): {1, 2},
+					utils.GetFeedKey(feeds[1]): {1, 1},
+					utils.GetFeedKey(feeds[2]): {},
+				},
+				feedOrders: []appsapi.FeedOrder{
+					{
+						Feed:            feeds[0],
+						DesiredReplicas: pointer.Int32(3),
+					},
+					{
+						Feed:            feeds[1],
+						DesiredReplicas: pointer.Int32(2),
+					},
+					{
+						Feed:            feeds[2],
+						DesiredReplicas: nil,
+					},
+				},
+			},
+			want: false,
+		},
+		{
+			name: "scale up",
+			args: args{
+				replicas: map[string][]int32{
+					utils.GetFeedKey(feeds[0]): {1, 2},
+					utils.GetFeedKey(feeds[1]): {1, 1},
+					utils.GetFeedKey(feeds[2]): {},
+				},
+				feedOrders: []appsapi.FeedOrder{
+					{
+						Feed:            feeds[0],
+						DesiredReplicas: pointer.Int32(5),
+					},
+					{
+						Feed:            feeds[1],
+						DesiredReplicas: pointer.Int32(2),
+					},
+					{
+						Feed:            feeds[2],
+						DesiredReplicas: nil,
+					},
+				},
+			},
+			want: true,
+		},
+		{
+			name: "scale down",
+			args: args{
+				replicas: map[string][]int32{
+					utils.GetFeedKey(feeds[0]): {1, 2},
+					utils.GetFeedKey(feeds[1]): {1, 1},
+					utils.GetFeedKey(feeds[2]): {},
+				},
+				feedOrders: []appsapi.FeedOrder{
+					{
+						Feed:            feeds[0],
+						DesiredReplicas: pointer.Int32(2),
+					},
+					{
+						Feed:            feeds[1],
+						DesiredReplicas: pointer.Int32(1),
+					},
+					{
+						Feed:            feeds[2],
+						DesiredReplicas: nil,
+					},
+				},
+			},
+			want: true,
+		},
+		{
+			name: "new feed",
+			args: args{
+				replicas: map[string][]int32{
+					utils.GetFeedKey(feeds[0]): {1, 2},
+					utils.GetFeedKey(feeds[2]): {},
+				},
+				feedOrders: []appsapi.FeedOrder{
+					{
+						Feed:            feeds[0],
+						DesiredReplicas: pointer.Int32(3),
+					},
+					{
+						Feed:            feeds[1],
+						DesiredReplicas: pointer.Int32(1),
+					},
+					{
+						Feed:            feeds[2],
+						DesiredReplicas: nil,
+					},
+				},
+			},
+			want: true,
+		},
+		{
+			name: "delete feed",
+			args: args{
+				replicas: map[string][]int32{
+					utils.GetFeedKey(feeds[0]): {1, 2},
+					utils.GetFeedKey(feeds[1]): {1, 1},
+				},
+				feedOrders: []appsapi.FeedOrder{
+					{
+						Feed:            feeds[0],
+						DesiredReplicas: pointer.Int32(2),
+					},
+					{
+						Feed:            feeds[1],
+						DesiredReplicas: pointer.Int32(2),
+					},
+					{
+						Feed:            feeds[2],
+						DesiredReplicas: nil,
+					},
+				},
+			},
+			want: true,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := isFeedChanged(&appsapi.Subscription{
+				Spec:   appsapi.SubscriptionSpec{Feeds: feeds},
+				Status: appsapi.SubscriptionStatus{Replicas: tt.args.replicas},
+			}, &appsapi.FeedInventory{
+				Spec: appsapi.FeedInventorySpec{Feeds: tt.args.feedOrders},
+			}); got != tt.want {
+				t.Errorf("isFeedChanged() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}