Federated ConfigMap controller

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

+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/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/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/util/wait:go_default_library",
+        "//vendor:github.com/stretchr/testify/assert",
+    ],
+)

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

+/*
+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/configmap.go

+/*
+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)
+}