Merge pull request #45993 from irfanurrehman/fed-hpa

federation/pkg/federatedtypes/BUILD

@@ -10,11 +10,16 @@ load(
 
 go_test(
     name = "go_default_test",
-    srcs = ["scheduling_test.go"],
+    srcs = [
+        "hpa_test.go",
+        "scheduling_test.go",
+    ],
     library = ":go_default_library",
     tags = ["automanaged"],
     deps = [
+        "//federation/pkg/federation-controller/util/test:go_default_library",
         "//vendor/github.com/stretchr/testify/assert:go_default_library",
+        "//vendor/k8s.io/api/autoscaling/v1:go_default_library",
         "//vendor/k8s.io/api/core/v1:go_default_library",
         "//vendor/k8s.io/api/extensions/v1beta1:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
@@ -29,6 +34,7 @@ go_library(
         "configmap.go",
         "daemonset.go",
         "deployment.go",
+        "hpa.go",
         "namespace.go",
         "qualifiedname.go",
         "registry.go",
@@ -48,12 +54,14 @@ go_library(
         "//pkg/api:go_default_library",
         "//pkg/controller/namespace/deletion:go_default_library",
         "//vendor/github.com/golang/glog:go_default_library",
+        "//vendor/k8s.io/api/autoscaling/v1:go_default_library",
         "//vendor/k8s.io/api/core/v1:go_default_library",
         "//vendor/k8s.io/api/extensions/v1beta1:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/api/meta:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/runtime:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/runtime/schema:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/util/sets:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/watch:go_default_library",
         "//vendor/k8s.io/client-go/dynamic:go_default_library",
         "//vendor/k8s.io/client-go/kubernetes:go_default_library",

federation/pkg/federatedtypes/deployment.go

@@ -40,16 +40,16 @@ func init() {
 }
 
 type DeploymentAdapter struct {
-	*schedulingAdapter
+	*replicaSchedulingAdapter
 	client federationclientset.Interface
 }
 
 func NewDeploymentAdapter(client federationclientset.Interface, config *restclient.Config) FederatedTypeAdapter {
-	schedulingAdapter := schedulingAdapter{
+	schedulingAdapter := replicaSchedulingAdapter{
 		preferencesAnnotationName: FedDeploymentPreferencesAnnotation,
-		updateStatusFunc: func(obj pkgruntime.Object, status interface{}) error {
+		updateStatusFunc: func(obj pkgruntime.Object, schedulingInfo interface{}) error {
 			deployment := obj.(*extensionsv1.Deployment)
-			typedStatus := status.(ReplicaSchedulingStatus)
+			typedStatus := schedulingInfo.(*ReplicaSchedulingInfo).Status
 			if typedStatus.Replicas != deployment.Status.Replicas || typedStatus.UpdatedReplicas != deployment.Status.UpdatedReplicas ||
 				typedStatus.ReadyReplicas != deployment.Status.ReadyReplicas || typedStatus.AvailableReplicas != deployment.Status.AvailableReplicas {
 				deployment.Status = extensionsv1.DeploymentStatus{

federation/pkg/federatedtypes/hpa_test.go

+/*
+Copyright 2017 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.
+*/
+
+package federatedtypes
+
+import (
+	"testing"
+	"time"
+
+	autoscalingv1 "k8s.io/api/autoscaling/v1"
+	apiv1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	pkgruntime "k8s.io/apimachinery/pkg/runtime"
+	. "k8s.io/kubernetes/federation/pkg/federation-controller/util/test"
+
+	"github.com/stretchr/testify/assert"
+)
+
+type replicas struct {
+	min int32
+	max int32
+}
+
+func TestGetHpaScheduleState(t *testing.T) {
+	defaultFedHpa := newHpaWithReplicas(NewInt32(1), NewInt32(70), 10)
+	testCases := map[string]struct {
+		fedHpa           *autoscalingv1.HorizontalPodAutoscaler
+		localHpas        map[string]pkgruntime.Object
+		expectedReplicas map[string]*replicas
+	}{
+		"Distribiutes replicas randomly if no existing hpa in any local cluster": {
+			localHpas: func() map[string]pkgruntime.Object {
+				hpas := make(map[string]pkgruntime.Object)
+				hpas["c1"] = nil
+				hpas["c2"] = nil
+				return hpas
+			}(),
+		},
+		"Cluster with no hpa gets replicas if other clusters have replicas": {
+			localHpas: func() map[string]pkgruntime.Object {
+				hpas := make(map[string]pkgruntime.Object)
+				hpas["c1"] = newHpaWithReplicas(NewInt32(1), NewInt32(70), 10)
+				hpas["c2"] = nil
+				return hpas
+			}(),
+			expectedReplicas: map[string]*replicas{
+				"c1": {
+					min: int32(1),
+					max: int32(9),
+				},
+				"c2": {
+					min: int32(1),
+					max: int32(1),
+				},
+			},
+		},
+		"Cluster needing max replicas gets it if there is another cluster to offer max": {
+			localHpas: func() map[string]pkgruntime.Object {
+				hpa1 := newHpaWithReplicas(NewInt32(1), NewInt32(70), 7)
+				hpa1 = updateHpaStatus(hpa1, NewInt32(50), 5, 5, true)
+				hpa2 := newHpaWithReplicas(NewInt32(1), NewInt32(70), 1)
+				hpa2 = updateHpaStatus(hpa2, NewInt32(70), 1, 1, true)
+				// include third object to ensure, it does not break the test
+				hpa3 := newHpaWithReplicas(NewInt32(1), NewInt32(70), 2)
+				hpa3 = updateHpaStatus(hpa3, NewInt32(70), 1, 1, false)
+				hpas := make(map[string]pkgruntime.Object)
+				hpas["c1"] = hpa1
+				hpas["c2"] = hpa2
+				hpas["c3"] = hpa3
+				return hpas
+			}(),
+			expectedReplicas: map[string]*replicas{
+				"c1": {
+					min: int32(1),
+					max: int32(6),
+				},
+				"c2": {
+					min: int32(1),
+					max: int32(2),
+				},
+				"c3": {
+					min: int32(1),
+					max: int32(2),
+				},
+			},
+		},
+		"Cluster needing max replicas does not get it if there is no cluster offerring max": {
+			localHpas: func() map[string]pkgruntime.Object {
+				hpa1 := newHpaWithReplicas(NewInt32(1), NewInt32(70), 9)
+				hpa1 = updateHpaStatus(hpa1, NewInt32(70), 9, 9, false)
+				hpa2 := newHpaWithReplicas(NewInt32(1), NewInt32(70), 1)
+				hpa2 = updateHpaStatus(hpa2, NewInt32(70), 1, 1, true)
+				hpas := make(map[string]pkgruntime.Object)
+				hpas["c1"] = hpa1
+				hpas["c2"] = hpa2
+				return hpas
+			}(),
+			expectedReplicas: map[string]*replicas{
+				"c1": {
+					min: int32(1),
+					max: int32(9),
+				},
+				"c2": {
+					min: int32(1),
+					max: int32(1),
+				},
+			},
+		},
+		"Cluster which can increase min replicas gets to increase min if there is a cluster offering min": {
+			fedHpa: newHpaWithReplicas(NewInt32(4), NewInt32(70), 10),
+			localHpas: func() map[string]pkgruntime.Object {
+				hpa1 := newHpaWithReplicas(NewInt32(3), NewInt32(70), 6)
+				hpa1 = updateHpaStatus(hpa1, NewInt32(50), 3, 3, true)
+				hpa2 := newHpaWithReplicas(NewInt32(1), NewInt32(70), 4)
+				hpa2 = updateHpaStatus(hpa2, NewInt32(50), 3, 3, true)
+				hpas := make(map[string]pkgruntime.Object)
+				hpas["c1"] = hpa1
+				hpas["c2"] = hpa2
+				return hpas
+			}(),
+			expectedReplicas: map[string]*replicas{
+				"c1": {
+					min: int32(2),
+					max: int32(6),
+				},
+				"c2": {
+					min: int32(2),
+					max: int32(4),
+				},
+			},
+		},
+		"Cluster which can increase min replicas does not increase if there are no clusters offering min": {
+			fedHpa: newHpaWithReplicas(NewInt32(4), NewInt32(70), 10),
+			localHpas: func() map[string]pkgruntime.Object {
+				hpa1 := newHpaWithReplicas(NewInt32(3), NewInt32(70), 6)
+				hpa1 = updateHpaStatus(hpa1, NewInt32(50), 4, 4, true)
+				hpa2 := newHpaWithReplicas(NewInt32(1), NewInt32(70), 4)
+				hpa2 = updateHpaStatus(hpa2, NewInt32(50), 3, 3, true)
+				hpas := make(map[string]pkgruntime.Object)
+				hpas["c1"] = hpa1
+				hpas["c2"] = hpa2
+				return hpas
+			}(),
+			expectedReplicas: map[string]*replicas{
+				"c1": {
+					min: int32(3),
+					max: int32(6),
+				},
+				"c2": {
+					min: int32(1),
+					max: int32(4),
+				},
+			},
+		},
+		"Increasing replicas on fed object increases the same on clusters": {
+			// Existing total of local min, max = 1+1, 5+5 decreasing to below
+			fedHpa: newHpaWithReplicas(NewInt32(4), NewInt32(70), 14),
+			localHpas: func() map[string]pkgruntime.Object {
+				// does not matter if scaleability is true
+				hpas := make(map[string]pkgruntime.Object)
+				hpas["c1"] = newHpaWithReplicas(NewInt32(1), NewInt32(70), 5)
+				hpas["c2"] = newHpaWithReplicas(NewInt32(1), NewInt32(70), 5)
+				return hpas
+			}(),
+			// We dont know which cluster gets how many, but the resultant total should match
+		},
+		"Decreasing replicas on fed object decreases the same on clusters": {
+			// Existing total of local min, max = 2+2, 8+8 decreasing to below
+			fedHpa: newHpaWithReplicas(NewInt32(3), NewInt32(70), 8),
+			localHpas: func() map[string]pkgruntime.Object {
+				// does not matter if scaleability is true
+				hpas := make(map[string]pkgruntime.Object)
+				hpas["c1"] = newHpaWithReplicas(NewInt32(2), NewInt32(70), 8)
+				hpas["c2"] = newHpaWithReplicas(NewInt32(2), NewInt32(70), 8)
+				return hpas
+			}(),
+			// We dont know which cluster gets how many, but the resultant total should match
+		},
+	}
+
+	for testName, testCase := range testCases {
+		t.Run(testName, func(t *testing.T) {
+			if testCase.fedHpa == nil {
+				testCase.fedHpa = defaultFedHpa
+			}
+			scheduledState := getHpaScheduleState(testCase.fedHpa, testCase.localHpas)
+			checkClusterConditions(t, testCase.fedHpa, scheduledState)
+			if testCase.expectedReplicas != nil {
+				for cluster, replicas := range testCase.expectedReplicas {
+					scheduledReplicas := scheduledState[cluster]
+					assert.Equal(t, replicas.min, scheduledReplicas.min)
+					assert.Equal(t, replicas.max, scheduledReplicas.max)
+				}
+			}
+		})
+	}
+}
+
+func updateHpaStatus(hpa *autoscalingv1.HorizontalPodAutoscaler, currentUtilisation *int32, current, desired int32, scaleable bool) *autoscalingv1.HorizontalPodAutoscaler {
+	hpa.Status.CurrentReplicas = current
+	hpa.Status.DesiredReplicas = desired
+	hpa.Status.CurrentCPUUtilizationPercentage = currentUtilisation
+	now := metav1.Now()
+	scaledTime := now
+	if scaleable {
+		// definitely more then 5 minutes ago
+		scaledTime = metav1.NewTime(now.Time.Add(-6 * time.Minute))
+	}
+	hpa.Status.LastScaleTime = &scaledTime
+	return hpa
+}
+
+func checkClusterConditions(t *testing.T, fedHpa *autoscalingv1.HorizontalPodAutoscaler, scheduled map[string]*replicaNums) {
+	minTotal := int32(0)
+	maxTotal := int32(0)
+	for _, replicas := range scheduled {
+		minTotal += replicas.min
+		maxTotal += replicas.max
+	}
+
+	// - Total of max matches the fed max
+	assert.Equal(t, fedHpa.Spec.MaxReplicas, maxTotal)
+	// - Total of min is not less then fed min
+	assert.Condition(t, func() bool {
+		if *fedHpa.Spec.MinReplicas <= minTotal {
+			return true
+		}
+		return false
+	})
+}
+
+func newHpaWithReplicas(min, targetUtilisation *int32, max int32) *autoscalingv1.HorizontalPodAutoscaler {
+	return &autoscalingv1.HorizontalPodAutoscaler{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "myhpa",
+			Namespace: apiv1.NamespaceDefault,
+			SelfLink:  "/api/mylink",
+		},
+		Spec: autoscalingv1.HorizontalPodAutoscalerSpec{
+			ScaleTargetRef: autoscalingv1.CrossVersionObjectReference{
+				Kind: "HorizontalPodAutoscaler",
+				Name: "target-",
+			},
+			MinReplicas:                    min,
+			MaxReplicas:                    max,
+			TargetCPUUtilizationPercentage: targetUtilisation,
+		},
+	}
+}

federation/pkg/federatedtypes/replicaset.go

@@ -40,16 +40,16 @@ func init() {
 }
 
 type ReplicaSetAdapter struct {
-	*schedulingAdapter
+	*replicaSchedulingAdapter
 	client federationclientset.Interface
 }
 
 func NewReplicaSetAdapter(client federationclientset.Interface, config *restclient.Config) FederatedTypeAdapter {
-	schedulingAdapter := schedulingAdapter{
+	replicaSchedulingAdapter := replicaSchedulingAdapter{
 		preferencesAnnotationName: FedReplicaSetPreferencesAnnotation,
-		updateStatusFunc: func(obj pkgruntime.Object, status interface{}) error {
+		updateStatusFunc: func(obj pkgruntime.Object, schedulingInfo interface{}) error {
 			rs := obj.(*extensionsv1.ReplicaSet)
-			typedStatus := status.(ReplicaSchedulingStatus)
+			typedStatus := schedulingInfo.(*ReplicaSchedulingInfo).Status
 			if typedStatus.Replicas != rs.Status.Replicas || typedStatus.FullyLabeledReplicas != rs.Status.FullyLabeledReplicas ||
 				typedStatus.ReadyReplicas != rs.Status.ReadyReplicas || typedStatus.AvailableReplicas != rs.Status.AvailableReplicas {
 				rs.Status = extensionsv1.ReplicaSetStatus{
@@ -64,7 +64,7 @@ func NewReplicaSetAdapter(client federationclientset.Interface, config *restclie
 			return nil
 		},
 	}
-	return &ReplicaSetAdapter{&schedulingAdapter, client}
+	return &ReplicaSetAdapter{&replicaSchedulingAdapter, client}
 }
 
 func (a *ReplicaSetAdapter) Kind() string {

federation/pkg/federatedtypes/scheduling.go

@@ -37,6 +37,16 @@ import (
 	"github.com/golang/glog"
 )
 
+// ScheduleAction is used by the interface ScheduleObject of SchedulingAdapter
+// to sync controller reconcile to convey the action type needed for the
+// particular cluster local object in ScheduleObject
+type ScheduleAction string
+
+const (
+	ActionAdd    = "add"
+	ActionDelete = "delete"
+)
+
 // ReplicaSchedulingStatus contains the status of the replica type objects (rs or deployment)
 // that are being scheduled into joined clusters.
 type ReplicaSchedulingStatus struct {
@@ -58,26 +68,26 @@ type ReplicaSchedulingInfo struct {
 // federated type that requires more complex synchronization logic.
 type SchedulingAdapter interface {
 	GetSchedule(obj pkgruntime.Object, key string, clusters []*federationapi.Cluster, informer fedutil.FederatedInformer) (interface{}, error)
-	ScheduleObject(cluster *federationapi.Cluster, clusterObj pkgruntime.Object, federationObjCopy pkgruntime.Object, schedulingInfo interface{}) (pkgruntime.Object, bool, error)
-	UpdateFederatedStatus(obj pkgruntime.Object, status interface{}) error
+	ScheduleObject(cluster *federationapi.Cluster, clusterObj pkgruntime.Object, federationObjCopy pkgruntime.Object, schedulingInfo interface{}) (pkgruntime.Object, ScheduleAction, error)
+	UpdateFederatedStatus(obj pkgruntime.Object, schedulingInfo interface{}) error
 
 	// EquivalentIgnoringSchedule returns whether obj1 and obj2 are
 	// equivalent ignoring differences due to scheduling.
 	EquivalentIgnoringSchedule(obj1, obj2 pkgruntime.Object) bool
 }
 
-// schedulingAdapter is meant to be embedded in other type adapters that require
-// workload scheduling.
-type schedulingAdapter struct {
+// replicaSchedulingAdapter is meant to be embedded in other type adapters that require
+// workload scheduling with actual pod replicas.
+type replicaSchedulingAdapter struct {
 	preferencesAnnotationName string
 	updateStatusFunc          func(pkgruntime.Object, interface{}) error
 }
 
-func (a *schedulingAdapter) IsSchedulingAdapter() bool {
+func (a *replicaSchedulingAdapter) IsSchedulingAdapter() bool {
 	return true
 }
 
-func (a *schedulingAdapter) GetSchedule(obj pkgruntime.Object, key string, clusters []*federationapi.Cluster, informer fedutil.FederatedInformer) (interface{}, error) {
+func (a *replicaSchedulingAdapter) GetSchedule(obj pkgruntime.Object, key string, clusters []*federationapi.Cluster, informer fedutil.FederatedInformer) (interface{}, error) {
 	var clusterNames []string
 	for _, cluster := range clusters {
 		clusterNames = append(clusterNames, cluster.Name)
@@ -128,7 +138,7 @@ func (a *schedulingAdapter) GetSchedule(obj pkgruntime.Object, key string, clust
 	}, nil
 }
 
-func (a *schedulingAdapter) ScheduleObject(cluster *federationapi.Cluster, clusterObj pkgruntime.Object, federationObjCopy pkgruntime.Object, schedulingInfo interface{}) (pkgruntime.Object, bool, error) {
+func (a *replicaSchedulingAdapter) ScheduleObject(cluster *federationapi.Cluster, clusterObj pkgruntime.Object, federationObjCopy pkgruntime.Object, schedulingInfo interface{}) (pkgruntime.Object, ScheduleAction, error) {
 	typedSchedulingInfo := schedulingInfo.(*ReplicaSchedulingInfo)
 	replicas, ok := typedSchedulingInfo.Schedule[cluster.Name]
 	if !ok {
@@ -152,11 +162,15 @@ func (a *schedulingAdapter) ScheduleObject(cluster *federationapi.Cluster, clust
 			}
 		}
 	}
-	return federationObjCopy, replicas > 0, nil
+	var action ScheduleAction = ""
+	if replicas > 0 {
+		action = ActionAdd
+	}
+	return federationObjCopy, action, nil
 }
 
-func (a *schedulingAdapter) UpdateFederatedStatus(obj pkgruntime.Object, status interface{}) error {
-	return a.updateStatusFunc(obj, status)
+func (a *replicaSchedulingAdapter) UpdateFederatedStatus(obj pkgruntime.Object, schedulingInfo interface{}) error {
+	return a.updateStatusFunc(obj, schedulingInfo)
 }
 
 func schedule(planner *planner.Planner, obj pkgruntime.Object, key string, clusterNames []string, currentReplicasPerCluster map[string]int64, estimatedCapacity map[string]int64) map[string]int64 {

federation/pkg/federation-controller/sync/controller.go

@@ -490,8 +490,7 @@ func syncToClusters(clustersAccessor clustersAccessorFunc, operationsAccessor op
 		if !ok {
 			glog.Fatalf("Adapter for kind %q does not properly implement SchedulingAdapter.", kind)
 		}
-		typedScheduleInfo := schedulingInfo.(*federatedtypes.ReplicaSchedulingInfo)
-		err = schedulingAdapter.UpdateFederatedStatus(obj, typedScheduleInfo.Status)
+		err = schedulingAdapter.UpdateFederatedStatus(obj, schedulingInfo)
 		if err != nil {
 			runtime.HandleError(fmt.Errorf("adapter.UpdateFinished() failed on adapter for %s %q: %v", kind, key, err))
 			return statusError
@@ -548,7 +547,7 @@ func clusterOperations(adapter federatedtypes.FederatedTypeAdapter, selectedClus
 			return nil, wrappedErr
 		}
 
-		shouldCreateIfNeeded := true
+		var scheduleAction federatedtypes.ScheduleAction = federatedtypes.ActionAdd
 		if adapter.IsSchedulingAdapter() {
 			schedulingAdapter, ok := adapter.(federatedtypes.SchedulingAdapter)
 			if !ok {
@@ -559,7 +558,7 @@ func clusterOperations(adapter federatedtypes.FederatedTypeAdapter, selectedClus
 			if clusterObj != nil {
 				clusterTypedObj = clusterObj.(pkgruntime.Object)
 			}
-			desiredObj, shouldCreateIfNeeded, err = schedulingAdapter.ScheduleObject(cluster, clusterTypedObj, desiredObj, schedulingInfo)
+			desiredObj, scheduleAction, err = schedulingAdapter.ScheduleObject(cluster, clusterTypedObj, desiredObj, schedulingInfo)
 			if err != nil {
 				runtime.HandleError(err)
 				return nil, err
@@ -568,11 +567,15 @@ func clusterOperations(adapter federatedtypes.FederatedTypeAdapter, selectedClus
 
 		var operationType util.FederatedOperationType = ""
 		if found {
-			clusterObj := clusterObj.(pkgruntime.Object)
-			if !adapter.Equivalent(desiredObj, clusterObj) {
-				operationType = util.OperationTypeUpdate
+			if scheduleAction == federatedtypes.ActionDelete {
+				operationType = util.OperationTypeDelete
+			} else {
+				clusterObj := clusterObj.(pkgruntime.Object)
+				if !adapter.Equivalent(desiredObj, clusterObj) {
+					operationType = util.OperationTypeUpdate
+				}
 			}
-		} else if shouldCreateIfNeeded {
+		} else if scheduleAction == federatedtypes.ActionAdd {
 			operationType = util.OperationTypeAdd
 		}
 

federation/pkg/federation-controller/util/test/test_helper.go

@@ -446,3 +446,9 @@ func AssertHasFinalizer(t *testing.T, obj runtime.Object, finalizer string) {
 	require.Nil(t, err)
 	assert.True(t, hasFinalizer)
 }
+
+func NewInt32(val int32) *int32 {
+	p := new(int32)
+	*p = val
+	return p
+}