VERSION

@@ -1 +1 @@
-2.12.3-2024.12.17-4d75d35f4
+2.12.3-2024.12.23-4a8e092c0

acr_controller/service/acr_service.go

@@ -93,7 +93,7 @@ func (c *acrService) ChangeRevision(ctx context.Context, a *application.Applicat
 	revisions := []string{*revision}
 
 	if app.Status.OperationState != nil && app.Status.OperationState.Operation.Sync != nil {
-		c.logger.Infof("Patch operation sync result for application %s", app.Name)
+		c.logger.Infof("Patch operation status for application %s", app.Name)
 		return c.patchOperationSyncResultWithChangeRevision(ctx, app, revisions)
 	}
 

changelog/CHANGELOG-2.12.3-2024.12.23-4a8e092c0.md

@@ -0,0 +1,2 @@
+### Features
+-  feat(event-reporter): using sources-server for getting application version 

changelog/CHANGELOG.md

@@ -1,2 +1,2 @@
 ### Features
--  feat(event-reporter): multisourced apps support improvements: reporting syncOperationRevisions, detecting correct resource sourceIdx, reporting correct git commit info 
+-  feat(event-reporter): using sources-server for getting application version 

cmd/event-reporter-server/commands/event_reporter_server.go

@@ -6,6 +6,8 @@ import (
 	"math"
 	"time"
 
+	"github.com/argoproj/argo-cd/v2/pkg/sources_server_client"
+
 	"github.com/argoproj/argo-cd/v2/event_reporter/reporter"
 
 	"github.com/argoproj/argo-cd/v2/event_reporter"
@@ -104,6 +106,8 @@ func NewCommand() *cobra.Command {
 		rateLimiterBucketSize   int
 		rateLimiterDuration     time.Duration
 		rateLimiterLearningMode bool
+		useSourcesServer        bool
+		sourcesServerBaseURL    string
 	)
 	command := &cobra.Command{
 		Use:               cliName,
@@ -192,6 +196,10 @@ func NewCommand() *cobra.Command {
 					Capacity:     rateLimiterBucketSize,
 					LearningMode: rateLimiterLearningMode,
 				},
+				UseSourcesServer: useSourcesServer,
+				SourcesServerConfig: &sources_server_client.SourcesServerConfig{
+					BaseURL: sourcesServerBaseURL,
+				},
 			}
 
 			log.Infof("Starting event reporter server with grpc transport %v", useGrpc)
@@ -245,6 +253,8 @@ func NewCommand() *cobra.Command {
 	command.Flags().IntVar(&rateLimiterBucketSize, "rate-limiter-bucket-size", env.ParseNumFromEnv("RATE_LIMITER_BUCKET_SIZE", math.MaxInt, 0, math.MaxInt), "The maximum amount of requests allowed per window.")
 	command.Flags().DurationVar(&rateLimiterDuration, "rate-limiter-period", env.ParseDurationFromEnv("RATE_LIMITER_DURATION", 24*time.Hour, 0, math.MaxInt64), "The rate limit window size.")
 	command.Flags().BoolVar(&rateLimiterLearningMode, "rate-limiter-learning-mode", env.ParseBoolFromEnv("RATE_LIMITER_LEARNING_MODE_ENABLED", false), "The rate limit enabled in learning mode ( not blocking sending to queue but logging it )")
+	command.Flags().BoolVar(&useSourcesServer, "use-sources-server", env.ParseBoolFromEnv("SOURCES_SERVER_ENABLED", false), "Use sources-server instead of repo-server fork")
+	command.Flags().StringVar(&sourcesServerBaseURL, "sources-server-base-url", env.StringFromEnv("SOURCES_SERVER_BASE_URL", common.DefaultSourcesServerAddr), "Sources-server base URL")
 	cacheSrc = servercache.AddCacheFlagsToCmd(command, cacheutil.Options{
 		OnClientCreated: func(client *redis.Client) {
 			redisClient = client

common/common.go

@@ -25,7 +25,8 @@ const (
 	// DefaultDexServerAddr is the HTTP address of the Dex OIDC server, which we run a reverse proxy against
 	DefaultDexServerAddr = "argocd-dex-server:5556"
 	// DefaultRedisAddr is the default redis address
-	DefaultRedisAddr = "argocd-redis:6379"
+	DefaultRedisAddr         = "argocd-redis:6379"
+	DefaultSourcesServerAddr = "sources-server:8090"
 )
 
 // Kubernetes ConfigMap and Secret resource names which hold Argo CD settings

event_reporter/controller/controller.go

@@ -6,6 +6,8 @@ import (
 	"strings"
 	"time"
 
+	"github.com/argoproj/argo-cd/v2/pkg/sources_server_client"
+
 	"github.com/argoproj/argo-cd/v2/util/db"
 
 	appclient "github.com/argoproj/argo-cd/v2/event_reporter/application"
@@ -45,15 +47,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, db db.ArgoDB) 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, useSourcesServer bool, sourcesServerConfig *sources_server_client.SourcesServerConfig) 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, db),
+		applicationEventReporter: reporter.NewApplicationEventReporter(cache, applicationServiceClient, appLister, codefreshConfig, metricsServer, db, useSourcesServer, sourcesServerConfig),
 		cache:                    cache,
 		settingsMgr:              settingsMgr,
 		applicationServiceClient: applicationServiceClient,

event_reporter/reporter/app_revision_test.go

@@ -56,6 +56,8 @@ func TestGetRevisionsDetails(t *testing.T) {
 			&metrics.MetricsServer{},
 			fakeArgoDb(),
 			"0.0.1",
+			false,
+			nil,
 		}
 
 		result, _ := reporter.getRevisionsDetails(context.Background(), &app, []string{expectedRevision})
@@ -122,6 +124,8 @@ func TestGetRevisionsDetails(t *testing.T) {
 			&metrics.MetricsServer{},
 			fakeArgoDb(),
 			"0.0.1",
+			false,
+			nil,
 		}
 
 		result, _ := reporter.getRevisionsDetails(context.Background(), &app, []string{expectedRevision1, expectedRevision2})
@@ -165,6 +169,8 @@ func TestGetRevisionsDetails(t *testing.T) {
 			&metrics.MetricsServer{},
 			fakeArgoDb(),
 			"0.0.1",
+			false,
+			nil,
 		}
 
 		result, _ := reporter.getRevisionsDetails(context.Background(), &app, []string{expectedRevision})

event_reporter/reporter/application_event_reporter.go

@@ -9,6 +9,8 @@ import (
 	"strings"
 	"time"
 
+	"github.com/argoproj/argo-cd/v2/pkg/sources_server_client"
+
 	"github.com/argoproj/argo-cd/v2/util/db"
 
 	"github.com/argoproj/argo-cd/v2/event_reporter/utils"
@@ -46,6 +48,8 @@ type applicationEventReporter struct {
 	metricsServer            *metrics.MetricsServer
 	db                       db.ArgoDB
 	runtimeVersion           string
+	useSourcesServer         bool
+	sourcesServerClient      sources_server_client.SourceServerClientInteface
 }
 
 type ApplicationEventReporter interface {
@@ -59,7 +63,7 @@ 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, db db.ArgoDB) ApplicationEventReporter {
+func NewApplicationEventReporter(cache *servercache.Cache, applicationServiceClient appclient.ApplicationClient, appLister applisters.ApplicationLister, codefreshConfig *codefresh.CodefreshConfig, metricsServer *metrics.MetricsServer, db db.ArgoDB, useSourcesServer bool, sourcesServerConfig *sources_server_client.SourcesServerConfig) ApplicationEventReporter {
 	return &applicationEventReporter{
 		cache:                    cache,
 		applicationServiceClient: applicationServiceClient,
@@ -68,6 +72,8 @@ func NewApplicationEventReporter(cache *servercache.Cache, applicationServiceCli
 		metricsServer:            metricsServer,
 		db:                       db,
 		runtimeVersion:           codefreshConfig.RuntimeVersion,
+		useSourcesServer:         useSourcesServer,
+		sourcesServerClient:      sources_server_client.NewSourceServerClient(sourcesServerConfig),
 	}
 }
 
@@ -265,7 +271,20 @@ func (s *applicationEventReporter) resolveApplicationVersions(ctx context.Contex
 		}
 
 		syncManifests, _ := s.getDesiredManifests(ctx, logCtx, a, nil, &sourcePositions, syncResultRevisions)
-		return syncManifests.GetApplicationVersions()
+
+		var applicationVersions *apiclient.ApplicationVersions
+		if s.useSourcesServer {
+			log.Infof("cfGetAppVersion. Getting version from sourcesserver")
+			if len(*syncResultRevisions) == 0 {
+				return nil
+			}
+			appVers := s.sourcesServerClient.GetAppVersion(a, &(*syncResultRevisions)[0])
+			applicationVersions = utils.SourcesAppVersionsToRepo(appVers, logCtx)
+		} else {
+			applicationVersions = syncManifests.GetApplicationVersions()
+		}
+
+		return applicationVersions
 	}
 
 	syncResultRevision := utils.GetOperationSyncResultRevision(a)
@@ -275,7 +294,16 @@ func (s *applicationEventReporter) resolveApplicationVersions(ctx context.Contex
 	}
 
 	syncManifests, _ := s.getDesiredManifests(ctx, logCtx, a, syncResultRevision, nil, nil)
-	return syncManifests.GetApplicationVersions()
+
+	var applicationVersions *apiclient.ApplicationVersions
+	if s.useSourcesServer {
+		log.Infof("cfGetAppVersion. Getting version from sourcesserver")
+		appVers := s.sourcesServerClient.GetAppVersion(a, syncResultRevision)
+		applicationVersions = utils.SourcesAppVersionsToRepo(appVers, logCtx)
+	} else {
+		applicationVersions = syncManifests.GetApplicationVersions()
+	}
+	return applicationVersions
 }
 
 func (s *applicationEventReporter) getAppForResourceReporting(
@@ -309,8 +337,8 @@ func (s *applicationEventReporter) processResource(
 	appEventProcessingStartedAt string,
 	desiredManifests *apiclient.ManifestResponse,
 	manifestGenErr bool,
-	originalApplication *appv1.Application, // passed onlu if resource is app
-	applicationVersions *apiclient.ApplicationVersions, // passed onlu if resource is app
+	originalApplication *appv1.Application, // passed only if resource is app
+	applicationVersions *apiclient.ApplicationVersions, // passed only if resource is app
 	reportedEntityParentApp *ReportedEntityParentApp,
 	argoTrackingMetadata *ArgoTrackingMetadata,
 ) error {

event_reporter/reporter/application_event_reporter_test.go

@@ -173,6 +173,8 @@ func fakeReporter(customAppServiceClient appclient.ApplicationClient) *applicati
 		metricsServ,
 		fakeArgoDb(),
 		"0.0.1",
+		false,
+		nil,
 	}
 }
 

event_reporter/server.go

@@ -10,6 +10,8 @@ import (
 	"strings"
 	"time"
 
+	"github.com/argoproj/argo-cd/v2/pkg/sources_server_client"
+
 	appclient "github.com/argoproj/argo-cd/v2/event_reporter/application"
 	"github.com/argoproj/argo-cd/v2/event_reporter/reporter"
 
@@ -95,6 +97,8 @@ type EventReporterServerOpts struct {
 	RootPath                 string
 	CodefreshConfig          *codefresh.CodefreshConfig
 	RateLimiterOpts          *reporter.RateLimiterOpts
+	UseSourcesServer         bool
+	SourcesServerConfig      *sources_server_client.SourcesServerConfig
 }
 
 type handlerSwitcher struct {
@@ -153,7 +157,7 @@ func (a *EventReporterServer) Init(ctx context.Context) {
 }
 
 func (a *EventReporterServer) RunController(ctx context.Context) {
-	controller := event_reporter.NewEventReporterController(a.appInformer, a.Cache, a.settingsMgr, a.ApplicationServiceClient, a.appLister, a.CodefreshConfig, a.serviceSet.MetricsServer, a.featureManager, a.RateLimiterOpts, a.db)
+	controller := event_reporter.NewEventReporterController(a.appInformer, a.Cache, a.settingsMgr, a.ApplicationServiceClient, a.appLister, a.CodefreshConfig, a.serviceSet.MetricsServer, a.featureManager, a.RateLimiterOpts, a.db, a.UseSourcesServer, a.SourcesServerConfig)
 	go controller.Run(ctx)
 }
 

event_reporter/utils/app_version.go

@@ -3,7 +3,10 @@ package utils
 import (
 	"encoding/json"
 
+	log "github.com/sirupsen/logrus"
+
 	"github.com/argoproj/argo-cd/v2/pkg/apiclient/events"
+	"github.com/argoproj/argo-cd/v2/pkg/sources_server_client"
 	"github.com/argoproj/argo-cd/v2/reposerver/apiclient"
 )
 
@@ -16,3 +19,17 @@ func RepoAppVersionsToEvent(applicationVersions *apiclient.ApplicationVersions)
 	}
 	return applicationVersionsEvents, nil
 }
+
+func SourcesAppVersionsToRepo(applicationVersions *sources_server_client.AppVersionResult, logCtx *log.Entry) *apiclient.ApplicationVersions {
+	if applicationVersions == nil {
+		return nil
+	}
+	applicationVersionsRepo := &apiclient.ApplicationVersions{}
+	applicationVersionsData, _ := json.Marshal(applicationVersions)
+	err := json.Unmarshal(applicationVersionsData, applicationVersionsRepo)
+	if err != nil {
+		logCtx.Errorf("can't unmarshal app version: %v", err)
+		return nil
+	}
+	return applicationVersionsRepo
+}

go.mod

@@ -84,12 +84,12 @@ require (
 	go.opentelemetry.io/otel v1.21.0
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.21.0
 	go.opentelemetry.io/otel/sdk v1.21.0
-	golang.org/x/crypto v0.23.0
+	golang.org/x/crypto v0.31.0
 	golang.org/x/exp v0.0.0-20230522175609-2e198f4a06a1
 	golang.org/x/net v0.25.0
 	golang.org/x/oauth2 v0.12.0
-	golang.org/x/sync v0.5.0
-	golang.org/x/term v0.20.0
+	golang.org/x/sync v0.10.0
+	golang.org/x/term v0.27.0
 	golang.org/x/time v0.5.0
 	google.golang.org/genproto/googleapis/api v0.0.0-20230822172742-b8732ec3820d
 	google.golang.org/grpc v1.59.0
@@ -150,10 +150,10 @@ require (
 	github.com/tidwall/pretty v1.2.0 // indirect
 	go.opencensus.io v0.24.0 // indirect
 	go.starlark.net v0.0.0-20230525235612-a134d8f9ddca // indirect
-	golang.org/x/mod v0.14.0 // indirect
-	golang.org/x/sys v0.20.0 // indirect
-	golang.org/x/text v0.15.0 // indirect
-	golang.org/x/tools v0.16.1 // indirect
+	golang.org/x/mod v0.17.0 // indirect
+	golang.org/x/sys v0.28.0 // indirect
+	golang.org/x/text v0.21.0 // indirect
+	golang.org/x/tools v0.21.1-0.20240508182429-e35e4ccd0d2d // indirect
 	google.golang.org/api v0.132.0 // indirect
 	google.golang.org/genproto v0.0.0-20230822172742-b8732ec3820d // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20230822172742-b8732ec3820d // indirect
@@ -190,7 +190,7 @@ require (
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/desertbit/timer v0.0.0-20180107155436-c41aec40b27f // indirect
 	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
-	github.com/dlclark/regexp2 v1.11.2
+	github.com/dlclark/regexp2 v1.11.4
 	github.com/emicklei/go-restful/v3 v3.11.0 // indirect
 	github.com/emirpasic/gods v1.18.1 // indirect
 	github.com/evanphx/json-patch/v5 v5.8.0 // indirect