feat: keep track of promotion state in promotion status

api/v1alpha1/promotion_types.go

@@ -143,6 +143,26 @@
 	HealthChecks []HealthCheckStep `json:"healthChecks,omitempty" protobuf:"bytes,8,rep,name=healthChecks"`
 	// FinishedAt is the time when the promotion was completed.
 	FinishedAt *metav1.Time `json:"finishedAt,omitempty" protobuf:"bytes,6,opt,name=finishedAt"`
+	// CurrentStep is the index of the current promotion step being executed. This
+	// permits steps that have already run successfully to be skipped on
+	// subsequent reconciliations attempts.
+	CurrentStep int64 `json:"currentStep,omitempty" protobuf:"varint,9,opt,name=currentStep"`
+	// State stores the state of the promotion process between reconciliation
+	// attempts.
+	State *apiextensionsv1.JSON `json:"state,omitempty" protobuf:"bytes,10,opt,name=state"`
+}
+
+// GetConfig returns the State field as unmarshalled YAML.
+func (s *PromotionStatus) GetState() map[string]any {
+	if s.State == nil {
+		return nil
+	}
+
+	var state map[string]any
+	if err := yaml.Unmarshal(s.State.Raw, &state); err != nil {
+		return nil
+	}
+	return state
 }
 
 // HealthCheckStep describes a health check directive which can be executed by

charts/kargo/resources/crds/kargo.akuity.io_promotions.yaml

@@ -109,6 +109,13 @@ spec:
               Status describes the current state of the transition represented by this
               Promotion.
             properties:
+              currentStep:
+                description: |-
+                  CurrentStep is the index of the current promotion step being executed. This
+                  permits steps that have already run successfully to be skipped on
+                  subsequent reconciliations attempts.
+                format: int64
+                type: integer
               finishedAt:
                 description: FinishedAt is the time when the promotion was completed.
                 format: date-time
@@ -501,6 +508,11 @@ spec:
                 description: Phase describes where the Promotion currently is in its
                   lifecycle.
                 type: string
+              state:
+                description: |-
+                  State stores the state of the promotion process between reconciliation
+                  attempts.
+                x-kubernetes-preserve-unknown-fields: true
             type: object
         required:
         - spec

charts/kargo/resources/crds/kargo.akuity.io_stages.yaml

@@ -1211,6 +1211,13 @@ spec:
                   status:
                     description: Status is the (optional) status of the Promotion.
                     properties:
+                      currentStep:
+                        description: |-
+                          CurrentStep is the index of the current promotion step being executed. This
+                          permits steps that have already run successfully to be skipped on
+                          subsequent reconciliations attempts.
+                        format: int64
+                        type: integer
                       finishedAt:
                         description: FinishedAt is the time when the promotion was
                           completed.
@@ -1611,6 +1618,11 @@ spec:
                         description: Phase describes where the Promotion currently
                           is in its lifecycle.
                         type: string
+                      state:
+                        description: |-
+                          State stores the state of the promotion process between reconciliation
+                          attempts.
+                        x-kubernetes-preserve-unknown-fields: true
                     type: object
                 required:
                 - name
@@ -2067,6 +2079,13 @@ spec:
                   status:
                     description: Status is the (optional) status of the Promotion.
                     properties:
+                      currentStep:
+                        description: |-
+                          CurrentStep is the index of the current promotion step being executed. This
+                          permits steps that have already run successfully to be skipped on
+                          subsequent reconciliations attempts.
+                        format: int64
+                        type: integer
                       finishedAt:
                         description: FinishedAt is the time when the promotion was
                           completed.
@@ -2467,6 +2486,11 @@ spec:
                         description: Phase describes where the Promotion currently
                           is in its lifecycle.
                         type: string
+                      state:
+                        description: |-
+                          State stores the state of the promotion process between reconciliation
+                          attempts.
+                        x-kubernetes-preserve-unknown-fields: true
                     type: object
                 required:
                 - name

internal/controller/promotions/promotions.go

@@ -511,25 +511,34 @@
 			})
 		}
 
