Merge pull request #41942 from csbell/fw-name

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

@@ -17,6 +17,7 @@ limitations under the License.
 package ingress
 
 import (
+	"crypto/md5"
 	"fmt"
 	"sync"
 	"time"
@@ -55,6 +56,7 @@ const (
 	uidConfigMapName        = "ingress-uid"                                 // Name of the config-map and key the ingress controller stores its uid in.
 	uidConfigMapNamespace   = "kube-system"
 	uidKey                  = "uid"
+	providerUidKey          = "provider-uid"
 	// Annotation on the ingress in federation control plane that is used to keep
 	// track of the first cluster in which we create ingress.
 	// We wait for ingress to be created in this cluster before creating it any
@@ -272,7 +274,7 @@ func NewIngressController(client federationclientset.Interface) *IngressControll
 			glog.V(4).Infof("Attempting to update ConfigMap: %v", configMap)
 			_, err := client.Core().ConfigMaps(configMap.Namespace).Update(configMap)
 			if err == nil {
-				glog.V(4).Infof("Successfully updated ConfigMap %q", configMapName)
+				glog.V(4).Infof("Successfully updated ConfigMap %q %v", configMapName, configMap)
 			} else {
 				glog.V(4).Infof("Failed to update ConfigMap %q: %v", configMapName, err)
 			}
@@ -523,7 +525,10 @@ func (ic *IngressController) reconcileConfigMapForCluster(clusterName string) {
 		uidConfigMapNamespacedName := types.NamespacedName{Name: uidConfigMapName, Namespace: uidConfigMapNamespace}
 		configMapObj, found, err := ic.configMapFederatedInformer.GetTargetStore().GetByKey(cluster.Name, uidConfigMapNamespacedName.String())
 		if !found || err != nil {
-			logmsg := fmt.Sprintf("Failed to get ConfigMap %q for cluster %q.  Will try again later: %v", uidConfigMapNamespacedName, cluster.Name, err)
+			logmsg := fmt.Sprintf("Failed to get ConfigMap %q for cluster %q.  Will try again later", uidConfigMapNamespacedName, cluster.Name)
+			if err != nil {
+				logmsg = fmt.Sprintf("%v: %v", logmsg, err)
+			}
 			if len(ic.ingressInformerStore.List()) > 0 { // Error-level if ingresses are active, Info-level otherwise.
 				glog.Errorf(logmsg)
 			} else {
@@ -543,6 +548,12 @@ func (ic *IngressController) reconcileConfigMapForCluster(clusterName string) {
 	}
 }
 
+// getProviderUid returns a provider ID based on the provided clusterName.
+func getProviderUid(clusterName string) string {
+	hashedName := md5.Sum([]byte(clusterName))
+	return fmt.Sprintf("%x", hashedName[:8])
+}
+
 /*
   reconcileConfigMap ensures that the configmap in the cluster has a UID
   consistent with the federation cluster's associated annotation.
@@ -570,12 +581,29 @@ func (ic *IngressController) reconcileConfigMap(cluster *federationapi.Cluster,
 	}
 
 	if !clusterIngressUIDExists || clusterIngressUID == "" {
-		ic.updateClusterIngressUIDToMasters(cluster, configMapUID) // Second argument is the fallback, in case this is the only cluster, in which case it becomes the master
-		return
+		glog.V(4).Infof("Cluster %q is the only master", cluster.Name)
+		// Second argument is the fallback, in case this is the only cluster, in which case it becomes the master
+		var err error
+		if clusterIngressUID, err = ic.updateClusterIngressUIDToMasters(cluster, configMapUID); err != nil {
+			return
+		}
+		// If we successfully update the Cluster Object, fallthrough and update the configMap.
 	}
-	if configMapUID != clusterIngressUID { // An update is required
-		glog.V(4).Infof("Ingress UID update is required: configMapUID %q not equal to clusterIngressUID %q", configMapUID, clusterIngressUID)
+
+	// Figure out providerUid.
+	providerUid := getProviderUid(cluster.Name)
+	configMapProviderUid := configMap.Data[providerUidKey]
+
+	if configMapUID == clusterIngressUID && configMapProviderUid == providerUid {
+		glog.V(4).Infof("Ingress configMap update is not required: UID %q and ProviderUid %q are equal", configMapUID, providerUid)
+	} else {
+		if configMapUID != clusterIngressUID {
+			glog.V(4).Infof("Ingress configMap update is required for UID: configMapUID %q not equal to clusterIngressUID %q", configMapUID, clusterIngressUID)
+		} else if configMapProviderUid != providerUid {
+			glog.V(4).Infof("Ingress configMap update is required: configMapProviderUid %q not equal to providerUid %q", configMapProviderUid, providerUid)
+		}
 		configMap.Data[uidKey] = clusterIngressUID
+		configMap.Data[providerUidKey] = providerUid
 		operations := []util.FederatedOperation{{
 			Type:        util.OperationTypeUpdate,
 			Obj:         configMap,
@@ -588,8 +616,6 @@ func (ic *IngressController) reconcileConfigMap(cluster *federationapi.Cluster,
 			glog.Errorf("Failed to execute update of ConfigMap %q on cluster %q: %v", configMapName, cluster.Name, err)
 			ic.configMapDeliverer.DeliverAfter(cluster.Name, nil, ic.configMapReviewDelay)
 		}
-	} else {
-		glog.V(4).Infof("Ingress UID update is not required: configMapUID %q is equal to clusterIngressUID %q", configMapUID, clusterIngressUID)
 	}
 }
 
@@ -619,13 +645,13 @@ func (ic *IngressController) getMasterCluster() (master *federationapi.Cluster,
   updateClusterIngressUIDToMasters takes the ingress UID annotation on the master cluster and applies it to cluster.
   If there is no master cluster, then fallbackUID is used (and hence this cluster becomes the master).
 */
-func (ic *IngressController) updateClusterIngressUIDToMasters(cluster *federationapi.Cluster, fallbackUID string) {
+func (ic *IngressController) updateClusterIngressUIDToMasters(cluster *federationapi.Cluster, fallbackUID string) (string, error) {
 	masterCluster, masterUID, err := ic.getMasterCluster()
 	clusterObj, clusterErr := api.Scheme.DeepCopy(cluster) // Make a clone so that we don't clobber our input param
 	cluster, ok := clusterObj.(*federationapi.Cluster)
 	if clusterErr != nil || !ok {
 		glog.Errorf("Internal error: Failed clone cluster resource while attempting to add master ingress UID annotation (%q = %q) from master cluster %q to cluster %q, will try again later: %v", uidAnnotationKey, masterUID, masterCluster.Name, cluster.Name, err)
-		return
+		return "", err
 	}
 	if err == nil {
 		if masterCluster.Name != cluster.Name { // We're not the master, need to get in sync
@@ -635,12 +661,14 @@ func (ic *IngressController) updateClusterIngressUIDToMasters(cluster *federatio
 			cluster.ObjectMeta.Annotations[uidAnnotationKey] = masterUID
 			if _, err = ic.federatedApiClient.Federation().Clusters().Update(cluster); err != nil {
 				glog.Errorf("Failed to add master ingress UID annotation (%q = %q) from master cluster %q to cluster %q, will try again later: %v", uidAnnotationKey, masterUID, masterCluster.Name, cluster.Name, err)
-				return
+				return "", err
 			} else {
 				glog.V(4).Infof("Successfully added master ingress UID annotation (%q = %q) from master cluster %q to cluster %q.", uidAnnotationKey, masterUID, masterCluster.Name, cluster.Name)
+				return masterUID, nil
 			}
 		} else {
 			glog.V(4).Infof("Cluster %q with ingress UID is already the master with annotation (%q = %q), no need to update.", cluster.Name, uidAnnotationKey, cluster.ObjectMeta.Annotations[uidAnnotationKey])
+			return cluster.ObjectMeta.Annotations[uidAnnotationKey], nil
 		}
 	} else {
 		glog.V(2).Infof("No master cluster found to source an ingress UID from for cluster %q.  Attempting to elect new master cluster %q with ingress UID %q = %q", cluster.Name, cluster.Name, uidAnnotationKey, fallbackUID)
@@ -651,11 +679,14 @@ func (ic *IngressController) updateClusterIngressUIDToMasters(cluster *federatio
 			cluster.ObjectMeta.Annotations[uidAnnotationKey] = fallbackUID
 			if _, err = ic.federatedApiClient.Federation().Clusters().Update(cluster); err != nil {
 				glog.Errorf("Failed to add ingress UID annotation (%q = %q) to cluster %q. No master elected. Will try again later: %v", uidAnnotationKey, fallbackUID, cluster.Name, err)
+				return "", err
 			} else {
 				glog.V(4).Infof("Successfully added ingress UID annotation (%q = %q) to cluster %q.", uidAnnotationKey, fallbackUID, cluster.Name)
+				return fallbackUID, nil
 			}
 		} else {
 			glog.Errorf("No master cluster exists, and fallbackUID for cluster %q is invalid (%q).  This probably means that no clusters have an ingress controller configmap with key %q.  Federated Ingress currently supports clusters running Google Loadbalancer Controller (\"GLBC\")", cluster.Name, fallbackUID, uidKey)
+			return "", err
 		}
 	}
 }

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

@@ -57,6 +57,7 @@ func TestIngressController(t *testing.T) {
 	cluster2 := NewCluster("cluster2", apiv1.ConditionTrue)
 	cfg1 := NewConfigMap("foo")
 	cfg2 := NewConfigMap("bar") // Different UID from cfg1, so that we can check that they get reconciled.
+	assert.NotEqual(t, cfg1.Data[uidKey], cfg2.Data[uidKey], fmt.Sprintf("ConfigMap in cluster 2 must initially not equal that in cluster 1 for this test - please fix test"))
 
 	t.Log("Creating fake infrastructure")
 	fedClient := &fakefedclientset.Clientset{}
@@ -74,6 +75,7 @@ func TestIngressController(t *testing.T) {
 	cluster1ConfigMapWatch := RegisterFakeWatch(configmaps, &cluster1Client.Fake)
 	cluster1IngressCreateChan := RegisterFakeCopyOnCreate(ingresses, &cluster1Client.Fake, cluster1IngressWatch)
 	cluster1IngressUpdateChan := RegisterFakeCopyOnUpdate(ingresses, &cluster1Client.Fake, cluster1IngressWatch)
+	cluster1ConfigMapUpdateChan := RegisterFakeCopyOnUpdate(configmaps, &cluster1Client.Fake, cluster1ConfigMapWatch)
 
 	cluster2Client := &fakekubeclientset.Clientset{}
 	RegisterFakeList(ingresses, &cluster2Client.Fake, &extensionsv1beta1.IngressList{Items: []extensionsv1beta1.Ingress{}})
@@ -138,11 +140,24 @@ func TestIngressController(t *testing.T) {
 	assert.NotNil(t, cluster)
 	assert.Equal(t, cluster.ObjectMeta.Annotations[uidAnnotationKey], cfg1.Data[uidKey])
 
+	t.Log("Adding cluster 2")
+	clusterWatch.Add(cluster2)
+	cluster2ConfigMapWatch.Add(cfg2)
+
+	t.Log("Checking that a configmap updates are propagated prior to Federated Ingress")
+	referenceUid := cfg1.Data[uidKey]
+	uid1, providerId1 := GetConfigMapUidAndProviderId(t, cluster1ConfigMapUpdateChan)
+	uid2, providerId2 := GetConfigMapUidAndProviderId(t, cluster2ConfigMapUpdateChan)
+	t.Logf("uid2 = %v and ref = %v", uid2, referenceUid)
+	assert.True(t, referenceUid == uid1, "Expected cluster1 configmap uid %q to be equal to referenceUid %q", uid1, referenceUid)
+	assert.True(t, referenceUid == uid2, "Expected cluster2 configmap uid %q to be equal to referenceUid %q", uid2, referenceUid)
+	assert.True(t, providerId1 != providerId2, "Expected cluster1 providerUid %q to be unique and different from cluster2 providerUid %q", providerId1, providerId2)
+
 	// Test add federated ingress.
 	t.Log("Adding Federated Ingress")
 	fedIngressWatch.Add(&fedIngress)
 
-	t.Logf("Checking that appropriate finalizers are added")
+	t.Log("Checking that appropriate finalizers are added")
 	// There should be an update to add both the finalizers.
 	updatedIngress := GetIngressFromChan(t, fedIngressUpdateChan)
 	assert.True(t, ingressController.hasFinalizerFunc(updatedIngress, deletionhelper.FinalizerDeleteFromUnderlyingClusters))
@@ -243,9 +258,7 @@ func TestIngressController(t *testing.T) {
 	fedIngress.Annotations[staticIPNameKeyWritable] = "foo" // Make sure that the base object has a static IP name first.
 	fedIngressWatch.Modify(&fedIngress)
 	clusterWatch.Add(cluster2)
-	// First check that the original values are not equal - see above comment
-	assert.NotEqual(t, cfg1.Data[uidKey], cfg2.Data[uidKey], fmt.Sprintf("ConfigMap in cluster 2 must initially not equal that in cluster 1 for this test - please fix test"))
-	cluster2ConfigMapWatch.Add(cfg2)
+
 	t.Log("Checking that the ingress got created in cluster 2")
 	createdIngress2 := GetIngressFromChan(t, cluster2IngressCreateChan)
 	assert.NotNil(t, createdIngress2)
@@ -253,17 +266,22 @@ func TestIngressController(t *testing.T) {
 	t.Logf("created meta: %v fed meta: %v", createdIngress2.ObjectMeta, fedIngress.ObjectMeta)
 	assert.True(t, util.ObjectMetaEquivalent(fedIngress.ObjectMeta, createdIngress2.ObjectMeta), "Metadata of created object is not equivalent")
 
-	t.Log("Checking that the configmap in cluster 2 got updated.")
-	updatedConfigMap2 := GetConfigMapFromChan(cluster2ConfigMapUpdateChan)
-	assert.NotNil(t, updatedConfigMap2, fmt.Sprintf("ConfigMap in cluster 2 was not updated (or more likely the test is broken and the API type written is wrong)"))
-	if updatedConfigMap2 != nil {
-		assert.Equal(t, cfg1.Data[uidKey], updatedConfigMap2.Data[uidKey],
-			fmt.Sprintf("UID's in configmaps in cluster's 1 and 2 are not equal (%q != %q)", cfg1.Data["uid"], updatedConfigMap2.Data["uid"]))
-	}
-
 	close(stop)
 }
 
+func GetConfigMapUidAndProviderId(t *testing.T, c chan runtime.Object) (string, string) {
+	updatedConfigMap := GetConfigMapFromChan(c)
+	assert.NotNil(t, updatedConfigMap, "ConfigMap should have received an update")
+	assert.NotNil(t, updatedConfigMap.Data, "ConfigMap data is empty")
+
+	for _, key := range []string{uidKey, providerUidKey} {
+		val, ok := updatedConfigMap.Data[key]
+		assert.True(t, ok, fmt.Sprintf("Didn't receive an update for key %v: %v", key, updatedConfigMap.Data))
+		assert.True(t, len(val) > 0, fmt.Sprintf("Received an empty update for key %v", key))
+	}
+	return updatedConfigMap.Data[uidKey], updatedConfigMap.Data[providerUidKey]
+}
+
 func GetIngressFromChan(t *testing.T, c chan runtime.Object) *extensionsv1beta1.Ingress {
 	obj := GetObjectFromChan(c)
 

test/e2e_federation/BUILD

@@ -42,6 +42,7 @@ go_library(
         "//vendor:k8s.io/apimachinery/pkg/api/errors",
         "//vendor:k8s.io/apimachinery/pkg/apis/meta/v1",
         "//vendor:k8s.io/apimachinery/pkg/util/intstr",
+        "//vendor:k8s.io/apimachinery/pkg/util/net",
         "//vendor:k8s.io/apimachinery/pkg/util/wait",
         "//vendor:k8s.io/client-go/rest",
         "//vendor:k8s.io/client-go/tools/clientcmd",

test/e2e_federation/util.go

@@ -48,6 +48,10 @@ var (
 	UserAgentName                 = "federation-e2e"
 	// We use this to decide how long to wait for our DNS probes to succeed.
 	DNSTTL = 180 * time.Second // TODO: make k8s.io/kubernetes/federation/pkg/federation-controller/service.minDnsTtl exported, and import it here.
+
+	// FederatedSvcNodePort is the node port on which the federated
+	// service shards are exposed in the underlying cluster.
+	FederatedSvcNodePort = int32(32256)
 )
 
 var FederationSuite common.Suite
@@ -259,6 +263,7 @@ func createService(clientset *fedclientset.Clientset, namespace, name string) (*
 					Protocol:   v1.ProtocolTCP,
 					Port:       80,
 					TargetPort: intstr.FromInt(8080),
+					NodePort:   FederatedSvcNodePort,
 				},
 			},
 			SessionAffinity: v1.ServiceAffinityNone,
@@ -279,10 +284,11 @@ func deleteServiceOrFail(clientset *fedclientset.Clientset, namespace string, se
 	if clientset == nil || len(namespace) == 0 || len(serviceName) == 0 {
 		Fail(fmt.Sprintf("Internal error: invalid parameters passed to deleteServiceOrFail: clientset: %v, namespace: %v, service: %v", clientset, namespace, serviceName))
 	}
+	framework.Logf("Deleting service %q in namespace %v", serviceName, namespace)
 	err := clientset.Services(namespace).Delete(serviceName, &metav1.DeleteOptions{OrphanDependents: orphanDependents})
 	framework.ExpectNoError(err, "Error deleting service %q from namespace %q", serviceName, namespace)
 	// Wait for the service to be deleted.
-	err = wait.Poll(5*time.Second, 3*wait.ForeverTestTimeout, func() (bool, error) {
+	err = wait.Poll(5*time.Second, 10*wait.ForeverTestTimeout, func() (bool, error) {
 		_, err := clientset.Core().Services(namespace).Get(serviceName, metav1.GetOptions{})
 		if err != nil && errors.IsNotFound(err) {
 			return true, nil