diff --git a/federation/BUILD b/federation/BUILD index e82b8a58a89..d2f17e9e70e 100644 --- a/federation/BUILD +++ b/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"], diff --git a/federation/cmd/federation-controller-manager/app/controllermanager.go b/federation/cmd/federation-controller-manager/app/controllermanager.go index ebc64c38cf9..78a71a2b2db 100644 --- a/federation/cmd/federation-controller-manager/app/controllermanager.go +++ b/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) { diff --git a/federation/pkg/federation-controller/cluster/clustercontroller.go b/federation/pkg/federation-controller/cluster/clustercontroller.go index 3710b2c6c03..422e0aabe58 100644 --- a/federation/pkg/federation-controller/cluster/clustercontroller.go +++ b/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) { diff --git a/federation/pkg/federation-controller/cluster/clustercontroller_test.go b/federation/pkg/federation-controller/cluster/clustercontroller_test.go index 0f87fb2ddf8..5d4b04ca590 100644 --- a/federation/pkg/federation-controller/cluster/clustercontroller_test.go +++ b/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) diff --git a/federation/pkg/federation-controller/secret/BUILD b/federation/pkg/federation-controller/secret/BUILD index 363202a3de5..3ad927bed97 100644 --- a/federation/pkg/federation-controller/secret/BUILD +++ b/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", diff --git a/federation/pkg/federation-controller/secret/secret_controller.go b/federation/pkg/federation-controller/secret/secret_controller.go index c1aff44c31d..c054e7c6157 100644 --- a/federation/pkg/federation-controller/secret/secret_controller.go +++ b/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) diff --git a/federation/pkg/federation-controller/secret/secret_controller_test.go b/federation/pkg/federation-controller/secret/secret_controller_test.go index bb71d343424..b7c68f87ce6 100644 --- a/federation/pkg/federation-controller/secret/secret_controller_test.go +++ b/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) diff --git a/federation/pkg/typeadapters/BUILD b/federation/pkg/typeadapters/BUILD new file mode 100644 index 00000000000..e2acb8be78b --- /dev/null +++ b/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"], +) diff --git a/federation/pkg/typeadapters/adapter.go b/federation/pkg/typeadapters/adapter.go new file mode 100644 index 00000000000..62bc2d86825 --- /dev/null +++ b/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 +} diff --git a/federation/pkg/typeadapters/crudtester/BUILD b/federation/pkg/typeadapters/crudtester/BUILD new file mode 100644 index 00000000000..f979146ed33 --- /dev/null +++ b/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"], +) diff --git a/federation/pkg/typeadapters/crudtester/crudtester.go b/federation/pkg/typeadapters/crudtester/crudtester.go new file mode 100644 index 00000000000..0ba8341a4ca --- /dev/null +++ b/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 +} diff --git a/federation/pkg/typeadapters/secret.go b/federation/pkg/typeadapters/secret.go new file mode 100644 index 00000000000..c7e2018f0ff --- /dev/null +++ b/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, + } +} diff --git a/test/integration/federation/BUILD b/test/integration/federation/BUILD index 69014eaa6b4..bc4ccc8bdf4 100644 --- a/test/integration/federation/BUILD +++ b/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", diff --git a/test/integration/federation/api_test.go b/test/integration/federation/api_test.go index 735cca705c7..4db1871a911 100644 --- a/test/integration/federation/api_test.go +++ b/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, diff --git a/test/integration/federation/crud_test.go b/test/integration/federation/crud_test.go new file mode 100644 index 00000000000..ce058b5e4c2 --- /dev/null +++ b/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) + }) + } +} diff --git a/test/integration/federation/framework/BUILD b/test/integration/federation/framework/BUILD index 4786d122133..e89c1705554 100644 --- a/test/integration/federation/framework/BUILD +++ b/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", ], ) diff --git a/test/integration/federation/framework/api.go b/test/integration/federation/framework/api.go index f4d92d8cd6f..3f1dc6c9684 100644 --- a/test/integration/federation/framework/api.go +++ b/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) diff --git a/test/integration/federation/framework/controller.go b/test/integration/federation/framework/controller.go new file mode 100644 index 00000000000..ce4633a845b --- /dev/null +++ b/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) +} diff --git a/test/integration/federation/framework/crudtester.go b/test/integration/federation/framework/crudtester.go new file mode 100644 index 00000000000..e90ae722fb2 --- /dev/null +++ b/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) +} diff --git a/test/integration/federation/framework/federation.go b/test/integration/federation/framework/federation.go new file mode 100644 index 00000000000..1b643b45fd2 --- /dev/null +++ b/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 + } +} diff --git a/test/integration/federation/framework/secret.go b/test/integration/federation/framework/secret.go new file mode 100644 index 00000000000..ab2d143da9a --- /dev/null +++ b/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 +} diff --git a/test/integration/federation/framework/util.go b/test/integration/federation/framework/util.go index 7cfaa7d29a4..6ce72607a61 100644 --- a/test/integration/federation/framework/util.go +++ b/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) } }