create new applicationPayload and resourcePayload

Dockerfile.dev

@@ -7,6 +7,7 @@ COPY argocd /usr/local/bin/
 USER root
 RUN ln -s /usr/local/bin/argocd /usr/local/bin/argocd-server && \
     ln -s /usr/local/bin/argocd /usr/local/bin/argocd-repo-server && \
+    ln -s /usr/local/bin/argocd /usr/local/bin/event-reporter-server && \
     ln -s /usr/local/bin/argocd /usr/local/bin/argocd-application-controller && \
     ln -s /usr/local/bin/argocd /usr/local/bin/argocd-dex && \
     ln -s /usr/local/bin/argocd /usr/local/bin/argocd-notifications && \

Makefile

@@ -145,12 +145,12 @@ endef
 PATH:=$(PATH):$(PWD)/hack
 
 # docker image publishing options
-DOCKER_PUSH?=false
-IMAGE_NAMESPACE?=quay.io/codefresh
+DOCKER_PUSH?=true
+IMAGE_NAMESPACE?=quay.io/noam_codefresh
 # perform static compilation
 STATIC_BUILD?=true
 # build development images
-DEV_IMAGE?=false
+DEV_IMAGE?=true
 ARGOCD_GPG_ENABLED?=true
 ARGOCD_E2E_APISERVER_PORT?=8080
 

event_reporter/controller/controller.go

@@ -69,7 +69,7 @@ func (c *eventReporterController) Run(ctx context.Context) {
 
 	// sendIfPermitted is a helper to send the application to the client's streaming channel if the
 	// caller has RBAC privileges permissions to view it
-	sendIfPermitted := func(ctx context.Context, a appv1.Application, eventType watch.EventType, eventProcessingStartedAt string, ignoreResourceCache bool) error {
+	sendIfPermitted := func(ctx context.Context, a appv1.Application, eventType watch.EventType, eventProcessingStartedAt time.Time, ignoreResourceCache bool) error {
 		if eventType == watch.Bookmark {
 			return nil // ignore this event
 		}
@@ -80,10 +80,17 @@ func (c *eventReporterController) Run(ctx context.Context) {
 		}
 		trackingMethod := argoutil.GetTrackingMethod(c.settingsMgr)
 
-		err = c.applicationEventReporter.StreamApplicationEvents(ctx, &a, eventProcessingStartedAt, ignoreResourceCache, &reporter.ArgoTrackingMetadata{
+		argoTrackingMetadata := &reporter.ArgoTrackingMetadata{
 			AppInstanceLabelKey: &appInstanceLabelKey,
 			TrackingMethod:      &trackingMethod,
-		})
+		}
+		err = c.applicationEventReporter.StreamApplicationEvents(
+			ctx,
+			eventProcessingStartedAt,
+			&a,
+			argoTrackingMetadata,
+			ignoreResourceCache,
+		)
 		if err != nil {
 			return err
 		}
@@ -92,6 +99,7 @@ func (c *eventReporterController) Run(ctx context.Context) {
 			logCtx.WithError(err).Error("failed to cache last sent application event")
 			return err
 		}
+
 		return nil
 	}
 
@@ -112,7 +120,8 @@ func (c *eventReporterController) Run(ctx context.Context) {
 				c.metricsServer.IncCachedIgnoredEventsCounter(metrics.MetricAppEventType, event.Application.Name)
 				continue
 			}
-			eventProcessingStartedAt := time.Now().Format("2006-01-02T15:04:05.000Z")
+
+			eventProcessingStartedAt := time.Now()
 			ctx, cancel := context.WithTimeout(ctx, 2*time.Minute)
 			err := sendIfPermitted(ctx, event.Application, event.Type, eventProcessingStartedAt, ignoreResourceCache)
 			if err != nil {
@@ -122,6 +131,7 @@ func (c *eventReporterController) Run(ctx context.Context) {
 					cancel()
 				}
 			}
+
 			cancel()
 		}
 	}

event_reporter/reporter/app_revision.go

