add federation replicaset controller

federation/cmd/federation-controller-manager/app/controllermanager.go

@@ -31,6 +31,7 @@ import (
 	"k8s.io/kubernetes/federation/pkg/dnsprovider"
 	clustercontroller "k8s.io/kubernetes/federation/pkg/federation-controller/cluster"
 	namespacecontroller "k8s.io/kubernetes/federation/pkg/federation-controller/namespace"
+	replicasetcontroller "k8s.io/kubernetes/federation/pkg/federation-controller/replicaset"
 	secretcontroller "k8s.io/kubernetes/federation/pkg/federation-controller/secret"
 	servicecontroller "k8s.io/kubernetes/federation/pkg/federation-controller/service"
 	"k8s.io/kubernetes/federation/pkg/federation-controller/util"
@@ -155,5 +156,9 @@ func StartControllers(s *options.CMServer, restClientCfg *restclient.Config) err
 	secretcontroller := secretcontroller.NewSecretController(secretcontrollerClientset)
 	secretcontroller.Run(wait.NeverStop)
 
+	replicaSetClientset := federationclientset.NewForConfigOrDie(restclient.AddUserAgent(restClientCfg, replicasetcontroller.UserAgentName))
+	replicaSetController := replicasetcontroller.NewReplicaSetController(replicaSetClientset)
+	go replicaSetController.Run(s.ConcurrentReplicaSetSyncs, wait.NeverStop)
+
 	select {}
 }

federation/cmd/federation-controller-manager/app/options/options.go

