No overview available.
Attributes
| Attribute | Type | Description |
|---|
| Scheme | *runtime.Scheme | Stores the runtime scheme for the TaskAction object, used for type-aware serialization and deserialization. |
| Recorder | events.EventRecorder | Records events related to the TaskAction object, providing an audit trail and debugging information. |
| PluginRegistry | *plugin.Registry | Manages and provides access to registered plugins, allowing the reconciler to extend its functionality. |
| DataStore | *storage.DataStore | Provides an interface for storing and retrieving data, enabling persistence for TaskAction-related information. |
| SecretManager | pluginsCore.SecretManager | Manages secrets, allowing the reconciler to securely access sensitive information required for TaskAction operations. |
| ResourceManager | pluginsCore.ResourceManager | Manages resources, enabling the reconciler to interact with and control external resources needed by TaskActions. |
| CatalogClient | catalog.AsyncClient | An asynchronous client for interacting with the catalog service, used for non-blocking catalog operations. |
| Catalog | catalog.Client | A synchronous client for interacting with the catalog service, used for blocking catalog operations. |
| MaxSystemFailures | uint32 | Defines the maximum number of system failures allowed before a TaskAction reconciliation attempt is considered unrecoverable. |
Methods
evaluateCacheBeforeExecution()
@classmethod
def evaluateCacheBeforeExecution(
ctx: context.Context,
taskAction: *flyteorgv1.TaskAction,
tCtx: pluginsCore.TaskExecutionContext
) - > (pluginsCore.Transition, bool, error)
Evaluates the cache before task execution to determine if the task can be short-circuited using cached results. This method checks for existing cached outputs to avoid redundant computation.
Parameters
| Name | Type | Description |
|---|
| ctx | context.Context | The context for the request, used for cancellation and deadlines. |
| taskAction | *flyteorgv1.TaskAction | The TaskAction object representing the task to be executed, containing its definition and current state. |
| tCtx | pluginsCore.TaskExecutionContext | The task execution context, providing access to task-specific information and utilities. |
Returns
| Type | Description |
|---|
(pluginsCore.Transition, bool, error) | A transition object indicating the next state, a boolean indicating if the cache was short-circuited, and an error if any occurred. |
finalizeCacheAfterExecution()
@classmethod
def finalizeCacheAfterExecution(
ctx: context.Context,
taskAction: *flyteorgv1.TaskAction,
tCtx: pluginsCore.TaskExecutionContext,
transition: pluginsCore.Transition,
cacheShortCircuited: bool
) - > (pluginsCore.Transition, error)
Finalizes the cache after task execution, potentially writing new outputs to the cache if the task completed successfully and was not short-circuited. This ensures future executions can benefit from the current task's results.
Parameters
| Name | Type | Description |
|---|
| ctx | context.Context | The context for the request, used for cancellation and deadlines. |
| taskAction | *flyteorgv1.TaskAction | The TaskAction object representing the task that was executed. |
| tCtx | pluginsCore.TaskExecutionContext | The task execution context, providing access to task-specific information and utilities. |
| transition | pluginsCore.Transition | The transition object representing the state change after task execution. |
| cacheShortCircuited | bool | A boolean indicating whether the task was short-circuited by the cache before execution. |
Returns
| Type | Description |
|---|
(pluginsCore.Transition, error) | A transition object indicating the next state and an error if any occurred during finalization. |
writeTaskOutputsToCache()
@classmethod
def writeTaskOutputsToCache(
ctx: context.Context,
tCtx: pluginsCore.TaskExecutionContext,
key: catalog.Key
) - > error
Writes the outputs of a completed task to the cache. This method is called when a task successfully finishes and its results should be stored for future reuse.
Parameters
| Name | Type | Description |
|---|
| ctx | context.Context | The context for the request, used for cancellation and deadlines. |
| tCtx | pluginsCore.TaskExecutionContext | The task execution context, providing access to task-specific information and outputs. |
| key | catalog.Key | The cache key under which the task outputs will be stored. |
Returns
| Type | Description |
|---|
error | An error if the write operation fails, otherwise nil. |
releaseCacheReservation()
@classmethod
def releaseCacheReservation(
ctx: context.Context,
cacheCfg: *taskCacheConfig
) - > error
Releases a previously acquired cache reservation. This is typically called after a task has completed and its cache entry has been finalized or if the reservation is no longer needed.
Parameters
| Name | Type | Description |
|---|
| ctx | context.Context | The context for the request, used for cancellation and deadlines. |
| cacheCfg | *taskCacheConfig | The configuration for the task cache, containing details about the reservation to be released. |
Returns
| Type | Description |
|---|
error | An error if releasing the reservation fails, otherwise nil. |
maxSystemFailures()
@classmethod
def maxSystemFailures() - > uint32
Retrieves the maximum number of system failures allowed for a task before it is marked as permanently failed. This value helps prevent infinite retries on persistent system issues.
Returns
| Type | Description |
|---|
uint32 | The maximum number of system failures allowed. |
resetPluginResource()
@classmethod
def resetPluginResource(
ctx: context.Context,
taskAction: *flyteorgv1.TaskAction,
p: pluginsCore.Plugin,
tCtx: pluginsCore.TaskExecutionContext
)
Resets the resources associated with a plugin for a given TaskAction. This is used to clean up or reinitialize plugin-specific state.
Parameters
| Name | Type | Description |
|---|
| ctx | context.Context | The context for the request, used for cancellation and deadlines. |
| taskAction | *flyteorgv1.TaskAction | The TaskAction object whose plugin resources need to be reset. |
| p | pluginsCore.Plugin | The plugin instance whose resources are to be reset. |
| tCtx | pluginsCore.TaskExecutionContext | The task execution context, providing access to task-specific information. |
recordSystemError()
@classmethod
def recordSystemError(
ctx: context.Context,
taskAction: *flyteorgv1.TaskAction,
original: *flyteorgv1.TaskAction,
pluginID: string,
handleErr: error
) - > (ctrl.Result, error)
Records a system error encountered during task execution and updates the TaskAction's status accordingly. This method handles errors that are not directly related to the user's task logic.
Parameters
| Name | Type | Description |
|---|
| ctx | context.Context | The context for the request, used for cancellation and deadlines. |
| taskAction | *flyteorgv1.TaskAction | The current TaskAction object that encountered the system error. |
| original | *flyteorgv1.TaskAction | The original TaskAction object before any modifications, used for comparison during status updates. |
| pluginID | string | The identifier of the plugin that reported the system error. |
| handleErr | error | The actual system error that occurred. |
Returns
| Type | Description |
|---|
(ctrl.Result, error) | A reconciliation result indicating whether to requeue the request and an error if the recording process itself failed. |
finalizePermanentFailure()
@classmethod
def finalizePermanentFailure(
ctx: context.Context,
taskAction: *flyteorgv1.TaskAction,
original: *flyteorgv1.TaskAction,
execErr: *core.ExecutionError
) - > (ctrl.Result, error)
Finalizes a TaskAction that has reached a permanent failure state. This method updates the TaskAction's status to indicate a terminal failure and prevents further retries.
Parameters
| Name | Type | Description |
|---|
| ctx | context.Context | The context for the request, used for cancellation and deadlines. |
| taskAction | *flyteorgv1.TaskAction | The current TaskAction object that is permanently failing. |
| original | *flyteorgv1.TaskAction | The original TaskAction object before any modifications, used for comparison during status updates. |
| execErr | *core.ExecutionError | The execution error details that led to the permanent failure. |
Returns
| Type | Description |
|---|
(ctrl.Result, error) | A reconciliation result indicating whether to requeue the request and an error if the finalization process failed. |
Reconcile()
@classmethod
def Reconcile(
ctx: context.Context,
req: ctrl.Request
) - > (ctrl.Result, error)
Reconciles a TaskAction object, driving its state transitions based on its current status and external events. This is the main entry point for the controller's reconciliation loop.
Parameters
| Name | Type | Description |
|---|
| ctx | context.Context | The context for the request, used for cancellation and deadlines. |
| req | ctrl.Request | The reconciliation request, containing the name and namespace of the TaskAction to reconcile. |
Returns
| Type | Description |
|---|
(ctrl.Result, error) | A reconciliation result indicating whether to requeue the request and an error if reconciliation failed. |
ensureTerminalLabels()
@classmethod
def ensureTerminalLabels(
ctx: context.Context,
taskAction: *flyteorgv1.TaskAction
) - > error
Ensures that a TaskAction object has the correct terminal labels applied. This is important for filtering and querying completed tasks.
Parameters
| Name | Type | Description |
|---|
| ctx | context.Context | The context for the request, used for cancellation and deadlines. |
| taskAction | *flyteorgv1.TaskAction | The TaskAction object to which terminal labels should be applied. |
Returns
| Type | Description |
|---|
error | An error if updating the labels fails, otherwise nil. |
handleAbortAndFinalize()
@classmethod
def handleAbortAndFinalize(
ctx: context.Context,
taskAction: *flyteorgv1.TaskAction
) - > (ctrl.Result, error)
Handles the abort request for a TaskAction and finalizes its state. This method ensures that an aborted task transitions to a terminal state gracefully.
Parameters
| Name | Type | Description |
|---|
| ctx | context.Context | The context for the request, used for cancellation and deadlines. |
| taskAction | *flyteorgv1.TaskAction | The TaskAction object that is being aborted. |
Returns
| Type | Description |
|---|
(ctrl.Result, error) | A reconciliation result indicating whether to requeue the request and an error if the abort or finalization process failed. |
removeFinalizer()
@classmethod
def removeFinalizer(
ctx: context.Context,
taskAction: *flyteorgv1.TaskAction
) - > (ctrl.Result, error)
Removes the finalizer from a TaskAction object. This is typically done after all cleanup operations for the TaskAction have been completed, allowing the object to be garbage collected.
Parameters
| Name | Type | Description |
|---|
| ctx | context.Context | The context for the request, used for cancellation and deadlines. |
| taskAction | *flyteorgv1.TaskAction | The TaskAction object from which the finalizer should be removed. |
Returns
| Type | Description |
|---|
(ctrl.Result, error) | A reconciliation result indicating whether to requeue the request and an error if removing the finalizer failed. |
updateTaskActionStatus()
@classmethod
def updateTaskActionStatus(
ctx: context.Context,
oldTaskAction: *flyteorgv1.TaskAction,
newTaskAction: *flyteorgv1.TaskAction,
phaseInfo: pluginsCore.PhaseInfo
) - > error
Updates the status of a TaskAction object based on the provided phase information. This method ensures that the TaskAction's status reflects its current state in the execution lifecycle.
Parameters
| Name | Type | Description |
|---|
| ctx | context.Context | The context for the request, used for cancellation and deadlines. |
| oldTaskAction | *flyteorgv1.TaskAction | The previous state of the TaskAction object, used for comparison to detect changes. |
| newTaskAction | *flyteorgv1.TaskAction | The new state of the TaskAction object with updated status information. |
| phaseInfo | pluginsCore.PhaseInfo | Information about the current phase of the task, including its state and any associated errors or details. |
Returns
| Type | Description |
|---|
error | An error if updating the status fails, otherwise nil. |
buildActionEvent()
@classmethod
def buildActionEvent(
ctx: context.Context,
taskAction: *flyteorgv1.TaskAction,
phaseInfo: pluginsCore.PhaseInfo
) - > *workflow.ActionEvent
Builds an ActionEvent object from a TaskAction and its current phase information. This event is used for external reporting and auditing of task lifecycle changes.
Parameters
| Name | Type | Description |
|---|
| ctx | context.Context | The context for the request, used for cancellation and deadlines. |
| taskAction | *flyteorgv1.TaskAction | The TaskAction object for which the event is being built. |
| phaseInfo | pluginsCore.PhaseInfo | Information about the current phase of the task, used to populate event details. |
Returns
| Type | Description |
|---|
*workflow.ActionEvent | The constructed ActionEvent object. |
SetupWithManager()
@classmethod
def SetupWithManager(
mgr: ctrl.Manager
) - > error
Sets up the TaskActionReconciler with a controller manager. This method registers the reconciler to watch for TaskAction objects and enqueue reconciliation requests.
Parameters
| Name | Type | Description |
|---|
| mgr | ctrl.Manager | The controller manager responsible for managing controllers and their watches. |
Returns
| Type | Description |
|---|
error | An error if the setup fails, otherwise nil. |