Merge pull request #46071 from emaildanwilson/fedClusterSelectorIntegration

federation/pkg/federatedtypes/crudtester/crudtester.go

@@ -74,7 +74,7 @@ func (c *FederatedTypeCRUDTester) CheckLifecycle(desiredObject pkgruntime.Object
 	c.CheckDelete(obj, &orphanDependents)
 }
 
-func (c *FederatedTypeCRUDTester) CheckCreate(desiredObject pkgruntime.Object) pkgruntime.Object {
+func (c *FederatedTypeCRUDTester) Create(desiredObject pkgruntime.Object) pkgruntime.Object {
 	namespace := c.adapter.ObjectMeta(desiredObject).Namespace
 	c.tl.Logf("Creating new federated %s in namespace %q", c.kind, namespace)
 
@@ -86,6 +86,12 @@ func (c *FederatedTypeCRUDTester) CheckCreate(desiredObject pkgruntime.Object) p
 	namespacedName := c.adapter.NamespacedName(obj)
 	c.tl.Logf("Created new federated %s %q", c.kind, namespacedName)
 
+	return obj
+}
+
+func (c *FederatedTypeCRUDTester) CheckCreate(desiredObject pkgruntime.Object) pkgruntime.Object {
+	obj := c.Create(desiredObject)
+
 	c.CheckPropagation(obj)
 
 	return obj
@@ -165,18 +171,25 @@ 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)
+	c.CheckPropagationForClients(obj, c.clusterClients, true)
 }
 
 // CheckPropagationForClients checks propagation for the provided clients
-func (c *FederatedTypeCRUDTester) CheckPropagationForClients(obj pkgruntime.Object, clusterClients []clientset.Interface) {
+func (c *FederatedTypeCRUDTester) CheckPropagationForClients(obj pkgruntime.Object, clusterClients []clientset.Interface, objExpected bool) {
 	namespacedName := c.adapter.NamespacedName(obj)
 
 	c.tl.Logf("Waiting for %s %q in %d clusters", c.kind, namespacedName, len(clusterClients))
 	for _, client := range clusterClients {
 		err := c.waitForResource(client, obj)
-		if err != nil {
+		switch {
+		case err == wait.ErrWaitTimeout:
+			if objExpected {
+				c.tl.Fatalf("Timeout verifying %s %q in a member cluster: %v", c.kind, namespacedName, err)
+			}
+		case err != nil:
 			c.tl.Fatalf("Failed to verify %s %q in a member cluster: %v", c.kind, namespacedName, err)
+		case err == nil && !objExpected:
+			c.tl.Fatalf("Found unexpected object %s %q in a member cluster: %v", c.kind, namespacedName, err)
 		}
 	}
 }

test/integration/federation/crud_test.go

@@ -23,6 +23,7 @@ import (
 	"github.com/pborman/uuid"
 
 	pkgruntime "k8s.io/apimachinery/pkg/runtime"
+	federationapi "k8s.io/kubernetes/federation/apis/federation/v1beta1"
 	"k8s.io/kubernetes/federation/pkg/federatedtypes"
 	"k8s.io/kubernetes/federation/pkg/federatedtypes/crudtester"
 	"k8s.io/kubernetes/test/integration/federation/framework"
@@ -37,7 +38,7 @@ func TestFederationCRUD(t *testing.T) {
 	federatedTypes := federatedtypes.FederatedTypes()
 	for kind, fedType := range federatedTypes {
 		t.Run(kind, func(t *testing.T) {
-			fixture, crudTester, obj := initCRUDTest(t, &fedFixture, fedType.AdapterFactory, kind)
+			fixture, crudTester, obj, _ := initCRUDTest(t, &fedFixture, fedType.AdapterFactory, kind)
 			defer fixture.TearDown(t)
 
 			crudTester.CheckLifecycle(obj)
@@ -56,7 +57,7 @@ func TestFederationCRUD(t *testing.T) {
 	}
 	for testName, orphanDependents := range testCases {
 		t.Run(testName, func(t *testing.T) {
-			fixture, crudTester, obj := initCRUDTest(t, &fedFixture, adapterFactory, kind)
+			fixture, crudTester, obj, _ := initCRUDTest(t, &fedFixture, adapterFactory, kind)
 			defer fixture.TearDown(t)
 
 			updatedObj := crudTester.CheckCreate(obj)
@@ -65,20 +66,40 @@ func TestFederationCRUD(t *testing.T) {
 	}
 
 	t.Run("Resource should be propagated to a newly added cluster", func(t *testing.T) {
-		fixture, crudTester, obj := initCRUDTest(t, &fedFixture, adapterFactory, kind)
+		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)
+		objectExpected := true
+		crudTester.CheckPropagationForClients(updatedObj, fedFixture.ClusterClients, objectExpected)
+	})
+
+	t.Run("Resource should only be propagated to the cluster with a matching selector", func(t *testing.T) {
+		fixture, crudTester, obj, adapter := initCRUDTest(t, &fedFixture, adapterFactory, kind)
+		defer fixture.TearDown(t)
+
+		// Set an annotation to specify that the object is isolated to cluster 1.
+		federatedtypes.SetAnnotation(adapter, obj, federationapi.FederationClusterSelectorAnnotation, `[{"key": "cluster", "operator": "==", "values": ["1"]}]`)
+
+		updatedObj := crudTester.Create(obj)
+
+		// Check propagation to the first cluster
+		objectExpected := true
+		crudTester.CheckPropagationForClients(updatedObj, fedFixture.ClusterClients[0:1], objectExpected)
+
+		// Verify the object is not sent to the second cluster
+		objectExpected = false
+		crudTester.CheckPropagationForClients(updatedObj, fedFixture.ClusterClients[1:2], objectExpected)
+
 	})
 }
 
 // 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) {
+	*framework.ControllerFixture, *crudtester.FederatedTypeCRUDTester, pkgruntime.Object, federatedtypes.FederatedTypeAdapter) {
 	config := fedFixture.APIFixture.NewConfig()
 	fixture := framework.NewControllerFixture(t, kind, adapterFactory, config)
 
@@ -89,5 +110,5 @@ func initCRUDTest(t *testing.T, fedFixture *framework.FederationFixture, adapter
 
 	obj := adapter.NewTestObject(uuid.New())
 
-	return fixture, crudTester, obj
+	return fixture, crudTester, obj, adapter
 }

test/integration/federation/framework/federation.go

@@ -92,7 +92,8 @@ func (f *FederationFixture) StartCluster(t *testing.T) {
 
 	cluster := &federationapi.Cluster{
 		ObjectMeta: metav1.ObjectMeta{
-			Name: fmt.Sprintf("cluster-%d", clusterId),
+			Name:   fmt.Sprintf("cluster-%d", clusterId),
+			Labels: map[string]string{"cluster": fmt.Sprintf("%d", clusterId)},
 		},
 		Spec: federationapi.ClusterSpec{
 			ServerAddressByClientCIDRs: []federationapi.ServerAddressByClientCIDR{