@@ -47,6 +47,10 @@ type ControllerManagerConfiguration struct {
 	// allowed to sync concurrently. Larger number = more responsive service
 	// management, but more CPU (and network) load.
 	ConcurrentServiceSyncs int `json:"concurrentServiceSyncs"`
+	// concurrentReplicaSetSyncs is the number of ReplicaSets that are
+	// allowed to sync concurrently. Larger number = more responsive service
+	// management, but more CPU (and network) load.
+	ConcurrentReplicaSetSyncs int `json:"concurrentReplicaSetSyncs"`
 	// clusterMonitorPeriod is the period for syncing ClusterStatus in cluster controller.
 	ClusterMonitorPeriod unversioned.Duration `json:"clusterMonitorPeriod"`
 	// APIServerQPS is the QPS to use while talking with federation apiserver.
@@ -78,13 +82,14 @@ const (
 func NewCMServer() *CMServer {
 	s := CMServer{
 		ControllerManagerConfiguration: ControllerManagerConfiguration{
-			Port:                   FederatedControllerManagerPort,
-			Address:                "0.0.0.0",
-			ConcurrentServiceSyncs: 10,
-			ClusterMonitorPeriod:   unversioned.Duration{Duration: 40 * time.Second},
-			APIServerQPS:           20.0,
-			APIServerBurst:         30,
-			LeaderElection:         leaderelection.DefaultLeaderElectionConfiguration(),
+			Port:                      FederatedControllerManagerPort,
+			Address:                   "0.0.0.0",
+			ConcurrentServiceSyncs:    10,
+			ConcurrentReplicaSetSyncs: 10,
+			ClusterMonitorPeriod:      unversioned.Duration{Duration: 40 * time.Second},
+			APIServerQPS:              20.0,
+			APIServerBurst:            30,
+			LeaderElection:            leaderelection.DefaultLeaderElectionConfiguration(),
 		},
 	}
 	return &s
@@ -97,6 +102,7 @@ func (s *CMServer) AddFlags(fs *pflag.FlagSet) {
 	fs.StringVar(&s.FederationName, "federation-name", s.FederationName, "Federation name.")
 	fs.StringVar(&s.ZoneName, "zone-name", s.ZoneName, "Zone name, like example.com.")
 	fs.IntVar(&s.ConcurrentServiceSyncs, "concurrent-service-syncs", s.ConcurrentServiceSyncs, "The number of service syncing operations that will be done concurrently. Larger number = faster endpoint updating, but more CPU (and network) load")
+	fs.IntVar(&s.ConcurrentReplicaSetSyncs, "concurrent-replicaset-syncs", s.ConcurrentReplicaSetSyncs, "The number of ReplicaSets syncing operations that will be done concurrently. Larger number = faster endpoint updating, but more CPU (and network) load")
 	fs.DurationVar(&s.ClusterMonitorPeriod.Duration, "cluster-monitor-period", s.ClusterMonitorPeriod.Duration, "The period for syncing ClusterStatus in ClusterController.")
 	fs.BoolVar(&s.EnableProfiling, "profiling", true, "Enable profiling via web interface host:port/debug/pprof/")
 	fs.StringVar(&s.Master, "master", s.Master, "The address of the federation API server (overrides any value in kubeconfig)")

federation/pkg/federation-controller/replicaset/replicasetcontroller_test.go

+/*
+Copyright 2016 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 replicaset
+
+import (
+	"fmt"
+	"github.com/stretchr/testify/assert"
+	fedv1 "k8s.io/kubernetes/federation/apis/federation/v1beta1"
+	fedclientfake "k8s.io/kubernetes/federation/client/clientset_generated/federation_release_1_4/fake"
+	apiv1 "k8s.io/kubernetes/pkg/api/v1"
+	extensionsv1 "k8s.io/kubernetes/pkg/apis/extensions/v1beta1"
+	"testing"
+	"time"
+	//kubeclientfake "k8s.io/kubernetes/pkg/client/clientset_generated/release_1_4/fake"
+	"k8s.io/kubernetes/federation/client/clientset_generated/federation_release_1_4"
+	fedutil "k8s.io/kubernetes/federation/pkg/federation-controller/util"
+	"k8s.io/kubernetes/pkg/client/testing/core"
+	"k8s.io/kubernetes/pkg/watch"
+)
+
+func TestReplicaSetController(t *testing.T) {
+
+	replicaSetReviewDelay = 10 * time.Millisecond
+	clusterAvailableDelay = 20 * time.Millisecond
+	clusterUnavailableDelay = 60 * time.Millisecond
+
+	fedclientset := fedclientfake.NewSimpleClientset()
+	fedrswatch := watch.NewFake()
+	fedclientset.PrependWatchReactor("replicasets", core.DefaultWatchReactor(fedrswatch, nil))
+
+	fedclientset.Federation().Clusters().Create(mkCluster("k8s-1", apiv1.ConditionTrue))
+	fedclientset.Federation().Clusters().Create(mkCluster("k8s-2", apiv1.ConditionTrue))
+
+	kube1clientset := fedclientfake.NewSimpleClientset()
+	kube1rswatch := watch.NewFake()
+	kube1clientset.PrependWatchReactor("replicasets", core.DefaultWatchReactor(kube1rswatch, nil))
+	kube2clientset := fedclientfake.NewSimpleClientset()
+	kube2rswatch := watch.NewFake()
+	kube2clientset.PrependWatchReactor("replicasets", core.DefaultWatchReactor(kube2rswatch, nil))
+
+	stopChan := make(chan struct{})
+	replicaSetController := NewReplicaSetController(fedclientset)
+	informer := toFederatedInformerForTestOnly(replicaSetController.fedInformer)
+	informer.SetClientFactory(func(cluster *fedv1.Cluster) (federation_release_1_4.Interface, error) {
+		switch cluster.Name {
+		case "k8s-1":
+			return kube1clientset, nil
+		case "k8s-2":
+			return kube2clientset, nil
+		default:
+			return nil, fmt.Errorf("Unknown cluster: %v", cluster.Name)
+		}
+	})
+
+	go replicaSetController.Run(1, stopChan)
+
+	rs := mkReplicaSet("rs", 9)
+	rs, _ = fedclientset.Extensions().ReplicaSets(apiv1.NamespaceDefault).Create(rs)
+	fedrswatch.Add(rs)
+	time.Sleep(1 * time.Second)
+
+	rs1, _ := kube1clientset.Extensions().ReplicaSets(apiv1.NamespaceDefault).Get(rs.Name)
+	rs1.Status.Replicas = *rs1.Spec.Replicas
+	rs1, _ = kube1clientset.Extensions().ReplicaSets(apiv1.NamespaceDefault).UpdateStatus(rs1)
+	kube1rswatch.Modify(rs1)
+
+	rs2, _ := kube2clientset.Extensions().ReplicaSets(apiv1.NamespaceDefault).Get(rs.Name)
+	rs2.Status.Replicas = *rs2.Spec.Replicas
+	rs2, _ = kube2clientset.Extensions().ReplicaSets(apiv1.NamespaceDefault).UpdateStatus(rs2)
+	kube2rswatch.Modify(rs2)
+
+	time.Sleep(1 * time.Second)
+	rs, _ = fedclientset.Extensions().ReplicaSets(apiv1.NamespaceDefault).Get(rs.Name)
+	assert.Equal(t, *rs.Spec.Replicas, *rs1.Spec.Replicas+*rs2.Spec.Replicas)
+	assert.Equal(t, rs.Status.Replicas, rs1.Status.Replicas+rs2.Status.Replicas)
+
+	var replicas int32 = 20
+	rs.Spec.Replicas = &replicas
+	rs, _ = fedclientset.Extensions().ReplicaSets(apiv1.NamespaceDefault).Update(rs)
+	fedrswatch.Modify(rs)
+
+	time.Sleep(2 * time.Second)
+	rs, _ = fedclientset.Extensions().ReplicaSets(apiv1.NamespaceDefault).Get(rs.Name)
+	rs1, _ = kube1clientset.Extensions().ReplicaSets(apiv1.NamespaceDefault).Get(rs.Name)
+	rs2, _ = kube2clientset.Extensions().ReplicaSets(apiv1.NamespaceDefault).Get(rs.Name)
+	assert.Equal(t, *rs.Spec.Replicas, *rs1.Spec.Replicas+*rs2.Spec.Replicas)
+	assert.Equal(t, rs.Status.Replicas, rs1.Status.Replicas+rs2.Status.Replicas)
+
+	close(stopChan)
+}
+
+func toFederatedInformerForTestOnly(informer fedutil.FederatedInformer) fedutil.FederatedInformerForTestOnly {
+	inter := informer.(interface{})
+	return inter.(fedutil.FederatedInformerForTestOnly)
+}
+
+func mkCluster(name string, readyStatus apiv1.ConditionStatus) *fedv1.Cluster {
+	return &fedv1.Cluster{
+		ObjectMeta: apiv1.ObjectMeta{
+			Name: name,
+		},
+		Status: fedv1.ClusterStatus{
+			Conditions: []fedv1.ClusterCondition{
+				{Type: fedv1.ClusterReady, Status: readyStatus},
+			},
+		},
+	}
+}
+
+func mkReplicaSet(name string, replicas int32) *extensionsv1.ReplicaSet {
+	return &extensionsv1.ReplicaSet{
+		ObjectMeta: apiv1.ObjectMeta{
+			Name:      name,
+			Namespace: apiv1.NamespaceDefault,
+		},
+		Spec: extensionsv1.ReplicaSetSpec{
+			Replicas: &replicas,
+		},
+	}
+}