Merge pull request kubewharf#265 from Poor12/support-custom-migration

config/sample/host/01-ftc.yaml

@@ -9,6 +9,8 @@ spec:
     pluralName: namespaces
     scope: Cluster
     version: v1
+  statusAggregation:
+    enabled: false
   controllers:
     - - kubeadmiral.io/nsautoprop-controller
     - - kubeadmiral.io/overridepolicy-controller
@@ -42,8 +44,6 @@ spec:
     pluralName: deployments
     scope: Namespaced
     version: v1
-  statusAggregation:
-    enabled: true
   autoMigration:
     enabled: true
   controllers:
@@ -297,8 +297,6 @@ spec:
     pluralName: jobs
     scope: Namespaced
     version: v1
-  statusAggregation:
-    enabled: true
   controllers:
     - - kubeadmiral.io/global-scheduler
     - - kubeadmiral.io/overridepolicy-controller

pkg/apis/core/v1alpha1/extensions_schedulingprofile.go

@@ -29,6 +29,7 @@ func GetDefaultEnabledPlugins() *fedcore.EnabledPlugins {
 		names.ClusterAffinity,
 		names.ClusterReady,
 		names.ClusterTerminating,
+		names.ClusterEvicted,
 	}
 
 	scorePlugins := []string{

pkg/apis/core/v1alpha1/types_federatedtypeconfig.go

@@ -53,8 +53,9 @@ type FederatedTypeConfigSpec struct {
 	// The API resource type to be federated.
 	SourceType APIResource `json:"sourceType"`
 
-	// Configuration for StatusAggregation. If left empty, the StatusAggregation feature will be disabled.
+	// Configuration for StatusAggregation.
 	// +optional
+	// +kubebuilder:default:={enabled:true}
 	StatusAggregation *StatusAggregationConfig `json:"statusAggregation,omitempty"`
 	// Configuration for StatusCollection. If left empty, the StatusCollection feature will be disabled.
 	// +optional
@@ -115,6 +116,7 @@ type StatusCollectionConfig struct {
 // StatusAggregationConfig defines the configurations for the StatusAggregation feature.
 type StatusAggregationConfig struct {
 	// Whether or not to enable status aggregation.
+	// +kubebuilder:default:=true
 	Enabled bool `json:"enabled"`
 }
 

pkg/controllers/common/constants.go

@@ -116,6 +116,11 @@ const (
 	TemplateGeneratorMergePatchAnnotation = FederateControllerPrefix + "template-generator-merge-patch"
 
 	LatestReplicasetDigestsAnnotation = DefaultPrefix + "latest-replicaset-digests"
+
+	// MigrationConfigurationAnnotation contains custom migration configuration from users.
+	MigrationConfigurationAnnotation = DefaultPrefix + "migration-configuration"
+	// AppliedMigrationConfigurationAnnotation contains the applied custom migration configuration.
+	AppliedMigrationConfigurationAnnotation = DefaultPrefix + "applied-migration-configuration"
 )
 
 // PropagatedAnnotationKeys and PropagatedLabelKeys are used to store the keys of annotations and labels that are present

pkg/controllers/federate/util.go

@@ -265,11 +265,13 @@ var (
 		scheduler.FollowsObjectAnnotation,
 		common.FollowersAnnotation,
 		common.DisableFollowingAnnotation,
+		common.MigrationConfigurationAnnotation,
 	)
 
 	// List of annotations that should be ignored on the source object
 	ignoredAnnotationSet = sets.New(
 		common.LatestReplicasetDigestsAnnotation,
+		common.AppliedMigrationConfigurationAnnotation,
 	)
 
 	federatedLabelSet = sets.New[string](

pkg/controllers/scheduler/constants.go

@@ -41,10 +41,11 @@ const (
 
 	DefaultSchedulingMode = fedcorev1a1.SchedulingModeDuplicate
 
-	EventReasonScheduleFederatedObject   = "ScheduleFederatedObject"
-	EventReasonInvalidFollowsObject      = "InvalidFollowsObject"
-	EventReasonWebhookConfigurationError = "WebhookConfigurationError"
-	EventReasonWebhookRegistered         = "WebhookRegistered"
+	EventReasonScheduleFederatedObject     = "ScheduleFederatedObject"
+	EventReasonInvalidFollowsObject        = "InvalidFollowsObject"
+	EventReasonWebhookConfigurationError   = "WebhookConfigurationError"
+	EventReasonWebhookRegistered           = "WebhookRegistered"
+	EventReasonParseMigrationConfiguration = "ParseMigrationConfiguration"
 
 	SchedulingTriggersAnnotation        = common.DefaultPrefix + "scheduling-triggers"
 	SchedulingDeferredReasonsAnnotation = common.DefaultPrefix + "scheduling-deferred-reasons"

pkg/controllers/scheduler/framework/plugins/clusterevicted/cluster_evicted.go

@@ -0,0 +1,65 @@
+/*
+Copyright 2023 The KubeAdmiral Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package clusterevicted
+
+import (
+	"context"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	fedcorev1a1 "github.com/kubewharf/kubeadmiral/pkg/apis/core/v1alpha1"
+	"github.com/kubewharf/kubeadmiral/pkg/controllers/scheduler/framework"
+	"github.com/kubewharf/kubeadmiral/pkg/controllers/scheduler/framework/plugins/names"
+)
+
+const (
+	// ErrReason for cluster evicted not matching.
+	ErrReason = "cluster(s) were evicted due to custom migration configuration"
+)
+
+type ClusterEvicted struct{}
+
+func NewClusterEvicted(_ framework.Handle) (framework.Plugin, error) {
+	return &ClusterEvicted{}, nil
+}
+
+func (pl *ClusterEvicted) Name() string {
+	return names.ClusterEvicted
+}
+
+// Filter invoked at the filter extension point.
+func (pl *ClusterEvicted) Filter(
+	ctx context.Context,
+	su *framework.SchedulingUnit,
+	cluster *fedcorev1a1.FederatedCluster,
+) *framework.Result {
+	err := framework.PreCheck(ctx, su, cluster)
+	if err != nil {
+		return framework.NewResult(framework.Error, err.Error())
+	}
+
+	if su.CustomMigration.Info != nil {
+		unavailableClusters := su.CustomMigration.Info.UnavailableClusters
+		for _, unavailableCluster := range unavailableClusters {
+			if cluster.Name == unavailableCluster.Cluster && metav1.Now().Time.Before(unavailableCluster.ValidUntil.Time) {
+				return framework.NewResult(framework.Unschedulable, ErrReason)
+			}
+		}
+	}
+
+	return framework.NewResult(framework.Success)
+}

pkg/controllers/scheduler/framework/plugins/clusterevicted/cluster_evicted_test.go

@@ -0,0 +1,153 @@
+/*
+Copyright 2023 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+
+This file may have been modified by The KubeAdmiral Authors
+("KubeAdmiral Modifications"). All KubeAdmiral Modifications
+are Copyright 2023 The KubeAdmiral Authors.
+*/
+
+package clusterevicted
+
+import (
+	"context"
+	"reflect"
+	"testing"
+	"time"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	fedcorev1a1 "github.com/kubewharf/kubeadmiral/pkg/apis/core/v1alpha1"
+	"github.com/kubewharf/kubeadmiral/pkg/controllers/scheduler/framework"
+	"github.com/kubewharf/kubeadmiral/pkg/controllers/scheduler/framework/plugins/names"
+)
+
+func makeCluster(clusterName string) *fedcorev1a1.FederatedCluster {
+	return &fedcorev1a1.FederatedCluster{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: clusterName,
+		},
+	}
+}
+
+func TestName(t *testing.T) {
+	p, _ := NewClusterEvicted(nil)
+	if name := p.Name(); !reflect.DeepEqual(name, names.ClusterEvicted) {
+		t.Errorf("Expected name %s, but got %s", names.ClusterEvicted, name)
+	}
+}
+
+func TestClusterEvictedPlugin(t *testing.T) {
+	tests := []struct {
+		name           string
+		su             *framework.SchedulingUnit
+		cluster        *fedcorev1a1.FederatedCluster
+		expectedResult *framework.Result
+	}{
+		{
+			name:           "su is nil",
+			su:             nil,
+			cluster:        makeCluster("cluster1"),
+			expectedResult: framework.NewResult(framework.Error),
+		},
+		{
+			"cluster should not be filtered when custom migration info is empty",
+			&framework.SchedulingUnit{
+				CustomMigration: framework.CustomMigrationSpec{
+					Info: nil,
+				},
+			},
+			makeCluster("cluster1"),
+			framework.NewResult(framework.Success),
+		},
+		{
+			"cluster should not be filtered when name not in custom migration info",
+			&framework.SchedulingUnit{
+				CustomMigration: framework.CustomMigrationSpec{
+					Info: &framework.CustomMigrationInfo{
+						UnavailableClusters: []framework.UnavailableCluster{
+							{
+								Cluster: "cluster2",
+							},
+						},
+					},
+				},
+			},
+			makeCluster("cluster1"),
+			framework.NewResult(framework.Success),
+		},
+		{
+			"cluster should not be filtered when replica migration in custom migration info",
+			&framework.SchedulingUnit{
+				CustomMigration: framework.CustomMigrationSpec{
+					Info: &framework.CustomMigrationInfo{
+						LimitedCapacity: map[string]int64{
+							"cluster1": 1,
+						},
+					},
+				},
+			},
+			makeCluster("cluster1"),
+			framework.NewResult(framework.Success),
+		},
+		{
+			name: "cluster should not be filtered when time has expired",
+			su: &framework.SchedulingUnit{
+				CustomMigration: framework.CustomMigrationSpec{
+					Info: &framework.CustomMigrationInfo{
+						UnavailableClusters: []framework.UnavailableCluster{
+							{
+								Cluster: "cluster1",
+								ValidUntil: metav1.Time{
+									Time: time.Now().Add(-1 * time.Hour),
+								},
+							},
+						},
+					},
+				},
+			},
+			cluster:        makeCluster("cluster1"),
+			expectedResult: framework.NewResult(framework.Success),
+		},
+		{
+			name: "cluster should be filtered",
+			su: &framework.SchedulingUnit{
+				CustomMigration: framework.CustomMigrationSpec{
+					Info: &framework.CustomMigrationInfo{
+						UnavailableClusters: []framework.UnavailableCluster{
+							{
+								Cluster: "cluster1",
+								ValidUntil: metav1.Time{
+									Time: time.Now().Add(1 * time.Hour),
+								},
+							},
+						},
+					},
+				},
+			},
+			cluster:        makeCluster("cluster1"),
+			expectedResult: framework.NewResult(framework.Unschedulable),
+		},
+	}
+
+	p, _ := NewClusterEvicted(nil)
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			result := p.(framework.FilterPlugin).Filter(context.TODO(), test.su, test.cluster)
+			if result.IsSuccess() != test.expectedResult.IsSuccess() {
+				t.Errorf("result does not match: %v, want %v", result, test.expectedResult)
+			}
+		})
+	}
+}

pkg/controllers/scheduler/framework/plugins/names/names.go

@@ -29,4 +29,5 @@ const (
 	ClusterResourcesMostAllocated      = "ClusterResourcesMostAllocated"
 	MaxCluster                         = "MaxCluster"
 	ClusterCapacityWeight              = "ClusterCapacityWeight"
+	ClusterEvicted                     = "ClusterEvicted"
 )

pkg/controllers/scheduler/framework/plugins/rsp/rsp.go

@@ -153,6 +153,11 @@ func (pl *ClusterCapacityWeight) ReplicaScheduling(
 		}
 	}
 
+	limitedCapacity := map[string]int64{}
+	if su.CustomMigration.Info != nil && su.CustomMigration.Info.LimitedCapacity != nil {
+		limitedCapacity = su.CustomMigration.Info.LimitedCapacity
+	}
+
 	scheduleResult, overflow, err := planner.Plan(
 		&planner.ReplicaSchedulingPreference{
 			Clusters: clusterPreferences,
@@ -161,6 +166,7 @@ func (pl *ClusterCapacityWeight) ReplicaScheduling(
 		ExtractClusterNames(clusters),
 		currentReplicas,
 		estimatedCapacity,
+		limitedCapacity,
 		su.Key(),
 		su.AvoidDisruption,
 		keepUnschedulableReplicas,