Merge pull request #45745 from marun/fed-test-cluster-addition

Automatic merge from submit-queue (batch tested with PRs 45990, 45544, 45745, 45742, 45678)

[Federation]  Add integration testing for cluster addition

This PR adds integration testing of the sync controller for cluster addition.  This ensures coverage equivalency between the integration tests and the old controller unit tests, so those tests are removed by this PR.

Resolves #45257

cc: @kubernetes/sig-federation-pr-reviews

federation/pkg/federatedtypes/crudtester/crudtester.go

@@ -163,11 +163,17 @@ func (c *FederatedTypeCRUDTester) CheckDelete(obj pkgruntime.Object, orphanDepen
 	}
 }
 
+// CheckPropagation checks propagation for the crud tester's clients
 func (c *FederatedTypeCRUDTester) CheckPropagation(obj pkgruntime.Object) {
+	c.CheckPropagationForClients(obj, c.clusterClients)
+}
+
+// CheckPropagationForClients checks propagation for the provided clients
+func (c *FederatedTypeCRUDTester) CheckPropagationForClients(obj pkgruntime.Object, clusterClients []clientset.Interface) {
 	namespacedName := c.adapter.NamespacedName(obj)
 
-	c.tl.Logf("Waiting for %s %q in %d clusters", c.kind, namespacedName, len(c.clusterClients))
+	c.tl.Logf("Waiting for %s %q in %d clusters", c.kind, namespacedName, len(clusterClients))
-	for _, client := range c.clusterClients {
+	for _, client := range clusterClients {
 		err := c.waitForResource(client, obj)
 		if err != nil {
 			c.tl.Fatalf("Failed to verify %s %q in a member cluster: %v", c.kind, namespacedName, err)

federation/pkg/federation-controller/sync/BUILD

@@ -39,6 +39,22 @@ go_library(
     ],
 )
 
+go_test(
+    name = "go_default_test",
+    srcs = ["controller_test.go"],
+    library = ":go_default_library",
+    tags = ["automanaged"],
+    deps = [
+        "//federation/apis/federation/v1beta1: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/api/v1:go_default_library",
+        "//vendor/github.com/stretchr/testify/require:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/runtime:go_default_library",
+    ],
+)
+
 filegroup(
     name = "package-srcs",
     srcs = glob(["**"]),
@@ -51,34 +67,3 @@ filegroup(
     srcs = [":package-srcs"],
     tags = ["automanaged"],
 )
-
-go_test(
-    name = "go_default_test",
-    srcs = [
-        "configmap_controller_test.go",
-        "controller_test.go",
-        "daemonset_controller_test.go",
-        "secret_controller_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/deletionhelper:go_default_library",
-        "//federation/pkg/federation-controller/util/test:go_default_library",
-        "//pkg/api/v1:go_default_library",
-        "//pkg/apis/extensions/v1beta1:go_default_library",
-        "//pkg/client/clientset_generated/clientset:go_default_library",
-        "//pkg/client/clientset_generated/clientset/fake:go_default_library",
-        "//vendor/github.com/golang/glog:go_default_library",
-        "//vendor/github.com/stretchr/testify/assert:go_default_library",
-        "//vendor/github.com/stretchr/testify/require: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/types:go_default_library",
-        "//vendor/k8s.io/apimachinery/pkg/util/wait:go_default_library",
-    ],
-)

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

@@ -1,190 +0,0 @@
-/*
-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 sync
-
-import (
-	"fmt"
-	"testing"
-	"time"
-
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/runtime"
-	"k8s.io/apimachinery/pkg/types"
-	"k8s.io/apimachinery/pkg/util/wait"
-	federationapi "k8s.io/kubernetes/federation/apis/federation/v1beta1"
-	fakefedclientset "k8s.io/kubernetes/federation/client/clientset_generated/federation_clientset/fake"
-	"k8s.io/kubernetes/federation/pkg/federatedtypes"
-	"k8s.io/kubernetes/federation/pkg/federation-controller/util"
-	"k8s.io/kubernetes/federation/pkg/federation-controller/util/deletionhelper"
-	. "k8s.io/kubernetes/federation/pkg/federation-controller/util/test"
-	apiv1 "k8s.io/kubernetes/pkg/api/v1"
-	kubeclientset "k8s.io/kubernetes/pkg/client/clientset_generated/clientset"
-	fakekubeclientset "k8s.io/kubernetes/pkg/client/clientset_generated/clientset/fake"
-
-	"github.com/golang/glog"
-	"github.com/stretchr/testify/assert"
-)
-
-func TestConfigMapController(t *testing.T) {
-	configmaps := "configmaps"
-	clusters := "clusters"
-	informerStoreErr := "configmap should have appeared in the informer store"
-
-	cluster1 := NewCluster("cluster1", apiv1.ConditionTrue)
-	cluster2 := NewCluster("cluster2", apiv1.ConditionTrue)
-
-	fakeClient := &fakefedclientset.Clientset{}
-	RegisterFakeList(clusters, &fakeClient.Fake, &federationapi.ClusterList{Items: []federationapi.Cluster{*cluster1}})
-	RegisterFakeList(configmaps, &fakeClient.Fake, &apiv1.ConfigMapList{Items: []apiv1.ConfigMap{}})
-	configmapWatch := RegisterFakeWatch(configmaps, &fakeClient.Fake)
-	configmapUpdateChan := RegisterFakeCopyOnUpdate(configmaps, &fakeClient.Fake, configmapWatch)
-	clusterWatch := RegisterFakeWatch(clusters, &fakeClient.Fake)
-
-	cluster1Client := &fakekubeclientset.Clientset{}
-	cluster1Watch := RegisterFakeWatch(configmaps, &cluster1Client.Fake)
-	RegisterFakeList(configmaps, &cluster1Client.Fake, &apiv1.ConfigMapList{Items: []apiv1.ConfigMap{}})
-	cluster1CreateChan := RegisterFakeCopyOnCreate(configmaps, &cluster1Client.Fake, cluster1Watch)
-	cluster1UpdateChan := RegisterFakeCopyOnUpdate(configmaps, &cluster1Client.Fake, cluster1Watch)
-
-	cluster2Client := &fakekubeclientset.Clientset{}
-	cluster2Watch := RegisterFakeWatch(configmaps, &cluster2Client.Fake)
-	RegisterFakeList(configmaps, &cluster2Client.Fake, &apiv1.ConfigMapList{Items: []apiv1.ConfigMap{}})
-	cluster2CreateChan := RegisterFakeCopyOnCreate(configmaps, &cluster2Client.Fake, cluster2Watch)
-
-	configmapController := newFederationSyncController(fakeClient, federatedtypes.NewConfigMapAdapter(fakeClient))
-	informer := ToFederatedInformerForTestOnly(configmapController.informer)
-	informer.SetClientFactory(func(cluster *federationapi.Cluster) (kubeclientset.Interface, error) {
-		switch cluster.Name {
-		case cluster1.Name:
-			return cluster1Client, nil
-		case cluster2.Name:
-			return cluster2Client, nil
-		default:
-			return nil, fmt.Errorf("Unknown cluster")
-		}
-	})
-
-	configmapController.minimizeLatency()
-
-	stop := make(chan struct{})
-	configmapController.Run(stop)
-
-	configmap1 := &apiv1.ConfigMap{
-		ObjectMeta: metav1.ObjectMeta{
-			Name:      "test-configmap",
-			Namespace: "ns",
-			SelfLink:  "/api/v1/namespaces/ns/configmaps/test-configmap",
-		},
-		Data: map[string]string{
-			"A": "ala ma kota",
-			"B": "quick brown fox",
-		},
-	}
-
-	// Test add federated configmap.
-	configmapWatch.Add(configmap1)
-	// There should be 2 updates to add both the finalizers.
-	updatedConfigMap := GetConfigMapFromChan(configmapUpdateChan)
-	AssertHasFinalizer(t, updatedConfigMap, deletionhelper.FinalizerDeleteFromUnderlyingClusters)
-	AssertHasFinalizer(t, updatedConfigMap, metav1.FinalizerOrphanDependents)
-
-	// Verify that the configmap is created in underlying cluster1.
-	createdConfigMap := GetConfigMapFromChan(cluster1CreateChan)
-	assert.NotNil(t, createdConfigMap)
-	assert.Equal(t, configmap1.Namespace, createdConfigMap.Namespace)
-	assert.Equal(t, configmap1.Name, createdConfigMap.Name)
-	assert.True(t, util.ConfigMapEquivalent(configmap1, createdConfigMap))
-
-	// Wait for the configmap to appear in the informer store
-	err := WaitForStoreUpdate(
-		configmapController.informer.GetTargetStore(),
-		cluster1.Name, types.NamespacedName{Namespace: configmap1.Namespace, Name: configmap1.Name}.String(), wait.ForeverTestTimeout)
-	assert.Nil(t, err, informerStoreErr)
-
-	// Test update federated configmap.
-	configmap1.Annotations = map[string]string{
-		"A": "B",
-	}
-	configmapWatch.Modify(configmap1)
-	updatedConfigMap = GetConfigMapFromChan(cluster1UpdateChan)
-	assert.NotNil(t, updatedConfigMap)
-	assert.Equal(t, configmap1.Name, updatedConfigMap.Name)
-	assert.Equal(t, configmap1.Namespace, updatedConfigMap.Namespace)
-	assert.True(t, util.ConfigMapEquivalent(configmap1, updatedConfigMap))
-
-	// Wait for the configmap to appear in the informer store
-	err = WaitForConfigMapStoreUpdate(
-		configmapController.informer.GetTargetStore(),
-		cluster1.Name, types.NamespacedName{Namespace: configmap1.Namespace, Name: configmap1.Name}.String(),
-		configmap1, wait.ForeverTestTimeout)
-	assert.Nil(t, err, informerStoreErr)
-
-	// Test update federated configmap.
-	configmap1.Data = map[string]string{
-		"config": "myconfigurationfile",
-	}
-
-	configmapWatch.Modify(configmap1)
-	for {
-		updatedConfigMap := GetConfigMapFromChan(cluster1UpdateChan)
-		assert.NotNil(t, updatedConfigMap)
-		if updatedConfigMap == nil {
-			break
-		}
-		assert.Equal(t, configmap1.Name, updatedConfigMap.Name)
-		assert.Equal(t, configmap1.Namespace, updatedConfigMap.Namespace)
-		if util.ConfigMapEquivalent(configmap1, updatedConfigMap) {
-			break
-		}
-	}
-
-	// Test add cluster
-	clusterWatch.Add(cluster2)
-	createdConfigMap2 := GetConfigMapFromChan(cluster2CreateChan)
-	assert.NotNil(t, createdConfigMap2)
-	assert.Equal(t, configmap1.Name, createdConfigMap2.Name)
-	assert.Equal(t, configmap1.Namespace, createdConfigMap2.Namespace)
-	assert.True(t, util.ConfigMapEquivalent(configmap1, createdConfigMap2))
-
-	close(stop)
-}
-
-func GetConfigMapFromChan(c chan runtime.Object) *apiv1.ConfigMap {
-	if configmap := GetObjectFromChan(c); configmap == nil {
-		return nil
-	} else {
-		return configmap.(*apiv1.ConfigMap)
-	}
-}
-
-// Wait till the store is updated with latest configmap.
-func WaitForConfigMapStoreUpdate(store util.FederatedReadOnlyStore, clusterName, key string, desiredConfigMap *apiv1.ConfigMap, timeout time.Duration) error {
-	retryInterval := 200 * time.Millisecond
-	err := wait.PollImmediate(retryInterval, timeout, func() (bool, error) {
-		obj, found, err := store.GetByKey(clusterName, key)
-		if !found || err != nil {
-			glog.Infof("%s is not in the store", key)
-			return false, err
-		}
-		equal := util.ConfigMapEquivalent(obj.(*apiv1.ConfigMap), desiredConfigMap)
-		if !equal {
-			glog.Infof("wrong content in the store expected:\n%v\nactual:\n%v\n", *desiredConfigMap, *obj.(*apiv1.ConfigMap))
-		}
-		return equal, err
-	})
-	return err
-}

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

@@ -1,168 +0,0 @@
-/*
-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 sync
-
-import (
-	"fmt"
-	"reflect"
-	"testing"
-
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/runtime"
-	"k8s.io/apimachinery/pkg/types"
-	"k8s.io/apimachinery/pkg/util/wait"
-	federationapi "k8s.io/kubernetes/federation/apis/federation/v1beta1"
-	fakefedclientset "k8s.io/kubernetes/federation/client/clientset_generated/federation_clientset/fake"
-	"k8s.io/kubernetes/federation/pkg/federatedtypes"
-	"k8s.io/kubernetes/federation/pkg/federation-controller/util"
-	"k8s.io/kubernetes/federation/pkg/federation-controller/util/deletionhelper"
-	. "k8s.io/kubernetes/federation/pkg/federation-controller/util/test"
-	apiv1 "k8s.io/kubernetes/pkg/api/v1"
-	extensionsv1 "k8s.io/kubernetes/pkg/apis/extensions/v1beta1"
-	kubeclientset "k8s.io/kubernetes/pkg/client/clientset_generated/clientset"
-	fakekubeclientset "k8s.io/kubernetes/pkg/client/clientset_generated/clientset/fake"
-
-	"github.com/stretchr/testify/assert"
-)
-
-func TestDaemonSetController(t *testing.T) {
-	daemonsets := "daemonsets"
-	clusters := "clusters"
-
-	cluster1 := NewCluster("cluster1", apiv1.ConditionTrue)
-	cluster2 := NewCluster("cluster2", apiv1.ConditionTrue)
-
-	fakeClient := &fakefedclientset.Clientset{}
-	RegisterFakeList(clusters, &fakeClient.Fake, &federationapi.ClusterList{Items: []federationapi.Cluster{*cluster1}})
-	RegisterFakeList(daemonsets, &fakeClient.Fake, &extensionsv1.DaemonSetList{Items: []extensionsv1.DaemonSet{}})
-	daemonsetWatch := RegisterFakeWatch(daemonsets, &fakeClient.Fake)
-	daemonsetUpdateChan := RegisterFakeCopyOnUpdate(daemonsets, &fakeClient.Fake, daemonsetWatch)
-	clusterWatch := RegisterFakeWatch("clusters", &fakeClient.Fake)
-
-	cluster1Client := &fakekubeclientset.Clientset{}
-	cluster1Watch := RegisterFakeWatch(daemonsets, &cluster1Client.Fake)
-	RegisterFakeList(daemonsets, &cluster1Client.Fake, &extensionsv1.DaemonSetList{Items: []extensionsv1.DaemonSet{}})
-	cluster1CreateChan := RegisterFakeCopyOnCreate(daemonsets, &cluster1Client.Fake, cluster1Watch)
-	cluster1UpdateChan := RegisterFakeCopyOnUpdate(daemonsets, &cluster1Client.Fake, cluster1Watch)
-
-	cluster2Client := &fakekubeclientset.Clientset{}
-	cluster2Watch := RegisterFakeWatch(daemonsets, &cluster2Client.Fake)
-	RegisterFakeList(daemonsets, &cluster2Client.Fake, &extensionsv1.DaemonSetList{Items: []extensionsv1.DaemonSet{}})
-	cluster2CreateChan := RegisterFakeCopyOnCreate(daemonsets, &cluster2Client.Fake, cluster2Watch)
-
-	daemonsetController := newFederationSyncController(fakeClient, federatedtypes.NewDaemonSetAdapter(fakeClient))
-	informer := ToFederatedInformerForTestOnly(daemonsetController.informer)
-	informer.SetClientFactory(func(cluster *federationapi.Cluster) (kubeclientset.Interface, error) {
-		switch cluster.Name {
-		case cluster1.Name:
-			return cluster1Client, nil
-		case cluster2.Name:
-			return cluster2Client, nil
-		default:
-			return nil, fmt.Errorf("Unknown cluster")
-		}
-	})
-
-	daemonsetController.minimizeLatency()
-
-	stop := make(chan struct{})
-	daemonsetController.Run(stop)
-
-	daemonset1 := extensionsv1.DaemonSet{
-		ObjectMeta: metav1.ObjectMeta{
-			Name:      "test-daemonset",
-			Namespace: "ns",
-			SelfLink:  "/api/v1/namespaces/ns/daemonsets/test-daemonset",
-		},
-		Spec: extensionsv1.DaemonSetSpec{
-			Selector: &metav1.LabelSelector{
-				MatchLabels: map[string]string{
-					"A": "xyz",
-				},
-			},
-		},
-	}
-
-	// Test add federated daemonset.
-	daemonsetWatch.Add(&daemonset1)
-
-	// There should be an update to add both the finalizers.
-	updatedDaemonSet := GetDaemonSetFromChan(daemonsetUpdateChan)
-	AssertHasFinalizer(t, updatedDaemonSet, deletionhelper.FinalizerDeleteFromUnderlyingClusters)
-	AssertHasFinalizer(t, updatedDaemonSet, metav1.FinalizerOrphanDependents)
-	daemonset1 = *updatedDaemonSet
-
-	createdDaemonSet := GetDaemonSetFromChan(cluster1CreateChan)
-	assert.NotNil(t, createdDaemonSet)
-	assert.Equal(t, daemonset1.Namespace, createdDaemonSet.Namespace)
-	assert.Equal(t, daemonset1.Name, createdDaemonSet.Name)
-	assert.True(t, daemonsetsEqual(daemonset1, *createdDaemonSet),
-		fmt.Sprintf("expected: %v, actual: %v", daemonset1, *createdDaemonSet))
-
-	daemonsetKey := types.NamespacedName{
-		Namespace: daemonset1.Namespace,
-		Name:      daemonset1.Name,
-	}.String()
-	// Wait for the daemonset to appear in the informer store
-	err := WaitForStoreUpdate(
-		daemonsetController.informer.GetTargetStore(),
-		cluster1.Name, daemonsetKey, wait.ForeverTestTimeout)
-	assert.Nil(t, err, "daemonset should have appeared in the informer store")
-
-	// TODO: Re-enable this when we have fixed these flaky tests: https://github.com/kubernetes/kubernetes/issues/36540.
-	// Test update federated daemonset.
-	daemonset1.Annotations = map[string]string{
-		"A": "B",
-	}
-	daemonsetWatch.Modify(&daemonset1)
-	updatedDaemonSet = GetDaemonSetFromChan(cluster1UpdateChan)
-	assert.NotNil(t, updatedDaemonSet)
-	assert.Equal(t, daemonset1.Name, updatedDaemonSet.Name)
-	assert.Equal(t, daemonset1.Namespace, updatedDaemonSet.Namespace)
-	assert.True(t, daemonsetsEqual(daemonset1, *updatedDaemonSet),
-		fmt.Sprintf("expected: %v, actual: %v", daemonset1, *updatedDaemonSet))
-
-	// Test update federated daemonset.
-	daemonset1.Spec.Template.Name = "TEST"
-	daemonsetWatch.Modify(&daemonset1)
-	err = CheckObjectFromChan(cluster1UpdateChan, MetaAndSpecCheckingFunction(&daemonset1))
-	assert.NoError(t, err)
-
-	// Test add cluster
-	clusterWatch.Add(cluster2)
-	createdDaemonSet2 := GetDaemonSetFromChan(cluster2CreateChan)
-	assert.NotNil(t, createdDaemonSet2)
-	assert.Equal(t, daemonset1.Name, createdDaemonSet2.Name)
-	assert.Equal(t, daemonset1.Namespace, createdDaemonSet2.Namespace)
-	assert.True(t, daemonsetsEqual(daemonset1, *createdDaemonSet2),
-		fmt.Sprintf("expected: %v, actual: %v", daemonset1, *createdDaemonSet2))
-
-	close(stop)
-}
-
-func daemonsetsEqual(a, b extensionsv1.DaemonSet) bool {
-	return util.ObjectMetaEquivalent(a.ObjectMeta, b.ObjectMeta) && reflect.DeepEqual(a.Spec, b.Spec)
-}
-
-func GetDaemonSetFromChan(c chan runtime.Object) *extensionsv1.DaemonSet {
-	if daemonset := GetObjectFromChan(c); daemonset == nil {
-		return nil
-	} else {
-		return daemonset.(*extensionsv1.DaemonSet)
-	}
-
-}

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

@@ -1,212 +0,0 @@
-/*
-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 sync
-
-import (
-	"fmt"
-	"reflect"
-	"testing"
-	"time"
-
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/runtime"
-	"k8s.io/apimachinery/pkg/types"
-	"k8s.io/apimachinery/pkg/util/wait"
-	federationapi "k8s.io/kubernetes/federation/apis/federation/v1beta1"
-	fakefedclientset "k8s.io/kubernetes/federation/client/clientset_generated/federation_clientset/fake"
-	"k8s.io/kubernetes/federation/pkg/federatedtypes"
-	"k8s.io/kubernetes/federation/pkg/federation-controller/util"
-	"k8s.io/kubernetes/federation/pkg/federation-controller/util/deletionhelper"
-	. "k8s.io/kubernetes/federation/pkg/federation-controller/util/test"
-	apiv1 "k8s.io/kubernetes/pkg/api/v1"
-	kubeclientset "k8s.io/kubernetes/pkg/client/clientset_generated/clientset"
-	fakekubeclientset "k8s.io/kubernetes/pkg/client/clientset_generated/clientset/fake"
-
-	"github.com/golang/glog"
-	"github.com/stretchr/testify/assert"
-)
-
-func TestSecretController(t *testing.T) {
-	clusters := "clusters"
-	secrets := "secrets"
-
-	cluster1 := NewCluster("cluster1", apiv1.ConditionTrue)
-	cluster2 := NewCluster("cluster2", apiv1.ConditionTrue)
-
-	fakeClient := &fakefedclientset.Clientset{}
-	RegisterFakeList(clusters, &fakeClient.Fake, &federationapi.ClusterList{Items: []federationapi.Cluster{*cluster1}})
-	RegisterFakeList(secrets, &fakeClient.Fake, &apiv1.SecretList{Items: []apiv1.Secret{}})
-	secretWatch := RegisterFakeWatch(secrets, &fakeClient.Fake)
-	secretUpdateChan := RegisterFakeCopyOnUpdate(secrets, &fakeClient.Fake, secretWatch)
-	clusterWatch := RegisterFakeWatch(clusters, &fakeClient.Fake)
-
-	cluster1Client := &fakekubeclientset.Clientset{}
-	cluster1Watch := RegisterFakeWatch(secrets, &cluster1Client.Fake)
-	RegisterFakeList(secrets, &cluster1Client.Fake, &apiv1.SecretList{Items: []apiv1.Secret{}})
-	cluster1CreateChan := RegisterFakeCopyOnCreate(secrets, &cluster1Client.Fake, cluster1Watch)
-	cluster1UpdateChan := RegisterFakeCopyOnUpdate(secrets, &cluster1Client.Fake, cluster1Watch)
-
-	cluster2Client := &fakekubeclientset.Clientset{}
-	cluster2Watch := RegisterFakeWatch(secrets, &cluster2Client.Fake)
-	RegisterFakeList(secrets, &cluster2Client.Fake, &apiv1.SecretList{Items: []apiv1.Secret{}})
-	cluster2CreateChan := RegisterFakeCopyOnCreate(secrets, &cluster2Client.Fake, cluster2Watch)
-
-	secretController := newFederationSyncController(fakeClient, federatedtypes.NewSecretAdapter(fakeClient))
-	informerClientFactory := func(cluster *federationapi.Cluster) (kubeclientset.Interface, error) {
-		switch cluster.Name {
-		case cluster1.Name:
-			return cluster1Client, nil
-		case cluster2.Name:
-			return cluster2Client, nil
-		default:
-			return nil, fmt.Errorf("Unknown cluster")
-		}
-	}
-	setClientFactory(secretController.informer, informerClientFactory)
-
-	secretController.minimizeLatency()
-
-	stop := make(chan struct{})
-	secretController.Run(stop)
-
-	secret1 := apiv1.Secret{
-		ObjectMeta: metav1.ObjectMeta{
-			Name:      "test-secret",
-			Namespace: "ns",
-			SelfLink:  "/api/v1/namespaces/ns/secrets/test-secret",
-		},
-		Data: map[string][]byte{
-			"A": []byte("ala ma kota"),
-			"B": []byte("quick brown fox"),
-		},
-		Type: apiv1.SecretTypeOpaque,
-	}
-
-	// Test add federated secret.
-	secretWatch.Add(&secret1)
-	// There should be an update to add both the finalizers.
-	updatedSecret := GetSecretFromChan(secretUpdateChan)
-	AssertHasFinalizer(t, updatedSecret, deletionhelper.FinalizerDeleteFromUnderlyingClusters)
-	AssertHasFinalizer(t, updatedSecret, metav1.FinalizerOrphanDependents)
-	secret1 = *updatedSecret
-
-	// Verify that the secret is created in underlying cluster1.
-	createdSecret := GetSecretFromChan(cluster1CreateChan)
-	assert.NotNil(t, createdSecret)
-	assert.Equal(t, secret1.Namespace, createdSecret.Namespace)
-	assert.Equal(t, secret1.Name, createdSecret.Name)
-	assert.True(t, secretsEqual(secret1, *createdSecret),
-		fmt.Sprintf("expected: %v, actual: %v", secret1, *createdSecret))
-
-	// Wait for the secret to appear in the informer store
-	err := WaitForStoreUpdate(
-		secretController.informer.GetTargetStore(),
-		cluster1.Name, types.NamespacedName{Namespace: secret1.Namespace, Name: secret1.Name}.String(), wait.ForeverTestTimeout)
-	assert.Nil(t, err, "secret should have appeared in the informer store")
-
-	checkAll := func(expected apiv1.Secret) CheckingFunction {
-		return func(obj runtime.Object) error {
-			glog.V(4).Infof("Checking %v", obj)
-			s := obj.(*apiv1.Secret)
-			if err := CompareObjectMeta(expected.ObjectMeta, s.ObjectMeta); err != nil {
-				return err
-			}
-			if !reflect.DeepEqual(expected.Data, s.Data) {
-				return fmt.Errorf("Data is different expected:%v actual:%v", expected.Data, s.Data)
-			}
-			if expected.Type != s.Type {
-				return fmt.Errorf("Type is different expected:%v actual:%v", expected.Type, s.Type)
-			}
-			return nil
-		}
-	}
-
-	// Test update federated secret.
-	secret1.Annotations = map[string]string{
-		"A": "B",
-	}
-	secretWatch.Modify(&secret1)
-	err = CheckObjectFromChan(cluster1UpdateChan, checkAll(secret1))
-	assert.NoError(t, err)
-
-	// Wait for the secret to be updated in the informer store.
-	err = WaitForSecretStoreUpdate(
-		secretController.informer.GetTargetStore(),
-		cluster1.Name, types.NamespacedName{Namespace: secret1.Namespace, Name: secret1.Name}.String(),
-		&secret1, wait.ForeverTestTimeout)
-	assert.NoError(t, err, "secret should have been updated in the informer store")
-
-	// Test update federated secret.
-	secret1.Data = map[string][]byte{
-		"config": []byte("myconfigurationfile"),
-	}
-	secretWatch.Modify(&secret1)
-	err = CheckObjectFromChan(cluster1UpdateChan, checkAll(secret1))
-	assert.NoError(t, err)
-
-	// Test add cluster
-	clusterWatch.Add(cluster2)
-	createdSecret2 := GetSecretFromChan(cluster2CreateChan)
-	assert.NotNil(t, createdSecret2)
-	assert.Equal(t, secret1.Name, createdSecret2.Name)
-	assert.Equal(t, secret1.Namespace, createdSecret2.Namespace)
-	assert.True(t, secretsEqual(secret1, *createdSecret2),
-		fmt.Sprintf("expected: %v, actual: %v", secret1, *createdSecret2))
-
-	close(stop)
-}
-
-func setClientFactory(informer util.FederatedInformer, informerClientFactory func(*federationapi.Cluster) (kubeclientset.Interface, error)) {
-	testInformer := ToFederatedInformerForTestOnly(informer)
-	testInformer.SetClientFactory(informerClientFactory)
-}
-
-func secretsEqual(a, b apiv1.Secret) bool {
-	// Clear the SelfLink and ObjectMeta.Finalizers since they will be different
-	// in resource in federation control plane and resource in underlying cluster.
-	a.SelfLink = ""
-	b.SelfLink = ""
-	a.ObjectMeta.Finalizers = []string{}
-	b.ObjectMeta.Finalizers = []string{}
-	return reflect.DeepEqual(a, b)
-}
-
-func GetSecretFromChan(c chan runtime.Object) *apiv1.Secret {
-	if secret := GetObjectFromChan(c); secret == nil {
-		return nil
-	} else {
-		return secret.(*apiv1.Secret)
-	}
-}
-
-// Wait till the store is updated with latest secret.
-func WaitForSecretStoreUpdate(store util.FederatedReadOnlyStore, clusterName, key string, desiredSecret *apiv1.Secret, timeout time.Duration) error {
-	retryInterval := 200 * time.Millisecond
-	err := wait.PollImmediate(retryInterval, timeout, func() (bool, error) {
-		obj, found, err := store.GetByKey(clusterName, key)
-		if !found || err != nil {
-			glog.Infof("%s is not in the store", key)
-			return false, err
-		}
-		equal := secretsEqual(*obj.(*apiv1.Secret), *desiredSecret)
-		if !equal {
-			glog.Infof("wrong content in the store expected:\n%v\nactual:\n%v\n", *desiredSecret, *obj.(*apiv1.Secret))
-		}
-		return equal, err
-	})
-	return err
-}

test/integration/federation/BUILD

@@ -20,6 +20,7 @@ go_test(
     deps = [
         "//federation/apis/federation/v1beta1:go_default_library",
         "//federation/pkg/federatedtypes:go_default_library",
+        "//federation/pkg/federatedtypes/crudtester:go_default_library",
         "//pkg/api/v1:go_default_library",
         "//pkg/apis/autoscaling/v1:go_default_library",
         "//pkg/apis/batch/v1:go_default_library",
@@ -28,6 +29,7 @@ go_test(
         "//vendor/github.com/pborman/uuid:go_default_library",
         "//vendor/github.com/stretchr/testify/assert: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",
     ],
 )

test/integration/federation/crud_test.go

@@ -22,7 +22,9 @@ import (
 
 	"github.com/pborman/uuid"
 
+	pkgruntime "k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/kubernetes/federation/pkg/federatedtypes"
+	"k8s.io/kubernetes/federation/pkg/federatedtypes/crudtester"
 	"k8s.io/kubernetes/test/integration/federation/framework"
 )
 
@@ -35,41 +37,57 @@ func TestFederationCRUD(t *testing.T) {
 	federatedTypes := federatedtypes.FederatedTypes()
 	for kind, fedType := range federatedTypes {
 		t.Run(kind, func(t *testing.T) {
-			config := fedFixture.APIFixture.NewConfig()
+			fixture, crudTester, obj := initCRUDTest(t, &fedFixture, fedType.AdapterFactory, kind)
-			fixture := framework.NewControllerFixture(t, kind, fedType.AdapterFactory, config)
 			defer fixture.TearDown(t)
 
-			client := fedFixture.APIFixture.NewClient(fmt.Sprintf("crud-test-%s", kind))
+			crudTester.CheckLifecycle(obj)
-			adapter := fedType.AdapterFactory(client)
-
-			crudtester := framework.NewFederatedTypeCRUDTester(t, adapter, fedFixture.ClusterClients)
-			obj := adapter.NewTestObject(uuid.New())
-			crudtester.CheckLifecycle(obj)
 		})
 	}
 
-	// Validate deletion handling where orphanDependents is true or nil for a single resource type since the
+	// The following tests target a single type since the underlying logic is common across all types.
-	// underlying logic is common across all types.
-	orphanedDependents := true
-	testCases := map[string]*bool{
-		"Resources should not be deleted from underlying clusters when OrphanDependents is true": &orphanedDependents,
-		"Resources should not be deleted from underlying clusters when OrphanDependents is nil":  nil,
-	}
 	kind := federatedtypes.SecretKind
 	adapterFactory := federatedtypes.NewSecretAdapter
+
+	// Validate deletion handling where orphanDependents is true or nil
+	orphanedDependents := true
+	testCases := map[string]*bool{
+		"Resource should not be deleted from underlying clusters when OrphanDependents is true": &orphanedDependents,
+		"Resource should not be deleted from underlying clusters when OrphanDependents is nil":  nil,
+	}
 	for testName, orphanDependents := range testCases {
 		t.Run(testName, func(t *testing.T) {
-			config := fedFixture.APIFixture.NewConfig()
+			fixture, crudTester, obj := initCRUDTest(t, &fedFixture, adapterFactory, kind)
-			fixture := framework.NewControllerFixture(t, kind, adapterFactory, config)
 			defer fixture.TearDown(t)
 
-			client := fedFixture.APIFixture.NewClient(fmt.Sprintf("deletion-test-%s", kind))
+			updatedObj := crudTester.CheckCreate(obj)
-			adapter := adapterFactory(client)
+			crudTester.CheckDelete(updatedObj, orphanDependents)
-
-			crudtester := framework.NewFederatedTypeCRUDTester(t, adapter, fedFixture.ClusterClients)
-			obj := adapter.NewTestObject(uuid.New())
-			updatedObj := crudtester.CheckCreate(obj)
-			crudtester.CheckDelete(updatedObj, orphanDependents)
 		})
 	}
+
+	t.Run("Resource should be propagated to a newly added cluster", func(t *testing.T) {
+		fixture, crudTester, obj := initCRUDTest(t, &fedFixture, adapterFactory, kind)
+		defer fixture.TearDown(t)
+
+		updatedObj := crudTester.CheckCreate(obj)
+		// Start a new cluster and validate that the resource is propagated to it.
+		fedFixture.StartCluster(t)
+		// Check propagation to the new cluster by providing the updated set of clients
+		crudTester.CheckPropagationForClients(updatedObj, fedFixture.ClusterClients)
+	})
+}
+
+// initCRUDTest initializes common elements of a crud test
+func initCRUDTest(t *testing.T, fedFixture *framework.FederationFixture, adapterFactory federatedtypes.AdapterFactory, kind string) (
+	*framework.ControllerFixture, *crudtester.FederatedTypeCRUDTester, pkgruntime.Object) {
+	config := fedFixture.APIFixture.NewConfig()
+	fixture := framework.NewControllerFixture(t, kind, adapterFactory, config)
+
+	client := fedFixture.APIFixture.NewClient(fmt.Sprintf("crud-test-%s", kind))
+	adapter := adapterFactory(client)
+
+	crudTester := framework.NewFederatedTypeCRUDTester(t, adapter, fedFixture.ClusterClients)
+
+	obj := adapter.NewTestObject(uuid.New())
+
+	return fixture, crudTester, obj
 }

test/integration/federation/framework/federation.go

@@ -23,6 +23,7 @@ import (
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	federationapi "k8s.io/kubernetes/federation/apis/federation/v1beta1"
+	federationclientset "k8s.io/kubernetes/federation/client/clientset_generated/federation_clientset"
 	clustercontroller "k8s.io/kubernetes/federation/pkg/federation-controller/cluster"
 	"k8s.io/kubernetes/pkg/client/clientset_generated/clientset"
 	"k8s.io/kubernetes/pkg/master"
@@ -43,6 +44,7 @@ type FederationFixture struct {
 	Clusters            []*MemberCluster
 	ClusterClients      []clientset.Interface
 	ClusterController   *clustercontroller.ClusterController
+	fedClient           federationclientset.Interface
 	stopChan            chan struct{}
 }
 
@@ -64,46 +66,46 @@ func (f *FederationFixture) SetUp(t *testing.T) {
 	monitorPeriod := 1 * time.Second
 	clustercontroller.StartClusterController(f.APIFixture.NewConfig(), f.stopChan, monitorPeriod)
 
-	f.startClusters()
+	f.fedClient = f.APIFixture.NewClient("federation-fixture")
+	for i := 0; i < f.DesiredClusterCount; i++ {
+		f.StartCluster(t)
+	}
 }
 
-func (f *FederationFixture) startClusters() {
+func (f *FederationFixture) StartCluster(t *testing.T) {
-	fedClient := f.APIFixture.NewClient("federation-fixture")
+	config := framework.NewMasterConfig()
-	for i := 0; i < f.DesiredClusterCount; i++ {
+	_, _, closeFn := framework.RunAMaster(config)
-		config := framework.NewMasterConfig()
+	host := config.GenericConfig.LoopbackClientConfig.Host
-		_, _, closeFn := framework.RunAMaster(config)
-		host := config.GenericConfig.LoopbackClientConfig.Host
 
-		// Use fmt to ensure the output will be visible when run with go test -v
+	clusterClient := clientset.NewForConfigOrDie(config.GenericConfig.LoopbackClientConfig)
-		fmt.Printf("Federated cluster %d serving on %s", i, host)
+	f.ClusterClients = append(f.ClusterClients, clusterClient)
+	f.Clusters = append(f.Clusters, &MemberCluster{
+		CloseFn: closeFn,
+		Config:  config,
+		Client:  clusterClient,
+		Host:    host,
+	})
 
-		clusterClient := clientset.NewForConfigOrDie(config.GenericConfig.LoopbackClientConfig)
+	clusterId := len(f.ClusterClients)
-		f.Clusters = append(f.Clusters, &MemberCluster{
-			CloseFn: closeFn,
-			Config:  config,
-			Client:  clusterClient,
-			Host:    host,
-		})
 
-		f.ClusterClients = append(f.ClusterClients, clusterClient)
+	t.Logf("Federated cluster %d serving on %s", clusterId, host)
 
-		cluster := &federationapi.Cluster{
+	cluster := &federationapi.Cluster{
-			ObjectMeta: metav1.ObjectMeta{
+		ObjectMeta: metav1.ObjectMeta{
-				Name: fmt.Sprintf("cluster-%d", i),
+			Name: fmt.Sprintf("cluster-%d", clusterId),
-			},
+		},
-			Spec: federationapi.ClusterSpec{
+		Spec: federationapi.ClusterSpec{
-				ServerAddressByClientCIDRs: []federationapi.ServerAddressByClientCIDR{
+			ServerAddressByClientCIDRs: []federationapi.ServerAddressByClientCIDR{
-					{
+				{
-						ClientCIDR:    "0.0.0.0/0",
+					ClientCIDR:    "0.0.0.0/0",
-						ServerAddress: host,
+					ServerAddress: host,
-					},
 				},
-				// Use insecure access
-				SecretRef: nil,
 			},
-		}
+			// Use insecure access
-		fedClient.FederationV1beta1().Clusters().Create(cluster)
+			SecretRef: nil,
+		},
 	}
+	f.fedClient.FederationV1beta1().Clusters().Create(cluster)
 }
 
 func (f *FederationFixture) TearDown(t *testing.T) {