@@ -6,8 +6,6 @@ import (
 	"github.com/argoproj/argo-cd/v2/event_reporter/utils"
 	"github.com/argoproj/argo-cd/v2/pkg/apiclient/application"
 	"github.com/argoproj/argo-cd/v2/pkg/apis/application/v1alpha1"
-
-	log "github.com/sirupsen/logrus"
 )
 
 // treats multi-sourced apps as single source and gets first revision details
@@ -63,8 +61,9 @@ func (s *applicationEventReporter) getRevisionsDetails(ctx context.Context, a *v
 	return rms, nil
 }
 
-func (s *applicationEventReporter) getApplicationRevisionsMetadata(ctx context.Context, logCtx *log.Entry, a *v1alpha1.Application) (*utils.AppSyncRevisionsMetadata, error) {
+func (s *applicationEventReporter) getApplicationRevisionsMetadata(ctx context.Context, a *v1alpha1.Application) (*utils.AppSyncRevisionsMetadata, error) {
 	result := &utils.AppSyncRevisionsMetadata{}
+	logCtx := utils.GetLogger(ctx)
 
 	if a.Status.Sync.Revision != "" || a.Status.Sync.Revisions != nil || (a.Status.History != nil && len(a.Status.History) > 0) {
 		// can be the latest revision of repository

event_reporter/reporter/application_errors_parser.go

@@ -4,21 +4,20 @@ import (
 	"fmt"
 	"strings"
 
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	appv1 "github.com/argoproj/argo-cd/v2/pkg/apis/application/v1alpha1"
+	"github.com/argoproj/argo-cd/v2/pkg/codefresh"
 
 	"github.com/argoproj/gitops-engine/pkg/health"
 	"github.com/argoproj/gitops-engine/pkg/sync/common"
-
-	"github.com/argoproj/argo-cd/v2/pkg/apiclient/events"
-	appv1 "github.com/argoproj/argo-cd/v2/pkg/apis/application/v1alpha1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
-func parseApplicationSyncResultErrors(os *appv1.OperationState) []*events.ObjectError {
-	var errors []*events.ObjectError
+func parseApplicationSyncResultErrors(os *appv1.OperationState) []*codefresh.EventError {
+	var errors []*codefresh.EventError
 	// mean that resource not found as sync result but application can contain error inside operation state itself,
 	// for example app created with invalid yaml
 	if os.Phase == common.OperationError || os.Phase == common.OperationFailed {
-		errors = append(errors, &events.ObjectError{
+		errors = append(errors, &codefresh.EventError{
 			Type:     "sync",
 			Level:    "error",
 			Message:  os.Message,
@@ -33,8 +32,8 @@ var (
 	syncTaskNotValidErrorMessage      = "one or more synchronization tasks are not valid"
 )
 
-func parseApplicationSyncResultErrorsFromConditions(status appv1.ApplicationStatus) []*events.ObjectError {
-	var errs []*events.ObjectError
+func parseApplicationSyncResultErrorsFromConditions(status appv1.ApplicationStatus) []*codefresh.EventError {
+	var errs []*codefresh.EventError
 	if status.Conditions == nil {
 		return errs
 	}
@@ -47,7 +46,7 @@ func parseApplicationSyncResultErrorsFromConditions(status appv1.ApplicationStat
 		}
 
 		if level := getConditionLevel(cnd); level != "" {
-			errs = append(errs, &events.ObjectError{
+			errs = append(errs, &codefresh.EventError{
 				Type:     "sync",
 				Level:    level,
 				Message:  cnd.Message,
@@ -75,8 +74,8 @@ func getConditionTime(cnd appv1.ApplicationCondition) metav1.Time {
 	return metav1.Now()
 }
 
-func parseResourceSyncResultErrors(rs *appv1.ResourceStatus, os *appv1.OperationState) []*events.ObjectError {
-	errors := []*events.ObjectError{}
+func parseResourceSyncResultErrors(rs *appv1.ResourceStatus, os *appv1.OperationState) []*codefresh.EventError {
+	errors := []*codefresh.EventError{}
 	if os.SyncResult == nil {
 		return errors
 	}
@@ -93,7 +92,7 @@ func parseResourceSyncResultErrors(rs *appv1.ResourceStatus, os *appv1.Operation
 		return errors
 	}
 
-	errors = append(errors, &events.ObjectError{
+	errors = append(errors, &codefresh.EventError{
 		Type:     "sync",
 		Level:    "error",
 		Message:  sr.Message,
@@ -103,8 +102,8 @@ func parseResourceSyncResultErrors(rs *appv1.ResourceStatus, os *appv1.Operation
 	return errors
 }
 
-func parseAggregativeHealthErrorsOfApplication(a *appv1.Application, appTree *appv1.ApplicationTree) []*events.ObjectError {
-	var errors []*events.ObjectError
+func parseAggregativeHealthErrorsOfApplication(a *appv1.Application, appTree *appv1.ApplicationTree) []*codefresh.EventError {
+	var errors []*codefresh.EventError
 	if a.Status.Resources == nil {
 		return errors
 	}
@@ -120,8 +119,8 @@ func parseAggregativeHealthErrorsOfApplication(a *appv1.Application, appTree *ap
 	return errors
 }
 
-func parseAggregativeHealthErrors(rs *appv1.ResourceStatus, apptree *appv1.ApplicationTree, addReference bool) []*events.ObjectError {
-	errs := make([]*events.ObjectError, 0)
+func parseAggregativeHealthErrors(rs *appv1.ResourceStatus, apptree *appv1.ApplicationTree, addReference bool) []*codefresh.EventError {
+	errs := make([]*codefresh.EventError, 0)
 
 	if apptree == nil {
 		return errs
@@ -149,20 +148,20 @@ func parseAggregativeHealthErrors(rs *appv1.ResourceStatus, apptree *appv1.Appli
 	return errs
 }
 
-func getNodeHealthError(node appv1.ResourceNode, managedResource *appv1.ResourceStatus, addReference bool) *events.ObjectError {
+func getNodeHealthError(node appv1.ResourceNode, managedResource *appv1.ResourceStatus, addReference bool) *codefresh.EventError {
 	if node.Health == nil || node.Health.Status != health.HealthStatusDegraded {
 		return nil
 	}
 
-	newErr := &events.ObjectError{
+	newErr := &codefresh.EventError{
 		Type:     "health",
 		Level:    "error",
 		Message:  node.Health.Message,
 		LastSeen: *node.CreatedAt,
 	}
 
 	if addReference {
-		newErr.SourceReference = &events.ErrorSourceReference{
+		newErr.SourceReference = &codefresh.ErrorSourceReference{
 			Group:     managedResource.Group,
 			Version:   managedResource.Version,
 			Kind:      managedResource.Kind,
@@ -173,26 +172,27 @@ func getNodeHealthError(node appv1.ResourceNode, managedResource *appv1.Resource
 	return newErr
 }
 
-func parseAggregativeResourcesSyncErrors(resourceResults appv1.ResourceResults) []*events.ObjectError {
-	var errs []*events.ObjectError
+func parseAggregativeResourcesSyncErrors(resourceResults appv1.ResourceResults) []*codefresh.EventError {
+	var errs []*codefresh.EventError
 
 	if resourceResults == nil {
 		return errs
 	}
 
 	for _, rr := range resourceResults {
 		if rr.Message != "" {
-			objectError := events.ObjectError{
+			objectError := &codefresh.EventError{
 				Type:     "sync",
 				Level:    "error",
 				LastSeen: metav1.Now(),
 				Message:  fmt.Sprintf("Resource %s(%s): \n %s", rr.Kind, rr.Name, rr.Message),
 			}
 			if rr.Status == common.ResultCodeSyncFailed {
-				errs = append(errs, &objectError)
+				errs = append(errs, objectError)
 			}
+
 			if rr.HookPhase == common.OperationFailed || rr.HookPhase == common.OperationError {
-				errs = append(errs, &objectError)
+				errs = append(errs, objectError)
 			}
 		}
 	}