Skip to main content

The Reconciliation Process

The reconciliation process in Flyte is the core control loop that manages the lifecycle of a TaskAction custom resource. The TaskActionReconciler (found in executor/pkg/controller/taskaction_controller.go) acts as a bridge between the Kubernetes controller-runtime and Flyte's internal plugin machinery, ensuring that tasks move from creation to completion while handling retries, failures, and resource cleanup.

Initialization and Registration​

The TaskActionReconciler is registered with the Kubernetes Manager during the executor's startup. It is configured to watch TaskAction resources and own the underlying pods it creates.

// executor/pkg/controller/taskaction_controller.go

func (r *TaskActionReconciler) SetupWithManager(mgr ctrl.Manager) error {
	return ctrl.NewControllerManagedBy(mgr).
		For(&flyteorgv1.TaskAction{}).
		Owns(&corev1.Pod{}).
		Named("taskaction").
		Complete(r)
}

When a TaskAction is created or updated, the Reconcile method is triggered. It first fetches the resource and checks if it is already in a terminal state (Succeeded, Failed, or Aborted). If terminal, it stamps the resource with garbage collection labels and exits.

Validation and Plugin Resolution​

Before any execution logic begins, the reconciler must ensure the TaskAction spec is valid and that a handler exists for the requested taskType. This is handled by the validateTaskAction function using the pluginResolver interface.

// executor/pkg/controller/taskaction_controller.go

p, reason, err := validateTaskAction(taskAction, r.PluginRegistry)
if err != nil {
    // ... handle validation error ...
    setCondition(taskAction, flyteorgv1.ConditionTypeFailed, metav1.ConditionTrue, reason, err.Error())
    r.Status().Update(ctx, taskAction)
    return ctrl.Result{}, nil // terminal — do not requeue
}

The pluginResolver (implemented by plugin.Registry) maps the taskType (e.g., "container", "spark") to a specific pluginsCore.Plugin implementation. If validation fails or the plugin cannot be resolved, the reconciler marks the task as failed immediately without adding a finalizer, allowing the resource to be deleted without manual intervention.

The Execution Loop​

Once validated, the reconciler ensures the flyte.org/plugin-finalizer is present and then prepares the execution environment.

Building the Execution Context​

The reconciler constructs a TaskExecutionContext using plugin.NewTaskExecutionContext. This context provides the plugin with access to:

  • DataStore: For reading inputs and writing outputs.
  • SecretManager: For accessing encrypted credentials.
  • ResourceManager: For managing external resource quotas.
  • PluginStateManager: For persisting plugin-specific state across reconciliation loops.

Invoking the Plugin​

The core of the reconciliation is the call to the plugin's Handle method:

// executor/pkg/controller/taskaction_controller.go

transition, err = p.Handle(ctx, tCtx)
if err != nil {
    return r.recordSystemError(ctx, taskAction, originalTaskActionInstance, p.GetID(), err)
}

The Handle method returns a Transition, which describes the next state of the task (e.g., Queued, Running, Success). The reconciler then maps this transition to Kubernetes conditions and updates the TaskAction status.

Error Handling and Retries​

Flyte distinguishes between two types of failures during reconciliation:

System Failures​

System failures are errors encountered by the Flyte infrastructure or the plugin itself (e.g., network timeouts, K8s API errors). These are tracked via Status.SystemFailures.

  • They do not consume the user's retry budget (Status.Attempts).
  • If SystemFailures exceeds the MaxSystemFailures threshold (default: 3), the task is marked as a PermanentFailure.
  • The counter is reset to 0 upon any successful non-system-error transition.

User Failures​

User failures occur when the task logic itself fails (e.g., a Python script exits with a non-zero code). These are reported by the plugin as PhaseRetryableFailure.

  • The reconciler increments Status.Attempts.
  • If attempts are still available, the reconciler calls p.Abort(ctx, tCtx) to clean up the current attempt and clears the PluginState to allow a fresh restart.
  • If attempts are exhausted, the task moves to PhasePermanentFailure.

Finalization and Cleanup​

When a TaskAction is deleted, the Kubernetes API sets a DeletionTimestamp. The reconciler detects this and enters the handleAbortAndFinalize flow.

// executor/pkg/controller/taskaction_controller.go

if !taskAction.DeletionTimestamp.IsZero() {
    return r.handleAbortAndFinalize(ctx, taskAction)
}

The reconciler calls the plugin's Abort and Finalize methods. This ensures that external resources (like Spark clusters or cloud-native jobs) are terminated and cleaned up before the TaskAction CRD is finally removed from the cluster. Once these calls succeed, the flyte.org/plugin-finalizer is removed.