Method operate

workflow/controller/operator.go:193–537

operate is the main operator logic of a workflow. It evaluates the current state of the workflow, and its pods and decides how to proceed down the execution path. TODO: an error returned by this method should result in requeuing the workflow to be retried at a later time As you must not call `persis

(ctx context.Context)

// later time
// As you must not call `persistUpdates` twice, you must not call `operate` twice.
func (woc *wfOperationCtx) operate(ctx context.Context) {
	defer argoruntime.RecoverFromPanic(ctx, woc.log)

	defer func() {
		woc.persistUpdates(ctx)
	}()
	defer func() {
		if r := recover(); r != nil {
			woc.log.WithFields(logging.Fields{"stack": string(debug.Stack()), "r": r}).Error(ctx, "Recovered from panic")
			if rerr, ok := r.(error); ok {
				woc.markWorkflowError(ctx, rerr)
			} else {
				woc.markWorkflowError(ctx, fmt.Errorf("%v", r))
			}
			woc.controller.metrics.OperationPanic(ctx)
		}
	}()

	woc.log.WithFields(logging.Fields{"phase": woc.wf.Status.Phase, "resourceVersion": woc.wf.ObjectMeta.ResourceVersion, "lastSeenVersion": woc.wf.GetAnnotations()[common.AnnotationKeyLastSeenVersion]}).Info(ctx, "Processing workflow")

	// Set the Execute workflow spec for execution
	// ExecWF is a runtime execution spec which merged from Wf, WFT and Wfdefault
	err := woc.setExecWorkflow(ctx)
	if err != nil {
		woc.log.WithError(err).Error(ctx, "Unable to set ExecWorkflow")
		return
	}

	if woc.wf.Status.ArtifactRepositoryRef == nil {
		ref, err := woc.controller.artifactRepositories.Resolve(ctx, woc.execWf.Spec.ArtifactRepositoryRef, woc.wf.Namespace)
		if err != nil {
			woc.markWorkflowError(ctx, fmt.Errorf("failed to resolve artifact repository: %w", err))
			return
		}
		woc.wf.Status.ArtifactRepositoryRef = ref
		woc.updated = true
	}

	repo, err := woc.controller.artifactRepositories.Get(ctx, woc.wf.Status.ArtifactRepositoryRef)
	if err != nil {
		woc.markWorkflowError(ctx, fmt.Errorf("failed to get artifact repository: %v", err))
		return
	}
	woc.artifactRepository = repo

	woc.addArtifactGCFinalizer(ctx)

	// Reconciliation of Outputs (Artifacts). See ReportOutputs() of executor.go.
	woc.taskResultReconciliation(ctx)

	// Do artifact GC if task result reconciliation is complete.
	if woc.wf.Status.Fulfilled() {
		if err := woc.garbageCollectArtifacts(ctx); err != nil {
			woc.log.WithError(err).Error(ctx, "failed to GC artifacts")
			return
		}
	} else {
		woc.log.Debug(ctx, "Skipping artifact GC")