diff --git a/controllers/eventreport.go b/controllers/eventreport.go index 6df1974..7aa0055 100644 --- a/controllers/eventreport.go +++ b/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 } diff --git a/controllers/eventtrigger_deployer.go b/controllers/eventtrigger_deployer.go index 7a9ccbc..de00747 100644 --- a/controllers/eventtrigger_deployer.go +++ b/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 := ¤tResource.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, diff --git a/controllers/eventtrigger_deployer_test.go b/controllers/eventtrigger_deployer_test.go index e4c81cb..6b1a96d 100644 --- a/controllers/eventtrigger_deployer_test.go +++ b/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) { diff --git a/controllers/export_test.go b/controllers/export_test.go index 7a71fc7..3101ec7 100644 --- a/controllers/export_test.go +++ b/controllers/export_test.go @@ -52,7 +52,6 @@ type ( ) var ( - RemoveClusterInfoEntry = removeClusterInfoEntry EventTriggerHash = eventTriggerHash RemoveStaleEventSources = removeStaleEventSources RemoveStaleEventReports = removeStaleEventReports diff --git a/controllers/utils.go b/controllers/utils.go index 643eadb..ddfd24e 100644 --- a/controllers/utils.go +++ b/controllers/utils.go @@ -124,3 +124,7 @@ func SetVersion(v string) { func getVersion() string { return version } + +func stringPtr(s string) *string { + return &s +} diff --git a/test/fv/cloudevent_test.go b/test/fv/cloudevent_test.go index daee37b..f753830 100644 --- a/test/fv/cloudevent_test.go +++ b/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)