Merge pull request #1052 from marun/vary-apiversion

Support varying the apiVersion of target resources

Author: k8s-ci-robot

CHANGELOG.md

@@ -1,4 +1,10 @@
 # Unreleased
+-  [#1052](https://github.com/kubernetes-sigs/kubefed/pull/1052)
+   Support has been added for varying the `apiVersion` of target
+   resources. This is intended to allow a federated type to manage
+   more than one version of the target type across member clusters.
+   `apiVersion` can be set either in the template of a federated
+   resource or via override.
 -  [#951](https://github.com/kubernetes-sigs/kubefed/issues/951)
    Propagation status for a namespaced federated resource whose
    containing namespace is not federated now indicates an unhealthy

pkg/client/generic/genericclient.go

@@ -34,6 +34,8 @@ type Client interface {
 	Delete(ctx context.Context, obj runtime.Object, namespace, name string) error
 	List(ctx context.Context, obj runtime.Object, namespace string) error
 	UpdateStatus(ctx context.Context, obj runtime.Object) error
+
+	ListWithOptions(ctx context.Context, opts *client.ListOptions, obj runtime.Object) error
 }
 
 type genericClient struct {
@@ -85,6 +87,10 @@ func (c *genericClient) List(ctx context.Context, obj runtime.Object, namespace
 	return c.client.List(ctx, &client.ListOptions{Namespace: namespace}, obj)
 }
 
+func (c *genericClient) ListWithOptions(ctx context.Context, opts *client.ListOptions, obj runtime.Object) error {
+	return c.client.List(ctx, opts, obj)
+}
+
 func (c *genericClient) UpdateStatus(ctx context.Context, obj runtime.Object) error {
 	return c.client.Status().Update(ctx, obj)
 }

pkg/controller/sync/controller.go

@@ -29,6 +29,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	pkgruntime "k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apimachinery/pkg/util/wait"
@@ -246,11 +247,12 @@ func (s *KubeFedSyncController) reconcile(qualifiedName util.QualifiedName) util
 		return util.StatusError
 	}
 	if possibleOrphan {
-		targetKind := s.typeConfig.GetTargetType().Kind
-		klog.V(2).Infof("Ensuring the removal of the label %q from %s %q in member clusters.", util.ManagedByKubeFedLabelKey, targetKind, qualifiedName)
-		err = s.removeManagedLabel(targetKind, qualifiedName)
+		apiResource := s.typeConfig.GetTargetType()
+		gvk := apiResourceToGVK(&apiResource)
+		klog.V(2).Infof("Ensuring the removal of the label %q from %s %q in member clusters.", util.ManagedByKubeFedLabelKey, gvk.Kind, qualifiedName)
+		err = s.removeManagedLabel(gvk, qualifiedName)
 		if err != nil {
-			wrappedErr := errors.Wrapf(err, "failed to remove the label %q from %s %q in member clusters", util.ManagedByKubeFedLabelKey, targetKind, qualifiedName)
+			wrappedErr := errors.Wrapf(err, "failed to remove the label %q from %s %q in member clusters", util.ManagedByKubeFedLabelKey, gvk.Kind, qualifiedName)
 			runtime.HandleError(wrappedErr)
 			return util.StatusError
 		}
@@ -453,7 +455,7 @@ func (s *KubeFedSyncController) ensureDeletion(fedResource FederatedResource) ut
 			return util.StatusError
 		}
 		klog.V(2).Infof("Initiating the removal of the label %q from resources previously managed by %s %q.", util.ManagedByKubeFedLabelKey, kind, key)
-		err = s.removeManagedLabel(fedResource.TargetKind(), fedResource.TargetName())
+		err = s.removeManagedLabel(fedResource.TargetGVK(), fedResource.TargetName())
 		if err != nil {
 			wrappedErr := errors.Wrapf(err, "failed to remove the label %q from all resources previously managed by %s %q", util.ManagedByKubeFedLabelKey, kind, key)
 			runtime.HandleError(wrappedErr)
@@ -477,8 +479,8 @@ func (s *KubeFedSyncController) ensureDeletion(fedResource FederatedResource) ut
 
 // removeManagedLabel attempts to remove the managed label from
 // resources with the given name in member clusters.
-func (s *KubeFedSyncController) removeManagedLabel(kind string, qualifiedName util.QualifiedName) error {
-	ok, err := s.handleDeletionInClusters(kind, qualifiedName, func(dispatcher dispatch.UnmanagedDispatcher, clusterName string, clusterObj *unstructured.Unstructured) {
+func (s *KubeFedSyncController) removeManagedLabel(gvk schema.GroupVersionKind, qualifiedName util.QualifiedName) error {
+	ok, err := s.handleDeletionInClusters(gvk, qualifiedName, func(dispatcher dispatch.UnmanagedDispatcher, clusterName string, clusterObj *unstructured.Unstructured) {
 		if clusterObj.GetDeletionTimestamp() != nil {
 			return
 		}
@@ -495,11 +497,11 @@ func (s *KubeFedSyncController) removeManagedLabel(kind string, qualifiedName ut
 }
 
 func (s *KubeFedSyncController) deleteFromClusters(fedResource FederatedResource) (bool, error) {
-	kind := fedResource.TargetKind()
+	gvk := fedResource.TargetGVK()
 	qualifiedName := fedResource.TargetName()
 
 	remainingClusters := []string{}
-	ok, err := s.handleDeletionInClusters(kind, qualifiedName, func(dispatcher dispatch.UnmanagedDispatcher, clusterName string, clusterObj *unstructured.Unstructured) {
+	ok, err := s.handleDeletionInClusters(gvk, qualifiedName, func(dispatcher dispatch.UnmanagedDispatcher, clusterName string, clusterObj *unstructured.Unstructured) {
 		// If the containing namespace of a FederatedNamespace is
 		// marked for deletion, it is impossible to require the
 		// removal of the namespace in advance of removal of the sync
@@ -557,7 +559,7 @@ func (s *KubeFedSyncController) ensureRemovedOrUnmanaged(fedResource FederatedRe
 		return errors.Wrap(err, "failed to get a list of clusters")
 	}
 
-	dispatcher := dispatch.NewCheckUnmanagedDispatcher(s.informer.GetClientForCluster, fedResource.TargetKind(), fedResource.TargetName())
+	dispatcher := dispatch.NewCheckUnmanagedDispatcher(s.informer.GetClientForCluster, fedResource.TargetGVK(), fedResource.TargetName())
 	unreadyClusters := []string{}
 	for _, cluster := range clusters {
 		if !util.IsClusterReady(&cluster.Status) {
@@ -581,15 +583,15 @@ func (s *KubeFedSyncController) ensureRemovedOrUnmanaged(fedResource FederatedRe
 
 // handleDeletionInClusters invokes the provided deletion handler for
 // each managed resource in member clusters.
-func (s *KubeFedSyncController) handleDeletionInClusters(kind string, qualifiedName util.QualifiedName,
+func (s *KubeFedSyncController) handleDeletionInClusters(gvk schema.GroupVersionKind, qualifiedName util.QualifiedName,
 	deletionFunc func(dispatcher dispatch.UnmanagedDispatcher, clusterName string, clusterObj *unstructured.Unstructured)) (bool, error) {
 
 	clusters, err := s.informer.GetClusters()
 	if err != nil {
 		return false, errors.Wrap(err, "failed to get a list of clusters")
 	}
 
-	dispatcher := dispatch.NewUnmanagedDispatcher(s.informer.GetClientForCluster, kind, qualifiedName)
+	dispatcher := dispatch.NewUnmanagedDispatcher(s.informer.GetClientForCluster, gvk, qualifiedName)
 	key := qualifiedName.String()
 	retrievalFailureClusters := []string{}
 	unreadyClusters := []string{}
@@ -603,7 +605,7 @@ func (s *KubeFedSyncController) handleDeletionInClusters(kind string, qualifiedN
 
 		rawClusterObj, _, err := s.informer.GetTargetStore().GetByKey(clusterName, key)
 		if err != nil {
-			wrappedErr := errors.Wrapf(err, "failed to retrieve %s %q for cluster %q", kind, key, clusterName)
+			wrappedErr := errors.Wrapf(err, "failed to retrieve %s %q for cluster %q", gvk.Kind, key, clusterName)
 			runtime.HandleError(wrappedErr)
 			retrievalFailureClusters = append(retrievalFailureClusters, clusterName)
 			continue

pkg/controller/sync/dispatch/checkunmanaged.go

@@ -17,14 +17,18 @@ limitations under the License.
 package dispatch
 
 import (
+	"context"
+
 	"github.com/pkg/errors"
 
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	pkgruntime "k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/klog"
 
+	"sigs.k8s.io/kubefed/pkg/client/generic"
 	"sigs.k8s.io/kubefed/pkg/controller/util"
 )
 
@@ -39,16 +43,16 @@ type CheckUnmanagedDispatcher interface {
 type checkUnmanagedDispatcherImpl struct {
 	dispatcher *operationDispatcherImpl
 
+	targetGVK  schema.GroupVersionKind
 	targetName util.QualifiedName
-	targetKind string
 }
 
-func NewCheckUnmanagedDispatcher(clientAccessor clientAccessorFunc, targetKind string, targetName util.QualifiedName) CheckUnmanagedDispatcher {
+func NewCheckUnmanagedDispatcher(clientAccessor clientAccessorFunc, targetGVK schema.GroupVersionKind, targetName util.QualifiedName) CheckUnmanagedDispatcher {
 	dispatcher := newOperationDispatcher(clientAccessor, nil)
 	return &checkUnmanagedDispatcherImpl{
 		dispatcher: dispatcher,
+		targetGVK:  targetGVK,
 		targetName: targetName,
-		targetKind: targetKind,
 	}
 }
 
@@ -63,10 +67,12 @@ func (d *checkUnmanagedDispatcherImpl) CheckRemovedOrUnlabeled(clusterName strin
 	d.dispatcher.incrementOperationsInitiated()
 	const op = "check for deletion of resource or removal of managed label from"
 	const opContinuous = "Checking for deletion of resource or removal of managed label from"
-	go d.dispatcher.clusterOperation(clusterName, op, func(client util.ResourceClient) util.ReconciliationStatus {
-		klog.V(2).Infof(eventTemplate, opContinuous, d.targetKind, d.targetName, clusterName)
+	go d.dispatcher.clusterOperation(clusterName, op, func(client generic.Client) util.ReconciliationStatus {
+		klog.V(2).Infof(eventTemplate, opContinuous, d.targetGVK.Kind, d.targetName, clusterName)
 
-		clusterObj, err := client.Resources(d.targetName.Namespace).Get(d.targetName.Name, metav1.GetOptions{})
+		clusterObj := &unstructured.Unstructured{}
+		clusterObj.SetGroupVersionKind(d.targetGVK)
+		err := client.Get(context.Background(), clusterObj, d.targetName.Namespace, d.targetName.Name)
 		if apierrors.IsNotFound(err) {
 			return util.StatusAllOK
 		}
@@ -95,5 +101,5 @@ func (d *checkUnmanagedDispatcherImpl) CheckRemovedOrUnlabeled(clusterName strin
 }
 
 func (d *checkUnmanagedDispatcherImpl) wrapOperationError(err error, clusterName, operation string) error {
-	return wrapOperationError(err, operation, d.targetKind, d.targetName.String(), clusterName)
+	return wrapOperationError(err, operation, d.targetGVK.Kind, d.targetName.String(), clusterName)
 }

pkg/controller/sync/dispatch/managed.go

@@ -17,18 +17,20 @@ limitations under the License.
 package dispatch
 
 import (
+	"context"
 	"fmt"
 	"strings"
 	"sync"
 
 	"github.com/pkg/errors"
 
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	pkgruntime "k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/util/sets"
 
+	"sigs.k8s.io/kubefed/pkg/client/generic"
 	"sigs.k8s.io/kubefed/pkg/controller/sync/status"
 	"sigs.k8s.io/kubefed/pkg/controller/util"
 )
@@ -38,6 +40,7 @@ import (
 type FederatedResourceForDispatch interface {
 	TargetName() util.QualifiedName
 	TargetKind() string
+	TargetGVK() schema.GroupVersionKind
 	Object() *unstructured.Unstructured
 	VersionForCluster(clusterName string) (string, error)
 	ObjectForCluster(clusterName string) (*unstructured.Unstructured, error)
@@ -80,7 +83,7 @@ func NewManagedDispatcher(clientAccessor clientAccessorFunc, fedResource Federat
 		skipAdoptingResources: skipAdoptingResources,
 	}
 	d.dispatcher = newOperationDispatcher(clientAccessor, d)
-	d.unmanagedDispatcher = newUnmanagedDispatcher(d.dispatcher, d, fedResource.TargetKind(), fedResource.TargetName())
+	d.unmanagedDispatcher = newUnmanagedDispatcher(d.dispatcher, d, fedResource.TargetGVK(), fedResource.TargetName())
 	return d
 }
 
@@ -126,7 +129,7 @@ func (d *managedDispatcherImpl) Create(clusterName string) {
 
 	d.dispatcher.incrementOperationsInitiated()
 	const op = "create"
-	go d.dispatcher.clusterOperation(clusterName, op, func(client util.ResourceClient) util.ReconciliationStatus {
+	go d.dispatcher.clusterOperation(clusterName, op, func(client generic.Client) util.ReconciliationStatus {
 		d.recordEvent(clusterName, op, "Creating")
 
 		obj, err := d.fedResource.ObjectForCluster(clusterName)
@@ -139,9 +142,9 @@ func (d *managedDispatcherImpl) Create(clusterName string) {
 			return d.recordOperationError(status.ApplyOverridesFailed, clusterName, op, err)
 		}
 
-		createdObj, err := client.Resources(obj.GetNamespace()).Create(obj, metav1.CreateOptions{})
+		err = client.Create(context.Background(), obj)
 		if err == nil {
-			version := util.ObjectVersion(createdObj)
+			version := util.ObjectVersion(obj)
 			d.recordVersion(clusterName, version)
 			return util.StatusAllOK
 		}
@@ -155,19 +158,19 @@ func (d *managedDispatcherImpl) Create(clusterName string) {
 
 		// Attempt to update the existing resource to ensure that it
 		// is labeled as a managed resource.
-		clusterObj, err := client.Resources(obj.GetNamespace()).Get(obj.GetName(), metav1.GetOptions{})
+		err = client.Get(context.Background(), obj, obj.GetNamespace(), obj.GetName())
 		if err != nil {
 			wrappedErr := errors.Wrapf(err, "failed to retrieve object potentially requiring adoption")
 			return d.recordOperationError(status.RetrievalFailed, clusterName, op, wrappedErr)
 		}
 
-		if d.skipAdoptingResources && !d.fedResource.IsNamespaceInHostCluster(clusterObj) {
+		if d.skipAdoptingResources && !d.fedResource.IsNamespaceInHostCluster(obj) {
 			_ = d.recordOperationError(status.AlreadyExists, clusterName, op, errors.Errorf("Resource pre-exist in cluster"))
 			return util.StatusAllOK
 		}
 
 		d.recordError(clusterName, op, errors.Errorf("An update will be attempted instead of a creation due to an existing resource"))
-		d.Update(clusterName, clusterObj)
+		d.Update(clusterName, obj)
 		return util.StatusAllOK
 	})
 }
@@ -177,7 +180,7 @@ func (d *managedDispatcherImpl) Update(clusterName string, clusterObj *unstructu
 
 	d.dispatcher.incrementOperationsInitiated()
 	const op = "update"
-	go d.dispatcher.clusterOperation(clusterName, op, func(client util.ResourceClient) util.ReconciliationStatus {
+	go d.dispatcher.clusterOperation(clusterName, op, func(client generic.Client) util.ReconciliationStatus {
 		obj, err := d.fedResource.ObjectForCluster(clusterName)
 		if err != nil {
 			return d.recordOperationError(status.ComputeResourceFailed, clusterName, op, err)
@@ -206,11 +209,11 @@ func (d *managedDispatcherImpl) Update(clusterName string, clusterObj *unstructu
 		// Only record an event if the resource is not current
 		d.recordEvent(clusterName, op, "Updating")
 
-		updatedObj, err := client.Resources(obj.GetNamespace()).Update(obj, metav1.UpdateOptions{})
+		err = client.Update(context.Background(), obj)
 		if err != nil {
 			return d.recordOperationError(status.UpdateFailed, clusterName, op, err)
 		}
-		version = util.ObjectVersion(updatedObj)
+		version = util.ObjectVersion(obj)
 		d.recordVersion(clusterName, version)
 		return util.StatusAllOK
 	})

pkg/controller/sync/dispatch/operation.go

@@ -24,11 +24,12 @@ import (
 
 	"k8s.io/apimachinery/pkg/util/runtime"
 
+	"sigs.k8s.io/kubefed/pkg/client/generic"
 	"sigs.k8s.io/kubefed/pkg/controller/sync/status"
 	"sigs.k8s.io/kubefed/pkg/controller/util"
 )
 
-type clientAccessorFunc func(clusterName string) (util.ResourceClient, error)
+type clientAccessorFunc func(clusterName string) (generic.Client, error)
 
 type dispatchRecorder interface {
 	recordEvent(clusterName, operation, operationContinuous string)
@@ -91,9 +92,8 @@ func (d *operationDispatcherImpl) Wait() (bool, error) {
 	return ok, nil
 }
 
-func (d *operationDispatcherImpl) clusterOperation(clusterName, op string, opFunc func(util.ResourceClient) util.ReconciliationStatus) {
-	// TODO(marun) Update to generic client and support cancellation
-	// on timeout.
+func (d *operationDispatcherImpl) clusterOperation(clusterName, op string, opFunc func(generic.Client) util.ReconciliationStatus) {
+	// TODO(marun) Support cancellation of client calls on timeout.
 	client, err := d.clientAccessor(clusterName)
 	if err != nil {
 		wrappedErr := errors.Wrapf(err, "Error retrieving client for cluster")