fix(hydrator): refresh by annotation instead of work queue (#22016) (#22067)

Signed-off-by: Michael Crenshaw <[email protected]>

Author: crenshaw-dev

controller/hydrator/hydrator.go

@@ -27,7 +27,7 @@ type Dependencies interface {
 	GetProcessableApps() (*appv1.ApplicationList, error)
 	GetRepoObjs(app *appv1.Application, source appv1.ApplicationSource, revision string, project *appv1.AppProject) ([]*unstructured.Unstructured, *apiclient.ManifestResponse, error)
 	GetWriteCredentials(ctx context.Context, repoURL string, project string) (*appv1.Repository, error)
-	RequestAppRefresh(appName string)
+	RequestAppRefresh(appName string, appNamespace string) error
 	// TODO: only allow access to the hydrator status
 	PersistAppHydratorStatus(orig *appv1.Application, newStatus *appv1.SourceHydratorStatus)
 	AddHydrationQueueItem(key HydrationQueueKey)
@@ -156,7 +156,10 @@ func (h *Hydrator) ProcessHydrationQueueItem(hydrationKey HydrationQueueKey) (pr
 		}
 		h.dependencies.PersistAppHydratorStatus(origApp, &app.Status.SourceHydrator)
 		// Request a refresh since we pushed a new commit.
-		h.dependencies.RequestAppRefresh(app.QualifiedName())
+		err := h.dependencies.RequestAppRefresh(app.Name, app.Namespace)
+		if err != nil {
+			logCtx.WithField("app", app.QualifiedName()).WithError(err).Error("Failed to request app refresh after hydration")
+		}
 	}
 	return
 }

controller/hydrator_dependencies.go

@@ -7,6 +7,7 @@ import (
 	"github.com/argoproj/argo-cd/v2/controller/hydrator"
 	appv1 "github.com/argoproj/argo-cd/v2/pkg/apis/application/v1alpha1"
 	"github.com/argoproj/argo-cd/v2/reposerver/apiclient"
+	argoutil "github.com/argoproj/argo-cd/v2/util/argo"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
@@ -65,8 +66,17 @@ func (ctrl *ApplicationController) GetWriteCredentials(ctx context.Context, repo
 	return ctrl.db.GetWriteRepository(ctx, repoURL, project)
 }
 
-func (ctrl *ApplicationController) RequestAppRefresh(appName string) {
-	ctrl.requestAppRefresh(appName, CompareWithLatest.Pointer(), nil)
+func (ctrl *ApplicationController) RequestAppRefresh(appName string, appNamespace string) error {
+	// We request a refresh by setting the annotation instead of by adding it to the refresh queue, because there is no
+	// guarantee that the hydrator is running on the same controller shard as is processing the application.
+
+	// This function is called for each app after a hydrate operation is completed so that the app controller can pick
+	// up the newly-hydrated changes. So we set hydrate=false to avoid a hydrate loop.
+	_, err := argoutil.RefreshApp(ctrl.applicationClientset.ArgoprojV1alpha1().Applications(appNamespace), appName, appv1.RefreshTypeNormal, false)
+	if err != nil {
+		return fmt.Errorf("failed to request app refresh: %w", err)
+	}
+	return nil
 }
 
 func (ctrl *ApplicationController) PersistAppHydratorStatus(orig *appv1.Application, newStatus *appv1.SourceHydratorStatus) {

pkg/apis/application/v1alpha1/types.go

@@ -464,6 +464,13 @@ const (
 	RefreshTypeHard   RefreshType = "hard"
 )
 
+type HydrateType string
+
+const (
+	// HydrateTypeNormal is a normal hydration
+	HydrateTypeNormal HydrateType = "normal"
+)
+
 type RefTarget struct {
 	Repo           Repository `protobuf:"bytes,1,opt,name=repo"`
 	TargetRevision string     `protobuf:"bytes,2,opt,name=targetRevision"`
@@ -3026,7 +3033,7 @@ func (app *Application) IsHydrateRequested() bool {
 	if !ok {
 		return false
 	}
-	if typeStr == "normal" {
+	if typeStr == string(HydrateTypeNormal) {
 		return true
 	}
 	return false

server/application/application.go

@@ -748,7 +748,7 @@ func (s *Server) Get(ctx context.Context, q *application.ApplicationQuery) (*app
 	})
 	defer unsubscribe()
 
-	app, err := argoutil.RefreshApp(appIf, appName, refreshType)
+	app, err := argoutil.RefreshApp(appIf, appName, refreshType, true)
 	if err != nil {
 		return nil, fmt.Errorf("error refreshing the app: %w", err)
 	}

util/argo/argo.go

@@ -226,15 +226,18 @@ func FilterByNameP(apps []*argoappv1.Application, name string) []*argoappv1.Appl
 }
 
 // RefreshApp updates the refresh annotation of an application to coerce the controller to process it
-func RefreshApp(appIf v1alpha1.ApplicationInterface, name string, refreshType argoappv1.RefreshType) (*argoappv1.Application, error) {
+func RefreshApp(appIf v1alpha1.ApplicationInterface, name string, refreshType argoappv1.RefreshType, hydrate bool) (*argoappv1.Application, error) {
 	metadata := map[string]interface{}{
 		"metadata": map[string]interface{}{
 			"annotations": map[string]string{
 				argoappv1.AnnotationKeyRefresh: string(refreshType),
-				argoappv1.AnnotationKeyHydrate: "normal",
 			},
 		},
 	}
+	if hydrate {
+		metadata["metadata"].(map[string]interface{})["annotations"].(map[string]string)[argoappv1.AnnotationKeyHydrate] = string(argoappv1.HydrateTypeNormal)
+	}
+
 	var err error
 	patch, err := json.Marshal(metadata)
 	if err != nil {

util/argo/argo_test.go

@@ -41,7 +41,7 @@ func TestRefreshApp(t *testing.T) {
 	testApp.Namespace = "default"
 	appClientset := appclientset.NewSimpleClientset(&testApp)
 	appIf := appClientset.ArgoprojV1alpha1().Applications("default")
-	_, err := RefreshApp(appIf, "test-app", argoappv1.RefreshTypeNormal)
+	_, err := RefreshApp(appIf, "test-app", argoappv1.RefreshTypeNormal, true)
 	require.NoError(t, err)
 	// For some reason, the fake Application interface doesn't reflect the patch status after Patch(),
 	// so can't verify it was set in unit tests.

util/webhook/webhook.go

@@ -311,7 +311,7 @@ func (a *ArgoCDWebhookHandler) HandleEvent(payload interface{}) {
 					refreshPaths := path.GetAppRefreshPaths(&app)
 					if path.AppFilesHaveChanged(refreshPaths, changedFiles) {
 						namespacedAppInterface := a.appClientset.ArgoprojV1alpha1().Applications(app.ObjectMeta.Namespace)
-						_, err = argo.RefreshApp(namespacedAppInterface, app.ObjectMeta.Name, v1alpha1.RefreshTypeNormal)
+						_, err = argo.RefreshApp(namespacedAppInterface, app.ObjectMeta.Name, v1alpha1.RefreshTypeNormal, true)
 						if err != nil {
 							log.Warnf("Failed to refresh app '%s' for controller reprocessing: %v", app.ObjectMeta.Name, err)
 							continue