Use retry logic in agentless mode to update ConfigMap

controllers/eventreport.go

@@ -25,6 +25,7 @@ import (
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/util/retry"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	"github.com/projectsveltos/event-manager/api/v1beta1"
@@ -73,33 +74,36 @@ func addEventSourceToConfigMap(ctx context.Context, c client.Client, clusterName
 	configMapName := mgmtagent.GetConfigMapName(clusterName, clusterType)
 	eventSourceEntryKey := mgmtagent.GetKeyForEventSource(et.Name, es.Name)
 
-	currentConfigMap := &corev1.ConfigMap{}
-	err := c.Get(ctx, types.NamespacedName{Namespace: configMapNamespace, Name: configMapName}, currentConfigMap)
-	if err != nil {
-		if apierrors.IsNotFound(err) {
-			currentConfigMap = &corev1.ConfigMap{
-				ObjectMeta: metav1.ObjectMeta{
-					Namespace: configMapNamespace,
-					Name:      configMapName,
-				},
-				Data: map[string]string{
-					eventSourceEntryKey: es.Name,
-				},
+	err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
+		currentConfigMap := &corev1.ConfigMap{}
+		err := c.Get(ctx, types.NamespacedName{Namespace: configMapNamespace, Name: configMapName}, currentConfigMap)
+		if err != nil {
+			if apierrors.IsNotFound(err) {
+				currentConfigMap = &corev1.ConfigMap{
+					ObjectMeta: metav1.ObjectMeta{
+						Namespace: configMapNamespace,
+						Name:      configMapName,
+					},
+					Data: map[string]string{
+						eventSourceEntryKey: es.Name,
+					},
+				}
+				logger.V(logs.LogDebug).Info(fmt.Sprintf("creating entry %s in ConfigMap %s/%s",
+					eventSourceEntryKey, configMapNamespace, configMapName))
+				return c.Create(ctx, currentConfigMap)
 			}
-			logger.V(logs.LogDebug).Info(fmt.Sprintf("creating entry %s in ConfigMap %s/%s",
-				eventSourceEntryKey, configMapNamespace, configMapName))
-			return c.Create(ctx, currentConfigMap)
+			return err
 		}
-		return err
-	}
 
-	if currentConfigMap.Data == nil {
-		currentConfigMap.Data = map[string]string{}
-	}
-	currentConfigMap.Data[eventSourceEntryKey] = es.Name
-	logger.V(logs.LogDebug).Info(fmt.Sprintf("creating entry %s in ConfigMap %s/%s",
-		eventSourceEntryKey, configMapNamespace, configMapName))
-	return c.Update(ctx, currentConfigMap)
+		if currentConfigMap.Data == nil {
+			currentConfigMap.Data = map[string]string{}
+		}
+		currentConfigMap.Data[eventSourceEntryKey] = es.Name
+		logger.V(logs.LogDebug).Info(fmt.Sprintf("creating entry %s in ConfigMap %s/%s",
+			eventSourceEntryKey, configMapNamespace, configMapName))
+		return c.Update(ctx, currentConfigMap)
+	})
+	return err
 }
 
 // When sveltos-agent is running in the management cluster, EventSources are not copied to the managed cluster.
@@ -111,27 +115,30 @@ func removeEventSourceFromConfigMap(ctx context.Context, c client.Client, cluste
 	configMapNamespace := clusterNamespace
 	configMapName := mgmtagent.GetConfigMapName(clusterName, clusterType)
 
-	currentConfigMap := &corev1.ConfigMap{}
-	err := c.Get(ctx, types.NamespacedName{Namespace: configMapNamespace, Name: configMapName}, currentConfigMap)
-	if err != nil {
-		if apierrors.IsNotFound(err) {
-			return nil
-		}
-
-		return err
-	}
-
 	logger.V(logs.LogDebug).Info(fmt.Sprintf("removing entries for eventTrigger %s in ConfigMap %s/%s execpt %q",
 		et.Name, configMapNamespace, configMapName, leaveEntry))
 
-	for k, v := range currentConfigMap.Data {
-		if v == leaveEntry {
-			continue
+	err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
+		currentConfigMap := &corev1.ConfigMap{}
+		err := c.Get(ctx, types.NamespacedName{Namespace: configMapNamespace, Name: configMapName}, currentConfigMap)
+		if err != nil {
+			if apierrors.IsNotFound(err) {
+				return nil
+			}
+
+			return err
 		}
-		if mgmtagent.IsEventSourceEntryForEventTrigger(k, et.Name) {
-			delete(currentConfigMap.Data, k)
+
+		for k, v := range currentConfigMap.Data {
+			if v == leaveEntry {
+				continue
+			}
+			if mgmtagent.IsEventSourceEntryForEventTrigger(k, et.Name) {
+				delete(currentConfigMap.Data, k)
+			}
 		}
-	}
 
-	return c.Update(ctx, currentConfigMap)
+		return c.Update(ctx, currentConfigMap)
+	})
+	return err
 }

controllers/eventtrigger_deployer.go

@@ -44,7 +44,6 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/dynamic"
 	"k8s.io/client-go/rest"
-	"k8s.io/client-go/util/retry"
 	clusterv1 "sigs.k8s.io/cluster-api/api/core/v1beta2"
 	"sigs.k8s.io/cluster-api/util"
 	"sigs.k8s.io/cluster-api/util/annotations"
@@ -498,7 +497,7 @@ func (r *EventTriggerReconciler) proceedProcessingEventTrigger(ctx context.Conte
 	clusterInfo := &libsveltosv1beta1.ClusterInfo{
 		Cluster:        *cluster,
 		Hash:           currentHash,
-		FailureMessage: nil,
+		FailureMessage: stringPtr(""),
 		Status:         libsveltosv1beta1.SveltosStatusProvisioning,
 	}
 
@@ -534,6 +533,7 @@ func (r *EventTriggerReconciler) proceedProcessingEventTrigger(ctx context.Conte
 				return r.proceedDeployingEventTriggerInPullMode(ctx, eScope, cluster, f, isConfigSame,
 					currentHash, logger)
 			}
+
 			return clusterInfo, nil
 		}
 
@@ -909,36 +909,6 @@ func isClusterInfoForCluster(clusterInfo *libsveltosv1beta1.ClusterInfo, cluster
 		clusterproxy.GetClusterType(&clusterInfo.Cluster) == clusterType
 }
 
-func removeClusterInfoEntry(ctx context.Context, c client.Client,
-	clusterNamespace, clusterName string, clusterType libsveltosv1beta1.ClusterType,
-	resource *v1beta1.EventTrigger, logger logr.Logger) error {
-
-	err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
-		currentResource := &v1beta1.EventTrigger{}
-		err := c.Get(ctx, types.NamespacedName{Name: resource.Name}, currentResource)
-		if err != nil {
-			return err
-		}
-
-		for i := range currentResource.Status.ClusterInfo {
-			cc := &currentResource.Status.ClusterInfo[i]
-			if isClusterInfoForCluster(cc, clusterNamespace, clusterName, clusterType) {
-				currentResource.Status.ClusterInfo = remove(currentResource.Status.ClusterInfo, i)
-				return c.Status().Update(context.TODO(), currentResource)
-			}
-		}
-
-		return nil
-	})
-
-	return err
-}
-
-func remove(s []libsveltosv1beta1.ClusterInfo, i int) []libsveltosv1beta1.ClusterInfo {
-	s[i] = s[len(s)-1]
-	return s[:len(s)-1]
-}
-
 // deployEventSource deploys (creates or updates) referenced EventSource.
 func deployEventSource(ctx context.Context, c client.Client,
 	clusterNamespace, clusterName string, clusterType libsveltosv1beta1.ClusterType,

controllers/eventtrigger_deployer_test.go

@@ -186,56 +186,6 @@ var _ = Describe("EventTrigger deployer", func() {
 			clusterType, false)).To(BeTrue())
 	})
 
