This class handles all etcd/K8s TaskAction CR operations for the Actions service. It manages watching for updates, deduplicates RecordAction calls, and uses a worker pool to process events for TaskAction resources while preserving per-resource ordering.
Methods
Enqueue()
@classmethod
def Enqueue(
ctx: context.Context,
action: *actions.Action,
runSpec: *task.RunSpec
) - > error
Enqueues a new action for processing. This method adds the action to a queue to be picked up by workers.
Parameters
| Name | Type | Description |
|---|
| ctx | context.Context | The context for the request, used for cancellation and deadlines. |
| action | *actions.Action | The action to be enqueued. |
| runSpec | *task.RunSpec | The run specification associated with the action. |
Returns
| Type | Description |
|---|
error | An error if the action could not be enqueued, otherwise nil. |
AbortAction()
@classmethod
def AbortAction(
ctx: context.Context,
actionID: *common.ActionIdentifier,
reason: *string
) - > error
Aborts a running action. This method signals the system to stop the specified action.
Parameters
| Name | Type | Description |
|---|
| ctx | context.Context | The context for the request, used for cancellation and deadlines. |
| actionID | *common.ActionIdentifier | The identifier of the action to abort. |
| reason | *string | The reason for aborting the action. |
Returns
| Type | Description |
|---|
error | An error if the action could not be aborted, otherwise nil. |
PutStatus()
@classmethod
def PutStatus(
ctx: context.Context,
actionID: *common.ActionIdentifier,
attempt: uint32,
status: *workflow.ActionStatus
) - > error
Updates the status of a specific attempt for an action. This method is used to report the current state of an action's execution.
Parameters
| Name | Type | Description |
|---|
| ctx | context.Context | The context for the request, used for cancellation and deadlines. |
| actionID | *common.ActionIdentifier | The identifier of the action whose status is being updated. |
| attempt | uint32 | The attempt number for which the status is being updated. |
| status | *workflow.ActionStatus | The new status to set for the action attempt. |
Returns
| Type | Description |
|---|
error | An error if the status could not be updated, otherwise nil. |
ListRunActions()
@classmethod
def ListRunActions(
ctx: context.Context,
runID: *common.RunIdentifier
) - > []*executorv1.TaskAction, error
Lists all actions associated with a specific run. This method retrieves all TaskActions that belong to a given run.
Parameters
| Name | Type | Description |
|---|
| ctx | context.Context | The context for the request, used for cancellation and deadlines. |
| runID | *common.RunIdentifier | The identifier of the run for which to list actions. |
Returns
| Type | Description |
|---|
[]*executorv1.TaskAction, error | A slice of TaskAction objects associated with the run and an error if the listing failed. |
ListChildActions()
@classmethod
def ListChildActions(
ctx: context.Context,
parentActionID: *common.ActionIdentifier
) - > []*executorv1.TaskAction, error
Lists all child actions for a given parent action. This method retrieves all TaskActions that are direct descendants of a specified parent action.
Parameters
| Name | Type | Description |
|---|
| ctx | context.Context | The context for the request, used for cancellation and deadlines. |
| parentActionID | *common.ActionIdentifier | The identifier of the parent action whose children are to be listed. |
Returns
| Type | Description |
|---|
[]*executorv1.TaskAction, error | A slice of child TaskAction objects and an error if the listing failed. |
GetTaskAction()
@classmethod
def GetTaskAction(
ctx: context.Context,
actionID: *common.ActionIdentifier
) - > *executorv1.TaskAction, error
Retrieves a specific TaskAction by its identifier. This method fetches the details of a single TaskAction.
Parameters
| Name | Type | Description |
|---|
| ctx | context.Context | The context for the request, used for cancellation and deadlines. |
| actionID | *common.ActionIdentifier | The identifier of the TaskAction to retrieve. |
Returns
| Type | Description |
|---|
*executorv1.TaskAction, error | The requested TaskAction object and an error if it could not be found or retrieved. |
Subscribe()
@classmethod
def Subscribe(
parentActionName: string
) - > chan *ActionUpdate
Subscribes to updates for actions under a specific parent action. Callers use this to receive real-time notifications about changes to child actions.
Parameters
| Name | Type | Description |
|---|
| parentActionName | string | The name of the parent action to subscribe to. |
Returns
| Type | Description |
|---|
chan *ActionUpdate | A channel that will receive ActionUpdate messages. |
Unsubscribe()
@classmethod
def Unsubscribe(
parentActionName: string,
ch: chan *ActionUpdate
)
Unsubscribes a channel from receiving updates for a parent action. This stops the flow of notifications to the specified channel.
Parameters
| Name | Type | Description |
|---|
| parentActionName | string | The name of the parent action from which to unsubscribe. |
| ch | chan *ActionUpdate | The channel to unsubscribe. |
StartWatching()
@classmethod
def StartWatching(
ctx: context.Context
) - > error
Initiates the watching process for TaskAction resources. This method sets up the necessary informers and workers to monitor changes in TaskActions.
Parameters
| Name | Type | Description |
|---|
| ctx | context.Context | The context for starting the watch, used for cancellation. |
Returns
| Type | Description |
|---|
error | An error if the watching process could not be started, otherwise nil. |
@classmethod
def setupInformer(
ctx: context.Context
) - > error
Sets up the Kubernetes informer for TaskAction resources. This internal method configures the mechanism to receive events about TaskAction changes.
Parameters
| Name | Type | Description |
|---|
| ctx | context.Context | The context for setting up the informer. |
Returns
| Type | Description |
|---|
error | An error if the informer could not be set up, otherwise nil. |
worker()
@classmethod
def worker(
ctx: context.Context,
ch: < -chan watch.Event,
stopCh: < -chan struct{}
)
Processes watch events for TaskActions. This method runs as a goroutine, handling events from the Kubernetes API and dispatching them.
Parameters
| Name | Type | Description |
|---|
| ctx | context.Context | The context for the worker, used for cancellation. |
| ch | < -chan watch.Event | The channel from which to receive watch events. |
| stopCh | < -chan struct{} | A channel that signals the worker to stop. |
dispatchEvent()
@classmethod
def dispatchEvent(
taskAction: *executorv1.TaskAction,
eventType: watch.EventType
)
Dispatches a TaskAction event to relevant subscribers. This method processes a TaskAction and notifies all listening channels.
Parameters
| Name | Type | Description |
|---|
| taskAction | *executorv1.TaskAction | The TaskAction object that triggered the event. |
| eventType | watch.EventType | The type of watch event (e.g., Added, Modified, Deleted). |
handleWatchEvent()
@classmethod
def handleWatchEvent(
ctx: context.Context,
event: watch.Event
)
Handles a single watch event received from the Kubernetes API. This method processes the event and dispatches it to the appropriate worker channel.
Parameters
| Name | Type | Description |
|---|
| ctx | context.Context | The context for handling the event. |
| event | watch.Event | The watch event to handle. |
handleTaskActionEvent()
@classmethod
def handleTaskActionEvent(
ctx: context.Context,
taskAction: *executorv1.TaskAction,
eventType: watch.EventType
)
Processes a TaskAction event, notifying subscribers and updating internal state. This method is called by workers to handle specific TaskAction changes.
Parameters
| Name | Type | Description |
|---|
| ctx | context.Context | The context for handling the TaskAction event. |
| taskAction | *executorv1.TaskAction | The TaskAction object associated with the event. |
| eventType | watch.EventType | The type of watch event (e.g., Added, Modified, Deleted). |
shouldSkipTaskAction()
@classmethod
def shouldSkipTaskAction(
taskAction: *executorv1.TaskAction
) - > bool
Determines if a given TaskAction should be skipped from further processing. This method applies filtering logic to avoid unnecessary work.
Parameters
| Name | Type | Description |
|---|
| taskAction | *executorv1.TaskAction | The TaskAction to evaluate for skipping. |
Returns
| Type | Description |
|---|
bool | True if the TaskAction should be skipped, false otherwise. |
notifySubscribers()
@classmethod
def notifySubscribers(
ctx: context.Context,
update: *ActionUpdate
)
Notifies all subscribed channels about an ActionUpdate. This method broadcasts updates to all listeners for a specific parent action.
Parameters
| Name | Type | Description |
|---|
| ctx | context.Context | The context for the notification. |
| update | *ActionUpdate | The update message to send to subscribers. |
notifyRunService()
@classmethod
def notifyRunService(
ctx: context.Context,
taskAction: *executorv1.TaskAction,
update: *ActionUpdate,
eventType: watch.EventType
)
Notifies the Run Service about a TaskAction update. This method ensures that the Run Service is aware of changes to TaskActions, especially terminal statuses.
Parameters
| Name | Type | Description |
|---|
| ctx | context.Context | The context for the notification. |
| taskAction | *executorv1.TaskAction | The TaskAction that has been updated. |
| update | *ActionUpdate | The update message containing the new state. |
| eventType | watch.EventType | The type of watch event that occurred. |
StopWatching()
@classmethod
def StopWatching()
Stops the watching process for TaskAction resources. This method cleans up resources and terminates all worker goroutines.
markTerminalStatusRecorded()
@classmethod
def markTerminalStatusRecorded(
ctx: context.Context,
taskAction: *executorv1.TaskAction
) - > error
Marks a TaskAction's terminal status as recorded to prevent duplicate notifications. This method updates the TaskAction's metadata to indicate that its final state has been processed.
Parameters
| Name | Type | Description |
|---|
| ctx | context.Context | The context for the request. |
| taskAction | *executorv1.TaskAction | The TaskAction whose terminal status needs to be marked as recorded. |
Returns
| Type | Description |
|---|
error | An error if the status could not be marked as recorded, otherwise nil. |