Merge pull request #35635 from mwielgus/configmap-ctrl

Automatic merge from submit-queue

Federated ConfigMap controller

Based on the secrets controller. E2e tests will come in the next PR.

**Release note**:

``` release-note
Federated ConfigMap controller. Supports all the API that regular ConfigMap has.
```

cc: @quinton-hoole @kubernetes/sig-cluster-federation

federation/cmd/federation-controller-manager/app/BUILD

@@ -24,6 +24,7 @@ go_library(
         "//federation/pkg/dnsprovider/providers/aws/route53:go_default_library",
         "//federation/pkg/dnsprovider/providers/google/clouddns:go_default_library",
         "//federation/pkg/federation-controller/cluster:go_default_library",
+        "//federation/pkg/federation-controller/configmap:go_default_library",
         "//federation/pkg/federation-controller/daemonset:go_default_library",
         "//federation/pkg/federation-controller/deployment:go_default_library",
         "//federation/pkg/federation-controller/ingress:go_default_library",

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

@@ -30,6 +30,7 @@ import (
 	"k8s.io/kubernetes/federation/cmd/federation-controller-manager/app/options"
 	"k8s.io/kubernetes/federation/pkg/dnsprovider"
 	clustercontroller "k8s.io/kubernetes/federation/pkg/federation-controller/cluster"
+	configmapcontroller "k8s.io/kubernetes/federation/pkg/federation-controller/configmap"
 	daemonset "k8s.io/kubernetes/federation/pkg/federation-controller/daemonset"
 	deploymentcontroller "k8s.io/kubernetes/federation/pkg/federation-controller/deployment"
 	ingresscontroller "k8s.io/kubernetes/federation/pkg/federation-controller/ingress"
@@ -158,6 +159,10 @@ func StartControllers(s *options.CMServer, restClientCfg *restclient.Config) err
 	secretcontroller := secretcontroller.NewSecretController(secretcontrollerClientset)
 	secretcontroller.Run(wait.NeverStop)
 
+	configmapcontrollerClientset := federationclientset.NewForConfigOrDie(restclient.AddUserAgent(restClientCfg, "configmap-controller"))
+	configmapcontroller := configmapcontroller.NewConfigMapController(configmapcontrollerClientset)
+	configmapcontroller.Run(wait.NeverStop)
+
 	daemonsetcontrollerClientset := federationclientset.NewForConfigOrDie(restclient.AddUserAgent(restClientCfg, "daemonset-controller"))
 	daemonsetcontroller := daemonset.NewDaemonSetController(daemonsetcontrollerClientset)
 	daemonsetcontroller.Run(wait.NeverStop)

federation/pkg/federation-controller/configmap/BUILD

@@ -0,0 +1,54 @@
+package(default_visibility = ["//visibility:public"])
+
+licenses(["notice"])
+
+load(
+    "@io_bazel_rules_go//go:def.bzl",
+    "go_binary",
+    "go_library",
+    "go_test",
+    "cgo_library",
+)
+
+go_library(
+    name = "go_default_library",
+    srcs = ["configmap_controller.go"],
+    tags = ["automanaged"],
+    deps = [
+        "//federation/apis/federation/v1beta1:go_default_library",
+        "//federation/client/clientset_generated/federation_release_1_5:go_default_library",
+        "//federation/pkg/federation-controller/util: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/cache:go_default_library",
+        "//pkg/client/clientset_generated/release_1_5:go_default_library",
+        "//pkg/client/record:go_default_library",
+        "//pkg/controller:go_default_library",
+        "//pkg/runtime:go_default_library",
+        "//pkg/types:go_default_library",
+        "//pkg/util/flowcontrol:go_default_library",
+        "//pkg/watch:go_default_library",
+        "//vendor:github.com/golang/glog",
+    ],
+)
+
+go_test(
+    name = "go_default_test",
+    srcs = ["configmap_controller_test.go"],
+    library = "go_default_library",
+    tags = ["automanaged"],
+    deps = [
+        "//federation/apis/federation/v1beta1:go_default_library",
+        "//federation/client/clientset_generated/federation_release_1_5/fake:go_default_library",
+        "//federation/pkg/federation-controller/util:go_default_library",
+        "//federation/pkg/federation-controller/util/test:go_default_library",
+        "//pkg/api/v1:go_default_library",
+        "//pkg/client/clientset_generated/release_1_5:go_default_library",
+        "//pkg/client/clientset_generated/release_1_5/fake:go_default_library",
+        "//pkg/runtime:go_default_library",
+        "//pkg/types:go_default_library",
+        "//pkg/util/wait:go_default_library",
+        "//vendor:github.com/stretchr/testify/assert",
+    ],
+)

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

@@ -0,0 +1,318 @@
+/*
+Copyright 2016 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 configmap
+
+import (
+	"time"
+
+	federation_api "k8s.io/kubernetes/federation/apis/federation/v1beta1"
+	federationclientset "k8s.io/kubernetes/federation/client/clientset_generated/federation_release_1_5"
+	"k8s.io/kubernetes/federation/pkg/federation-controller/util"
+	"k8s.io/kubernetes/federation/pkg/federation-controller/util/eventsink"
+	"k8s.io/kubernetes/pkg/api"
+	api_v1 "k8s.io/kubernetes/pkg/api/v1"
+	"k8s.io/kubernetes/pkg/client/cache"
+	kubeclientset "k8s.io/kubernetes/pkg/client/clientset_generated/release_1_5"
+	"k8s.io/kubernetes/pkg/client/record"
+	"k8s.io/kubernetes/pkg/controller"
+	pkg_runtime "k8s.io/kubernetes/pkg/runtime"
+	"k8s.io/kubernetes/pkg/types"
+	"k8s.io/kubernetes/pkg/util/flowcontrol"
+	"k8s.io/kubernetes/pkg/watch"
+
+	"github.com/golang/glog"
+)
+
+const (
+	allClustersKey = "ALL_CLUSTERS"
+)
+
+type ConfigMapController struct {
+	// For triggering single configmap reconciliation. This is used when there is an
+	// add/update/delete operation on a configmap in either federated API server or
+	// in some member of the federation.
+	configmapDeliverer *util.DelayingDeliverer
+
+	// For triggering all configmaps reconciliation. This is used when
+	// a new cluster becomes available.
+	clusterDeliverer *util.DelayingDeliverer
+
+	// Contains configmaps present in members of federation.
+	configmapFederatedInformer util.FederatedInformer
+	// For updating members of federation.
+	federatedUpdater util.FederatedUpdater
+	// Definitions of configmaps that should be federated.
+	configmapInformerStore cache.Store
+	// Informer controller for configmaps that should be federated.
+	configmapInformerController cache.ControllerInterface
+
+	// Client to federated api server.
+	federatedApiClient federationclientset.Interface
+
+	// Backoff manager for configmaps
+	configmapBackoff *flowcontrol.Backoff
+
+	// For events
+	eventRecorder record.EventRecorder
+
+	configmapReviewDelay  time.Duration
+	clusterAvailableDelay time.Duration
+	smallDelay            time.Duration
+	updateTimeout         time.Duration
+}
+
+// NewConfigMapController returns a new configmap controller
+func NewConfigMapController(client federationclientset.Interface) *ConfigMapController {
+	broadcaster := record.NewBroadcaster()
+	broadcaster.StartRecordingToSink(eventsink.NewFederatedEventSink(client))
+	recorder := broadcaster.NewRecorder(api.EventSource{Component: "federated-configmaps-controller"})
+
+	configmapcontroller := &ConfigMapController{
+		federatedApiClient:    client,
+		configmapReviewDelay:  time.Second * 10,
+		clusterAvailableDelay: time.Second * 20,
+		smallDelay:            time.Second * 3,
+		updateTimeout:         time.Second * 30,
+		configmapBackoff:      flowcontrol.NewBackOff(5*time.Second, time.Minute),
+		eventRecorder:         recorder,
+	}
+
+	// Build delivereres for triggering reconciliations.
+	configmapcontroller.configmapDeliverer = util.NewDelayingDeliverer()
+	configmapcontroller.clusterDeliverer = util.NewDelayingDeliverer()
+
+	// Start informer on federated API servers on configmaps that should be federated.
+	configmapcontroller.configmapInformerStore, configmapcontroller.configmapInformerController = cache.NewInformer(
+		&cache.ListWatch{
+			ListFunc: func(options api.ListOptions) (pkg_runtime.Object, error) {
+				versionedOptions := util.VersionizeV1ListOptions(options)
+				return client.Core().ConfigMaps(api_v1.NamespaceAll).List(versionedOptions)
+			},
+			WatchFunc: func(options api.ListOptions) (watch.Interface, error) {
+				versionedOptions := util.VersionizeV1ListOptions(options)
+				return client.Core().ConfigMaps(api_v1.NamespaceAll).Watch(versionedOptions)
+			},
+		},
+		&api_v1.ConfigMap{},
+		controller.NoResyncPeriodFunc(),
+		util.NewTriggerOnAllChanges(func(obj pkg_runtime.Object) { configmapcontroller.deliverConfigMapObj(obj, 0, false) }))
+
+	// Federated informer on configmaps in members of federation.
+	configmapcontroller.configmapFederatedInformer = util.NewFederatedInformer(
+		client,
+		func(cluster *federation_api.Cluster, targetClient kubeclientset.Interface) (cache.Store, cache.ControllerInterface) {
+			return cache.NewInformer(
+				&cache.ListWatch{
+					ListFunc: func(options api.ListOptions) (pkg_runtime.Object, error) {
+						versionedOptions := util.VersionizeV1ListOptions(options)
+						return targetClient.Core().ConfigMaps(api_v1.NamespaceAll).List(versionedOptions)
+					},
+					WatchFunc: func(options api.ListOptions) (watch.Interface, error) {
+						versionedOptions := util.VersionizeV1ListOptions(options)
+						return targetClient.Core().ConfigMaps(api_v1.NamespaceAll).Watch(versionedOptions)
+					},
+				},
+				&api_v1.ConfigMap{},
+				controller.NoResyncPeriodFunc(),
+				// Trigger reconciliation whenever something in federated cluster is changed. In most cases it
+				// would be just confirmation that some configmap opration succeeded.
+				util.NewTriggerOnAllChanges(
+					func(obj pkg_runtime.Object) {
+						configmapcontroller.deliverConfigMapObj(obj, configmapcontroller.configmapReviewDelay, false)
+					},
+				))
+		},
+
+		&util.ClusterLifecycleHandlerFuncs{
+			ClusterAvailable: func(cluster *federation_api.Cluster) {
+				// When new cluster becomes available process all the configmaps again.
+				configmapcontroller.clusterDeliverer.DeliverAt(allClustersKey, nil, time.Now().Add(configmapcontroller.clusterAvailableDelay))
+			},
+		},
+	)
+
+	// Federated updater along with Create/Update/Delete operations.
+	configmapcontroller.federatedUpdater = util.NewFederatedUpdater(configmapcontroller.configmapFederatedInformer,
+		func(client kubeclientset.Interface, obj pkg_runtime.Object) error {
+			configmap := obj.(*api_v1.ConfigMap)
+			_, err := client.Core().ConfigMaps(configmap.Namespace).Create(configmap)
+			return err
+		},
+		func(client kubeclientset.Interface, obj pkg_runtime.Object) error {
+			configmap := obj.(*api_v1.ConfigMap)
+			_, err := client.Core().ConfigMaps(configmap.Namespace).Update(configmap)
+			return err
+		},
+		func(client kubeclientset.Interface, obj pkg_runtime.Object) error {
+			configmap := obj.(*api_v1.ConfigMap)
+			err := client.Core().ConfigMaps(configmap.Namespace).Delete(configmap.Name, &api_v1.DeleteOptions{})
+			return err
+		})
+	return configmapcontroller
+}
+
+func (configmapcontroller *ConfigMapController) Run(stopChan <-chan struct{}) {
+	go configmapcontroller.configmapInformerController.Run(stopChan)
+	configmapcontroller.configmapFederatedInformer.Start()
+	go func() {
+		<-stopChan
+		configmapcontroller.configmapFederatedInformer.Stop()
+	}()
+	configmapcontroller.configmapDeliverer.StartWithHandler(func(item *util.DelayingDelivererItem) {
+		configmap := item.Value.(*types.NamespacedName)
+		configmapcontroller.reconcileConfigMap(*configmap)
+	})
+	configmapcontroller.clusterDeliverer.StartWithHandler(func(_ *util.DelayingDelivererItem) {
+		configmapcontroller.reconcileConfigMapsOnClusterChange()
+	})
+	util.StartBackoffGC(configmapcontroller.configmapBackoff, stopChan)
+}
+
+func (configmapcontroller *ConfigMapController) deliverConfigMapObj(obj interface{}, delay time.Duration, failed bool) {
+	configmap := obj.(*api_v1.ConfigMap)
+	configmapcontroller.deliverConfigMap(types.NamespacedName{Namespace: configmap.Namespace, Name: configmap.Name}, delay, failed)
+}
+
+// Adds backoff to delay if this delivery is related to some failure. Resets backoff if there was no failure.
+func (configmapcontroller *ConfigMapController) deliverConfigMap(configmap types.NamespacedName, delay time.Duration, failed bool) {
+	key := configmap.String()
+	if failed {
+		configmapcontroller.configmapBackoff.Next(key, time.Now())
+		delay = delay + configmapcontroller.configmapBackoff.Get(key)
+	} else {
+		configmapcontroller.configmapBackoff.Reset(key)
+	}
+	configmapcontroller.configmapDeliverer.DeliverAfter(key, &configmap, delay)
+}
+
+// Check whether all data stores are in sync. False is returned if any of the informer/stores is not yet
+// synced with the corresponding api server.
+func (configmapcontroller *ConfigMapController) isSynced() bool {
+	if !configmapcontroller.configmapFederatedInformer.ClustersSynced() {
+		glog.V(2).Infof("Cluster list not synced")
+		return false
+	}
+	clusters, err := configmapcontroller.configmapFederatedInformer.GetReadyClusters()
+	if err != nil {
+		glog.Errorf("Failed to get ready clusters: %v", err)
+		return false
+	}
+	if !configmapcontroller.configmapFederatedInformer.GetTargetStore().ClustersSynced(clusters) {
+		return false
+	}
+	return true
+}
+
+// The function triggers reconciliation of all federated configmaps.
+func (configmapcontroller *ConfigMapController) reconcileConfigMapsOnClusterChange() {
+	if !configmapcontroller.isSynced() {
+		glog.V(4).Infof("Configmap controller not synced")
+		configmapcontroller.clusterDeliverer.DeliverAt(allClustersKey, nil, time.Now().Add(configmapcontroller.clusterAvailableDelay))
+	}
+	for _, obj := range configmapcontroller.configmapInformerStore.List() {
+		configmap := obj.(*api_v1.ConfigMap)
+		configmapcontroller.deliverConfigMap(types.NamespacedName{Namespace: configmap.Namespace, Name: configmap.Name},
+			configmapcontroller.smallDelay, false)
+	}
+}
+
+func (configmapcontroller *ConfigMapController) reconcileConfigMap(configmap types.NamespacedName) {
+
+	if !configmapcontroller.isSynced() {
+		glog.V(4).Infof("Configmap controller not synced")
+		configmapcontroller.deliverConfigMap(configmap, configmapcontroller.clusterAvailableDelay, false)
+		return
+	}
+
+	key := configmap.String()
+	baseConfigMapObj, exist, err := configmapcontroller.configmapInformerStore.GetByKey(key)
+	if err != nil {
+		glog.Errorf("Failed to query main configmap store for %v: %v", key, err)
+		configmapcontroller.deliverConfigMap(configmap, 0, true)
+		return
+	}
+
+	if !exist {
+		// Not federated configmap, ignoring.
+		glog.V(8).Infof("Skipping not federated config map: %s", key)
+		return
+	}
+	baseConfigMap := baseConfigMapObj.(*api_v1.ConfigMap)
+
+	clusters, err := configmapcontroller.configmapFederatedInformer.GetReadyClusters()
+	if err != nil {
+		glog.Errorf("Failed to get cluster list: %v, retrying shortly", err)
+		configmapcontroller.deliverConfigMap(configmap, configmapcontroller.clusterAvailableDelay, false)
+		return
+	}
+
+	operations := make([]util.FederatedOperation, 0)
+	for _, cluster := range clusters {
+		clusterConfigMapObj, found, err := configmapcontroller.configmapFederatedInformer.GetTargetStore().GetByKey(cluster.Name, key)
+		if err != nil {
+			glog.Errorf("Failed to get %s from %s: %v, retrying shortly", key, cluster.Name, err)
+			configmapcontroller.deliverConfigMap(configmap, 0, true)
+			return
+		}
+
+		desiredConfigMap := &api_v1.ConfigMap{
+			ObjectMeta: util.CopyObjectMeta(baseConfigMap.ObjectMeta),
+			Data:       baseConfigMap.Data,
+		}
+
+		if !found {
+			configmapcontroller.eventRecorder.Eventf(baseConfigMap, api.EventTypeNormal, "CreateInCluster",
+				"Creating configmap in cluster %s", cluster.Name)
+
+			operations = append(operations, util.FederatedOperation{
+				Type:        util.OperationTypeAdd,
+				Obj:         desiredConfigMap,
+				ClusterName: cluster.Name,
+			})
+		} else {
+			clusterConfigMap := clusterConfigMapObj.(*api_v1.ConfigMap)
+
+			// Update existing configmap, if needed.
+			if !util.ConfigMapEquivalent(desiredConfigMap, clusterConfigMap) {
+				configmapcontroller.eventRecorder.Eventf(baseConfigMap, api.EventTypeNormal, "UpdateInCluster",
+					"Updating configmap in cluster %s", cluster.Name)
+				operations = append(operations, util.FederatedOperation{
+					Type:        util.OperationTypeUpdate,
+					Obj:         desiredConfigMap,
+					ClusterName: cluster.Name,
+				})
+			}
+		}
+	}
+
+	if len(operations) == 0 {
+		// Everything is in order
+		glog.V(8).Infof("No operations needed for %s", key)
+		return
+	}
+	err = configmapcontroller.federatedUpdater.UpdateWithOnError(operations, configmapcontroller.updateTimeout,
+		func(op util.FederatedOperation, operror error) {
+			configmapcontroller.eventRecorder.Eventf(baseConfigMap, api.EventTypeNormal, "UpdateInClusterFailed",
+				"ConfigMap update in cluster %s failed: %v", op.ClusterName, operror)
+		})
+
+	if err != nil {
+		glog.Errorf("Failed to execute updates for %s: %v, retrying shortly", key, err)
+		configmapcontroller.deliverConfigMap(configmap, 0, true)
+		return
+	}
+}

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

@@ -0,0 +1,142 @@
+/*
+Copyright 2016 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 configmap
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	federation_api "k8s.io/kubernetes/federation/apis/federation/v1beta1"
+	fake_fedclientset "k8s.io/kubernetes/federation/client/clientset_generated/federation_release_1_5/fake"
+	"k8s.io/kubernetes/federation/pkg/federation-controller/util"
+	. "k8s.io/kubernetes/federation/pkg/federation-controller/util/test"
+	api_v1 "k8s.io/kubernetes/pkg/api/v1"
+	kubeclientset "k8s.io/kubernetes/pkg/client/clientset_generated/release_1_5"
+	fake_kubeclientset "k8s.io/kubernetes/pkg/client/clientset_generated/release_1_5/fake"
+	"k8s.io/kubernetes/pkg/runtime"
+	"k8s.io/kubernetes/pkg/types"
+	"k8s.io/kubernetes/pkg/util/wait"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestConfigMapController(t *testing.T) {
+	cluster1 := NewCluster("cluster1", api_v1.ConditionTrue)
+	cluster2 := NewCluster("cluster2", api_v1.ConditionTrue)
+
+	fakeClient := &fake_fedclientset.Clientset{}
+	RegisterFakeList("clusters", &fakeClient.Fake, &federation_api.ClusterList{Items: []federation_api.Cluster{*cluster1}})
+	RegisterFakeList("configmaps", &fakeClient.Fake, &api_v1.ConfigMapList{Items: []api_v1.ConfigMap{}})
+	configmapWatch := RegisterFakeWatch("configmaps", &fakeClient.Fake)
+	clusterWatch := RegisterFakeWatch("clusters", &fakeClient.Fake)
+
+	cluster1Client := &fake_kubeclientset.Clientset{}
+	cluster1Watch := RegisterFakeWatch("configmaps", &cluster1Client.Fake)
+	RegisterFakeList("configmaps", &cluster1Client.Fake, &api_v1.ConfigMapList{Items: []api_v1.ConfigMap{}})
+	cluster1CreateChan := RegisterFakeCopyOnCreate("configmaps", &cluster1Client.Fake, cluster1Watch)
+	cluster1UpdateChan := RegisterFakeCopyOnUpdate("configmaps", &cluster1Client.Fake, cluster1Watch)
+
+	cluster2Client := &fake_kubeclientset.Clientset{}
+	cluster2Watch := RegisterFakeWatch("configmaps", &cluster2Client.Fake)
+	RegisterFakeList("configmaps", &cluster2Client.Fake, &api_v1.ConfigMapList{Items: []api_v1.ConfigMap{}})
+	cluster2CreateChan := RegisterFakeCopyOnCreate("configmaps", &cluster2Client.Fake, cluster2Watch)
+
+	configmapController := NewConfigMapController(fakeClient)
+	informer := ToFederatedInformerForTestOnly(configmapController.configmapFederatedInformer)
+	informer.SetClientFactory(func(cluster *federation_api.Cluster) (kubeclientset.Interface, error) {
+		switch cluster.Name {
+		case cluster1.Name:
+			return cluster1Client, nil
+		case cluster2.Name:
+			return cluster2Client, nil
+		default:
+			return nil, fmt.Errorf("Unknown cluster")
+		}
+	})
+
+	configmapController.clusterAvailableDelay = time.Second
+	configmapController.configmapReviewDelay = 50 * time.Millisecond
+	configmapController.smallDelay = 20 * time.Millisecond
+	configmapController.updateTimeout = 5 * time.Second
+
+	stop := make(chan struct{})
+	configmapController.Run(stop)
+
+	configmap1 := &api_v1.ConfigMap{
+		ObjectMeta: api_v1.ObjectMeta{
+			Name:      "test-configmap",
+			Namespace: "ns",
+			SelfLink:  "/api/v1/namespaces/ns/configmaps/test-configmap",
+		},
+		Data: map[string]string{
+			"A": "ala ma kota",
+			"B": "quick brown fox",
+		},
+	}
+
+	// Test add federated configmap.
+	configmapWatch.Add(configmap1)
+	createdConfigMap := GetConfigMapFromChan(cluster1CreateChan)
+	assert.NotNil(t, createdConfigMap)
+	assert.Equal(t, configmap1.Namespace, createdConfigMap.Namespace)
+	assert.Equal(t, configmap1.Name, createdConfigMap.Name)
+	assert.True(t, util.ConfigMapEquivalent(configmap1, createdConfigMap))
+
+	// Wait for the configmap to appear in the informer store
+	err := WaitForStoreUpdate(
+		configmapController.configmapFederatedInformer.GetTargetStore(),
+		cluster1.Name, types.NamespacedName{Namespace: configmap1.Namespace, Name: configmap1.Name}.String(), wait.ForeverTestTimeout)
+	assert.Nil(t, err, "configmap should have appeared in the informer store")
+
+	// Test update federated configmap.
+	configmap1.Annotations = map[string]string{
+		"A": "B",
+	}
+	configmapWatch.Modify(configmap1)
+	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))
+
+	// Test update federated configmap.
+	configmap1.Data = map[string]string{
+		"config": "myconfigurationfile",
+	}
+	configmapWatch.Modify(configmap1)
+	updatedConfigMap2 := 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))
+
+	// Test add cluster
+	clusterWatch.Add(cluster2)
+	createdConfigMap2 := GetConfigMapFromChan(cluster2CreateChan)
+	assert.NotNil(t, createdConfigMap2)
+	assert.Equal(t, configmap1.Name, createdConfigMap2.Name)
+	assert.Equal(t, configmap1.Namespace, createdConfigMap2.Namespace)
+	assert.True(t, util.ConfigMapEquivalent(configmap1, createdConfigMap2))
+
+	close(stop)
+}
+
+func GetConfigMapFromChan(c chan runtime.Object) *api_v1.ConfigMap {
+	configmap := GetObjectFromChan(c).(*api_v1.ConfigMap)
+	return configmap
+}

federation/pkg/federation-controller/util/BUILD

@@ -15,6 +15,7 @@ go_library(
     srcs = [
         "backoff.go",
         "cluster_util.go",
+        "configmap.go",
         "delaying_deliverer.go",
         "federated_informer.go",
         "federated_updater.go",

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

@@ -0,0 +1,31 @@
+/*
+Copyright 2016 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 util
+
+import (
+	"reflect"
+
+	api_v1 "k8s.io/kubernetes/pkg/api/v1"
+)
+
+// Checks if cluster-independent, user provided data in two given ConfigMapss are eqaul. If in
+// the future the ConfigMap structure is expanded then any field that is not populated.
+// by the api server should be included here.
+func ConfigMapEquivalent(s1, s2 *api_v1.ConfigMap) bool {
+	return ObjectMetaEquivalent(s1.ObjectMeta, s2.ObjectMeta) &&
+		reflect.DeepEqual(s1.Data, s2.Data)
+}

test/test_owners.csv

@@ -477,6 +477,7 @@ k8s.io/kubernetes/federation/pkg/dnsprovider,sttts,1
 k8s.io/kubernetes/federation/pkg/dnsprovider/providers/aws/route53,cjcullen,1
 k8s.io/kubernetes/federation/pkg/dnsprovider/providers/google/clouddns,jsafrane,1
 k8s.io/kubernetes/federation/pkg/federation-controller/cluster,nikhiljindal,0
+k8s.io/kubernetes/federation/pkg/federation-controller/configmap,mwielgus,0
 k8s.io/kubernetes/federation/pkg/federation-controller/daemonset,childsb,1
 k8s.io/kubernetes/federation/pkg/federation-controller/deployment,zmerlynn,1
 k8s.io/kubernetes/federation/pkg/federation-controller/ingress,vishh,1
@@ -800,6 +801,7 @@ k8s.io/kubernetes/pkg/util/sets,quinton-hoole,0
 k8s.io/kubernetes/pkg/util/slice,quinton-hoole,0
 k8s.io/kubernetes/pkg/util/strategicpatch,brendandburns,1
 k8s.io/kubernetes/pkg/util/strings,quinton-hoole,0
+k8s.io/kubernetes/pkg/util/system,mwielgus,0
 k8s.io/kubernetes/pkg/util/term,davidopp,1
 k8s.io/kubernetes/pkg/util/testing,jlowdermilk,1
 k8s.io/kubernetes/pkg/util/threading,roberthbailey,1