-		workDir := filepath.Join(os.TempDir(), "promotion-"+string(workingPromo.UID))
-		if err := os.MkdirAll(workDir, 0o700); err != nil && !os.IsExist(err) {
+		promoCtx := directives.PromotionContext{
+			WorkDir:         filepath.Join(os.TempDir(), "promotion-"+string(workingPromo.UID)),
+			Project:         stageNamespace,
+			Stage:           stageName,
+			FreightRequests: stage.Spec.RequestedFreight,
+			Freight:         *workingPromo.Status.FreightCollection.DeepCopy(),
+			StartFromStep:   promo.Status.CurrentStep,
+			State:           directives.State(workingPromo.Status.GetState()),
+		}
+		if err := os.Mkdir(promoCtx.WorkDir, 0o700); err == nil {
+			// If we're working with a fresh directory, we should start the promotion
+			// process again from the beginning.
+			promoCtx.StartFromStep = 0
+			promoCtx.State = nil
+		} else if !os.IsExist(err) {
 			return nil, fmt.Errorf("error creating working directory: %w", err)
 		}
 		defer func() {
 			if workingPromo.Status.Phase.IsTerminal() {
-				if err := os.RemoveAll(workDir); err != nil {
+				if err := os.RemoveAll(promoCtx.WorkDir); err != nil {
 					logger.Error(err, "could not remove working directory")
 				}
 			}
 		}()
 
-		res, err := r.directivesEngine.Promote(ctx, directives.PromotionContext{
-			WorkDir:         workDir,
-			Project:         stageNamespace,
-			Stage:           stageName,
-			FreightRequests: stage.Spec.RequestedFreight,
-			Freight:         *workingPromo.Status.FreightCollection.DeepCopy(),
-		}, steps)
+		res, err := r.directivesEngine.Promote(ctx, promoCtx, steps)
+		workingPromo.Status.CurrentStep = res.CurrentStep
+		workingPromo.Status.State = &apiextensionsv1.JSON{Raw: res.State.ToJSON()}
 		switch res.Status {
 		case directives.PromotionStatusPending:
 			workingPromo.Status.Phase = kargoapi.PromotionPhaseRunning
@@ -541,7 +550,6 @@
 					Config: &apiextensionsv1.JSON{Raw: step.Config.ToJSON()},
 				})
 			}
-
 			workingPromo.Status.Phase = kargoapi.PromotionPhaseSucceeded
 			workingPromo.Status.HealthChecks = healthChecks
 		case directives.PromotionStatusFailure:

internal/directives/git_pr_waiter.go

@@ -131,35 +131,42 @@
 }
 
 func getPRNumber(sharedState State, cfg GitWaitForPRConfig) (int64, error) {
-	prNumber := cfg.PRNumber
-	if cfg.PRNumberFromOpen != "" {
-		stepOutput, exists := sharedState.Get(cfg.PRNumberFromOpen)
-		if !exists {
-			return 0, fmt.Errorf(
-				"no output found from step with alias %q",
-				cfg.PRNumberFromOpen,
-			)
-		}
-		stepOutputMap, ok := stepOutput.(map[string]any)
-		if !ok {
-			return 0, fmt.Errorf(
-				"output from step with alias %q is not a map[string]any",
-				cfg.PRNumberFromOpen,
-			)
-		}
-		prNumberAny, exists := stepOutputMap[prNumberKey]
-		if !exists {
-			return 0, fmt.Errorf(
-				"no PR number found in output from step with alias %q",
-				cfg.PRNumberFromOpen,
-			)
-		}
-		if prNumber, ok = prNumberAny.(int64); !ok {
-			return 0, fmt.Errorf(
-				"PR number in output from step with alias %q is not an int64",
-				cfg.PRNumberFromOpen,
-			)
-		}
+	if cfg.PRNumberFromOpen == "" {
+		return cfg.PRNumber, nil
+	}
+	stepOutput, exists := sharedState.Get(cfg.PRNumberFromOpen)
+	if !exists {
+		return 0, fmt.Errorf(
+			"no output found from step with alias %q",
+			cfg.PRNumberFromOpen,
+		)
+	}
+	stepOutputMap, ok := stepOutput.(map[string]any)
+	if !ok {
+		return 0, fmt.Errorf(
+			"output from step with alias %q is not a map[string]any",
+			cfg.PRNumberFromOpen,
+		)
+	}
+	prNumberAny, exists := stepOutputMap[prNumberKey]
+	if !exists {
+		return 0, fmt.Errorf(
+			"no PR number found in output from step with alias %q",
+			cfg.PRNumberFromOpen,
+		)
+	}
+	// If the state was rehydrated from PromotionStatus, which makes use of
+	// apiextensions.JSON, the PR number will be a float64. Otherwise, it will be
+	// an int64. We need to handle both cases.
+	switch prNumber := prNumberAny.(type) {
+	case int64:
+		return prNumber, nil
+	case float64:
+		return int64(prNumber), nil
+	default:
+		return 0, fmt.Errorf(
+			"PR number in output from step with alias %q is not an int64",
+			cfg.PRNumberFromOpen,
+		)
 	}
-	return prNumber, nil
 }

internal/directives/promotions.go

@@ -44,6 +44,11 @@ type PromotionContext struct {
 	// well as any Freight that has been inherited from the target Stage's current
 	// state.
 	Freight kargoapi.FreightCollection
+	// SharedState is the index of the step from which the promotion should begin
+	// execution.
+	StartFromStep int64
+	// State is the current state of the promotion process.
+	State State
 }
 
 // PromotionStep describes a single step in a user-defined promotion process.
@@ -79,6 +84,12 @@ type PromotionResult struct {
 	// PromotionStepRunners. This configuration can later be used as input to
 	// health check processes.
 	HealthCheckSteps []HealthCheckStep
+	// If the promotion process remains in-progress, perhaps waiting for a change
+	// in some external state, the value of this field will indicated where to
+	// resume the process in the next reconciliation.
+	CurrentStep int64
+	// State is the current state of the promotion process.
+	State State
 }
 
 // PromotionStatus is a type that represents the high-level outcome of the

internal/directives/simple_engine.go

@@ -48,24 +48,40 @@
 		var err error
 		workDir, err = os.MkdirTemp("", "run-")
 		if err != nil {
-			return PromotionResult{Status: PromotionStatusFailure},
+			return PromotionResult{
+					Status:      PromotionStatusFailure,
+					CurrentStep: 0,
+				},
 				fmt.Errorf("temporary working directory creation failed: %w", err)
 		}
 		defer os.RemoveAll(workDir)
 	}
 
 	// Initialize the shared state that will be passed to each step.