-	It("removeClusterInfoEntry removes cluster entry", func() {
-		clusterNamespace := randomString()
-		clusterName := randomString()
-		clusterType := libsveltosv1beta1.ClusterTypeCapi
-
-		resource := &v1beta1.EventTrigger{
-			ObjectMeta: metav1.ObjectMeta{
-				Name: randomString(),
-			},
-			Spec: v1beta1.EventTriggerSpec{
-				EventSourceName: randomString(),
-			},
-		}
-
-		Expect(testEnv.Create(context.TODO(), resource)).To(Succeed())
-		Expect(waitForObject(context.TODO(), testEnv.Client, resource)).To(Succeed())
-
-		resource.Status = v1beta1.EventTriggerStatus{
-			ClusterInfo: []libsveltosv1beta1.ClusterInfo{
-				*getClusterInfo(clusterNamespace, clusterName, clusterType),
-				*getClusterInfo(clusterNamespace, randomString(), clusterType),
-				*getClusterInfo(randomString(), clusterName, clusterType),
-				*getClusterInfo(clusterNamespace, clusterName, libsveltosv1beta1.ClusterTypeSveltos),
-			},
-		}
-		Expect(testEnv.Status().Update(context.TODO(), resource)).To(Succeed())
-		Eventually(func() bool {
-			currentChc := &v1beta1.EventTrigger{}
-			err := testEnv.Get(context.TODO(), types.NamespacedName{Name: resource.Name}, currentChc)
-			if err != nil {
-				return false
-			}
-			return len(currentChc.Status.ClusterInfo) != 0
-		}, timeout, pollingInterval).Should(BeTrue())
-
-		length := len(resource.Status.ClusterInfo)
-
-		Expect(controllers.RemoveClusterInfoEntry(context.TODO(), testEnv.Client, clusterNamespace, clusterName,
-			clusterType, resource, logger)).To(Succeed())
-
-		Eventually(func() bool {
-			currentChc := &v1beta1.EventTrigger{}
-			err := testEnv.Get(context.TODO(), types.NamespacedName{Name: resource.Name}, currentChc)
-			if err != nil {
-				return false
-			}
-			return len(currentChc.Status.ClusterInfo) == length-1
-		}, timeout, pollingInterval).Should(BeTrue())
-	})
-
 	It("eventTriggerHash returns current EventAddBasedAddOn hash", func() {
 		clusterNamespace := randomString()
 
@@ -2178,27 +2128,6 @@ data:
 	})
 })
 
