Add finalizers to federated configmaps

federation/pkg/federation-controller/configmap/BUILD

@@ -16,12 +16,14 @@ go_library(
         "//federation/apis/federation/v1beta1:go_default_library",
         "//federation/client/clientset_generated/federation_clientset:go_default_library",
         "//federation/pkg/federation-controller/util:go_default_library",
+        "//federation/pkg/federation-controller/util/deletionhelper:go_default_library",
         "//federation/pkg/federation-controller/util/eventsink:go_default_library",
         "//pkg/api:go_default_library",
         "//pkg/api/v1:go_default_library",
         "//pkg/client/clientset_generated/clientset:go_default_library",
         "//pkg/controller:go_default_library",
         "//vendor:github.com/golang/glog",
+        "//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/types",
@@ -42,10 +44,12 @@ go_test(
         "//federation/apis/federation/v1beta1:go_default_library",
         "//federation/client/clientset_generated/federation_clientset/fake:go_default_library",
         "//federation/pkg/federation-controller/util:go_default_library",
+        "//federation/pkg/federation-controller/util/deletionhelper:go_default_library",
         "//federation/pkg/federation-controller/util/test:go_default_library",
         "//pkg/api/v1:go_default_library",
         "//pkg/client/clientset_generated/clientset:go_default_library",
         "//pkg/client/clientset_generated/clientset/fake:go_default_library",
+        "//vendor:github.com/golang/glog",
         "//vendor:github.com/stretchr/testify/assert",
         "//vendor:k8s.io/apimachinery/pkg/apis/meta/v1",
         "//vendor:k8s.io/apimachinery/pkg/runtime",

federation/pkg/federation-controller/configmap/configmap_controller.go

@@ -17,8 +17,10 @@ limitations under the License.
 package configmap
 
 import (
+	"fmt"
 	"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/types"
@@ -30,6 +32,7 @@ import (
 	federationapi "k8s.io/kubernetes/federation/apis/federation/v1beta1"
 	federationclientset "k8s.io/kubernetes/federation/client/clientset_generated/federation_clientset"
 	"k8s.io/kubernetes/federation/pkg/federation-controller/util"
+	"k8s.io/kubernetes/federation/pkg/federation-controller/util/deletionhelper"
 	"k8s.io/kubernetes/federation/pkg/federation-controller/util/eventsink"
 	"k8s.io/kubernetes/pkg/api"
 	apiv1 "k8s.io/kubernetes/pkg/api/v1"
@@ -71,6 +74,9 @@ type ConfigMapController struct {
 	// For events
 	eventRecorder record.EventRecorder
 
+	// Finalizers
+	deletionHelper *deletionhelper.DeletionHelper
+
 	configmapReviewDelay  time.Duration
 	clusterAvailableDelay time.Duration
 	smallDelay            time.Duration
@@ -160,9 +166,71 @@ func NewConfigMapController(client federationclientset.Interface) *ConfigMapCont
 			err := client.Core().ConfigMaps(configmap.Namespace).Delete(configmap.Name, &metav1.DeleteOptions{})
 			return err
 		})
+
+	configmapcontroller.deletionHelper = deletionhelper.NewDeletionHelper(
+		configmapcontroller.hasFinalizerFunc,
+		configmapcontroller.removeFinalizerFunc,
+		configmapcontroller.addFinalizerFunc,
+		// objNameFunc
+		func(obj pkgruntime.Object) string {
+			configmap := obj.(*apiv1.ConfigMap)
+			return configmap.Name
+		},
+		configmapcontroller.updateTimeout,
+		configmapcontroller.eventRecorder,
+		configmapcontroller.configmapFederatedInformer,
+		configmapcontroller.federatedUpdater,
+	)
+
 	return configmapcontroller
 }
 
+// hasFinalizerFunc returns true if the given object has the given finalizer in its ObjectMeta.
+func (configmapcontroller *ConfigMapController) hasFinalizerFunc(obj pkgruntime.Object, finalizer string) bool {
+	configmap := obj.(*apiv1.ConfigMap)
+	for i := range configmap.ObjectMeta.Finalizers {
+		if string(configmap.ObjectMeta.Finalizers[i]) == finalizer {
+			return true
+		}
+	}
+	return false
+}
+
+// removeFinalizerFunc removes the finalizer from the given objects ObjectMeta. Assumes that the given object is a configmap.
+func (configmapcontroller *ConfigMapController) removeFinalizerFunc(obj pkgruntime.Object, finalizer string) (pkgruntime.Object, error) {
+	configmap := obj.(*apiv1.ConfigMap)
+	newFinalizers := []string{}
+	hasFinalizer := false
+	for i := range configmap.ObjectMeta.Finalizers {
+		if string(configmap.ObjectMeta.Finalizers[i]) != finalizer {
+			newFinalizers = append(newFinalizers, configmap.ObjectMeta.Finalizers[i])
+		} else {
+			hasFinalizer = true
+		}
+	}
+	if !hasFinalizer {
+		// Nothing to do.
+		return obj, nil
+	}
+	configmap.ObjectMeta.Finalizers = newFinalizers
+	configmap, err := configmapcontroller.federatedApiClient.Core().ConfigMaps(configmap.Namespace).Update(configmap)
+	if err != nil {
+		return nil, fmt.Errorf("failed to remove finalizer %s from configmap %s: %v", finalizer, configmap.Name, err)
+	}
+	return configmap, nil
+}
+
+// addFinalizerFunc adds the given finalizer to the given objects ObjectMeta. Assumes that the given object is a configmap.
+func (configmapcontroller *ConfigMapController) addFinalizerFunc(obj pkgruntime.Object, finalizers []string) (pkgruntime.Object, error) {
+	configmap := obj.(*apiv1.ConfigMap)
+	configmap.ObjectMeta.Finalizers = append(configmap.ObjectMeta.Finalizers, finalizers...)
+	configmap, err := configmapcontroller.federatedApiClient.Core().ConfigMaps(configmap.Namespace).Update(configmap)
+	if err != nil {
+		return nil, fmt.Errorf("failed to add finalizers %v to configmap %s: %v", finalizers, configmap.Name, err)
+	}
+	return configmap, nil
+}
+
 func (configmapcontroller *ConfigMapController) Run(stopChan <-chan struct{}) {
 	go configmapcontroller.configmapInformerController.Run(stopChan)
 	configmapcontroller.configmapFederatedInformer.Start()
@@ -251,6 +319,31 @@ func (configmapcontroller *ConfigMapController) reconcileConfigMap(configmap typ
 	}
 	baseConfigMap := baseConfigMapObj.(*apiv1.ConfigMap)
 
+	// Check if deletion has been requested.
+	if baseConfigMap.DeletionTimestamp != nil {
+		if err := configmapcontroller.delete(baseConfigMap); err != nil {
+			glog.Errorf("Failed to delete %s: %v", configmap, err)
+			configmapcontroller.eventRecorder.Eventf(baseConfigMap, api.EventTypeNormal, "DeleteFailed",
+				"ConfigMap delete failed: %v", err)
+			configmapcontroller.deliverConfigMap(configmap, 0, true)
+		}
+		return
+	}
+
+	glog.V(3).Infof("Ensuring delete object from underlying clusters finalizer for configmap: %s",
+		baseConfigMap.Name)
+	// Add the required finalizers before creating a configmap in underlying clusters.
+	updatedConfigMapObj, err := configmapcontroller.deletionHelper.EnsureFinalizers(baseConfigMap)
+	if err != nil {
+		glog.Errorf("Failed to ensure delete object from underlying clusters finalizer in configmap %s: %v",
+			baseConfigMap.Name, err)
+		configmapcontroller.deliverConfigMap(configmap, 0, false)
+		return
+	}
+	baseConfigMap = updatedConfigMapObj.(*apiv1.ConfigMap)
+
+	glog.V(3).Infof("Syncing configmap %s in underlying clusters", baseConfigMap.Name)
+
 	clusters, err := configmapcontroller.configmapFederatedInformer.GetReadyClusters()
 	if err != nil {
 		glog.Errorf("Failed to get cluster list: %v, retrying shortly", err)
@@ -315,3 +408,23 @@ func (configmapcontroller *ConfigMapController) reconcileConfigMap(configmap typ
 		return
 	}
 }
+
+// delete deletes the given configmap or returns error if the deletion was not complete.
+func (configmapcontroller *ConfigMapController) delete(configmap *apiv1.ConfigMap) error {
+	glog.V(3).Infof("Handling deletion of configmap: %v", *configmap)
+	_, err := configmapcontroller.deletionHelper.HandleObjectInUnderlyingClusters(configmap)
+	if err != nil {
+		return err
+	}
+
+	err = configmapcontroller.federatedApiClient.Core().ConfigMaps(configmap.Namespace).Delete(configmap.Name, nil)
+	if err != nil {
+		// Its all good if the error is not found error. That means it is deleted already and we do not have to do anything.
+		// This is expected when we are processing an update as a result of configmap finalizer deletion.
+		// The process that deleted the last finalizer is also going to delete the configmap and we do not have to do anything.
+		if !errors.IsNotFound(err) {
+			return fmt.Errorf("failed to delete configmap: %v", err)
+		}
+	}
+	return nil
+}

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

@@ -28,11 +28,13 @@ import (
 	federationapi "k8s.io/kubernetes/federation/apis/federation/v1beta1"
 	fakefedclientset "k8s.io/kubernetes/federation/client/clientset_generated/federation_clientset/fake"
 	"k8s.io/kubernetes/federation/pkg/federation-controller/util"
+	"k8s.io/kubernetes/federation/pkg/federation-controller/util/deletionhelper"
 	. "k8s.io/kubernetes/federation/pkg/federation-controller/util/test"
 	apiv1 "k8s.io/kubernetes/pkg/api/v1"
 	kubeclientset "k8s.io/kubernetes/pkg/client/clientset_generated/clientset"
 	fakekubeclientset "k8s.io/kubernetes/pkg/client/clientset_generated/clientset/fake"
 
+	"github.com/golang/glog"
 	"github.com/stretchr/testify/assert"
 )
 
@@ -44,6 +46,7 @@ func TestConfigMapController(t *testing.T) {
 	RegisterFakeList("clusters", &fakeClient.Fake, &federationapi.ClusterList{Items: []federationapi.Cluster{*cluster1}})
 	RegisterFakeList("configmaps", &fakeClient.Fake, &apiv1.ConfigMapList{Items: []apiv1.ConfigMap{}})
 	configmapWatch := RegisterFakeWatch("configmaps", &fakeClient.Fake)
+	configmapUpdateChan := RegisterFakeCopyOnUpdate("configmaps", &fakeClient.Fake, configmapWatch)
 	clusterWatch := RegisterFakeWatch("clusters", &fakeClient.Fake)
 
 	cluster1Client := &fakekubeclientset.Clientset{}
@@ -92,6 +95,12 @@ func TestConfigMapController(t *testing.T) {
 
 	// Test add federated configmap.
 	configmapWatch.Add(configmap1)
+	// There should be 2 updates to add both the finalizers.
+	updatedConfigMap := GetConfigMapFromChan(configmapUpdateChan)
+	assert.True(t, configmapController.hasFinalizerFunc(updatedConfigMap, deletionhelper.FinalizerDeleteFromUnderlyingClusters))
+	assert.True(t, configmapController.hasFinalizerFunc(updatedConfigMap, metav1.FinalizerOrphan))
+
+	// Verify that the configmap is created in underlying cluster1.
 	createdConfigMap := GetConfigMapFromChan(cluster1CreateChan)
 	assert.NotNil(t, createdConfigMap)
 	assert.Equal(t, configmap1.Namespace, createdConfigMap.Namespace)
@@ -109,22 +118,30 @@ func TestConfigMapController(t *testing.T) {
 		"A": "B",
 	}
 	configmapWatch.Modify(configmap1)
-	updatedConfigMap := GetConfigMapFromChan(cluster1UpdateChan)
+	updatedConfigMap = GetConfigMapFromChan(cluster1UpdateChan)
 	assert.NotNil(t, updatedConfigMap)
 	assert.Equal(t, configmap1.Name, updatedConfigMap.Name)
 	assert.Equal(t, configmap1.Namespace, updatedConfigMap.Namespace)
 	assert.True(t, util.ConfigMapEquivalent(configmap1, updatedConfigMap))
 
+	// Wait for the configmap to appear in the informer store
+	err = WaitForConfigMapStoreUpdate(
+		configmapController.configmapFederatedInformer.GetTargetStore(),
+		cluster1.Name, types.NamespacedName{Namespace: configmap1.Namespace, Name: configmap1.Name}.String(),
+		configmap1, wait.ForeverTestTimeout)
+	assert.Nil(t, err, "configmap should have appeared in the informer store")
+
 	// Test update federated configmap.
 	configmap1.Data = map[string]string{
 		"config": "myconfigurationfile",
 	}
+
 	configmapWatch.Modify(configmap1)
-	updatedConfigMap2 := GetConfigMapFromChan(cluster1UpdateChan)
+	updatedConfigMap = GetConfigMapFromChan(cluster1UpdateChan)
 	assert.NotNil(t, updatedConfigMap)
 	assert.Equal(t, configmap1.Name, updatedConfigMap.Name)
 	assert.Equal(t, configmap1.Namespace, updatedConfigMap.Namespace)
-	assert.True(t, util.ConfigMapEquivalent(configmap1, updatedConfigMap2))
+	assert.True(t, util.ConfigMapEquivalent(configmap1, updatedConfigMap))
 
 	// Test add cluster
 	clusterWatch.Add(cluster2)
@@ -141,3 +158,21 @@ func GetConfigMapFromChan(c chan runtime.Object) *apiv1.ConfigMap {
 	configmap := GetObjectFromChan(c).(*apiv1.ConfigMap)
 	return configmap
 }
+
+// Wait till the store is updated with latest configmap.
+func WaitForConfigMapStoreUpdate(store util.FederatedReadOnlyStore, clusterName, key string, desiredConfigMap *apiv1.ConfigMap, timeout time.Duration) error {
+	retryInterval := 200 * time.Millisecond
+	err := wait.PollImmediate(retryInterval, timeout, func() (bool, error) {
+		obj, found, err := store.GetByKey(clusterName, key)
+		if !found || err != nil {
+			glog.Infof("%s is not in the store", key)
+			return false, err
+		}
+		equal := util.ConfigMapEquivalent(obj.(*apiv1.ConfigMap), desiredConfigMap)
+		if !equal {
+			glog.Infof("wrong content in the store expected:\n%v\nactual:\n%v\n", *desiredConfigMap, *obj.(*apiv1.ConfigMap))
+		}
+		return equal, err
+	})
+	return err
+}