feat: implement history and rollback support for source hydrator (#27327)

Signed-off-by: liketosweep <liketosweep@gmail.com>

controller/appcontroller.go

@@ -1474,6 +1474,24 @@ func (ctrl *ApplicationController) processRequestedAppOperation(app *appv1.Appli
 		logCtx = logCtx.WithField("time_ms", time.Since(ts.StartTime).Milliseconds())
 		logCtx.Debug("Finished processing requested app operation")
 	}()
+
+	if app.Spec.SourceHydrator != nil && app.Operation != nil && app.Operation.Sync != nil {
+		revision := app.Operation.Sync.Revision
+		err := ctrl.hydrator.RollbackApp(context.Background(), app, revision)
+		if err != nil {
+			ctrl.setOperationState(app, &appv1.OperationState{
+				Phase:   synccommon.OperationFailed,
+				Message: err.Error(),
+			})
+			return
+		}
+		ctrl.setOperationState(app, &appv1.OperationState{
+			Phase:   synccommon.OperationSucceeded,
+			Message: fmt.Sprintf("rolled back to %s", revision),
+		})
+		return
+	}
+
 	terminatingCause := ""
 	if isOperationInProgress(app) {
 		state = app.Status.OperationState.DeepCopy()

controller/hydrator/hydrator.go

@@ -581,3 +581,84 @@ func IsRootPath(path string) bool {
 	clean := filepath.Clean(path)
 	return clean == "" || clean == "." || clean == string(filepath.Separator)
 }
+
+func (h *Hydrator) RollbackApp(ctx context.Context, app *appv1.Application, hydratedRevision string) error {
+	proj, err := h.dependencies.GetProcessableAppProj(app)
+	if err != nil {
+		return fmt.Errorf("failed to get project for %s: %w", app.QualifiedName(), err)
+	}
+
+	repoURL := app.Spec.SourceHydrator.DrySource.RepoURL
+	syncBranch := app.Spec.SourceHydrator.SyncSource.TargetBranch
+	syncPath := app.Spec.SourceHydrator.SyncSource.Path
+
+	rollbackSource := appv1.ApplicationSource{
+		RepoURL:        repoURL,
+		Path:           syncPath,
+		TargetRevision: hydratedRevision,
+	}
+
+	objs, _, err := h.dependencies.GetRepoObjs(ctx, app, rollbackSource, hydratedRevision, proj)
+	if err != nil {
+		return fmt.Errorf("failed to get manifests at revision %s: %w", hydratedRevision, err)
+	}
+
+	manifestDetails := make([]*commitclient.HydratedManifestDetails, len(objs))
+	for i, obj := range objs {
+		objJSON, err := json.Marshal(obj)
+		if err != nil {
+			return fmt.Errorf("failed to marshal object: %w", err)
+		}
+		manifestDetails[i] = &commitclient.HydratedManifestDetails{ManifestJSON: string(objJSON)}
+	}
+
+	repo, err := h.dependencies.GetWriteCredentials(ctx, repoURL, proj.Name)
+	if err != nil {
+		return fmt.Errorf("failed to get write credentials: %w", err)
+	}
+	if repo == nil {
+		repo = &appv1.Repository{Repo: repoURL}
+	}
+
+	authorName, err := h.dependencies.GetCommitAuthorName()
+	if err != nil {
+		return fmt.Errorf("failed to get commit author name: %w", err)
+	}
+	authorEmail, err := h.dependencies.GetCommitAuthorEmail()
+	if err != nil {
+		return fmt.Errorf("failed to get commit author email: %w", err)
+	}
+
+	revShort := hydratedRevision
+	if len(revShort) > 7 {
+		revShort = revShort[:7]
+	}
+
+	manifestsRequest := commitclient.CommitHydratedManifestsRequest{
+		Repo:          repo,
+		SyncBranch:    syncBranch,
+		TargetBranch:  syncBranch,
+		DrySha:        hydratedRevision,
+		CommitMessage: fmt.Sprintf("rollback %s to %s", app.Name, revShort),
+		Paths: []*commitclient.PathDetails{{
+			Path:      syncPath,
+			Manifests: manifestDetails,
+		}},
+		AuthorName:  authorName,
+		AuthorEmail: authorEmail,
+	}
+
+	closer, commitService, err := h.commitClientset.NewCommitServerClient()
+	if err != nil {
+		return fmt.Errorf("failed to create commit service: %w", err)
+	}
+	defer utilio.Close(closer)
+
+	_, err = commitService.CommitHydratedManifests(ctx, &manifestsRequest)
+	if err != nil {
+		return fmt.Errorf("failed to commit rollback: %w", err)
+	}
+
+	h.dependencies.RequestAppRefresh(app.Name, app.Namespace)
+	return nil
+}

server/application/application.go

@@ -2249,6 +2249,31 @@ func (s *Server) Rollback(ctx context.Context, rollbackReq *application.Applicat
 	if deploymentInfo == nil {
 		return nil, status.Errorf(codes.InvalidArgument, "application %s does not have deployment with id %v", a.QualifiedName(), rollbackReq.GetId())
 	}
+
+	if a.Spec.SourceHydrator != nil {
+		if deploymentInfo.Revision == "" {
+			return nil, status.Errorf(codes.FailedPrecondition,
+				"application %s history entry %d has no hydrated revision",
+				a.QualifiedName(), rollbackReq.GetId())
+		}
+		op := v1alpha1.Operation{
+			Sync: &v1alpha1.SyncOperation{
+				Revision: deploymentInfo.Revision,
+			},
+			InitiatedBy: v1alpha1.OperationInitiator{Username: session.Username(ctx)},
+		}
+		appName := rollbackReq.GetName()
+		appNs := s.appNamespaceOrDefault(rollbackReq.GetAppNamespace())
+		appIf := s.appclientset.ArgoprojV1alpha1().Applications(appNs)
+		a, err = argo.SetAppOperation(appIf, appName, &op)
+		if err != nil {
+			return nil, fmt.Errorf("error setting app operation: %w", err)
+		}
+		s.logAppEvent(ctx, a, argo.EventReasonOperationStarted,
+			fmt.Sprintf("initiated source hydrator rollback to %d", rollbackReq.GetId()))
+		return a, nil
+	}
+
 	if deploymentInfo.Source.IsZero() && deploymentInfo.Sources.IsZero() {
 		// Since source type was introduced to history starting with v0.12, and is now required for
 		// rollback, we cannot support rollback to revisions deployed using Argo CD v0.11 or below