-func getClusterInfo(clusterNamespace, clusterName string, clusterType libsveltosv1beta1.ClusterType) *libsveltosv1beta1.ClusterInfo {
-	var apiVersion, kind string
-	if clusterType == libsveltosv1beta1.ClusterTypeCapi {
-		apiVersion = clusterv1.GroupVersion.String()
-		kind = ClusterKind
-	} else {
-		apiVersion = libsveltosv1beta1.GroupVersion.String()
-		kind = libsveltosv1beta1.SveltosClusterKind
-	}
-
-	return &libsveltosv1beta1.ClusterInfo{
-		Cluster: corev1.ObjectReference{
-			Namespace:  clusterNamespace,
-			Name:       clusterName,
-			Kind:       kind,
-			APIVersion: apiVersion,
-		},
-		Hash: []byte(randomString()),
-	}
-}
-
 func validateLabels(labels map[string]string, clusterRef *corev1.ObjectReference,
 	eventTriggerName string, referencedResource client.Object) {
 

controllers/export_test.go

@@ -52,7 +52,6 @@ type (
 )
 
 var (
-	RemoveClusterInfoEntry          = removeClusterInfoEntry
 	EventTriggerHash                = eventTriggerHash
 	RemoveStaleEventSources         = removeStaleEventSources
 	RemoveStaleEventReports         = removeStaleEventReports

controllers/utils.go

@@ -124,3 +124,7 @@ func SetVersion(v string) {
 func getVersion() string {
 	return version
 }
+
+func stringPtr(s string) *string {
+	return &s
+}

test/fv/cloudevent_test.go

@@ -143,6 +143,35 @@ var _ = Describe("CloudEvents", func() {
 				currentEventReport)
 		}, timeout, pollingInterval).Should(BeNil())
 
+		if isAgentLessMode() {
+			erName := getEventReportName(eventSource.Name)
+			Byf("Verifying EventReports %s status is processed", erName)
+			Eventually(func() bool {
+				currentEventReport := &libsveltosv1beta1.EventReport{}
+				err := k8sClient.Get(context.TODO(),
+					types.NamespacedName{Namespace: kindWorkloadCluster.GetNamespace(), Name: erName},
+					currentEventReport)
+				if err != nil {
+					return false
+				}
+				return currentEventReport.Status.Phase != nil &&
+					*currentEventReport.Status.Phase == libsveltosv1beta1.ReportProcessed
+			}, timeout, pollingInterval).Should(BeTrue())
+		} else {
+			erName := eventSource.Name
+			Eventually(func() bool {
+				currentEventReport := &libsveltosv1beta1.EventReport{}
+				err := workloadClient.Get(context.TODO(),
+					types.NamespacedName{Namespace: projectsveltos, Name: erName},
+					currentEventReport)
+				if err != nil {
+					return false
+				}
+				return currentEventReport.Status.Phase != nil &&
+					*currentEventReport.Status.Phase == libsveltosv1beta1.ReportProcessed
+			}, timeout, pollingInterval).Should(BeTrue())
+		}
+
 		subject := randomString()
 		//nolint: lll // line with cloudEvent
 		jsonString := fmt.Sprintf(`{"specversion":"1.0","id":"10001","source":"my.source","type":"my.type","subject":%q,"datacontenttype":"application/json","data":{"message":"hello"}}`, subject)