[Federation] Record events in federated updater

federation/pkg/federation-controller/deployment/deploymentcontroller.go

@@ -188,7 +188,7 @@ func NewDeploymentController(federationClient fedclientset.Interface) *Deploymen
 		),
 	)
 
-	fdc.fedUpdater = fedutil.NewFederatedUpdater(fdc.fedDeploymentInformer,
+	fdc.fedUpdater = fedutil.NewFederatedUpdater(fdc.fedDeploymentInformer, "deployment", fdc.eventRecorder,
 		func(client kubeclientset.Interface, obj runtime.Object) error {
 			rs := obj.(*extensionsv1.Deployment)
 			_, err := client.Extensions().Deployments(rs.Namespace).Create(rs)
@@ -211,10 +211,9 @@ func NewDeploymentController(federationClient fedclientset.Interface) *Deploymen
 		// objNameFunc
 		func(obj runtime.Object) string {
 			deployment := obj.(*extensionsv1.Deployment)
-			return deployment.Name
+			return fmt.Sprintf("%s/%s", deployment.Namespace, deployment.Name)
 		},
 		updateTimeout,
-		fdc.eventRecorder,
 		fdc.fedDeploymentInformer,
 		fdc.fedUpdater,
 	)
@@ -526,13 +525,11 @@ func (fdc *DeploymentController) reconcileDeployment(key string) (reconciliation
 
 		if !exists {
 			if replicas > 0 {
-				fdc.eventRecorder.Eventf(fd, api.EventTypeNormal, "CreateInCluster",
-					"Creating deployment in cluster %s", clusterName)
-
 				operations = append(operations, fedutil.FederatedOperation{
 					Type:        fedutil.OperationTypeAdd,
 					Obj:         ld,
 					ClusterName: clusterName,
+					Key:         key,
 				})
 			}
 		} else {
@@ -541,13 +538,11 @@ func (fdc *DeploymentController) reconcileDeployment(key string) (reconciliation
 			currentLd := ldObj.(*extensionsv1.Deployment)
 			// Update existing replica set, if needed.
 			if !fedutil.DeploymentEquivalent(ld, currentLd) {
-				fdc.eventRecorder.Eventf(fd, api.EventTypeNormal, "UpdateInCluster",
-					"Updating deployment in cluster %s", clusterName)
-
 				operations = append(operations, fedutil.FederatedOperation{
 					Type:        fedutil.OperationTypeUpdate,
 					Obj:         ld,
 					ClusterName: clusterName,
+					Key:         key,
 				})
 				glog.Infof("Updating %s in %s", currentLd.Name, clusterName)
 			}
@@ -572,10 +567,7 @@ func (fdc *DeploymentController) reconcileDeployment(key string) (reconciliation
 		// Everything is in order
 		return statusAllOk, nil
 	}
-	err = fdc.fedUpdater.UpdateWithOnError(operations, updateTimeout, func(op fedutil.FederatedOperation, operror error) {
-		fdc.eventRecorder.Eventf(fd, api.EventTypeWarning, "FailedUpdateInCluster",
-			"Deployment update in cluster %s failed: %v", op.ClusterName, operror)
-	})
+	err = fdc.fedUpdater.Update(operations, updateTimeout)
 	if err != nil {
 		glog.Errorf("Failed to execute updates for %s: %v", key, err)
 		return statusError, err

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

@@ -229,7 +229,7 @@ func NewIngressController(client federationclientset.Interface) *IngressControll
 	)
 
 	// Federated ingress updater along with Create/Update/Delete operations.
-	ic.federatedIngressUpdater = util.NewFederatedUpdater(ic.ingressFederatedInformer,
+	ic.federatedIngressUpdater = util.NewFederatedUpdater(ic.ingressFederatedInformer, "ingress", ic.eventRecorder,
 		func(client kubeclientset.Interface, obj pkgruntime.Object) error {
 			ingress := obj.(*extensionsv1beta1.Ingress)
 			glog.V(4).Infof("Attempting to create Ingress: %v", ingress)
@@ -261,7 +261,7 @@ func NewIngressController(client federationclientset.Interface) *IngressControll
 		})
 
 	// Federated configmap updater along with Create/Update/Delete operations.  Only Update should ever be called.
-	ic.federatedConfigMapUpdater = util.NewFederatedUpdater(ic.configMapFederatedInformer,
+	ic.federatedConfigMapUpdater = util.NewFederatedUpdater(ic.configMapFederatedInformer, "configmap", ic.eventRecorder,
 		func(client kubeclientset.Interface, obj pkgruntime.Object) error {
 			configMap := obj.(*v1.ConfigMap)
 			configMapName := types.NamespacedName{Name: configMap.Name, Namespace: configMap.Namespace}
@@ -294,10 +294,9 @@ func NewIngressController(client federationclientset.Interface) *IngressControll
 		// objNameFunc
 		func(obj pkgruntime.Object) string {
 			ingress := obj.(*extensionsv1beta1.Ingress)
-			return ingress.Name
+			return fmt.Sprintf("%s/%s", ingress.Namespace, ingress.Name)
 		},
 		ic.updateTimeout,
-		ic.eventRecorder,
 		ic.ingressFederatedInformer,
 		ic.federatedIngressUpdater,
 	)
@@ -566,12 +565,12 @@ func (ic *IngressController) reconcileConfigMap(cluster *federationapi.Cluster,
 			Type:        util.OperationTypeUpdate,
 			Obj:         configMap,
 			ClusterName: cluster.Name,
+			Key:         configMapNsName.String(),
 		}}
 		glog.V(4).Infof("Calling federatedConfigMapUpdater.Update() - operations: %v", operations)
 		err := ic.federatedConfigMapUpdater.Update(operations, ic.updateTimeout)
 		if err != nil {
-			configMapName := types.NamespacedName{Name: configMap.Name, Namespace: configMap.Namespace}
-			glog.Errorf("Failed to execute update of ConfigMap %q on cluster %q: %v", configMapName, cluster.Name, err)
+			glog.Errorf("Failed to execute update of ConfigMap %q on cluster %q: %v", configMapNsName, cluster.Name, err)
 			ic.configMapDeliverer.DeliverAfter(cluster.Name, nil, ic.configMapReviewDelay)
 		}
 	}
@@ -770,8 +769,6 @@ func (ic *IngressController) reconcileIngress(ingress types.NamespacedName) {
 			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)
-			ic.eventRecorder.Eventf(baseIngress, api.EventTypeNormal, "CreateInCluster",
-				"Creating ingress in cluster %s", cluster.Name)
 
 			// We always first create an ingress in the first available cluster. Once that ingress
 			// has been created and allocated a global IP (visible via an annotation),
@@ -797,6 +794,7 @@ func (ic *IngressController) reconcileIngress(ingress types.NamespacedName) {
 					Type:        util.OperationTypeAdd,
 					Obj:         desiredIngress,
 					ClusterName: cluster.Name,
+					Key:         key,
 				})
 			} 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)
@@ -867,13 +865,12 @@ func (ic *IngressController) reconcileIngress(ingress types.NamespacedName) {
 				for key, val := range baseIngress.ObjectMeta.Labels {
 					desiredIngress.ObjectMeta.Labels[key] = val
 				}
-				ic.eventRecorder.Eventf(baseIngress, api.EventTypeNormal, "UpdateInCluster",
-					"Updating ingress in cluster %s", cluster.Name)
 
 				operations = append(operations, util.FederatedOperation{
 					Type:        util.OperationTypeUpdate,
 					Obj:         desiredIngress,
 					ClusterName: cluster.Name,
+					Key:         key,
 				})
 				// TODO: Transfer any readonly (target-proxy, url-map etc) annotations from the master cluster to the federation, if this is the master cluster.
 				// This is only for consistency, so that the federation ingress metadata matches the underlying clusters.  It's not actually required				}
@@ -887,10 +884,7 @@ func (ic *IngressController) reconcileIngress(ingress types.NamespacedName) {
 		return
 	}
 	glog.V(4).Infof("Calling federatedUpdater.Update() - operations: %v", operations)
-	err = ic.federatedIngressUpdater.UpdateWithOnError(operations, ic.updateTimeout, func(op util.FederatedOperation, operror error) {
-		ic.eventRecorder.Eventf(baseIngress, api.EventTypeWarning, "FailedClusterUpdate",
-			"Ingress update in cluster %s failed: %v", op.ClusterName, operror)
-	})
+	err = ic.federatedIngressUpdater.Update(operations, ic.updateTimeout)
 	if err != nil {
 		glog.Errorf("Failed to execute updates for %s: %v", ingress, err)
 		ic.deliverIngress(ingress, ic.ingressReviewDelay, true)

federation/pkg/federation-controller/namespace/namespace_controller.go

@@ -156,7 +156,7 @@ func NewNamespaceController(client federationclientset.Interface, dynamicClientP
 	)
 
 	// Federated updater along with Create/Update/Delete operations.
-	nc.federatedUpdater = util.NewFederatedUpdater(nc.namespaceFederatedInformer,
+	nc.federatedUpdater = util.NewFederatedUpdater(nc.namespaceFederatedInformer, "namespace", nc.eventRecorder,
 		func(client kubeclientset.Interface, obj runtime.Object) error {
 			namespace := obj.(*apiv1.Namespace)
 			_, err := client.Core().Namespaces().Create(namespace)
@@ -183,10 +183,9 @@ func NewNamespaceController(client federationclientset.Interface, dynamicClientP
 		// objNameFunc
 		func(obj runtime.Object) string {
 			namespace := obj.(*apiv1.Namespace)
-			return namespace.Name
+			return fmt.Sprintf("%s/%s", namespace.Namespace, namespace.Name)
 		},
 		nc.updateTimeout,
-		nc.eventRecorder,
 		nc.namespaceFederatedInformer,
 		nc.federatedUpdater,
 	)
@@ -370,26 +369,22 @@ func (nc *NamespaceController) reconcileNamespace(namespace string) {
 		glog.V(5).Infof("Desired namespace in underlying clusters: %+v", desiredNamespace)
 
 		if !found {
-			nc.eventRecorder.Eventf(baseNamespace, api.EventTypeNormal, "CreateInCluster",
-				"Creating namespace in cluster %s", cluster.Name)
-
 			operations = append(operations, util.FederatedOperation{
 				Type:        util.OperationTypeAdd,
 				Obj:         desiredNamespace,
 				ClusterName: cluster.Name,
+				Key:         namespace,
 			})
 		} else {
 			clusterNamespace := clusterNamespaceObj.(*apiv1.Namespace)
 
 			// Update existing namespace, if needed.
 			if !util.ObjectMetaAndSpecEquivalent(desiredNamespace, clusterNamespace) {
-				nc.eventRecorder.Eventf(baseNamespace, api.EventTypeNormal, "UpdateInCluster",
-					"Updating namespace in cluster %s. Desired: %+v\n Actual: %+v\n", cluster.Name, desiredNamespace, clusterNamespace)
-
 				operations = append(operations, util.FederatedOperation{
 					Type:        util.OperationTypeUpdate,
 					Obj:         desiredNamespace,
 					ClusterName: cluster.Name,
+					Key:         namespace,
 				})
 			}
 		}
@@ -401,10 +396,7 @@ func (nc *NamespaceController) reconcileNamespace(namespace string) {
 	}
 	glog.V(2).Infof("Updating namespace %s in underlying clusters. Operations: %d", baseNamespace.Name, len(operations))
 
-	err = nc.federatedUpdater.UpdateWithOnError(operations, nc.updateTimeout, func(op util.FederatedOperation, operror error) {
-		nc.eventRecorder.Eventf(baseNamespace, api.EventTypeWarning, "UpdateInClusterFailed",
-			"Namespace update in cluster %s failed: %v", op.ClusterName, operror)
-	})
+	err = nc.federatedUpdater.Update(operations, nc.updateTimeout)
 	if err != nil {
 		glog.Errorf("Failed to execute updates for %s: %v", namespace, err)
 		nc.deliverNamespace(namespace, 0, true)

federation/pkg/federation-controller/replicaset/replicasetcontroller.go

@@ -196,7 +196,7 @@ func NewReplicaSetController(federationClient fedclientset.Interface) *ReplicaSe
 	)
 	frsc.replicaSetLister = extensionslisters.NewReplicaSetLister(replicaSetIndexer)
 
-	frsc.fedUpdater = fedutil.NewFederatedUpdater(frsc.fedReplicaSetInformer,
+	frsc.fedUpdater = fedutil.NewFederatedUpdater(frsc.fedReplicaSetInformer, "replicaset", frsc.eventRecorder,
 		func(client kubeclientset.Interface, obj runtime.Object) error {
 			rs := obj.(*extensionsv1.ReplicaSet)
 			_, err := client.Extensions().ReplicaSets(rs.Namespace).Create(rs)
@@ -219,10 +219,9 @@ func NewReplicaSetController(federationClient fedclientset.Interface) *ReplicaSe
 		// objNameFunc
 		func(obj runtime.Object) string {
 			replicaset := obj.(*extensionsv1.ReplicaSet)
-			return replicaset.Name
+			return fmt.Sprintf("%s/%s", replicaset.Namespace, replicaset.Name)
 		},
 		updateTimeout,
-		frsc.eventRecorder,
 		frsc.fedReplicaSetInformer,
 		frsc.fedUpdater,
 	)
@@ -543,26 +542,22 @@ func (frsc *ReplicaSetController) reconcileReplicaSet(key string) (reconciliatio
 
 		if !exists {
 			if replicas > 0 {
-				frsc.eventRecorder.Eventf(frs, api.EventTypeNormal, "CreateInCluster",
-					"Creating replicaset in cluster %s", clusterName)
-
 				operations = append(operations, fedutil.FederatedOperation{
 					Type:        fedutil.OperationTypeAdd,
 					Obj:         lrs,
 					ClusterName: clusterName,
+					Key:         key,
 				})
 			}
 		} else {
 			currentLrs := lrsObj.(*extensionsv1.ReplicaSet)
 			// Update existing replica set, if needed.
 			if !fedutil.ObjectMetaAndSpecEquivalent(lrs, currentLrs) {
-				frsc.eventRecorder.Eventf(frs, api.EventTypeNormal, "UpdateInCluster",
-					"Updating replicaset in cluster %s", clusterName)
-
 				operations = append(operations, fedutil.FederatedOperation{
 					Type:        fedutil.OperationTypeUpdate,
 					Obj:         lrs,
 					ClusterName: clusterName,
+					Key:         key,
 				})
 			}
 			fedStatus.Replicas += currentLrs.Status.Replicas
@@ -584,10 +579,7 @@ func (frsc *ReplicaSetController) reconcileReplicaSet(key string) (reconciliatio
 		// Everything is in order
 		return statusAllOk, nil
 	}
-	err = frsc.fedUpdater.UpdateWithOnError(operations, updateTimeout, func(op fedutil.FederatedOperation, operror error) {
-		frsc.eventRecorder.Eventf(frs, api.EventTypeWarning, "FailedUpdateInCluster",
-			"Replicaset update in cluster %s failed: %v", op.ClusterName, operror)
-	})
+	err = frsc.fedUpdater.Update(operations, updateTimeout)
 	if err != nil {
 		glog.Errorf("Failed to execute updates for %s: %v", key, err)
 		return statusError, err

federation/pkg/federation-controller/service/servicecontroller.go

@@ -185,7 +185,7 @@ func New(federationClient fedclientset.Interface, dns dnsprovider.Interface,
 
 	s.federatedInformer = fedutil.NewFederatedInformer(federationClient, fedInformerFactory, &clusterLifecycle)
 
-	s.federatedUpdater = fedutil.NewFederatedUpdater(s.federatedInformer,
+	s.federatedUpdater = fedutil.NewFederatedUpdater(s.federatedInformer, "service", s.eventRecorder,
 		func(client kubeclientset.Interface, obj pkgruntime.Object) error {
 			svc := obj.(*v1.Service)
 			_, err := client.Core().Services(svc.Namespace).Create(svc)
@@ -240,10 +240,9 @@ func New(federationClient fedclientset.Interface, dns dnsprovider.Interface,
 		// objNameFunc
 		func(obj pkgruntime.Object) string {
 			service := obj.(*v1.Service)
-			return service.Name
+			return fmt.Sprintf("%s/%s", service.Namespace, service.Name)
 		},
 		updateTimeout,
-		s.eventRecorder,
 		s.federatedInformer,
 		s.federatedUpdater,
 	)
@@ -601,11 +600,7 @@ func (s *ServiceController) reconcileService(key string) reconciliationStatus {
 	}
 
 	if len(operations) != 0 {
-		err = s.federatedUpdater.UpdateWithOnError(operations, s.updateTimeout,
-			func(op fedutil.FederatedOperation, operror error) {
-				runtime.HandleError(fmt.Errorf("Service update in cluster %s failed: %v", op.ClusterName, operror))
-				s.eventRecorder.Eventf(fedService, api.EventTypeWarning, "UpdateInClusterFailed", "Service update in cluster %s failed: %v", op.ClusterName, operror)
-			})
+		err = s.federatedUpdater.Update(operations, s.updateTimeout)
 		if err != nil {
 			if !errors.IsAlreadyExists(err) {
 				runtime.HandleError(fmt.Errorf("Failed to execute updates for %s: %v", key, err))
@@ -642,12 +637,12 @@ func (s *ServiceController) getOperationsToPerformOnCluster(cluster *v1beta1.Clu
 		desiredService.ResourceVersion = ""
 
 		glog.V(4).Infof("Creating service in underlying cluster %s: %+v", cluster.Name, desiredService)
-		s.eventRecorder.Eventf(fedService, api.EventTypeNormal, "CreateInCluster", "Creating service in cluster %s", cluster.Name)
 
 		operation = &fedutil.FederatedOperation{
 			Type:        fedutil.OperationTypeAdd,
 			Obj:         desiredService,
 			ClusterName: cluster.Name,
+			Key:         key,
 		}
 	} else {
 		clusterService, ok := clusterServiceObj.(*v1.Service)
@@ -674,7 +669,6 @@ func (s *ServiceController) getOperationsToPerformOnCluster(cluster *v1beta1.Clu
 		// Update existing service, if needed.
 		if !Equivalent(desiredService, clusterService) {
 			glog.V(4).Infof("Service in underlying cluster %s does not match, Desired: %+v, Existing: %+v", cluster.Name, desiredService, clusterService)
-			s.eventRecorder.Eventf(fedService, api.EventTypeNormal, "UpdateInCluster", "Updating service in cluster %s. Desired: %+v\n Actual: %+v\n", cluster.Name, desiredService, clusterService)
 
 			// ResourceVersion of cluster service can be different from federated service,
 			// so do not update ResourceVersion while updating cluster service
@@ -684,6 +678,7 @@ func (s *ServiceController) getOperationsToPerformOnCluster(cluster *v1beta1.Clu
 				Type:        fedutil.OperationTypeUpdate,
 				Obj:         desiredService,
 				ClusterName: cluster.Name,
+				Key:         key,
 			}
 		} else {
 			glog.V(5).Infof("Service in underlying cluster %s is up to date: %+v", cluster.Name, desiredService)