Merge pull request #46453 from emaildanwilson/fedClusterSelectorIngress

Automatic merge from submit-queue (batch tested with PRs 46450, 46272, 46453, 46019, 46367)

Add ClusterSelector to Ingress Controller

This pull request adds ClusterSelector to the Federated Ingress Controller ref: design #29887 
This back ports the same functionality from the sync controller (merged pull #40234) in order to make this feature available across all Controllers for the 1.7 release.

cc: @kubernetes/sig-federation-pr-reviews @shashidharatd

**Release note**:
```
The annotation `federation.alpha.kubernetes.io/cluster-selector` can be used with Ingress objects to target federated clusters by label.
```

federation/pkg/federation-controller/ingress/BUILD

@@ -16,6 +16,7 @@ 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/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/ingress/ingress_controller.go

@@ -35,6 +35,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/clusterselector"
 	"k8s.io/kubernetes/federation/pkg/federation-controller/util/deletionhelper"
 	"k8s.io/kubernetes/federation/pkg/federation-controller/util/eventsink"
 	"k8s.io/kubernetes/pkg/api"
@@ -765,7 +766,14 @@ func (ic *IngressController) reconcileIngress(ingress types.NamespacedName) {
 		desiredIngress.Spec = *objSpecCopy
 		glog.V(4).Infof("Desired Ingress: %v", desiredIngress)
 
-		if !clusterIngressFound {
+		send, err := clusterselector.SendToCluster(cluster.Labels, desiredIngress.ObjectMeta.Annotations)
+		if err != nil {
+			glog.Errorf("Error processing ClusterSelector cluster: %s for Ingress map: %s error: %s", cluster.Name, key, err.Error())
+			return
+		}
+
+		switch {
+		case !clusterIngressFound && send:
 			glog.V(4).Infof("No existing Ingress %s in cluster %s - checking if appropriate to queue a create operation", ingress, cluster.Name)
 			// We can't supply server-created fields when creating a new object.
 			desiredIngress.ObjectMeta = util.DeepCopyRelevantObjectMeta(baseIngress.ObjectMeta)
@@ -799,7 +807,15 @@ func (ic *IngressController) reconcileIngress(ingress types.NamespacedName) {
 			} else {
 				glog.V(4).Infof("No annotation %q exists on ingress %q in federation and waiting for ingress in cluster %s. Not queueing create operation for ingress until annotation exists", staticIPNameKeyWritable, ingress, firstClusterName)
 			}
-		} else {
+		case clusterIngressFound && !send:
+			glog.V(5).Infof("Removing Ingress: %s from cluster: %s reason: cluster selectors do not match: %-v %-v", key, cluster.Name, cluster.ObjectMeta.Labels, desiredIngress.ObjectMeta.Annotations[federationapi.FederationClusterSelectorAnnotation])
+			operations = append(operations, util.FederatedOperation{
+				Type:        util.OperationTypeDelete,
+				Obj:         desiredIngress,
+				ClusterName: cluster.Name,
+				Key:         key,
+			})
+		case clusterIngressFound && send:
 			clusterIngress := clusterIngressObj.(*extensionsv1beta1.Ingress)
 			glog.V(4).Infof("Found existing Ingress %s in cluster %s - checking if update is required (in either direction)", ingress, cluster.Name)
 			clusterIPName, clusterIPNameExists := clusterIngress.ObjectMeta.Annotations[staticIPNameKeyReadonly]

federation/pkg/federation-controller/ingress/ingress_controller_test.go

@@ -233,6 +233,7 @@ func TestIngressController(t *testing.T) {
 		fedIngress.ObjectMeta.Annotations = make(map[string]string)
 	}
 	fedIngress.ObjectMeta.Annotations["A"] = "B"
+	fedIngress.ObjectMeta.Annotations[federationapi.FederationClusterSelectorAnnotation] = `[{"key": "cluster", "operator": "in", "values": ["cluster1","cluster2"]}]`
 	t.Log("Modifying Federated Ingress")
 	fedIngressWatch.Modify(&fedIngress)
 	t.Log("Checking that Ingress was correctly updated in cluster 1")

federation/pkg/federation-controller/util/test/test_helper.go

@@ -366,6 +366,7 @@ func NewCluster(name string, readyStatus apiv1.ConditionStatus) *federationapi.C
 		ObjectMeta: metav1.ObjectMeta{
 			Name:        name,
 			Annotations: map[string]string{},
+			Labels:      map[string]string{"cluster": name},
 		},
 		Status: federationapi.ClusterStatus{
 			Conditions: []federationapi.ClusterCondition{