[Federation] Convert the ReplicaSet controller to a sync controller.

federation/cmd/federation-controller-manager/app/BUILD

@@ -26,7 +26,6 @@ go_library(
         "//federation/pkg/federation-controller/deployment:go_default_library",
         "//federation/pkg/federation-controller/ingress:go_default_library",
         "//federation/pkg/federation-controller/namespace:go_default_library",
-        "//federation/pkg/federation-controller/replicaset:go_default_library",
         "//federation/pkg/federation-controller/service:go_default_library",
         "//federation/pkg/federation-controller/service/dns:go_default_library",
         "//federation/pkg/federation-controller/sync:go_default_library",

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

@@ -40,7 +40,6 @@ import (
 	deploymentcontroller "k8s.io/kubernetes/federation/pkg/federation-controller/deployment"
 	ingresscontroller "k8s.io/kubernetes/federation/pkg/federation-controller/ingress"
 	namespacecontroller "k8s.io/kubernetes/federation/pkg/federation-controller/namespace"
-	replicasetcontroller "k8s.io/kubernetes/federation/pkg/federation-controller/replicaset"
 	servicecontroller "k8s.io/kubernetes/federation/pkg/federation-controller/service"
 	servicednscontroller "k8s.io/kubernetes/federation/pkg/federation-controller/service/dns"
 	synccontroller "k8s.io/kubernetes/federation/pkg/federation-controller/sync"
@@ -167,14 +166,6 @@ func StartControllers(s *options.CMServer, restClientCfg *restclient.Config) err
 		}
 	}
 
