Execution Engine

The Execution Engine in Flyte is the core component responsible for the lifecycle of TaskAction resources. It operates as a Kubernetes controller that reconciles the state of TaskAction Custom Resources (CRDs) by delegating the actual work to specific plugins. The engine manages plugin execution, state persistence, error handling, and resource cleanup.

TaskAction Reconciliation Loop

The TaskActionReconciler, located in executor/pkg/controller/taskaction_controller.go, implements the primary reconciliation logic. For every TaskAction resource, the reconciler follows a structured lifecycle:

1. Validation and Plugin Resolution: The reconciler first validates the TaskAction spec and resolves the appropriate plugin using the PluginRegistry. If validation fails or the plugin cannot be found, the resource is marked as terminal immediately.
2. Finalizer Management: Once validation passes, the reconciler adds the flyte.org/plugin-finalizer to the resource. This ensures that the plugin has an opportunity to clean up external resources (like Kubernetes Pods or cloud jobs) before the TaskAction is deleted.
3. Context Construction: The engine builds a TaskExecutionContext and a PluginStateManager. These provide the plugin with necessary abstractions for storage, secrets, and state persistence.
4. Cache Evaluation: Before invoking the plugin, the engine checks the Flyte Catalog for existing outputs. If a cache hit occurs, the engine short-circuits execution and moves directly to a success state.
5. Plugin Invocation: The engine calls the plugin's Handle method. The plugin returns a Transition indicating the next phase of execution (e.g., Running, Success, RetryableFailure).
6. Status Persistence: The engine updates the TaskAction status with the new phase, persists any plugin-specific state, and reports events back to the Flyte backend.

// executor/pkg/controller/taskaction_controller.go

func (r *TaskActionReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
    // ... fetch taskAction ...
    
    // 1. Validate and resolve plugin
    p, reason, err := validateTaskAction(taskAction, r.PluginRegistry)
    if err != nil {
        // handle validation failure and mark terminal
        return ctrl.Result{}, nil 
    }

    // 2. Ensure finalizer is present
    if !controllerutil.ContainsFinalizer(taskAction, taskActionFinalizer) {
        controllerutil.AddFinalizer(taskAction, taskActionFinalizer)
        return ctrl.Result{}, r.Update(ctx, taskAction)
    }

    // 3. Build Context and State Manager
    stateMgr := plugin.NewPluginStateManager(taskAction.Status.PluginState, taskAction.Status.PluginStateVersion)
    tCtx, _ := plugin.NewTaskExecutionContext(taskAction, r.DataStore, stateMgr, r.SecretManager, r.ResourceManager, r.CatalogClient)

    // 4. Cache Check
    transition, cacheShortCircuited, err := r.evaluateCacheBeforeExecution(ctx, taskAction, tCtx)
    
    // 5. Plugin Handle
    if !cacheShortCircuited {
        transition, err = p.Handle(ctx, tCtx)
    }

    // 6. Update Status and Persist State
    // ... map transition to conditions and update status ...
}

Error Handling and Retries

Flyte distinguishes between two types of failures during execution: System Failures and User Failures.

System Failures

System failures are transient errors caused by the infrastructure (e.g., Kubernetes API timeouts or network issues). The engine tracks these using Status.SystemFailures.

• The engine allows a maximum number of consecutive system failures, defined by DefaultMaxSystemFailures (default: 3).
• If the threshold is exceeded, the TaskAction is moved to a PermanentFailure state with the code MaxSystemFailuresExceeded.
• On a system failure, the engine resets the plugin resource and clears the persisted state to ensure the next attempt starts fresh.

User Retries

User failures are errors returned by the task itself (e.g., a Python script crashing).

• If a plugin returns a PhaseRetryableFailure of kind USER, the engine performs an in-place retry.
• It increments Status.Attempts, aborts the current plugin resource (e.g., deletes the current Pod), and clears the PluginState.
• The next reconciliation loop will see the incremented attempt count and trigger the plugin to start a new execution.

Task Execution Context

The TaskExecutionContext, implemented in executor/pkg/plugin/task_exec_context.go, abstracts the environment for plugins. It provides:

• Input/Output Readers: Plugins use InputReader to fetch task inputs and OutputWriter to persist results.
• Storage Sharding: The engine uses ComputeActionOutputPath to generate deterministic, sharded paths for task outputs. This avoids S3 hot-spots by inserting a base-36 shard prefix derived from the TaskAction name and namespace.
• Secret Manager: Provides access to secrets required by the task.
• Plugin State Manager: Manages the serialization of plugin-specific data using gob encoding, allowing plugins to resume execution across multiple reconciliation loops.

Garbage Collection

To prevent CRD bloat in the Kubernetes cluster, Flyte includes a GarbageCollector (found in executor/pkg/controller/garbage_collector.go).

When a TaskAction reaches a terminal state (Succeeded, Failed, or Aborted), the reconciler stamps it with two labels:

1. flyte.org/termination-status: Set to terminated.
2. flyte.org/completed-time: Set to the UTC time of completion in YYYY-MM-DD.HH-mm format.

The GarbageCollector runs as a background process and periodically lists resources with these labels. It uses lexicographical string comparison on the completed-time label against a calculated cutoff to identify and delete expired resources.

// executor/pkg/controller/garbage_collector.go

func (gc *GarbageCollector) collect(ctx context.Context) error {
    cutoff := time.Now().UTC().Add(-gc.maxTTL).Format(labelTimeFormat)
    listOpts := []client.ListOption{
        client.MatchingLabels{LabelTerminationStatus: LabelValueTerminated},
        client.HasLabels{LabelCompletedTime},
        client.Limit(gcPageSize),
    }
    // ... list and delete items where completedTime < cutoff ...
}