use clusterselector w/ federated configmap deploys

federation/apis/federation/v1beta1/generated.proto

@@ -83,6 +83,27 @@ message ClusterList {
   repeated Cluster items = 2;
 }
 
+// ClusterSelectorRequirement contains values, a key, and an operator that relates the key and values.
+// The zero value of ClusterSelectorRequirement is invalid.
+// ClusterSelectorRequirement implements both set based match and exact match
+message ClusterSelectorRequirement {
+  // +patchMergeKey=key
+  // +patchStrategy=merge
+  optional string key = 1;
+
+  // The Operator defines how the Key is matched to the Values. One of "in", "notin",
+  // "exists", "!", "=", "!=", "gt" or "lt".
+  optional string operator = 2;
+
+  // An array of string values. If the operator is "in" or "notin",
+  // the values array must be non-empty. If the operator is "exists" or "!",
+  // the values array must be empty. If the operator is "gt" or "lt", the values
+  // array must have a single element, which will be interpreted as an integer.
+  // This array is replaced during a strategic merge patch.
+  // +optional
+  repeated string values = 3;
+}
+
 // ClusterSpec describes the attributes of a kubernetes cluster.
 message ClusterSpec {
   // A map of client CIDR to server address.

federation/apis/federation/v1beta1/types.go

@@ -123,7 +123,32 @@ type ClusterList struct {
 	Items []Cluster `json:"items" protobuf:"bytes,2,rep,name=items"`
 }
 
+// Expressed as value of annotation for selecting the clusters on which a resource is created.
+type ClusterSelector []ClusterSelectorRequirement
+
+// ClusterSelectorRequirement contains values, a key, and an operator that relates the key and values.
+// The zero value of ClusterSelectorRequirement is invalid.
+// ClusterSelectorRequirement implements both set based match and exact match
+type ClusterSelectorRequirement struct {
+	// +patchMergeKey=key
+	// +patchStrategy=merge
+	Key string `json:"key" patchStrategy:"merge" patchMergeKey:"key" protobuf:"bytes,1,opt,name=key"`
+	// The Operator defines how the Key is matched to the Values. One of "in", "notin",
+	// "exists", "!", "=", "!=", "gt" or "lt".
+	Operator string `json:"operator" protobuf:"bytes,2,opt,name=operator"`
+	// An array of string values. If the operator is "in" or "notin",
+	// the values array must be non-empty. If the operator is "exists" or "!",
+	// the values array must be empty. If the operator is "gt" or "lt", the values
+	// array must have a single element, which will be interpreted as an integer.
+	// This array is replaced during a strategic merge patch.
+	// +optional
+	Values []string `json:"values,omitempty" protobuf:"bytes,3,rep,name=values"`
+}
+
 const (
 	// FederationNamespaceSystem is the system namespace where we place federation control plane components.
 	FederationNamespaceSystem string = "federation-system"
+
+	// FederationClusterSelectorAnnotation is used to determine placement of objects on federated clusters
+	FederationClusterSelectorAnnotation string = "federation.alpha.kubernetes.io/cluster-selector"
 )

federation/apis/federation/v1beta1/types_swagger_doc_generated.go

@@ -62,6 +62,16 @@ func (ClusterList) SwaggerDoc() map[string]string {
 	return map_ClusterList
 }
 
+var map_ClusterSelectorRequirement = map[string]string{
+	"":         "ClusterSelectorRequirement contains values, a key, and an operator that relates the key and values. The zero value of ClusterSelectorRequirement is invalid. ClusterSelectorRequirement implements both set based match and exact match",
+	"operator": "The Operator defines how the Key is matched to the Values. One of \"in\", \"notin\", \"exists\", \"!\", \"=\", \"!=\", \"gt\" or \"lt\".",
+	"values":   "An array of string values. If the operator is \"in\" or \"notin\", the values array must be non-empty. If the operator is \"exists\" or \"!\", the values array must be empty. If the operator is \"gt\" or \"lt\", the values array must have a single element, which will be interpreted as an integer. This array is replaced during a strategic merge patch.",
+}
+
+func (ClusterSelectorRequirement) SwaggerDoc() map[string]string {
+	return map_ClusterSelectorRequirement
+}
+
 var map_ClusterSpec = map[string]string{
 	"": "ClusterSpec describes the attributes of a kubernetes cluster.",
 	"serverAddressByClientCIDRs": "A map of client CIDR to server address. This is to help clients reach servers in the most network-efficient way possible. Clients can use the appropriate server address as per the CIDR that they match. In case of multiple matches, clients should use the longest matching CIDR.",

federation/apis/federation/v1beta1/zz_generated.deepcopy.go

@@ -39,6 +39,7 @@ func RegisterDeepCopies(scheme *runtime.Scheme) error {
 		conversion.GeneratedDeepCopyFunc{Fn: DeepCopy_v1beta1_Cluster, InType: reflect.TypeOf(&Cluster{})},
 		conversion.GeneratedDeepCopyFunc{Fn: DeepCopy_v1beta1_ClusterCondition, InType: reflect.TypeOf(&ClusterCondition{})},
 		conversion.GeneratedDeepCopyFunc{Fn: DeepCopy_v1beta1_ClusterList, InType: reflect.TypeOf(&ClusterList{})},
+		conversion.GeneratedDeepCopyFunc{Fn: DeepCopy_v1beta1_ClusterSelectorRequirement, InType: reflect.TypeOf(&ClusterSelectorRequirement{})},
 		conversion.GeneratedDeepCopyFunc{Fn: DeepCopy_v1beta1_ClusterSpec, InType: reflect.TypeOf(&ClusterSpec{})},
 		conversion.GeneratedDeepCopyFunc{Fn: DeepCopy_v1beta1_ClusterStatus, InType: reflect.TypeOf(&ClusterStatus{})},
 		conversion.GeneratedDeepCopyFunc{Fn: DeepCopy_v1beta1_ServerAddressByClientCIDR, InType: reflect.TypeOf(&ServerAddressByClientCIDR{})},
@@ -97,6 +98,21 @@ func DeepCopy_v1beta1_ClusterList(in interface{}, out interface{}, c *conversion
 	}
 }
 
+// DeepCopy_v1beta1_ClusterSelectorRequirement is an autogenerated deepcopy function.
+func DeepCopy_v1beta1_ClusterSelectorRequirement(in interface{}, out interface{}, c *conversion.Cloner) error {
+	{
+		in := in.(*ClusterSelectorRequirement)
+		out := out.(*ClusterSelectorRequirement)
+		*out = *in
+		if in.Values != nil {
+			in, out := &in.Values, &out.Values
+			*out = make([]string, len(*in))
+			copy(*out, *in)
+		}
+		return nil
+	}
+}
+
 // DeepCopy_v1beta1_ClusterSpec is an autogenerated deepcopy function.
 func DeepCopy_v1beta1_ClusterSpec(in interface{}, out interface{}, c *conversion.Cloner) error {
 	{

federation/pkg/federation-controller/sync/BUILD

@@ -17,6 +17,7 @@ go_library(
         "//federation/client/clientset_generated/federation_clientset:go_default_library",
         "//federation/pkg/federatedtypes:go_default_library",
         "//federation/pkg/federation-controller/util:go_default_library",
+        "//federation/pkg/federation-controller/util/clusterselector: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",

federation/pkg/federation-controller/sync/controller.go

@@ -37,6 +37,7 @@ import (
 	federationclientset "k8s.io/kubernetes/federation/client/clientset_generated/federation_clientset"
 	"k8s.io/kubernetes/federation/pkg/federatedtypes"
 	"k8s.io/kubernetes/federation/pkg/federation-controller/util"
+	"k8s.io/kubernetes/federation/pkg/federation-controller/util/clusterselector"
 	"k8s.io/kubernetes/federation/pkg/federation-controller/util/deletionhelper"
 	"k8s.io/kubernetes/federation/pkg/federation-controller/util/eventsink"
 	"k8s.io/kubernetes/pkg/api"
@@ -356,10 +357,15 @@ func (s *FederationSyncController) reconcile(namespacedName types.NamespacedName
 	}
 
 	operationsAccessor := func(adapter federatedtypes.FederatedTypeAdapter, clusters []*federationapi.Cluster, obj pkgruntime.Object) ([]util.FederatedOperation, error) {
-		return clusterOperations(adapter, clusters, obj, func(clusterName string) (interface{}, bool, error) {
+		operations, err := clusterOperations(adapter, clusters, obj, key, func(clusterName string) (interface{}, bool, error) {
 			return s.informer.GetTargetStore().GetByKey(clusterName, key)
-		})
+		}, clusterselector.SendToCluster)
+		if err != nil {
+			s.eventRecorder.Eventf(obj, api.EventTypeWarning, "FedClusterOperationsError", "Error obtaining sync operations for %s: %s error: %s", kind, key, err.Error())
+		}
+		return operations, err
 	}
+
 	return syncToClusters(
 		s.informer.GetReadyClusters,
 		operationsAccessor,
@@ -451,11 +457,16 @@ func syncToClusters(clustersAccessor clustersAccessorFunc, operationsAccessor op
 }
 
 type clusterObjectAccessorFunc func(clusterName string) (interface{}, bool, error)
+type clusterSelectorFunc func(map[string]string, map[string]string) (bool, error)
 
 // clusterOperations returns the list of operations needed to synchronize the state of the given object to the provided clusters
-func clusterOperations(adapter federatedtypes.FederatedTypeAdapter, clusters []*federationapi.Cluster, obj pkgruntime.Object, accessor clusterObjectAccessorFunc) ([]util.FederatedOperation, error) {
-	key := federatedtypes.ObjectKey(adapter, obj)
+func clusterOperations(adapter federatedtypes.FederatedTypeAdapter, clusters []*federationapi.Cluster, obj pkgruntime.Object, key string, accessor clusterObjectAccessorFunc, selector clusterSelectorFunc) ([]util.FederatedOperation, error) {
+	// The data should not be modified.
+	desiredObj := adapter.Copy(obj)
+	objMeta := adapter.ObjectMeta(desiredObj)
+	kind := adapter.Kind()
 	operations := make([]util.FederatedOperation, 0)
+
 	for _, cluster := range clusters {
 		clusterObj, found, err := accessor(cluster.Name)
 		if err != nil {
@@ -463,18 +474,28 @@ func clusterOperations(adapter federatedtypes.FederatedTypeAdapter, clusters []*
 			runtime.HandleError(wrappedErr)
 			return nil, wrappedErr
 		}
-		// The data should not be modified.
-		desiredObj := adapter.Copy(obj)
+
+		send, err := selector(cluster.Labels, objMeta.Annotations)
+		if err != nil {
+			glog.Errorf("Error processing ClusterSelector cluster: %s for %s map: %s error: %s", cluster.Name, kind, key, err.Error())
+			return nil, err
+		} else if !send {
+			glog.V(5).Infof("Skipping cluster: %s for %s: %s reason: cluster selectors do not match: %-v %-v", cluster.Name, kind, key, cluster.ObjectMeta.Labels, objMeta.Annotations[federationapi.FederationClusterSelectorAnnotation])
+		}
 
 		var operationType util.FederatedOperationType = ""
-		if found {
+		switch {
+		case found && send:
 			clusterObj := clusterObj.(pkgruntime.Object)
 			if !adapter.Equivalent(desiredObj, clusterObj) {
 				operationType = util.OperationTypeUpdate
 			}
-		} else {
+		case found && !send:
+			operationType = util.OperationTypeDelete
+		case !found && send:
 			operationType = util.OperationTypeAdd
 		}
+
 		if len(operationType) > 0 {
 			operations = append(operations, util.FederatedOperation{
 				Type:        operationType,

federation/pkg/federation-controller/sync/controller_test.go

@@ -101,34 +101,55 @@ func TestClusterOperations(t *testing.T) {
 	federatedtypes.SetAnnotation(adapter, differingObj, "foo", "bar")
 
 	testCases := map[string]struct {
-		clusterObject pkgruntime.Object
-		expectedErr   bool
+		clusterObject   pkgruntime.Object
+		expectedErr     bool
+		expectedSendErr bool
+		sendToCluster   bool
+
 		operationType util.FederatedOperationType
 	}{
 		"Accessor error returned": {
 			expectedErr: true,
 		},
+		"sendToCluster error returned": {
+			expectedSendErr: true,
+		},
 		"Missing cluster object should result in add operation": {
 			operationType: util.OperationTypeAdd,
+			sendToCluster: true,
 		},
 		"Differing cluster object should result in update operation": {
 			clusterObject: differingObj,
 			operationType: util.OperationTypeUpdate,
+			sendToCluster: true,
+		},
+		"Matching object and not matching ClusterSelector should result in delete operation": {
+			clusterObject: obj,
+			operationType: util.OperationTypeDelete,
+			sendToCluster: false,
 		},
 		"Matching cluster object should not result in an operation": {
 			clusterObject: obj,
+			sendToCluster: true,
 		},
 	}
 	for testName, testCase := range testCases {
 		t.Run(testName, func(t *testing.T) {
 			clusters := []*federationapi.Cluster{fedtest.NewCluster("cluster1", apiv1.ConditionTrue)}
-			operations, err := clusterOperations(adapter, clusters, obj, func(string) (interface{}, bool, error) {
+			key := federatedtypes.ObjectKey(adapter, obj)
+
+			operations, err := clusterOperations(adapter, clusters, obj, key, func(string) (interface{}, bool, error) {
 				if testCase.expectedErr {
 					return nil, false, awfulError
 				}
 				return testCase.clusterObject, (testCase.clusterObject != nil), nil
+			}, func(map[string]string, map[string]string) (bool, error) {
+				if testCase.expectedSendErr {
+					return false, awfulError
+				}
+				return testCase.sendToCluster, nil
 			})
-			if testCase.expectedErr {
+			if testCase.expectedErr || testCase.expectedSendErr {
 				require.Error(t, err, "An error was expected")
 			} else {
 				require.NoError(t, err, "An error was not expected")

federation/pkg/federation-controller/util/BUILD

@@ -88,6 +88,7 @@ filegroup(
     name = "all-srcs",
     srcs = [
         ":package-srcs",
+        "//federation/pkg/federation-controller/util/clusterselector:all-srcs",
         "//federation/pkg/federation-controller/util/deletionhelper:all-srcs",
         "//federation/pkg/federation-controller/util/eventsink:all-srcs",
         "//federation/pkg/federation-controller/util/finalizers:all-srcs",

federation/pkg/federation-controller/util/clusterselector/BUILD

+package(default_visibility = ["//visibility:public"])
+
+licenses(["notice"])
+
+load(
+    "@io_bazel_rules_go//go:def.bzl",
+    "go_library",
+    "go_test",
+)
+
+go_test(
+    name = "go_default_test",
+    srcs = ["clusterselector_test.go"],
+    library = ":go_default_library",
+    tags = ["automanaged"],
+    deps = [
+        "//federation/apis/federation/v1beta1:go_default_library",
+        "//vendor/github.com/stretchr/testify/require:go_default_library",
+    ],
+)
+
+go_library(
+    name = "go_default_library",
+    srcs = ["clusterselector.go"],
+    tags = ["automanaged"],
+    deps = [
+        "//federation/apis/federation/v1beta1:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/labels:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/selection:go_default_library",
+    ],
+)
+
+filegroup(
+    name = "package-srcs",
+    srcs = glob(["**"]),
+    tags = ["automanaged"],
+    visibility = ["//visibility:private"],
+)
+
+filegroup(
+    name = "all-srcs",
+    srcs = [":package-srcs"],
+    tags = ["automanaged"],
+)

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

+/*
+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 clusterselector
+
+import (
+	"encoding/json"
+
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/selection"
+	federation_v1beta1 "k8s.io/kubernetes/federation/apis/federation/v1beta1"
+)
+
+// Parses the cluster selector annotation to find out if the object with that annotation should be forwarded to a cluster with the given clusterLabels.
+func SendToCluster(clusterLabels map[string]string, annotations map[string]string) (bool, error) {
+	// Check if a ClusterSelector annotation exists and send to all clusters when it does not exist
+	val, ok := annotations[federation_v1beta1.FederationClusterSelectorAnnotation]
+	if !ok {
+		return true, nil
+	}
+
+	selector, err := getSelector(val)
+	if err != nil {
+		return false, err
+	}
+	return selector.Matches(labels.Set(clusterLabels)), nil
+}
+
+func getSelector(annotation string) (labels.Selector, error) {
+	selector := labels.NewSelector()
+	requirements := make([]federation_v1beta1.ClusterSelectorRequirement, 0)
+	err := json.Unmarshal([]byte(annotation), &requirements)
+	if err != nil {
+		return nil, err
+	}
+
+	for _, requirement := range requirements {
+		r, err := labels.NewRequirement(requirement.Key, ConvertOperator(requirement.Operator), requirement.Values)
+		if err != nil {
+			// Stop processing and assume failure since we have no way of knowing the end users intent for this or any other clusters.
+			return nil, err
+		}
+		selector = selector.Add(*r)
+	}
+
+	return selector, nil
+}
+
+// ConvertOperator converts a string operator into selection.Operator type
+func ConvertOperator(source string) selection.Operator {
+	var op selection.Operator
+	switch source {
+	case "!", "DoesNotExist":
+		op = selection.DoesNotExist
+	case "=":
+		op = selection.Equals
+	case "==":
+		op = selection.DoubleEquals
+	case "in", "In":
+		op = selection.In
+	case "!=":
+		op = selection.NotEquals
+	case "notin", "NotIn":
+		op = selection.NotIn
+	case "exists", "Exists":
+		op = selection.Exists
+	case "gt", "Gt", ">":
+		op = selection.GreaterThan
+	case "lt", "Lt", "<":
+		op = selection.LessThan
+	}
+	return op
+}

federation/pkg/federation-controller/util/clusterselector/clusterselector_test.go

+/*
+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 clusterselector
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	federationapi "k8s.io/kubernetes/federation/apis/federation/v1beta1"
+)
+
+func TestSendToCluster(t *testing.T) {
+
+	clusterLabels := map[string]string{
+		"location":    "europe",
+		"environment": "prod",
+		"version":     "15",
+	}
+
+	testCases := map[string]struct {
+		objectAnnotations map[string]string
+		expectedResult    bool
+		expectedErr       bool
+	}{
+		"match with single annotation": {
+			objectAnnotations: map[string]string{
+				federationapi.FederationClusterSelectorAnnotation: `[{"key": "location", "operator": "in", "values": ["europe"]}]`,
+			},
+			expectedResult: true,
+		},
+		"match on multiple annotations": {
+			objectAnnotations: map[string]string{
+				federationapi.FederationClusterSelectorAnnotation: `[{"key": "location", "operator": "in", "values": ["europe"]}, {"key": "environment", "operator": "==", "values": ["prod"]}]`,
+			},
+			expectedResult: true,
+		},
+		"mismatch on one annotation": {
+			objectAnnotations: map[string]string{
+				federationapi.FederationClusterSelectorAnnotation: `[{"key": "location", "operator": "in", "values": ["europe"]}, {"key": "environment", "operator": "==", "values": ["test"]}]`,
+			},
+			expectedResult: false,
+		},
+		"match on not equal annotation": {
+			objectAnnotations: map[string]string{
+				federationapi.FederationClusterSelectorAnnotation: `[{"key": "location", "operator": "!=", "values": ["usa"]}, {"key": "environment", "operator": "in", "values": ["prod"]}]`,
+			},
+			expectedResult: true,
+		},
+		"match on greater than annotation": {
+			objectAnnotations: map[string]string{
+				federationapi.FederationClusterSelectorAnnotation: `[{"key": "version", "operator": ">", "values": ["14"]}]`,
+			},
+			expectedResult: true,
+		},
+		"mismatch on greater than annotation": {
+			objectAnnotations: map[string]string{
+				federationapi.FederationClusterSelectorAnnotation: `[{"key": "version", "operator": ">", "values": ["15"]}]`,
+			},
+			expectedResult: false,
+		},
+		"unable to parse annotation": {
+			objectAnnotations: map[string]string{
+				federationapi.FederationClusterSelectorAnnotation: `[{"not able to parse",}]`,
+			},
+			expectedResult: false,
+			expectedErr:    true,
+		},
+	}
+
+	for testName, testCase := range testCases {
+		t.Run(testName, func(t *testing.T) {
+			result, err := SendToCluster(clusterLabels, testCase.objectAnnotations)
+
+			if testCase.expectedErr {
+				require.Error(t, err, "An error was expected")
+			} else {
+				require.NoError(t, err, "An error was not expected")
+			}
+
+			require.Equal(t, testCase.expectedResult, result, "Unexpected response from SendToCluster")
+		})
+	}
+}