-	if controllerEnabled(s.Controllers, serverResources, replicasetcontroller.ControllerName, replicasetcontroller.RequiredResources, true) {
-		glog.V(3).Infof("Loading client config for replica set controller %q", replicasetcontroller.UserAgentName)
-		replicaSetClientset := federationclientset.NewForConfigOrDie(restclient.AddUserAgent(restClientCfg, replicasetcontroller.UserAgentName))
-		replicaSetController := replicasetcontroller.NewReplicaSetController(replicaSetClientset)
-		glog.V(3).Infof("Running replica set controller")
-		go replicaSetController.Run(s.ConcurrentReplicaSetSyncs, wait.NeverStop)
-	}
-
 	if controllerEnabled(s.Controllers, serverResources, deploymentcontroller.ControllerName, deploymentcontroller.RequiredResources, true) {
 		glog.V(3).Infof("Loading client config for deployment controller %q", deploymentcontroller.UserAgentName)
 		deploymentClientset := federationclientset.NewForConfigOrDie(restclient.AddUserAgent(restClientCfg, deploymentcontroller.UserAgentName))

federation/pkg/federatedtypes/BUILD

@@ -5,6 +5,7 @@ licenses(["notice"])
 load(
     "@io_bazel_rules_go//go:def.bzl",
     "go_library",
+    "go_test",
 )
 
 go_library(
@@ -14,15 +15,21 @@ go_library(
         "configmap.go",
         "daemonset.go",
         "registry.go",
+        "replicaset.go",
         "scheduling.go",
         "secret.go",
     ],
     tags = ["automanaged"],
     deps = [
+        "//federation/apis/federation:go_default_library",
         "//federation/apis/federation/v1beta1:go_default_library",
         "//federation/client/clientset_generated/federation_clientset:go_default_library",
         "//federation/pkg/federation-controller/util:go_default_library",
+        "//federation/pkg/federation-controller/util/planner:go_default_library",
+        "//federation/pkg/federation-controller/util/podanalyzer:go_default_library",
+        "//federation/pkg/federation-controller/util/replicapreferences:go_default_library",
         "//pkg/client/clientset_generated/clientset:go_default_library",
+        "//vendor/github.com/golang/glog: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",
@@ -48,3 +55,16 @@ filegroup(
     ],
     tags = ["automanaged"],
 )
+
+go_test(
+    name = "go_default_test",
+    srcs = ["replicaset_test.go"],
+    library = ":go_default_library",
+    tags = ["automanaged"],
+    deps = [
+        "//vendor/github.com/stretchr/testify/assert: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",
+    ],
+)

federation/pkg/federatedtypes/crudtester/crudtester.go

@@ -197,8 +197,17 @@ func (c *FederatedTypeCRUDTester) CheckPropagationForClients(obj pkgruntime.Obje
 func (c *FederatedTypeCRUDTester) waitForResource(client clientset.Interface, obj pkgruntime.Object) error {
 	namespacedName := c.adapter.NamespacedName(obj)
 	err := wait.PollImmediate(c.waitInterval, c.clusterWaitTimeout, func() (bool, error) {
+		equivalenceFunc := c.adapter.Equivalent
+		if c.adapter.IsSchedulingAdapter() {
+			schedulingAdapter, ok := c.adapter.(federatedtypes.SchedulingAdapter)
+			if !ok {
+				c.tl.Fatalf("Adapter for kind %q does not properly implement SchedulingAdapter.", c.adapter.Kind())
+			}
+			equivalenceFunc = schedulingAdapter.EquivalentIgnoringSchedule
+		}
+
 		clusterObj, err := c.adapter.ClusterGet(client, namespacedName)
-		if err == nil && c.adapter.Equivalent(clusterObj, obj) {
+		if err == nil && equivalenceFunc(clusterObj, obj) {
 			return true, nil
 		}
 		if errors.IsNotFound(err) {

federation/pkg/federatedtypes/replicaset_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 (
+	"fmt"
+	"testing"
+	"time"
+
+	apiv1 "k8s.io/api/core/v1"
+	extensionsv1 "k8s.io/api/extensions/v1beta1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	"github.com/stretchr/testify/assert"
+)
+
+const (
+	pods        = "pods"
+	replicasets = "replicasets"
+	k8s1        = "k8s-1"
+	k8s2        = "k8s-2"
+)
+
+func TestClusterReplicaState(t *testing.T) {
+	uncalledPodsGetter := func(clusterName string, replicaSet *extensionsv1.ReplicaSet) (*apiv1.PodList, error) {
+		t.Fatal("podsGetter should not be called when replica sets are all ready.")
+		return nil, nil
+	}
+
+	podsByReplicaSet := make(map[*extensionsv1.ReplicaSet][]*apiv1.Pod)
+	podsGetter := func(clusterName string, replicaSet *extensionsv1.ReplicaSet) (*apiv1.PodList, error) {
+		pods, ok := podsByReplicaSet[replicaSet]
+		if !ok {
+			t.Fatalf("No pods found in test data for replica set named %v", replicaSet.Name)
+			return nil, fmt.Errorf("Not found")
+		}
+		var podListPods []apiv1.Pod
+		for _, pod := range pods {
+			podListPods = append(podListPods, *pod)
+		}
+		return &apiv1.PodList{Items: podListPods}, nil
+	}
+
+	readyCondition := apiv1.PodCondition{Type: apiv1.PodReady}
+	unschedulableCondition := apiv1.PodCondition{
+		Type:               apiv1.PodScheduled,
+		Status:             apiv1.ConditionFalse,
+		Reason:             apiv1.PodReasonUnschedulable,
+		LastTransitionTime: metav1.NewTime(time.Now().Add(-1 * time.Hour)),
+	}
+
+	one := int64(1)
+	two := int64(2)
+
+	tests := map[string]struct {
+		rs1Replicas                   int32
+		rs2Replicas                   int32
+		rs1ReadyReplicas              int32
+		rs2ReadyReplicas              int32
+		podsGetter                    func(clusterName string, replicaSet *extensionsv1.ReplicaSet) (*apiv1.PodList, error)
+		pod1Phase                     apiv1.PodPhase
+		pod1Condition                 apiv1.PodCondition
+		pod2Phase                     apiv1.PodPhase
+		pod2Condition                 apiv1.PodCondition
+		cluster1Replicas              *int64
+		cluster2Replicas              *int64
+		cluster1UnschedulableReplicas *int64
+		cluster2UnschedulableReplicas *int64
+	}{
+		"All replica sets have an equal number of requested and ready replicas.": {rs1Replicas: 2, rs2Replicas: 2, rs1ReadyReplicas: 2, rs2ReadyReplicas: 2, podsGetter: uncalledPodsGetter, cluster1Replicas: &two, cluster2Replicas: &two},
+		"One replica set has a pending schedulable pod":                          {rs1Replicas: 2, rs2Replicas: 2, rs1ReadyReplicas: 1, rs2ReadyReplicas: 2, podsGetter: podsGetter, pod1Phase: apiv1.PodRunning, pod1Condition: readyCondition, pod2Phase: apiv1.PodPending, cluster1Replicas: &one, cluster2Replicas: &two},
+		"One replica set has an unschedulable pod":                               {rs1Replicas: 2, rs2Replicas: 2, rs1ReadyReplicas: 1, rs2ReadyReplicas: 2, podsGetter: podsGetter, pod1Phase: apiv1.PodRunning, pod1Condition: readyCondition, pod2Phase: apiv1.PodPending, pod2Condition: unschedulableCondition, cluster1Replicas: &one, cluster2Replicas: &two, cluster1UnschedulableReplicas: &one},
+	}
+
+	for name, tt := range tests {
+		t.Run(name, func(t *testing.T) {
+			clusters := []string{"one", "two"}
+			replicaSetsByCluster := make(map[string]*extensionsv1.ReplicaSet)
+			replicaSetGetter := func(clusterName string, key string) (interface{}, bool, error) {
+				rs, ok := replicaSetsByCluster[clusterName]
+				if !ok {
+					t.Fatalf("No replica set found in test data for %v", clusterName)
+					return nil, false, fmt.Errorf("Not found")
+				}
+				return rs, true, nil
+			}
+			rs1 := newReplicaSetWithReplicas("one", tt.rs1Replicas)
+			rs2 := newReplicaSetWithReplicas("two", tt.rs2Replicas)
+			rs1.Spec.Replicas = &tt.rs1Replicas
+			rs2.Spec.Replicas = &tt.rs2Replicas
+			rs1.Status.ReadyReplicas = tt.rs1ReadyReplicas
+			rs2.Status.ReadyReplicas = tt.rs2ReadyReplicas
+
+			replicaSetsByCluster["one"] = rs1
+			replicaSetsByCluster["two"] = rs2
+
+			pod1 := newPod("one")
+			pod2 := newPod("two")
+			podThree := newPod("three")
+			podFour := newPod("four")
+
+			pod1.Status.Phase = tt.pod1Phase
+			pod2.Status.Phase = tt.pod2Phase
+			pod1.Status.Conditions = []apiv1.PodCondition{tt.pod1Condition}
+			pod2.Status.Conditions = []apiv1.PodCondition{tt.pod2Condition}
+
+			podsByReplicaSet[rs1] = []*apiv1.Pod{pod1, pod2}
+			podsByReplicaSet[rs2] = []*apiv1.Pod{podThree, podFour}
+
+			current, estimatedCapacity, err := clustersReplicaState(clusters, "", replicaSetGetter, tt.podsGetter)
+
+			assert.Nil(t, err)
+
+			wantedCurrent := make(map[string]int64)
+			if tt.cluster1Replicas != nil {
+				wantedCurrent["one"] = *tt.cluster1Replicas
+			}
+			if tt.cluster2Replicas != nil {
+				wantedCurrent["two"] = *tt.cluster2Replicas
+			}
+			assert.Equal(t, wantedCurrent, current)
+
+			wantedEstimatedCapacity := make(map[string]int64)
+			if tt.cluster1UnschedulableReplicas != nil {
+				wantedEstimatedCapacity["one"] = *tt.cluster1UnschedulableReplicas
+			}
+			if tt.cluster2UnschedulableReplicas != nil {
+				wantedEstimatedCapacity["two"] = *tt.cluster2UnschedulableReplicas
+			}
+			assert.Equal(t, wantedEstimatedCapacity, estimatedCapacity)
+		})
+	}
+}
+
+func newReplicaSetWithReplicas(name string, replicas int32) *extensionsv1.ReplicaSet {
+	return &extensionsv1.ReplicaSet{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name,
+			Namespace: metav1.NamespaceDefault,
+			SelfLink:  "/api/v1/namespaces/default/replicasets/name",
+		},
+		Spec: extensionsv1.ReplicaSetSpec{
+			Replicas: &replicas,
+		},
+	}
+}
+
+func newPod(name string) *apiv1.Pod {
+	return &apiv1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name,
+			Namespace: metav1.NamespaceDefault,
+		},
+	}
+}

federation/pkg/federatedtypes/scheduling.go

@@ -44,4 +44,8 @@ type SchedulingAdapter interface {
 	GetSchedule(obj pkgruntime.Object, key string, clusters []*federationapi.Cluster, informer fedutil.FederatedInformer) (*SchedulingInfo, error)
 	ScheduleObject(cluster *federationapi.Cluster, clusterObj pkgruntime.Object, federationObjCopy pkgruntime.Object, schedulingInfo *SchedulingInfo) (pkgruntime.Object, bool, error)
 	UpdateFederatedStatus(obj pkgruntime.Object, status SchedulingStatus) error
+
+	// EquivalentIgnoringSchedule returns whether obj1 and obj2 are
+	// equivalent ignoring differences due to scheduling.
+	EquivalentIgnoringSchedule(obj1, obj2 pkgruntime.Object) bool
 }

federation/pkg/federation-controller/BUILD

@@ -28,7 +28,6 @@ filegroup(
         "//federation/pkg/federation-controller/deployment:all-srcs",
         "//federation/pkg/federation-controller/ingress:all-srcs",
         "//federation/pkg/federation-controller/namespace:all-srcs",
-        "//federation/pkg/federation-controller/replicaset:all-srcs",
         "//federation/pkg/federation-controller/service:all-srcs",
         "//federation/pkg/federation-controller/sync:all-srcs",
         "//federation/pkg/federation-controller/util:all-srcs",

federation/pkg/federation-controller/deployment/BUILD

@@ -20,7 +20,6 @@ go_library(
         "//federation/pkg/federation-controller/util/deletionhelper:go_default_library",
         "//federation/pkg/federation-controller/util/eventsink:go_default_library",
         "//federation/pkg/federation-controller/util/planner:go_default_library",
-        "//federation/pkg/federation-controller/util/podanalyzer:go_default_library",
         "//federation/pkg/federation-controller/util/replicapreferences:go_default_library",
         "//pkg/api:go_default_library",
         "//pkg/client/clientset_generated/clientset:go_default_library",
@@ -30,6 +29,7 @@ go_library(
         "//vendor/k8s.io/api/extensions/v1beta1:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/api/errors:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/labels: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/wait:go_default_library",

federation/pkg/federation-controller/deployment/deploymentcontroller.go

@@ -29,6 +29,7 @@ import (
 	extensionsv1 "k8s.io/api/extensions/v1beta1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/util/wait"
@@ -44,7 +45,6 @@ import (
 	"k8s.io/kubernetes/federation/pkg/federation-controller/util/deletionhelper"
 	"k8s.io/kubernetes/federation/pkg/federation-controller/util/eventsink"
 	"k8s.io/kubernetes/federation/pkg/federation-controller/util/planner"
-	"k8s.io/kubernetes/federation/pkg/federation-controller/util/podanalyzer"
 	"k8s.io/kubernetes/federation/pkg/federation-controller/util/replicapreferences"
 	"k8s.io/kubernetes/pkg/api"
 	kubeclientset "k8s.io/kubernetes/pkg/client/clientset_generated/clientset"
@@ -352,6 +352,62 @@ func (fdc *DeploymentController) worker() {
 	}
 }
 
+type podAnalysisResult struct {
+	// Total number of pods created.
+	total int
+	// Number of pods that are running and ready.
+	runningAndReady int
+	// Number of pods that have been in unschedulable state for UnshedulableThreshold seconds.
+	unschedulable int
+
+	// TODO: Handle other scenarios like pod waiting too long for scheduler etc.
+}
+
+const (
+	// TODO: make it configurable
+	unschedulableThreshold = 60 * time.Second
+)
+
+// A function that calculates how many pods from the list are in one of
+// the meaningful (from the replica set perspective) states. This function is
+// a temporary workaround against the current lack of ownerRef in pods.
+// TODO(perotinus): Unify this with the ReplicaSet controller.
+func analyzePods(selectorv1 *metav1.LabelSelector, allPods []fedutil.FederatedObject, currentTime time.Time) (map[string]podAnalysisResult, error) {
+	selector, err := metav1.LabelSelectorAsSelector(selectorv1)
+	if err != nil {
+		return nil, fmt.Errorf("invalid selector: %v", err)
+	}
+	result := make(map[string]podAnalysisResult)
+
+	for _, fedObject := range allPods {
+		pod, isPod := fedObject.Object.(*apiv1.Pod)
+		if !isPod {
+			return nil, fmt.Errorf("invalid arg content - not a *pod")
+		}
+		if !selector.Empty() && selector.Matches(labels.Set(pod.Labels)) {
+			status := result[fedObject.ClusterName]
+			status.total++
+			for _, condition := range pod.Status.Conditions {
+				if pod.Status.Phase == apiv1.PodRunning {
+					if condition.Type == apiv1.PodReady {
+						status.runningAndReady++
+					}
+				} else {
+					if condition.Type == apiv1.PodScheduled &&
+						condition.Status == apiv1.ConditionFalse &&
+						condition.Reason == apiv1.PodReasonUnschedulable &&
+						condition.LastTransitionTime.Add(unschedulableThreshold).Before(currentTime) {
+
+						status.unschedulable++
+					}
+				}
+			}
+			result[fedObject.ClusterName] = status
+		}
+	}
+	return result, nil
+}
+
 func (fdc *DeploymentController) schedule(fd *extensionsv1.Deployment, clusters []*fedv1.Cluster,
 	current map[string]int64, estimatedCapacity map[string]int64) map[string]int64 {
 	// TODO: integrate real scheduler
@@ -471,10 +527,7 @@ func (fdc *DeploymentController) reconcileDeployment(key string) (reconciliation
 	if err != nil {
 		return statusError, err
 	}
-	podStatus, err := podanalyzer.AnalysePods(fd.Spec.Selector, allPods, time.Now())
-	if err != nil {
-		return statusError, err
-	}
+	podStatus, err := analyzePods(fd.Spec.Selector, allPods, time.Now())
 	current := make(map[string]int64)
 	estimatedCapacity := make(map[string]int64)
 	for _, cluster := range clusters {
@@ -484,8 +537,8 @@ func (fdc *DeploymentController) reconcileDeployment(key string) (reconciliation
 		}
 		if exists {
 			ld := ldObj.(*extensionsv1.Deployment)
-			current[cluster.Name] = int64(podStatus[cluster.Name].RunningAndReady) // include pending as well?
-			unschedulable := int64(podStatus[cluster.Name].Unschedulable)
+			current[cluster.Name] = int64(podStatus[cluster.Name].runningAndReady) // include pending as well?
+			unschedulable := int64(podStatus[cluster.Name].unschedulable)
 			if unschedulable > 0 {
 				estimatedCapacity[cluster.Name] = int64(*ld.Spec.Replicas) - unschedulable
 			}

federation/pkg/federation-controller/replicaset/BUILD

-package(default_visibility = ["//visibility:public"])
-
-licenses(["notice"])
-
-load(
-    "@io_bazel_rules_go//go:def.bzl",
-    "go_library",
-    "go_test",
-)
-
-go_library(
-    name = "go_default_library",
-    srcs = ["replicasetcontroller.go"],
-    tags = ["automanaged"],
-    deps = [
-        "//federation/apis/federation:go_default_library",
-        "//federation/apis/federation/v1beta1:go_default_library",
-        "//federation/client/clientset_generated/federation_clientset:go_default_library",
-        "//federation/pkg/federation-controller/util:go_default_library",
-        "//federation/pkg/federation-controller/util/deletionhelper:go_default_library",
-        "//federation/pkg/federation-controller/util/eventsink:go_default_library",
-        "//federation/pkg/federation-controller/util/planner:go_default_library",
-        "//federation/pkg/federation-controller/util/podanalyzer:go_default_library",
-        "//federation/pkg/federation-controller/util/replicapreferences:go_default_library",
-        "//pkg/api:go_default_library",
-        "//pkg/client/clientset_generated/clientset:go_default_library",
-        "//pkg/controller:go_default_library",
-        "//vendor/github.com/golang/glog: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/errors: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/wait:go_default_library",
-        "//vendor/k8s.io/apimachinery/pkg/watch:go_default_library",
-        "//vendor/k8s.io/client-go/tools/cache:go_default_library",
-        "//vendor/k8s.io/client-go/tools/record:go_default_library",
-        "//vendor/k8s.io/client-go/util/flowcontrol:go_default_library",
-        "//vendor/k8s.io/client-go/util/workqueue:go_default_library",
-    ],
-)
-
-go_test(
-    name = "go_default_test",
-    srcs = ["replicasetcontroller_test.go"],
-    library = ":go_default_library",
-    tags = ["automanaged"],
-    deps = [
-        "//federation/apis/federation/v1beta1:go_default_library",
-        "//federation/client/clientset_generated/federation_clientset/fake:go_default_library",
-        "//federation/pkg/federation-controller/util/test:go_default_library",
-        "//pkg/client/clientset_generated/clientset:go_default_library",
-        "//pkg/client/clientset_generated/clientset/fake:go_default_library",
-        "//vendor/github.com/stretchr/testify/assert: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",
-        "//vendor/k8s.io/apimachinery/pkg/watch:go_default_library",
-        "//vendor/k8s.io/client-go/testing:go_default_library",
-    ],
-)
-
-filegroup(
-    name = "package-srcs",
-    srcs = glob(["**"]),
-    tags = ["automanaged"],
-    visibility = ["//visibility:private"],
-)
-
-filegroup(
-    name = "all-srcs",
-    srcs = [":package-srcs"],
-    tags = ["automanaged"],
-)

federation/pkg/federation-controller/sync/BUILD

@@ -42,18 +42,28 @@ go_library(
 
 go_test(
     name = "go_default_test",
-    srcs = ["controller_test.go"],
+    srcs = [
+        "controller_test.go",
+        "replicasetcontroller_test.go",
+    ],
     library = ":go_default_library",
     tags = ["automanaged"],
     deps = [
         "//federation/apis/federation/v1beta1:go_default_library",
+        "//federation/client/clientset_generated/federation_clientset/fake:go_default_library",
         "//federation/pkg/federatedtypes:go_default_library",
         "//federation/pkg/federation-controller/util:go_default_library",
         "//federation/pkg/federation-controller/util/test:go_default_library",
+        "//pkg/client/clientset_generated/clientset:go_default_library",
+        "//pkg/client/clientset_generated/clientset/fake:go_default_library",
+        "//vendor/github.com/stretchr/testify/assert:go_default_library",
         "//vendor/github.com/stretchr/testify/require: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",
         "//vendor/k8s.io/apimachinery/pkg/runtime:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/watch:go_default_library",
+        "//vendor/k8s.io/client-go/testing:go_default_library",
     ],
 )
 

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

@@ -84,10 +84,11 @@ type FederationSyncController struct {
 
 	deletionHelper *deletionhelper.DeletionHelper
 
-	reviewDelay           time.Duration
-	clusterAvailableDelay time.Duration
-	smallDelay            time.Duration
-	updateTimeout         time.Duration
+	reviewDelay             time.Duration
+	clusterAvailableDelay   time.Duration
+	clusterUnavailableDelay time.Duration
+	smallDelay              time.Duration
+	updateTimeout           time.Duration
 
 	adapter federatedtypes.FederatedTypeAdapter
 }
@@ -112,14 +113,15 @@ func newFederationSyncController(client federationclientset.Interface, adapter f
 	recorder := broadcaster.NewRecorder(api.Scheme, clientv1.EventSource{Component: fmt.Sprintf("federation-%v-controller", adapter.Kind())})
 
 	s := &FederationSyncController{
-		reviewDelay:           time.Second * 10,
-		clusterAvailableDelay: time.Second * 20,
-		smallDelay:            time.Second * 3,
-		updateTimeout:         time.Second * 30,
-		workQueue:             workqueue.New(),
-		backoff:               flowcontrol.NewBackOff(5*time.Second, time.Minute),
-		eventRecorder:         recorder,
-		adapter:               adapter,
+		reviewDelay:             time.Second * 10,
+		clusterAvailableDelay:   time.Second * 20,
+		clusterUnavailableDelay: time.Second * 60,
+		smallDelay:              time.Second * 3,
+		updateTimeout:           time.Second * 30,
+		workQueue:               workqueue.New(),
+		backoff:                 flowcontrol.NewBackOff(5*time.Second, time.Minute),
+		eventRecorder:           recorder,
+		adapter:                 adapter,
 	}
 
 	// Build delivereres for triggering reconciliations.
@@ -169,6 +171,10 @@ func newFederationSyncController(client federationclientset.Interface, adapter f
 				// When new cluster becomes available process all the target resources again.
 				s.clusterDeliverer.DeliverAt(allClustersKey, nil, time.Now().Add(s.clusterAvailableDelay))
 			},
+			// When a cluster becomes unavailable process all the target resources again.
+			ClusterUnavailable: func(cluster *federationapi.Cluster, _ []interface{}) {
+				s.clusterDeliverer.DeliverAt(allClustersKey, nil, time.Now().Add(s.clusterUnavailableDelay))
+			},
 		},
 	)
 
@@ -205,6 +211,7 @@ func newFederationSyncController(client federationclientset.Interface, adapter f
 // minimizeLatency reduces delays and timeouts to make the controller more responsive (useful for testing).
 func (s *FederationSyncController) minimizeLatency() {
 	s.clusterAvailableDelay = time.Second
+	s.clusterUnavailableDelay = time.Second
 	s.reviewDelay = 50 * time.Millisecond
 	s.smallDelay = 20 * time.Millisecond
 	s.updateTimeout = 5 * time.Second
@@ -328,6 +335,10 @@ func (s *FederationSyncController) reconcile(namespacedName types.NamespacedName
 	kind := s.adapter.Kind()
 	key := namespacedName.String()
 
+	glog.V(4).Infof("Starting to reconcile %v %v", kind, key)
+	startTime := time.Now()
+	defer glog.V(4).Infof("Finished reconciling %v %v (duration: %v)", kind, key, time.Now().Sub(startTime))
+
 	obj, err := s.objFromCache(kind, key)
 	if err != nil {
 		return statusError

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

@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package replicaset
+package sync
 
 import (
 	"flag"
@@ -29,7 +29,8 @@ import (
 	core "k8s.io/client-go/testing"
 	fedv1 "k8s.io/kubernetes/federation/apis/federation/v1beta1"
 	fedclientfake "k8s.io/kubernetes/federation/client/clientset_generated/federation_clientset/fake"
-	"k8s.io/kubernetes/federation/pkg/federation-controller/util/test"
+	"k8s.io/kubernetes/federation/pkg/federatedtypes"
+	testutil "k8s.io/kubernetes/federation/pkg/federation-controller/util/test"
 	kubeclientset "k8s.io/kubernetes/pkg/client/clientset_generated/clientset"
 	kubeclientfake "k8s.io/kubernetes/pkg/client/clientset_generated/clientset/fake"
 
@@ -48,11 +49,6 @@ func TestReplicaSetController(t *testing.T) {
 	flag.Set("v", "5")
 	flag.Parse()
 
-	replicaSetReviewDelay = 10 * time.Millisecond
-	clusterAvailableDelay = 20 * time.Millisecond
-	clusterUnavailableDelay = 60 * time.Millisecond
-	allReplicaSetReviewDelay = 120 * time.Millisecond
-
 	fedclientset := fedclientfake.NewSimpleClientset()
 	fedrswatch := watch.NewFake()
 	fedclientset.PrependWatchReactor(replicasets, core.DefaultWatchReactor(fedrswatch, nil))
@@ -81,20 +77,19 @@ func TestReplicaSetController(t *testing.T) {
 			return nil, fmt.Errorf("Unknown cluster: %v", cluster.Name)
 		}
 	}
-	replicaSetController := NewReplicaSetController(fedclientset)
-	rsFedinformer := testutil.ToFederatedInformerForTestOnly(replicaSetController.fedReplicaSetInformer)
+	replicaSetController := newFederationSyncController(fedclientset, federatedtypes.NewReplicaSetAdapter(fedclientset))
+	replicaSetController.minimizeLatency()
+	rsFedinformer := testutil.ToFederatedInformerForTestOnly(replicaSetController.informer)
 	rsFedinformer.SetClientFactory(fedInformerClientFactory)
-	podFedinformer := testutil.ToFederatedInformerForTestOnly(replicaSetController.fedPodInformer)
-	podFedinformer.SetClientFactory(fedInformerClientFactory)
 
 	stopChan := make(chan struct{})
 	defer close(stopChan)
-	go replicaSetController.Run(1, stopChan)
+	go replicaSetController.Run(stopChan)
 
 	rs := newReplicaSetWithReplicas("rs", 9)
 	rs, _ = fedclientset.Extensions().ReplicaSets(metav1.NamespaceDefault).Create(rs)
 	fedrswatch.Add(rs)
-	time.Sleep(1 * time.Second)
+	time.Sleep(2 * time.Second)
 
 	rs1, _ := kube1clientset.Extensions().ReplicaSets(metav1.NamespaceDefault).Get(rs.Name, metav1.GetOptions{})
 	kube1rswatch.Add(rs1)
@@ -114,7 +109,7 @@ func TestReplicaSetController(t *testing.T) {
 	rs2, _ = kube2clientset.Extensions().ReplicaSets(metav1.NamespaceDefault).UpdateStatus(rs2)
 	kube2rswatch.Modify(rs2)
 
-	time.Sleep(1 * time.Second)
+	time.Sleep(2 * time.Second)
 	rs, _ = fedclientset.Extensions().ReplicaSets(metav1.NamespaceDefault).Get(rs.Name, metav1.GetOptions{})
 	assert.Equal(t, *rs.Spec.Replicas, *rs1.Spec.Replicas+*rs2.Spec.Replicas)
 	assert.Equal(t, rs.Status.Replicas, rs1.Status.Replicas+rs2.Status.Replicas)
@@ -126,7 +121,7 @@ func TestReplicaSetController(t *testing.T) {
 	rs.Spec.Replicas = &replicas
 	rs, _ = fedclientset.Extensions().ReplicaSets(metav1.NamespaceDefault).Update(rs)
 	fedrswatch.Modify(rs)
-	time.Sleep(1 * time.Second)
+	time.Sleep(2 * time.Second)
 
 	rs1, _ = kube1clientset.Extensions().ReplicaSets(metav1.NamespaceDefault).Get(rs.Name, metav1.GetOptions{})
 	rs1.Status.Replicas = *rs1.Spec.Replicas

federation/pkg/federation-controller/util/podanalyzer/BUILD

@@ -12,12 +12,7 @@ go_library(
     name = "go_default_library",
     srcs = ["pod_helper.go"],
     tags = ["automanaged"],
-    deps = [
-        "//federation/pkg/federation-controller/util:go_default_library",
-        "//vendor/k8s.io/api/core/v1:go_default_library",
-        "//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
-        "//vendor/k8s.io/apimachinery/pkg/labels:go_default_library",
-    ],
+    deps = ["//vendor/k8s.io/api/core/v1:go_default_library"],
 )
 
 go_test(
@@ -26,7 +21,6 @@ go_test(
     library = ":go_default_library",
     tags = ["automanaged"],
     deps = [
-        "//federation/pkg/federation-controller/util:go_default_library",
         "//vendor/github.com/stretchr/testify/assert:go_default_library",
         "//vendor/k8s.io/api/core/v1:go_default_library",
         "//vendor/k8s.io/api/extensions/v1beta1:go_default_library",

federation/pkg/federation-controller/util/podanalyzer/pod_helper.go

@@ -17,13 +17,9 @@ limitations under the License.
 package podanalyzer
 
 import (
-	"fmt"
 	"time"
 
 	api_v1 "k8s.io/api/core/v1"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/labels"
-	"k8s.io/kubernetes/federation/pkg/federation-controller/util"
 )
 
 type PodAnalysisResult struct {
@@ -42,41 +38,26 @@ const (
 	UnschedulableThreshold = 60 * time.Second
 )
 
-// A function that calculates how many pods from the list are in one of
+// AnalyzePods calculates how many pods from the list are in one of
 // the meaningful (from the replica set perspective) states. This function is
 // a temporary workaround against the current lack of ownerRef in pods.
-func AnalysePods(selectorv1 *metav1.LabelSelector, allPods []util.FederatedObject, currentTime time.Time) (map[string]PodAnalysisResult, error) {
-	selector, err := metav1.LabelSelectorAsSelector(selectorv1)
-	if err != nil {
-		return nil, fmt.Errorf("invalid selector: %v", err)
-	}
-	result := make(map[string]PodAnalysisResult)
-
-	for _, fedObject := range allPods {
-		pod, isPod := fedObject.Object.(*api_v1.Pod)
-		if !isPod {
-			return nil, fmt.Errorf("invalid arg content - not a *pod")
-		}
-		if !selector.Empty() && selector.Matches(labels.Set(pod.Labels)) {
-			status := result[fedObject.ClusterName]
-			status.Total++
-			for _, condition := range pod.Status.Conditions {
-				if pod.Status.Phase == api_v1.PodRunning {
-					if condition.Type == api_v1.PodReady {
-						status.RunningAndReady++
-					}
-				} else {
-					if condition.Type == api_v1.PodScheduled &&
-						condition.Status == api_v1.ConditionFalse &&
-						condition.Reason == api_v1.PodReasonUnschedulable &&
-						condition.LastTransitionTime.Add(UnschedulableThreshold).Before(currentTime) {
-
-						status.Unschedulable++
-					}
+func AnalyzePods(pods *api_v1.PodList, currentTime time.Time) PodAnalysisResult {
+	result := PodAnalysisResult{}
+	for _, pod := range pods.Items {
+		result.Total++
+		for _, condition := range pod.Status.Conditions {
+			if pod.Status.Phase == api_v1.PodRunning {
+				if condition.Type == api_v1.PodReady {
+					result.RunningAndReady++
 				}
+			} else if condition.Type == api_v1.PodScheduled &&
+				condition.Status == api_v1.ConditionFalse &&
+				condition.Reason == api_v1.PodReasonUnschedulable &&
+				condition.LastTransitionTime.Add(UnschedulableThreshold).Before(currentTime) {
+
+				result.Unschedulable++
 			}
-			result[fedObject.ClusterName] = status
 		}
 	}
-	return result, nil
+	return result
 }

federation/pkg/federation-controller/util/podanalyzer/pod_helper_test.go

@@ -23,16 +23,13 @@ import (
 	api_v1 "k8s.io/api/core/v1"
 	"k8s.io/api/extensions/v1beta1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/kubernetes/federation/pkg/federation-controller/util"
 
 	"github.com/stretchr/testify/assert"
 )
 
 func TestAnalyze(t *testing.T) {
 	now := time.Now()
-	replicaSet := newReplicaSet(map[string]string{"A": "B"})
-	replicaSet2 := newReplicaSet(map[string]string{"C": "D"})
-	podRunning := newPod("p1", replicaSet,
+	podRunning := newPod("p1",
 		api_v1.PodStatus{
 			Phase: api_v1.PodRunning,
 			Conditions: []api_v1.PodCondition{
@@ -42,7 +39,7 @@ func TestAnalyze(t *testing.T) {
 				},
 			},
 		})
-	podUnschedulable := newPod("pU", replicaSet,
+	podUnschedulable := newPod("pU",
 		api_v1.PodStatus{
 			Phase: api_v1.PodPending,
 			Conditions: []api_v1.PodCondition{
@@ -54,42 +51,25 @@ func TestAnalyze(t *testing.T) {
 				},
 			},
 		})
-	podOther := newPod("pO", replicaSet,
-		api_v1.PodStatus{
-			Phase:      api_v1.PodPending,
-			Conditions: []api_v1.PodCondition{},
-		})
-	podOtherRS := newPod("pO", replicaSet2,
+	podOther := newPod("pO",
 		api_v1.PodStatus{
 			Phase:      api_v1.PodPending,
 			Conditions: []api_v1.PodCondition{},
 		})
 
-	federatedObjects := []util.FederatedObject{
-		{ClusterName: "c1", Object: podRunning},
-		{ClusterName: "c1", Object: podRunning},
-		{ClusterName: "c1", Object: podRunning},
-		{ClusterName: "c1", Object: podUnschedulable},
-		{ClusterName: "c1", Object: podUnschedulable},
-		{ClusterName: "c2", Object: podOther},
-		{ClusterName: "c2", Object: podOtherRS},
-	}
-
-	raport, err := AnalysePods(replicaSet.Spec.Selector, federatedObjects, now)
-	assert.NoError(t, err)
-	assert.Equal(t, 2, len(raport))
-	c1Raport := raport["c1"]
-	c2Raport := raport["c2"]
+	result := AnalyzePods(&api_v1.PodList{Items: []api_v1.Pod{*podRunning, *podRunning, *podRunning, *podUnschedulable, *podUnschedulable}}, now)
 	assert.Equal(t, PodAnalysisResult{
 		Total:           5,
 		RunningAndReady: 3,
 		Unschedulable:   2,
-	}, c1Raport)
+	}, result)
+
+	result = AnalyzePods(&api_v1.PodList{Items: []api_v1.Pod{*podOther}}, now)
 	assert.Equal(t, PodAnalysisResult{
 		Total:           1,
 		RunningAndReady: 0,
 		Unschedulable:   0,
-	}, c2Raport)
+	}, result)
 }
 
 func newReplicaSet(selectorMap map[string]string) *v1beta1.ReplicaSet {
@@ -97,7 +77,7 @@ func newReplicaSet(selectorMap map[string]string) *v1beta1.ReplicaSet {
 	rs := &v1beta1.ReplicaSet{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      "foobar",
-			Namespace: "default",
+			Namespace: metav1.NamespaceDefault,
 		},
 		Spec: v1beta1.ReplicaSetSpec{
 			Replicas: &replicas,
@@ -107,12 +87,11 @@ func newReplicaSet(selectorMap map[string]string) *v1beta1.ReplicaSet {
 	return rs
 }
 
-func newPod(name string, rs *v1beta1.ReplicaSet, status api_v1.PodStatus) *api_v1.Pod {
+func newPod(name string, status api_v1.PodStatus) *api_v1.Pod {
 	return &api_v1.Pod{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      name,
-			Namespace: rs.Namespace,
-			Labels:    rs.Spec.Selector.MatchLabels,
+			Namespace: metav1.NamespaceDefault,
 		},
 		Status: status,
 	}

test/e2e_federation/BUILD

@@ -29,7 +29,6 @@ go_library(
         "//federation/client/clientset_generated/federation_clientset:go_default_library",
         "//federation/client/clientset_generated/federation_clientset/typed/core/v1:go_default_library",
         "//federation/pkg/federatedtypes:go_default_library",
-        "//federation/pkg/federation-controller/replicaset:go_default_library",
         "//federation/pkg/federation-controller/util:go_default_library",
         "//pkg/api:go_default_library",
         "//pkg/api/v1:go_default_library",

test/e2e_federation/replicaset.go

@@ -37,7 +37,7 @@ import (
 	"k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/util/uuid"
 	"k8s.io/kubernetes/federation/apis/federation"
-	fedreplicsetcontroller "k8s.io/kubernetes/federation/pkg/federation-controller/replicaset"
+	federatedtypes "k8s.io/kubernetes/federation/pkg/federatedtypes"
 )
 
 const (
@@ -513,7 +513,7 @@ func newReplicaSetObj(namespace string, replicas int32, pref *federation.Replica
 	if pref != nil {
 		prefBytes, _ := json.Marshal(pref)
 		prefString := string(prefBytes)
-		rs.Annotations[fedreplicsetcontroller.FedReplicaSetPreferencesAnnotation] = prefString
+		rs.Annotations[federatedtypes.FedReplicaSetPreferencesAnnotation] = prefString
 	}
 	return rs