feat(event-reporter): multi-sourced apps - resource git details reporting

assets/swagger.json

@@ -5294,6 +5294,14 @@
         "sourceType": {
           "type": "string"
         },
+        "sourcesManifestsStartingIdx": {
+          "type": "array",
+          "title": "for multisourced apps will be [0,12,20], so this means that 0-11 - from first app source, 12-19 from second one, 20-x - third one",
+          "items": {
+            "type": "integer",
+            "format": "int32"
+          }
+        },
         "verifyResult": {
           "type": "string",
           "title": "Raw response of git verify-commit operation (always the empty string for Helm)"

changelog/CHANGELOG.md

@@ -1,2 +1,3 @@
-### Fixed
--  fix: failures in update revision for path should not affect sync
+### Updates
+- feat(event-reporter): multisourced apps support improvements: reporting syncOperationRevisions, detecting correct resource sourceIdx, reporting correct git commit info 
+- fix(repo-server): fixed repeated resources generation for ApplicationSourceTypeDirectory

event_reporter/application/client.go

@@ -126,6 +126,16 @@
 	if in.Revision != nil {
 		params = fmt.Sprintf("%s&revision=%s", params, *in.Revision)
 	}
+	if in.SourcePositions != nil && len(in.SourcePositions) > 0 {
+		for _, sourcePosition := range in.SourcePositions {
+			params = fmt.Sprintf("%s&sourcePositions=%s", params, sourcePosition)
+		}
+	}
+	if in.Revisions != nil && len(in.Revisions) > 0 {
+		for _, revision := range in.Revisions {
+			params = fmt.Sprintf("%s&revisions=%s", params, revision)
+		}
+	}
 	url := fmt.Sprintf("%s/api/v1/applications/%s/manifests%s", c.baseUrl, *in.Name, params)
 
 	manifest := &repoapiclient.ManifestResponse{}

event_reporter/controller/controller.go

@@ -2,6 +2,7 @@ package controller
 
 import (
 	"context"
+	"github.com/argoproj/argo-cd/v2/util/db"
 	"math"
 	"strings"
 	"time"
@@ -43,15 +44,15 @@ type eventReporterController struct {
 	metricsServer            *metrics.MetricsServer
 }
 
-func NewEventReporterController(appInformer cache.SharedIndexInformer, cache *servercache.Cache, settingsMgr *settings.SettingsManager, applicationServiceClient appclient.ApplicationClient, appLister applisters.ApplicationLister, codefreshConfig *codefresh.CodefreshConfig, metricsServer *metrics.MetricsServer, featureManager *reporter.FeatureManager, rateLimiterOpts *reporter.RateLimiterOpts) EventReporterController {
+func NewEventReporterController(appInformer cache.SharedIndexInformer, cache *servercache.Cache, settingsMgr *settings.SettingsManager, applicationServiceClient appclient.ApplicationClient, appLister applisters.ApplicationLister, codefreshConfig *codefresh.CodefreshConfig, metricsServer *metrics.MetricsServer, featureManager *reporter.FeatureManager, rateLimiterOpts *reporter.RateLimiterOpts, db db.ArgoDB) EventReporterController {
 	appBroadcaster := reporter.NewBroadcaster(featureManager, metricsServer, rateLimiterOpts)
 	_, err := appInformer.AddEventHandler(appBroadcaster)
 	if err != nil {
 		log.Error(err)
 	}
 	return &eventReporterController{
 		appBroadcaster:           appBroadcaster,
-		applicationEventReporter: reporter.NewApplicationEventReporter(cache, applicationServiceClient, appLister, codefreshConfig, metricsServer),
+		applicationEventReporter: reporter.NewApplicationEventReporter(cache, applicationServiceClient, appLister, codefreshConfig, metricsServer, db),
 		cache:                    cache,
 		settingsMgr:              settingsMgr,
 		applicationServiceClient: applicationServiceClient,

event_reporter/reporter/app_revision_test.go

@@ -54,6 +54,7 @@ func TestGetRevisionsDetails(t *testing.T) {
 			newAppLister(),
 			appServiceClient,
 			&metrics.MetricsServer{},
+			fakeArgoDb(),
 		}
 
 		result, _ := reporter.getRevisionsDetails(context.Background(), &app, []string{expectedRevision})
@@ -118,6 +119,7 @@ func TestGetRevisionsDetails(t *testing.T) {
 			newAppLister(),
 			appServiceClient,
 			&metrics.MetricsServer{},
+			fakeArgoDb(),
 		}
 
 		result, _ := reporter.getRevisionsDetails(context.Background(), &app, []string{expectedRevision1, expectedRevision2})
@@ -159,6 +161,7 @@ func TestGetRevisionsDetails(t *testing.T) {
 			newAppLister(),
 			appServiceClient,
 			&metrics.MetricsServer{},
+			fakeArgoDb(),
 		}
 
 		result, _ := reporter.getRevisionsDetails(context.Background(), &app, []string{expectedRevision})

event_reporter/reporter/application_event_reporter.go

@@ -9,8 +9,12 @@ import (
 	"strings"
 	"time"
 
+	"github.com/argoproj/argo-cd/v2/util/db"
+
 	"github.com/argoproj/argo-cd/v2/event_reporter/utils"
 
+	argoutils "github.com/argoproj/argo-cd/v2/util/argo"
+
 	"github.com/argoproj/argo-cd/v2/reposerver/apiclient"
 
 	argocommon "github.com/argoproj/argo-cd/v2/common"
@@ -40,6 +44,7 @@ type applicationEventReporter struct {
 	appLister                applisters.ApplicationLister
 	applicationServiceClient appclient.ApplicationClient
 	metricsServer            *metrics.MetricsServer
+	db                       db.ArgoDB
 }
 
 type ApplicationEventReporter interface {
@@ -53,13 +58,14 @@ type ApplicationEventReporter interface {
 	ShouldSendApplicationEvent(ae *appv1.ApplicationWatchEvent) (shouldSend bool, syncStatusChanged bool)
 }
 
-func NewApplicationEventReporter(cache *servercache.Cache, applicationServiceClient appclient.ApplicationClient, appLister applisters.ApplicationLister, codefreshConfig *codefresh.CodefreshConfig, metricsServer *metrics.MetricsServer) ApplicationEventReporter {
+func NewApplicationEventReporter(cache *servercache.Cache, applicationServiceClient appclient.ApplicationClient, appLister applisters.ApplicationLister, codefreshConfig *codefresh.CodefreshConfig, metricsServer *metrics.MetricsServer, db db.ArgoDB) ApplicationEventReporter {
 	return &applicationEventReporter{
 		cache:                    cache,
 		applicationServiceClient: applicationServiceClient,
 		codefreshClient:          codefresh.NewCodefreshClient(codefreshConfig),
 		appLister:                appLister,
 		metricsServer:            metricsServer,
+		db:                       db,
 	}
 }
 
@@ -87,15 +93,28 @@ func (s *applicationEventReporter) shouldSendResourceEvent(a *appv1.Application,
 	return true
 }
 
-func (r *applicationEventReporter) getDesiredManifests(ctx context.Context, a *appv1.Application, revision *string, logCtx *log.Entry) (*apiclient.ManifestResponse, bool) {
+func (r *applicationEventReporter) getDesiredManifests(
+	ctx context.Context,
+	logCtx *log.Entry,
+	a *appv1.Application,
+	revision *string,
+	sourcePositions *[]int64,
+	revisions *[]string,
+) (*apiclient.ManifestResponse, bool) {
 	// get the desired state manifests of the application
 	project := a.Spec.GetProject()
-	desiredManifests, err := r.applicationServiceClient.GetManifests(ctx, &application.ApplicationManifestQuery{
+	query := application.ApplicationManifestQuery{
 		Name:         &a.Name,
 		AppNamespace: &a.Namespace,
 		Revision:     revision,
 		Project:      &project,
-	})
+	}
+	if sourcePositions != nil && query.Revisions != nil {
+		query.SourcePositions = *sourcePositions
+		query.Revisions = *revisions
+	}
+
+	desiredManifests, err := r.applicationServiceClient.GetManifests(ctx, &query)
 	if err != nil {
 		// if it's manifest generation error we need to still report the actual state
 		// of the resources, but since we can't get the desired state, we will report
@@ -137,7 +156,7 @@ func (s *applicationEventReporter) StreamApplicationEvents(
 
 	logCtx.Info("getting desired manifests")
 
-	desiredManifests, manifestGenErr := s.getDesiredManifests(ctx, a, nil, logCtx)
+	desiredManifests, manifestGenErr := s.getDesiredManifests(ctx, logCtx, a, nil, nil, nil)
 
 	applicationVersions := s.resolveApplicationVersions(ctx, a, logCtx)
 
@@ -158,17 +177,21 @@ func (s *applicationEventReporter) StreamApplicationEvents(
 		rs := utils.GetAppAsResource(a)
 		utils.SetHealthStatusIfMissing(rs)
 
-		parentDesiredManifests, manifestGenErr := s.getDesiredManifests(ctx, parentApplicationEntity, nil, logCtx)
+		parentDesiredManifests, manifestGenErr := s.getDesiredManifests(ctx, logCtx, parentApplicationEntity, nil, nil, nil)
 
 		parentAppSyncRevisionsMetadata, err := s.getApplicationRevisionsMetadata(ctx, logCtx, parentApplicationEntity)
 		if err != nil {
 			logCtx.WithError(err).Warn("failed to get parent application's revision metadata, resuming")
 		}
 
+		validatedDestination := parentApplicationEntity.Spec.Destination.DeepCopy()
+		_ = argoutils.ValidateDestination(ctx, validatedDestination, s.db) // resolves server field if missing
+
 		err = s.processResource(ctx, *rs, logCtx, eventProcessingStartedAt, parentDesiredManifests, manifestGenErr, a, applicationVersions, &ReportedEntityParentApp{
-			app:               parentApplicationEntity,
-			appTree:           appTree,
-			revisionsMetadata: parentAppSyncRevisionsMetadata,
+			app:                  parentApplicationEntity,
+			appTree:              appTree,
+			revisionsMetadata:    parentAppSyncRevisionsMetadata,
+			validatedDestination: validatedDestination,
 		}, argoTrackingMetadata)
 		if err != nil {
 			s.metricsServer.IncErroredEventsCounter(metrics.MetricChildAppEventType, metrics.MetricEventUnknownErrorType, a.Name)
@@ -197,6 +220,9 @@ func (s *applicationEventReporter) StreamApplicationEvents(
 		s.metricsServer.ObserveEventProcessingDurationHistogramDuration(a.Name, metrics.MetricParentAppEventType, metricTimer.Duration())
 	}
 
+	validatedDestination := a.Spec.Destination.DeepCopy()
+	_ = argoutils.ValidateDestination(ctx, validatedDestination, s.db) // resolves server field if missing
+
 	revisionsMetadata, _ := s.getApplicationRevisionsMetadata(ctx, logCtx, a)
 	// for each resource in the application get desired and actual state,
 	// then stream the event
@@ -210,9 +236,10 @@ func (s *applicationEventReporter) StreamApplicationEvents(
 			continue
 		}
 		err := s.processResource(ctx, rs, logCtx, eventProcessingStartedAt, desiredManifests, manifestGenErr, nil, nil, &ReportedEntityParentApp{
-			app:               a,
-			appTree:           appTree,
-			revisionsMetadata: revisionsMetadata,
+			app:                  a,
+			appTree:              appTree,
+			revisionsMetadata:    revisionsMetadata,
+			validatedDestination: validatedDestination,
 		}, argoTrackingMetadata)
 		if err != nil {
 			s.metricsServer.IncErroredEventsCounter(metrics.MetricResourceEventType, metrics.MetricEventUnknownErrorType, a.Name)
@@ -222,13 +249,30 @@ func (s *applicationEventReporter) StreamApplicationEvents(
 	return nil
 }
 
+// returns appVersion from first non-ref source for multisourced apps
 func (s *applicationEventReporter) resolveApplicationVersions(ctx context.Context, a *appv1.Application, logCtx *log.Entry) *apiclient.ApplicationVersions {
-	syncRevision := utils.GetOperationStateRevision(a)
-	if syncRevision == nil {
+	if a.Spec.HasMultipleSources() {
+		syncResultRevisions := utils.GetOperationSyncResultRevisions(a)
+		if syncResultRevisions == nil {
+			return nil
+		}
+
+		var sourcePositions []int64
+		for i := 0; i < len(*syncResultRevisions); i++ {
+			sourcePositions = append(sourcePositions, int64(i+1))
+		}
+
+		syncManifests, _ := s.getDesiredManifests(ctx, logCtx, a, nil, &sourcePositions, syncResultRevisions)
+		return syncManifests.GetApplicationVersions()
+	}
+
+	syncResultRevision := utils.GetOperationSyncResultRevision(a)
+
+	if syncResultRevision == nil {
 		return nil
 	}
 
-	syncManifests, _ := s.getDesiredManifests(ctx, a, syncRevision, logCtx)
+	syncManifests, _ := s.getDesiredManifests(ctx, logCtx, a, syncResultRevision, nil, nil)
 	return syncManifests.GetApplicationVersions()
 }
 
@@ -279,7 +323,7 @@ func (s *applicationEventReporter) processResource(
 	})
 
 	// get resource desired state
-	desiredState := getResourceDesiredState(&rs, desiredManifests, logCtx)
+	desiredState, appSourceIdx := getResourceDesiredState(&rs, desiredManifests, logCtx)
 
 	actualState, err := s.getResourceActualState(ctx, logCtx, metricsEventType, rs, reportedEntityParentApp.app, originalApplication)
 	if err != nil {
@@ -304,16 +348,19 @@ func (s *applicationEventReporter) processResource(
 			actualState:    actualState,
 			desiredState:   desiredState,
 			manifestGenErr: manifestGenErr,
+			appSourceIdx:   appSourceIdx,
 			rsAsAppInfo: &ReportedResourceAsApp{
 				app:                 originalApplication,
 				revisionsMetadata:   originalAppRevisionMetadata,
 				applicationVersions: applicationVersions,
 			},
 		},
 		&ReportedEntityParentApp{
-			app:               parentApplicationToReport,
-			appTree:           reportedEntityParentApp.appTree,
-			revisionsMetadata: revisionMetadataToReport,
+			app:                  parentApplicationToReport,
+			appTree:              reportedEntityParentApp.appTree,
+			revisionsMetadata:    revisionMetadataToReport,
+			validatedDestination: reportedEntityParentApp.validatedDestination,
+			desiredManifests:     reportedEntityParentApp.desiredManifests,
 		},
 		argoTrackingMetadata,
 	)
@@ -474,11 +521,11 @@ func applicationMetadataChanged(ae *appv1.ApplicationWatchEvent, cachedApp *appv
 	return !reflect.DeepEqual(newEventAppMeta, cachedAppMeta)
 }
 
-func getResourceDesiredState(rs *appv1.ResourceStatus, ds *apiclient.ManifestResponse, logger *log.Entry) *apiclient.Manifest {
+func getResourceDesiredState(rs *appv1.ResourceStatus, ds *apiclient.ManifestResponse, logger *log.Entry) (manifest *apiclient.Manifest, sourceIdx int32) {
 	if ds == nil {
-		return &apiclient.Manifest{}
+		return &apiclient.Manifest{}, 0
 	}
-	for _, m := range ds.Manifests {
+	for idx, m := range ds.Manifests {
 		u, err := appv1.UnmarshalToUnstructured(m.CompiledManifest)
 		if err != nil {
 			logger.WithError(err).Warnf("failed to unmarshal compiled manifest")
@@ -498,11 +545,27 @@ func getResourceDesiredState(rs *appv1.ResourceStatus, ds *apiclient.ManifestRes
 				m.RawManifest = m.CompiledManifest
 			}
 
-			return m
+			return m, getResourceSourceIdxFromManifestResponse(idx, ds)
 		}
 	}
 
 	// no desired state for resource
 	// it's probably deleted from git
-	return &apiclient.Manifest{}
+	return &apiclient.Manifest{}, 0
+}
+
+func getResourceSourceIdxFromManifestResponse(rsIdx int, ds *apiclient.ManifestResponse) int32 {
+	if ds.SourcesManifestsStartingIdx == nil {
+		return -1
+	}
+
+	sourceIdx := int32(-1)
+
+	for currentSourceIdx, sourceStartingIdx := range ds.SourcesManifestsStartingIdx {
+		if int32(rsIdx) >= sourceStartingIdx {
+			sourceIdx = int32(currentSourceIdx)
+		}
+	}
+
+	return sourceIdx
 }

event_reporter/reporter/application_event_reporter_test.go

@@ -4,6 +4,9 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"github.com/argoproj/argo-cd/v2/util/db"
+	"github.com/argoproj/argo-cd/v2/util/settings"
+	"k8s.io/client-go/kubernetes/fake"
 	"net/http"
 	"testing"
 	"time"
@@ -19,8 +22,8 @@ import (
 	"github.com/argoproj/argo-cd/v2/pkg/apiclient"
 	apiclientapppkg "github.com/argoproj/argo-cd/v2/pkg/apiclient/application"
 	appv1reg "github.com/argoproj/argo-cd/v2/pkg/apis/application"
+	repoapiclient "github.com/argoproj/argo-cd/v2/reposerver/apiclient"
 	"github.com/argoproj/argo-cd/v2/util/io"
-
 	"google.golang.org/grpc"
 	"k8s.io/apimachinery/pkg/runtime"
 
@@ -93,6 +96,12 @@ func fakeAppServiceClient() apiclientapppkg.ApplicationServiceClient {
 	return applicationServiceClient
 }
 
+func fakeArgoDb() db.ArgoDB {
+	clientset := fake.NewSimpleClientset()
+
+	return db.NewDB("", settings.NewSettingsManager(context.Background(), clientset, testNamespace), clientset)
+}
+
 func fakeReporter(customAppServiceClient appclient.ApplicationClient) *applicationEventReporter {
 	guestbookApp := &appsv1.Application{
 		TypeMeta: metav1.TypeMeta{
@@ -159,6 +168,7 @@ func fakeReporter(customAppServiceClient appclient.ApplicationClient) *applicati
 		appLister,
 		customAppServiceClient,
 		metricsServ,
+		fakeArgoDb(),
 	}
 }
 
@@ -459,3 +469,20 @@ func TestGetResourceActualState(t *testing.T) {
 		assert.Equal(t, "", ptr.ToString(res.Manifest))
 	})
 }
+
+func TestGetResourceSourceIdxFromManifestResponse(t *testing.T) {
+	t.Run("should return correct sourceIdx", func(t *testing.T) {
+		sourcesManifestsStartingIdx := []int32{0, 5, 15}
+		desiredManifests := &repoapiclient.ManifestResponse{
+			SourcesManifestsStartingIdx: sourcesManifestsStartingIdx,
+		}
+
+		assert.Equal(t, int32(0), getResourceSourceIdxFromManifestResponse(0, desiredManifests))
+		assert.Equal(t, int32(0), getResourceSourceIdxFromManifestResponse(1, desiredManifests))
+		assert.Equal(t, int32(1), getResourceSourceIdxFromManifestResponse(5, desiredManifests))
+		assert.Equal(t, int32(1), getResourceSourceIdxFromManifestResponse(6, desiredManifests))
+		assert.Equal(t, int32(2), getResourceSourceIdxFromManifestResponse(15, desiredManifests))
+		assert.Equal(t, int32(2), getResourceSourceIdxFromManifestResponse(16, desiredManifests))
+		assert.Equal(t, int32(-1), getResourceSourceIdxFromManifestResponse(2, &repoapiclient.ManifestResponse{}))
+	})
+}