-	state := make(State)
+	state := promoCtx.State.DeepCopy()
+	if state == nil {
+		state = make(State)
+	}
 	var healthCheckSteps []HealthCheckStep
-	for _, step := range steps {
+
+	for i := promoCtx.StartFromStep; i < int64(len(steps)); i++ {
+		step := steps[i]
 		select {
 		case <-ctx.Done():
-			return PromotionResult{Status: PromotionStatusFailure}, ctx.Err()
+			return PromotionResult{
+				Status:      PromotionStatusFailure,
+				CurrentStep: i,
+				State:       state,
+			}, ctx.Err()
 		default:
 		}
 		reg, err := e.registry.GetPromotionStepRunnerRegistration(step.Kind)
 		if err != nil {
-			return PromotionResult{Status: PromotionStatusFailure},
+			return PromotionResult{
+					Status:      PromotionStatusFailure,
+					CurrentStep: i,
+					State:       state,
+				},
 				fmt.Errorf("no runner registered for step kind %q: %w", step.Kind, err)
 		}
 
@@ -93,17 +109,24 @@
 		}
 
 		result, err := reg.Runner.RunPromotionStep(ctx, stepCtx)
+		if step.Alias != "" {
+			state[step.Alias] = result.Output
+		}
 		if err != nil {
-			return PromotionResult{Status: PromotionStatusFailure},
+			return PromotionResult{
+					Status:      PromotionStatusFailure,
+					CurrentStep: i,
+					State:       state,
+				},
 				fmt.Errorf("failed to run step %q: %w", step.Kind, err)
 		}
 
 		if result.Status != PromotionStatusSuccess {
-			return PromotionResult{Status: result.Status}, nil
-		}
-
-		if step.Alias != "" {
-			state[step.Alias] = result.Output
+			return PromotionResult{
+				Status:      result.Status,
+				CurrentStep: i,
+				State:       state,
+			}, nil
 		}
 
 		if result.HealthCheckStep != nil {
@@ -113,6 +136,8 @@
 	return PromotionResult{
 		Status:           PromotionStatusSuccess,
 		HealthCheckSteps: healthCheckSteps,
+		CurrentStep:      0,
+		State:            state,
 	}, nil
 }
 

internal/directives/state.go

@@ -1,6 +1,10 @@
 package directives
 
-import "k8s.io/apimachinery/pkg/runtime"
+import (
+	"encoding/json"
+
+	"k8s.io/apimachinery/pkg/runtime"
+)
 
 // State is a type that represents shared state between executions of
 // PromotionSteps. It is not safe for concurrent use at present, as we expect
@@ -28,3 +32,12 @@
 	// consider writing our own implementation in the future.
 	return runtime.DeepCopyJSON(*s)
 }
+
+// ToJSON marshals the State to JSON.
+func (s State) ToJSON() []byte {
+	if len(s) == 0 {
+		return nil
+	}
+	b, _ := json.Marshal(s)
+	return b
+}