Merge pull request #42025 from marun/fed-crud-interation-test

Automatic merge from submit-queue

[Federation] Add integration test for secrets

This PR adds an integration test for secrets that:

 - performs create/read/update/delete on federation resources and validates that the changes are propagated to member clusters.  
 - uses an abstraction layer (fixture and adapter) to minimize the code required to support each federated type
   - It should be possible to replace a test-specific adapter with a runtime adapter in the future (as per #41050)
 - reuses fixture (federation api and clusters) across different resource types to minimize setup overhead
   - on a fast machine, setup takes ~4s, and validating each type takes ~2s  
 - uses the [Subtest feature added in Go 1.7](https://blog.golang.org/subtests) to allow the test for a specific controller to be run in isolation
   - ``make test-integration WHAT="federation  -test.run=TestFederationCRUD/secret"``

Once this PR merges the test can be extended to target other federated types.

This PR targets #40705

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

federation/BUILD

@@ -29,6 +29,7 @@ filegroup(
         "//federation/pkg/dnsprovider:all-srcs",
         "//federation/pkg/federation-controller:all-srcs",
         "//federation/pkg/kubefed:all-srcs",
+        "//federation/pkg/typeadapters:all-srcs",
         "//federation/registry/cluster:all-srcs",
     ],
     tags = ["automanaged"],

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

@@ -156,9 +156,8 @@ func Run(s *options.CMServer) error {
 }
 
 func StartControllers(s *options.CMServer, restClientCfg *restclient.Config) error {
-	glog.Infof("Loading client config for cluster controller %q", "cluster-controller")
-	ccClientset := federationclientset.NewForConfigOrDie(restclient.AddUserAgent(restClientCfg, "cluster-controller"))
-	glog.Infof("Running cluster controller")
+	stopChan := wait.NeverStop
+	minimizeLatency := false
 
 	discoveryClient := discovery.NewDiscoveryClientForConfigOrDie(restClientCfg)
 	serverResources, err := discoveryClient.ServerResources()
@@ -166,7 +165,7 @@ func StartControllers(s *options.CMServer, restClientCfg *restclient.Config) err
 		glog.Fatalf("Could not find resources from API Server: %v", err)
 	}
 
-	go clustercontroller.NewclusterController(ccClientset, s.ClusterMonitorPeriod.Duration).Run()
+	clustercontroller.StartClusterController(restClientCfg, stopChan, s.ClusterMonitorPeriod.Duration)
 
 	if controllerEnabled(s.Controllers, serverResources, servicecontroller.ControllerName, servicecontroller.RequiredResources, true) {
 		dns, err := dnsprovider.InitDnsProvider(s.DnsProvider, s.DnsConfigFile)
@@ -191,9 +190,7 @@ func StartControllers(s *options.CMServer, restClientCfg *restclient.Config) err
 	}
 
 	if controllerEnabled(s.Controllers, serverResources, secretcontroller.ControllerName, secretcontroller.RequiredResources, true) {
-		secretcontrollerClientset := federationclientset.NewForConfigOrDie(restclient.AddUserAgent(restClientCfg, "secret-controller"))
-		secretcontroller := secretcontroller.NewSecretController(secretcontrollerClientset)
-		secretcontroller.Run(wait.NeverStop)
+		secretcontroller.StartSecretController(restClientCfg, stopChan, minimizeLatency)
 	}
 
 	if controllerEnabled(s.Controllers, serverResources, configmapcontroller.ControllerName, configmapcontroller.RequiredResources, true) {

federation/pkg/federation-controller/cluster/clustercontroller.go

@@ -27,6 +27,7 @@ import (
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/apimachinery/pkg/watch"
+	restclient "k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/cache"
 	federationv1beta1 "k8s.io/kubernetes/federation/apis/federation/v1beta1"
 	clustercache "k8s.io/kubernetes/federation/client/cache"
@@ -53,8 +54,17 @@ type ClusterController struct {
 	clusterStore      clustercache.StoreToClusterLister
 }
 
-// NewclusterController returns a new cluster controller
-func NewclusterController(federationClient federationclientset.Interface, clusterMonitorPeriod time.Duration) *ClusterController {
+// StartClusterController starts a new cluster controller
+func StartClusterController(config *restclient.Config, stopChan <-chan struct{}, clusterMonitorPeriod time.Duration) {
+	restclient.AddUserAgent(config, "cluster-controller")
+	client := federationclientset.NewForConfigOrDie(config)
+	controller := newClusterController(client, clusterMonitorPeriod)
+	glog.Infof("Starting cluster controller")
+	controller.Run(stopChan)
+}
+
+// newClusterController returns a new cluster controller
+func newClusterController(federationClient federationclientset.Interface, clusterMonitorPeriod time.Duration) *ClusterController {
 	cc := &ClusterController{
 		knownClusterSet:         make(sets.String),
 		federationClient:        federationClient,
@@ -112,15 +122,15 @@ func (cc *ClusterController) addToClusterSet(obj interface{}) {
 }
 
 // Run begins watching and syncing.
-func (cc *ClusterController) Run() {
+func (cc *ClusterController) Run(stopChan <-chan struct{}) {
 	defer utilruntime.HandleCrash()
-	go cc.clusterController.Run(wait.NeverStop)
+	go cc.clusterController.Run(stopChan)
 	// monitor cluster status periodically, in phase 1 we just get the health state from "/healthz"
 	go wait.Until(func() {
 		if err := cc.UpdateClusterStatus(); err != nil {
 			glog.Errorf("Error monitoring cluster status: %v", err)
 		}
-	}, cc.clusterMonitorPeriod, wait.NeverStop)
+	}, cc.clusterMonitorPeriod, stopChan)
 }
 
 func (cc *ClusterController) GetClusterClient(cluster *federationv1beta1.Cluster) (*ClusterClient, error) {

federation/pkg/federation-controller/cluster/clustercontroller_test.go

@@ -132,7 +132,7 @@ func TestUpdateClusterStatusOK(t *testing.T) {
 		}
 	}
 
-	manager := NewclusterController(federationClientSet, 5)
+	manager := newClusterController(federationClientSet, 5)
 	err = manager.UpdateClusterStatus()
 	if err != nil {
 		t.Errorf("Failed to Update Cluster Status: %v", err)

federation/pkg/federation-controller/secret/BUILD

@@ -30,6 +30,7 @@ go_library(
         "//vendor:k8s.io/apimachinery/pkg/types",
         "//vendor:k8s.io/apimachinery/pkg/watch",
         "//vendor:k8s.io/client-go/pkg/api/v1",
+        "//vendor:k8s.io/client-go/rest",
         "//vendor:k8s.io/client-go/tools/cache",
         "//vendor:k8s.io/client-go/tools/record",
         "//vendor:k8s.io/client-go/util/flowcontrol",

federation/pkg/federation-controller/secret/secret_controller.go

@@ -27,6 +27,7 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/watch"
 	clientv1 "k8s.io/client-go/pkg/api/v1"
+	restclient "k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/client-go/tools/record"
 	"k8s.io/client-go/util/flowcontrol"
@@ -88,8 +89,20 @@ type SecretController struct {
 	updateTimeout         time.Duration
 }
 
-// NewSecretController returns a new secret controller
-func NewSecretController(client federationclientset.Interface) *SecretController {
+// StartSecretController starts a new secret controller
+func StartSecretController(config *restclient.Config, stopChan <-chan struct{}, minimizeLatency bool) {
+	restclient.AddUserAgent(config, "secret-controller")
+	client := federationclientset.NewForConfigOrDie(config)
+	controller := newSecretController(client)
+	if minimizeLatency {
+		controller.minimizeLatency()
+	}
+	glog.Infof("Starting Secret controller")
+	controller.Run(stopChan)
+}
+
+// newSecretController returns a new secret controller
+func newSecretController(client federationclientset.Interface) *SecretController {
 	broadcaster := record.NewBroadcaster()
 	broadcaster.StartRecordingToSink(eventsink.NewFederatedEventSink(client))
 	recorder := broadcaster.NewRecorder(api.Scheme, clientv1.EventSource{Component: "federated-secrets-controller"})
@@ -191,6 +204,14 @@ func NewSecretController(client federationclientset.Interface) *SecretController
 	return secretcontroller
 }
 
+// minimizeLatency reduces delays and timeouts to make the controller more responsive (useful for testing).
+func (secretcontroller *SecretController) minimizeLatency() {
+	secretcontroller.clusterAvailableDelay = time.Second
+	secretcontroller.secretReviewDelay = 50 * time.Millisecond
+	secretcontroller.smallDelay = 20 * time.Millisecond
+	secretcontroller.updateTimeout = 5 * time.Second
+}
+
 // Returns true if the given object has the given finalizer in its ObjectMeta.
 func (secretcontroller *SecretController) hasFinalizerFunc(obj pkgruntime.Object, finalizer string) bool {
 	secret := obj.(*apiv1.Secret)

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

@@ -66,7 +66,7 @@ func TestSecretController(t *testing.T) {
 	RegisterFakeList(secrets, &cluster2Client.Fake, &apiv1.SecretList{Items: []apiv1.Secret{}})
 	cluster2CreateChan := RegisterFakeCopyOnCreate(secrets, &cluster2Client.Fake, cluster2Watch)
 
-	secretController := NewSecretController(fakeClient)
+	secretController := newSecretController(fakeClient)
 	informerClientFactory := func(cluster *federationapi.Cluster) (kubeclientset.Interface, error) {
 		switch cluster.Name {
 		case cluster1.Name:
@@ -79,10 +79,7 @@ func TestSecretController(t *testing.T) {
 	}
 	setClientFactory(secretController.secretFederatedInformer, informerClientFactory)
 
-	secretController.clusterAvailableDelay = time.Second
-	secretController.secretReviewDelay = 50 * time.Millisecond
-	secretController.smallDelay = 20 * time.Millisecond
-	secretController.updateTimeout = 5 * time.Second
+	secretController.minimizeLatency()
 
 	stop := make(chan struct{})
 	secretController.Run(stop)

federation/pkg/typeadapters/BUILD

@@ -0,0 +1,42 @@
+package(default_visibility = ["//visibility:public"])
+
+licenses(["notice"])
+
+load(
+    "@io_bazel_rules_go//go:def.bzl",
+    "go_library",
+)
+
+go_library(
+    name = "go_default_library",
+    srcs = [
+        "adapter.go",
+        "secret.go",
+    ],
+    tags = ["automanaged"],
+    deps = [
+        "//federation/client/clientset_generated/federation_clientset:go_default_library",
+        "//federation/pkg/federation-controller/util:go_default_library",
+        "//pkg/api/v1:go_default_library",
+        "//pkg/client/clientset_generated/clientset:go_default_library",
+        "//vendor:k8s.io/apimachinery/pkg/apis/meta/v1",
+        "//vendor:k8s.io/apimachinery/pkg/runtime",
+        "//vendor:k8s.io/apimachinery/pkg/types",
+    ],
+)
+
+filegroup(
+    name = "package-srcs",
+    srcs = glob(["**"]),
+    tags = ["automanaged"],
+    visibility = ["//visibility:private"],
+)
+
+filegroup(
+    name = "all-srcs",
+    srcs = [
+        ":package-srcs",
+        "//federation/pkg/typeadapters/crudtester:all-srcs",
+    ],
+    tags = ["automanaged"],
+)

federation/pkg/typeadapters/adapter.go

@@ -0,0 +1,48 @@
+/*
+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 typeadapters
+
+import (
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	pkgruntime "k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
+	federationclientset "k8s.io/kubernetes/federation/client/clientset_generated/federation_clientset"
+	"k8s.io/kubernetes/pkg/client/clientset_generated/clientset"
+)
+
+// FederatedTypeAdapter defines operations for interacting with a
+// federated type.  Code written to this interface can then target any
+// type for which an implementation of this interface exists.
+type FederatedTypeAdapter interface {
+	SetClient(client federationclientset.Interface)
+
+	Kind() string
+	Equivalent(obj1, obj2 pkgruntime.Object) bool
+	ObjectMeta(obj pkgruntime.Object) *metav1.ObjectMeta
+	NamespacedName(obj pkgruntime.Object) types.NamespacedName
+
+	// Fed* operations target the federation control plane
+	FedCreate(obj pkgruntime.Object) (pkgruntime.Object, error)
+	FedGet(namespacedName types.NamespacedName) (pkgruntime.Object, error)
+	FedUpdate(obj pkgruntime.Object) (pkgruntime.Object, error)
+	FedDelete(namespacedName types.NamespacedName, options *metav1.DeleteOptions) error
+
+	// The following operations are intended to target a cluster that is a member of a federation
+	ClusterGet(client clientset.Interface, namespacedName types.NamespacedName) (pkgruntime.Object, error)
+
+	NewTestObject(namespace string) pkgruntime.Object
+}

federation/pkg/typeadapters/crudtester/BUILD

@@ -0,0 +1,35 @@
+package(default_visibility = ["//visibility:public"])
+
+licenses(["notice"])
+
+load(
+    "@io_bazel_rules_go//go:def.bzl",
+    "go_library",
+)
+
+go_library(
+    name = "go_default_library",
+    srcs = ["crudtester.go"],
+    tags = ["automanaged"],
+    deps = [
+        "//federation/pkg/typeadapters:go_default_library",
+        "//pkg/client/clientset_generated/clientset:go_default_library",
+        "//vendor:k8s.io/apimachinery/pkg/api/errors",
+        "//vendor:k8s.io/apimachinery/pkg/apis/meta/v1",
+        "//vendor:k8s.io/apimachinery/pkg/runtime",
+        "//vendor:k8s.io/apimachinery/pkg/util/wait",
+    ],
+)
+
+filegroup(
+    name = "package-srcs",
+    srcs = glob(["**"]),
+    tags = ["automanaged"],
+    visibility = ["//visibility:private"],
+)
+
+filegroup(
+    name = "all-srcs",
+    srcs = [":package-srcs"],
+    tags = ["automanaged"],
+)

federation/pkg/typeadapters/crudtester/crudtester.go

@@ -0,0 +1,217 @@
+/*
+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 crudtester
+
+import (
+	"time"
+
+	"k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	pkgruntime "k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/kubernetes/federation/pkg/typeadapters"
+	"k8s.io/kubernetes/pkg/client/clientset_generated/clientset"
+)
+
+const (
+	AnnotationTestFederationCRUDUpdate string = "federation.kubernetes.io/test-federation-crud-update"
+)
+
+// TestLogger defines operations common across different types of testing
+type TestLogger interface {
+	Fatalf(format string, args ...interface{})
+	Fatal(msg string)
+	Logf(format string, args ...interface{})
+}
+
+// FederatedTypeCRUDTester exercises Create/Read/Update/Delete operations for
+// federated types via the Federation API and validates that the
+// results of those operations are propagated to clusters that are
+// members of a federation.
+type FederatedTypeCRUDTester struct {
+	tl             TestLogger
+	adapter        typeadapters.FederatedTypeAdapter
+	kind           string
+	clusterClients []clientset.Interface
+	waitInterval   time.Duration
+	// Federation operations will use wait.ForeverTestTimeout.  Any
+	// operation that involves member clusters may take longer due to
+	// propagation latency.
+	clusterWaitTimeout time.Duration
+}
+
+func NewFederatedTypeCRUDTester(testLogger TestLogger, adapter typeadapters.FederatedTypeAdapter, clusterClients []clientset.Interface, waitInterval, clusterWaitTimeout time.Duration) *FederatedTypeCRUDTester {
+	return &FederatedTypeCRUDTester{
+		tl:                 testLogger,
+		adapter:            adapter,
+		kind:               adapter.Kind(),
+		clusterClients:     clusterClients,
+		waitInterval:       waitInterval,
+		clusterWaitTimeout: clusterWaitTimeout,
+	}
+}
+
+func (c *FederatedTypeCRUDTester) CheckLifecycle(desiredObject pkgruntime.Object) {
+	obj := c.CheckCreate(desiredObject)
+	c.CheckUpdate(obj)
+
+	// Validate the golden path - removal of dependents
+	orphanDependents := false
+	c.CheckDelete(obj, &orphanDependents)
+}
+
+func (c *FederatedTypeCRUDTester) CheckCreate(desiredObject pkgruntime.Object) pkgruntime.Object {
+	namespace := c.adapter.ObjectMeta(desiredObject).Namespace
+	c.tl.Logf("Creating new federated %s in namespace %q", c.kind, namespace)
+
+	obj, err := c.adapter.FedCreate(desiredObject)
+	if err != nil {
+		c.tl.Fatalf("Error creating federated %s in namespace %q : %v", c.kind, namespace, err)
+	}
+
+	namespacedName := c.adapter.NamespacedName(obj)
+	c.tl.Logf("Created new federated %s %q", c.kind, namespacedName)
+
+	c.CheckPropagation(obj)
+
+	return obj
+}
+
+func (c *FederatedTypeCRUDTester) CheckUpdate(obj pkgruntime.Object) {
+	namespacedName := c.adapter.NamespacedName(obj)
+
+	var initialAnnotation string
+	meta := c.adapter.ObjectMeta(obj)
+	if meta.Annotations != nil {
+		initialAnnotation = meta.Annotations[AnnotationTestFederationCRUDUpdate]
+	}
+
+	c.tl.Logf("Updating federated %s %q", c.kind, namespacedName)
+	updatedObj, err := c.updateFedObject(obj)
+	if err != nil {
+		c.tl.Fatalf("Error updating federated %s %q: %v", c.kind, namespacedName, err)
+	}
+
+	// updateFedObject is expected to have changed the value of the annotation
+	meta = c.adapter.ObjectMeta(updatedObj)
+	updatedAnnotation := meta.Annotations[AnnotationTestFederationCRUDUpdate]
+	if updatedAnnotation == initialAnnotation {
+		c.tl.Fatalf("Federated %s %q not mutated", c.kind, namespacedName)
+	}
+
+	c.CheckPropagation(updatedObj)
+}
+
+func (c *FederatedTypeCRUDTester) CheckDelete(obj pkgruntime.Object, orphanDependents *bool) {
+	namespacedName := c.adapter.NamespacedName(obj)
+
+	c.tl.Logf("Deleting federated %s %q", c.kind, namespacedName)
+	err := c.adapter.FedDelete(namespacedName, &metav1.DeleteOptions{OrphanDependents: orphanDependents})
+	if err != nil {
+		c.tl.Fatalf("Error deleting federated %s %q: %v", c.kind, namespacedName, err)
+	}
+
+	deletingInCluster := (orphanDependents != nil && *orphanDependents == false)
+
+	waitTimeout := wait.ForeverTestTimeout
+	if deletingInCluster {
+		// May need extra time to delete both federation and cluster resources
+		waitTimeout = c.clusterWaitTimeout
+	}
+
+	// Wait for deletion.  The federation resource will only be removed once orphan deletion has been
+	// completed or deemed unnecessary.
+	err = wait.PollImmediate(c.waitInterval, waitTimeout, func() (bool, error) {
+		_, err := c.adapter.FedGet(namespacedName)
+		if errors.IsNotFound(err) {
+			return true, nil
+		}
+		return false, err
+	})
+	if err != nil {
+		c.tl.Fatalf("Error deleting federated %s %q: %v", c.kind, namespacedName, err)
+	}
+
+	var stateMsg string = "present"
+	if deletingInCluster {
+		stateMsg = "not present"
+	}
+	for _, client := range c.clusterClients {
+		_, err := c.adapter.ClusterGet(client, namespacedName)
+		switch {
+		case !deletingInCluster && errors.IsNotFound(err):
+			c.tl.Fatalf("Federated %s %q was unexpectedly deleted from a member cluster", c.kind, namespacedName)
+		case deletingInCluster && err == nil:
+			c.tl.Fatalf("Federated %s %q was unexpectedly orphaned in a member cluster", c.kind, namespacedName)
+		case err != nil && !errors.IsNotFound(err):
+			c.tl.Fatalf("Error while checking whether %s %q is %s in member clusters: %v", c.kind, namespacedName, stateMsg, err)
+		}
+	}
+}
+
+func (c *FederatedTypeCRUDTester) CheckPropagation(obj pkgruntime.Object) {
+	namespacedName := c.adapter.NamespacedName(obj)
+
+	c.tl.Logf("Waiting for %s %q in %d clusters", c.kind, namespacedName, len(c.clusterClients))
+	for _, client := range c.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)
+		}
+	}
+}
+
+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) {
+		clusterObj, err := c.adapter.ClusterGet(client, namespacedName)
+		if err == nil && c.adapter.Equivalent(clusterObj, obj) {
+			return true, nil
+		}
+		if errors.IsNotFound(err) {
+			return false, nil
+		}
+		return false, err
+	})
+	return err
+}
+
+func (c *FederatedTypeCRUDTester) updateFedObject(obj pkgruntime.Object) (pkgruntime.Object, error) {
+	err := wait.PollImmediate(c.waitInterval, wait.ForeverTestTimeout, func() (bool, error) {
+		// Target the metadata for simplicity (it's type-agnostic)
+		meta := c.adapter.ObjectMeta(obj)
+		if meta.Annotations == nil {
+			meta.Annotations = make(map[string]string)
+		}
+		meta.Annotations[AnnotationTestFederationCRUDUpdate] = "updated"
+
+		_, err := c.adapter.FedUpdate(obj)
+		if errors.IsConflict(err) {
+			// The resource was updated by the federation controller.
+			// Get the latest version and retry.
+			namespacedName := c.adapter.NamespacedName(obj)
+			obj, err = c.adapter.FedGet(namespacedName)
+			return false, err
+		}
+		// Be tolerant of a slow server
+		if errors.IsServerTimeout(err) {
+			return false, nil
+		}
+		return (err == nil), err
+	})
+	return obj, err
+}

federation/pkg/typeadapters/secret.go

@@ -0,0 +1,93 @@
+/*
+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 typeadapters
+
+import (
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	pkgruntime "k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
+	federationclientset "k8s.io/kubernetes/federation/client/clientset_generated/federation_clientset"
+	"k8s.io/kubernetes/federation/pkg/federation-controller/util"
+	apiv1 "k8s.io/kubernetes/pkg/api/v1"
+	"k8s.io/kubernetes/pkg/client/clientset_generated/clientset"
+)
+
+type SecretAdapter struct {
+	client federationclientset.Interface
+}
+
+func NewSecretAdapter(client federationclientset.Interface) *SecretAdapter {
+	return &SecretAdapter{client: client}
+}
+
+func (a *SecretAdapter) SetClient(client federationclientset.Interface) {
+	a.client = client
+}
+
+func (a *SecretAdapter) Kind() string {
+	return "secret"
+}
+
+func (a *SecretAdapter) Equivalent(obj1, obj2 pkgruntime.Object) bool {
+	secret1 := obj1.(*apiv1.Secret)
+	secret2 := obj2.(*apiv1.Secret)
+	return util.SecretEquivalent(*secret1, *secret2)
+}
+
+func (a *SecretAdapter) ObjectMeta(obj pkgruntime.Object) *metav1.ObjectMeta {
+	return &obj.(*apiv1.Secret).ObjectMeta
+}
+
+func (a *SecretAdapter) NamespacedName(obj pkgruntime.Object) types.NamespacedName {
+	secret := obj.(*apiv1.Secret)
+	return types.NamespacedName{Namespace: secret.Namespace, Name: secret.Name}
+}
+
+func (a *SecretAdapter) FedCreate(obj pkgruntime.Object) (pkgruntime.Object, error) {
+	secret := obj.(*apiv1.Secret)
+	return a.client.CoreV1().Secrets(secret.Namespace).Create(secret)
+}
+
+func (a *SecretAdapter) FedGet(namespacedName types.NamespacedName) (pkgruntime.Object, error) {
+	return a.client.CoreV1().Secrets(namespacedName.Namespace).Get(namespacedName.Name, metav1.GetOptions{})
+}
+
+func (a *SecretAdapter) FedUpdate(obj pkgruntime.Object) (pkgruntime.Object, error) {
+	secret := obj.(*apiv1.Secret)
+	return a.client.CoreV1().Secrets(secret.Namespace).Update(secret)
+}
+
+func (a *SecretAdapter) FedDelete(namespacedName types.NamespacedName, options *metav1.DeleteOptions) error {
+	return a.client.CoreV1().Secrets(namespacedName.Namespace).Delete(namespacedName.Name, options)
+}
+
+func (a *SecretAdapter) ClusterGet(client clientset.Interface, namespacedName types.NamespacedName) (pkgruntime.Object, error) {
+	return client.CoreV1().Secrets(namespacedName.Namespace).Get(namespacedName.Name, metav1.GetOptions{})
+}
+
+func (a *SecretAdapter) NewTestObject(namespace string) pkgruntime.Object {
+	return &apiv1.Secret{
+		ObjectMeta: metav1.ObjectMeta{
+			GenerateName: "test-secret-",
+			Namespace:    namespace,
+		},
+		Data: map[string][]byte{
+			"A": []byte("ala ma kota"),
+		},
+		Type: apiv1.SecretTypeOpaque,
+	}
+}

test/integration/federation/BUILD

@@ -9,7 +9,10 @@ load(
 
 go_test(
     name = "go_default_test",
-    srcs = ["api_test.go"],
+    srcs = [
+        "api_test.go",
+        "crud_test.go",
+    ],
     tags = ["automanaged"],
     deps = [
         "//federation/apis/federation/v1beta1:go_default_library",
@@ -18,6 +21,7 @@ go_test(
         "//pkg/apis/batch/v1:go_default_library",
         "//pkg/apis/extensions/v1beta1:go_default_library",
         "//test/integration/federation/framework:go_default_library",
+        "//vendor:github.com/pborman/uuid",
         "//vendor:github.com/stretchr/testify/assert",
         "//vendor:k8s.io/apimachinery/pkg/apis/meta/v1",
         "//vendor:k8s.io/apimachinery/pkg/runtime/schema",

test/integration/federation/api_test.go

@@ -46,8 +46,8 @@ type apiTestFunc func(t *testing.T, host string)
 
 func TestFederationAPI(t *testing.T) {
 	f := &framework.FederationAPIFixture{}
-	f.Setup(t)
-	defer f.Teardown(t)
+	f.SetUp(t)
+	defer f.TearDown(t)
 
 	testCases := map[string]apiTestFunc{
 		"swaggerSpec":     testSwaggerSpec,

test/integration/federation/crud_test.go

@@ -0,0 +1,68 @@
+/*
+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 federation
+
+import (
+	"testing"
+
+	"github.com/pborman/uuid"
+
+	"k8s.io/kubernetes/test/integration/federation/framework"
+)
+
+// TestFederationCRUD validates create/read/update/delete operations for federated resource types.
+func TestFederationCRUD(t *testing.T) {
+	fedFixture := framework.FederationFixture{DesiredClusterCount: 2}
+	fedFixture.SetUp(t)
+	defer fedFixture.TearDown(t)
+
+	controllerFixtures := []framework.ControllerFixture{
+		&framework.SecretFixture{},
+	}
+	for _, fixture := range controllerFixtures {
+		t.Run(fixture.Kind(), func(t *testing.T) {
+			framework.SetUpControllerFixture(t, fedFixture.APIFixture, fixture)
+			defer fixture.TearDown(t)
+
+			adapter := fixture.Adapter()
+			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
+	// 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,
+	}
+	for testName, orphanDependents := range testCases {
+		t.Run(testName, func(t *testing.T) {
+			fixture := &framework.SecretFixture{}
+			framework.SetUpControllerFixture(t, fedFixture.APIFixture, fixture)
+			defer fixture.TearDown(t)
+
+			adapter := fixture.Adapter()
+			crudtester := framework.NewFederatedTypeCRUDTester(t, adapter, fedFixture.ClusterClients)
+			obj := adapter.NewTestObject(uuid.New())
+			updatedObj := crudtester.CheckCreate(obj)
+			crudtester.CheckDelete(updatedObj, orphanDependents)
+		})
+	}
+}

test/integration/federation/framework/BUILD

@@ -11,15 +11,29 @@ go_library(
     name = "go_default_library",
     srcs = [
         "api.go",
+        "controller.go",
+        "crudtester.go",
+        "federation.go",
+        "secret.go",
         "util.go",
     ],
     tags = ["automanaged"],
     deps = [
+        "//federation/apis/federation/v1beta1:go_default_library",
+        "//federation/client/clientset_generated/federation_clientset:go_default_library",
         "//federation/cmd/federation-apiserver/app:go_default_library",
         "//federation/cmd/federation-apiserver/app/options:go_default_library",
+        "//federation/pkg/federation-controller/cluster:go_default_library",
+        "//federation/pkg/federation-controller/secret:go_default_library",
+        "//federation/pkg/typeadapters:go_default_library",
+        "//federation/pkg/typeadapters/crudtester:go_default_library",
+        "//pkg/client/clientset_generated/clientset:go_default_library",
+        "//pkg/master:go_default_library",
         "//test/integration/framework:go_default_library",
         "//vendor:github.com/pborman/uuid",
+        "//vendor:k8s.io/apimachinery/pkg/apis/meta/v1",
         "//vendor:k8s.io/apimachinery/pkg/util/wait",
+        "//vendor:k8s.io/client-go/rest",
     ],
 )
 

test/integration/federation/framework/api.go

@@ -20,20 +20,18 @@ import (
 	"fmt"
 	"net/http"
 	"testing"
-	"time"
 
 	"github.com/pborman/uuid"
 
 	"k8s.io/apimachinery/pkg/util/wait"
+	restclient "k8s.io/client-go/rest"
+	federationclientset "k8s.io/kubernetes/federation/client/clientset_generated/federation_clientset"
 	"k8s.io/kubernetes/federation/cmd/federation-apiserver/app"
 	"k8s.io/kubernetes/federation/cmd/federation-apiserver/app/options"
 	"k8s.io/kubernetes/test/integration/framework"
 )
 
-const (
-	apiNoun      = "federation apiserver"
-	waitInterval = 50 * time.Millisecond
-)
+const apiNoun = "federation apiserver"
 
 func getRunOptions() *options.ServerRunOptions {
 	r := options.NewServerRunOptions()
@@ -51,11 +49,11 @@ type FederationAPIFixture struct {
 	stopChan chan struct{}
 }
 
-func (f *FederationAPIFixture) Setup(t *testing.T) {
+func (f *FederationAPIFixture) SetUp(t *testing.T) {
 	if f.stopChan != nil {
-		t.Fatal("Setup() already called")
+		t.Fatal("SetUp() already called")
 	}
-	defer TeardownOnPanic(t, f)
+	defer TearDownOnPanic(t, f)
 
 	f.stopChan = make(chan struct{})
 
@@ -74,15 +72,25 @@ func (f *FederationAPIFixture) Setup(t *testing.T) {
 	}
 }
 
-func (f *FederationAPIFixture) Teardown(t *testing.T) {
+func (f *FederationAPIFixture) TearDown(t *testing.T) {
 	if f.stopChan != nil {
 		close(f.stopChan)
 		f.stopChan = nil
 	}
 }
 
+func (f *FederationAPIFixture) NewConfig() *restclient.Config {
+	return &restclient.Config{Host: f.Host}
+}
+
+func (f *FederationAPIFixture) NewClient(userAgent string) federationclientset.Interface {
+	config := f.NewConfig()
+	restclient.AddUserAgent(config, userAgent)
+	return federationclientset.NewForConfigOrDie(config)
+}
+
 func startServer(t *testing.T, runOptions *options.ServerRunOptions, stopChan <-chan struct{}) error {
-	err := wait.PollImmediate(waitInterval, wait.ForeverTestTimeout, func() (bool, error) {
+	err := wait.PollImmediate(DefaultWaitInterval, wait.ForeverTestTimeout, func() (bool, error) {
 		port, err := framework.FindFreeLocalPort()
 		if err != nil {
 			t.Logf("Error allocating an ephemeral port: %v", err)
@@ -105,7 +113,7 @@ func startServer(t *testing.T, runOptions *options.ServerRunOptions, stopChan <-
 }
 
 func waitForServer(t *testing.T, host string) error {
-	err := wait.PollImmediate(waitInterval, wait.ForeverTestTimeout, func() (bool, error) {
+	err := wait.PollImmediate(DefaultWaitInterval, wait.ForeverTestTimeout, func() (bool, error) {
 		_, err := http.Get(host)
 		if err != nil {
 			t.Logf("Error when trying to contact the API: %v", err)

test/integration/federation/framework/controller.go

@@ -0,0 +1,46 @@
+/*
+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 framework
+
+import (
+	"fmt"
+	"testing"
+
+	restclient "k8s.io/client-go/rest"
+	federationclientset "k8s.io/kubernetes/federation/client/clientset_generated/federation_clientset"
+	"k8s.io/kubernetes/federation/pkg/typeadapters"
+)
+
+// ControllerFixture defines operations for managing a federation
+// controller.  Tests written to this interface can then target any
+// controller for which an implementation of this interface exists.
+type ControllerFixture interface {
+	TestFixture
+
+	SetUp(t *testing.T, testClient federationclientset.Interface, config *restclient.Config)
+
+	Kind() string
+
+	Adapter() typeadapters.FederatedTypeAdapter
+}
+
+// SetUpControllerFixture configures the given resource fixture to target the provided api fixture
+func SetUpControllerFixture(t *testing.T, apiFixture *FederationAPIFixture, controllerFixture ControllerFixture) {
+	client := apiFixture.NewClient(fmt.Sprintf("test-%s", controllerFixture.Kind()))
+	config := apiFixture.NewConfig()
+	controllerFixture.SetUp(t, client, config)
+}

test/integration/federation/framework/crudtester.go

@@ -0,0 +1,47 @@
+/*
+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 framework
+
+import (
+	"testing"
+
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/kubernetes/federation/pkg/typeadapters"
+	"k8s.io/kubernetes/federation/pkg/typeadapters/crudtester"
+	"k8s.io/kubernetes/pkg/client/clientset_generated/clientset"
+)
+
+type IntegrationLogger struct {
+	t *testing.T
+}
+
+func (l *IntegrationLogger) Logf(format string, args ...interface{}) {
+	l.t.Logf(format, args...)
+}
+
+func (l *IntegrationLogger) Fatalf(format string, args ...interface{}) {
+	l.t.Fatalf(format, args...)
+}
+
+func (l *IntegrationLogger) Fatal(msg string) {
+	l.t.Fatal(msg)
+}
+
+func NewFederatedTypeCRUDTester(t *testing.T, adapter typeadapters.FederatedTypeAdapter, clusterClients []clientset.Interface) *crudtester.FederatedTypeCRUDTester {
+	logger := &IntegrationLogger{t}
+	return crudtester.NewFederatedTypeCRUDTester(logger, adapter, clusterClients, DefaultWaitInterval, wait.ForeverTestTimeout)
+}

test/integration/federation/framework/federation.go

@@ -0,0 +1,123 @@
+/*
+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 framework
+
+import (
+	"fmt"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	federationapi "k8s.io/kubernetes/federation/apis/federation/v1beta1"
+	clustercontroller "k8s.io/kubernetes/federation/pkg/federation-controller/cluster"
+	"k8s.io/kubernetes/pkg/client/clientset_generated/clientset"
+	"k8s.io/kubernetes/pkg/master"
+	"k8s.io/kubernetes/test/integration/framework"
+)
+
+type MemberCluster struct {
+	Server *httptest.Server
+	Config *master.Config
+	Client clientset.Interface
+	Host   string
+}
+
+// FederationFixture manages a federation api server and a set of member clusters
+type FederationFixture struct {
+	APIFixture          *FederationAPIFixture
+	DesiredClusterCount int
+	Clusters            []*MemberCluster
+	ClusterClients      []clientset.Interface
+	ClusterController   *clustercontroller.ClusterController
+	stopChan            chan struct{}
+}
+
+func (f *FederationFixture) SetUp(t *testing.T) {
+	if f.APIFixture != nil {
+		t.Fatal("Fixture already started")
+	}
+	if f.DesiredClusterCount < 1 {
+		f.DesiredClusterCount = 1
+	}
+	defer TearDownOnPanic(t, f)
+
+	t.Logf("Starting a federation of %d clusters", f.DesiredClusterCount)
+
+	f.APIFixture = &FederationAPIFixture{}
+	f.APIFixture.SetUp(t)
+
+	f.stopChan = make(chan struct{})
+	monitorPeriod := 1 * time.Second
+	clustercontroller.StartClusterController(f.APIFixture.NewConfig(), f.stopChan, monitorPeriod)
+
+	f.startClusters()
+}
+
+func (f *FederationFixture) startClusters() {
+	fedClient := f.APIFixture.NewClient("federation-fixture")
+	for i := 0; i < f.DesiredClusterCount; i++ {
+		config := framework.NewMasterConfig()
+		_, server := framework.RunAMaster(config)
+		host := config.GenericConfig.LoopbackClientConfig.Host
+
+		// Use fmt to ensure the output will be visible when run with go test -v
+		fmt.Printf("Federated cluster %d serving on %s", i, host)
+
+		clusterClient := clientset.NewForConfigOrDie(config.GenericConfig.LoopbackClientConfig)
+		f.Clusters = append(f.Clusters, &MemberCluster{
+			Server: server,
+			Config: config,
+			Client: clusterClient,
+			Host:   host,
+		})
+
+		f.ClusterClients = append(f.ClusterClients, clusterClient)
+
+		cluster := &federationapi.Cluster{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: fmt.Sprintf("cluster-%d", i),
+			},
+			Spec: federationapi.ClusterSpec{
+				ServerAddressByClientCIDRs: []federationapi.ServerAddressByClientCIDR{
+					{
+						ClientCIDR:    "0.0.0.0/0",
+						ServerAddress: host,
+					},
+				},
+				// Use insecure access
+				SecretRef: nil,
+			},
+		}
+		fedClient.FederationV1beta1().Clusters().Create(cluster)
+	}
+}
+
+func (f *FederationFixture) TearDown(t *testing.T) {
+	if f.stopChan != nil {
+		close(f.stopChan)
+		f.stopChan = nil
+	}
+	for _, cluster := range f.Clusters {
+		cluster.Server.Close()
+	}
+	f.Clusters = nil
+	if f.APIFixture != nil {
+		f.APIFixture.TearDown(t)
+		f.APIFixture = nil
+	}
+}

test/integration/federation/framework/secret.go

@@ -0,0 +1,49 @@
+/*
+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 framework
+
+import (
+	"testing"
+
+	restclient "k8s.io/client-go/rest"
+	federationclientset "k8s.io/kubernetes/federation/client/clientset_generated/federation_clientset"
+	secretcontroller "k8s.io/kubernetes/federation/pkg/federation-controller/secret"
+	"k8s.io/kubernetes/federation/pkg/typeadapters"
+)
+
+type SecretFixture struct {
+	adapter  *typeadapters.SecretAdapter
+	stopChan chan struct{}
+}
+
+func (f *SecretFixture) SetUp(t *testing.T, client federationclientset.Interface, config *restclient.Config) {
+	f.adapter = typeadapters.NewSecretAdapter(client)
+	f.stopChan = make(chan struct{})
+	secretcontroller.StartSecretController(config, f.stopChan, true)
+}
+
+func (f *SecretFixture) TearDown(t *testing.T) {
+	close(f.stopChan)
+}
+func (f *SecretFixture) Kind() string {
+	adapter := &typeadapters.SecretAdapter{}
+	return adapter.Kind()
+}
+
+func (f *SecretFixture) Adapter() typeadapters.FederatedTypeAdapter {
+	return f.adapter
+}

test/integration/federation/framework/util.go

@@ -18,18 +18,23 @@ package framework
 
 import (
 	"testing"
+	"time"
 )
 
-// Setup is likely to be fixture-specific, but Teardown needs to be
-// consistent to enable TeardownOnPanic.
+const (
+	DefaultWaitInterval = 50 * time.Millisecond
+)
+
+// SetUp is likely to be fixture-specific, but TearDown needs to be
+// consistent to enable TearDownOnPanic.
 type TestFixture interface {
-	Teardown(t *testing.T)
+	TearDown(t *testing.T)
 }
 
-// TeardownOnPanic can be used to ensure cleanup on setup failure.
-func TeardownOnPanic(t *testing.T, f TestFixture) {
+// TearDownOnPanic can be used to ensure cleanup on setup failure.
+func TearDownOnPanic(t *testing.T, f TestFixture) {
 	if r := recover(); r != nil {
-		f.Teardown(t)
+		f.TearDown(t)
 		panic(r)
 	}
 }