Interacting with the Actions Service

The Actions Service in Flyte provides a gRPC and Connect-based interface for managing the lifecycle of TaskAction resources. It acts as the primary gateway for external clients—such as the Flyte UI or SDKs—to enqueue new tasks, monitor their progress through streaming updates, and abort executions when necessary.

Internally, the service is split into two layers: the Actions Service (actions.service.ActionsService), which handles the gRPC/Connect protocol and streaming logic, and the Actions Client (actions.k8s.ActionsClient), which manages the underlying Kubernetes Custom Resources (CRs) and event distribution.

Enqueuing Actions

When you trigger a task execution, the Actions Service creates a TaskAction Custom Resource in Kubernetes. This CR contains the task's specification, including its type, input URIs, and environment variables.

To enqueue an action, you call the Enqueue method on the ActionsService. Internally, this delegates to the ActionsClient.Enqueue method in actions/k8s/client.go:

func (c *ActionsClient) Enqueue(ctx context.Context, action *actions.Action, runSpec *task.RunSpec) error {
    // ... logic to determine taskActionName and namespace ...
    taskAction := &executorv1.TaskAction{
        ObjectMeta: metav1.ObjectMeta{
            Name:      taskActionName,
            Namespace: flyteNamespace,
            Labels: map[string]string{
                "flyte.org/project": actionID.Run.Project,
                "flyte.org/domain":  actionID.Run.Domain,
                "flyte.org/run":     actionID.Run.Name,
                "flyte.org/action":  actionID.Name,
            },
        },
        Spec: executorv1.TaskActionSpec{
            // ... populated from action and runSpec ...
        },
    }
    
    // If this is a child action, it inherits context and sets OwnerReferences
    if parentTaskAction != nil {
        inheritRunContextFromParentTaskAction(taskAction, parentTaskAction)
        taskAction.OwnerReferences = []metav1.OwnerReference{
            *metav1.NewControllerRef(parentTaskAction, executorv1.SchemeGroupVersion.WithKind("TaskAction")),
        }
    }

    return c.k8sClient.Create(ctx, taskAction)
}

Flyte uses Kubernetes OwnerReferences to maintain the hierarchy between parent and child actions. This ensures that if a parent action is deleted, Kubernetes automatically cleans up all descendant actions.

Monitoring Progress with Streaming Updates

The Actions Service provides a streaming API, WatchForUpdates, which allows clients to receive real-time state changes for a specific parent action and all its children.

The Subscription Pattern

To ensure no updates are missed during the transition from the initial state to the live stream, ActionsService.WatchForUpdates follows a "Subscribe-then-List" pattern:

1. Subscribe: It first calls s.client.Subscribe(parentActionID.Name) to start buffering events in a channel.
2. Snapshot: It then calls s.client.ListChildActions(ctx, parentActionID) to get the current state of all existing actions.
3. Stream: It sends the snapshot to the client, followed by a sentinel message, and then begins draining the subscription channel.

// From actions/service/actions_service.go
updateCh := s.client.Subscribe(parentActionID.Name)
defer s.client.Unsubscribe(parentActionID.Name, updateCh)

// Send initial state snapshot
childActions, err := s.client.ListChildActions(ctx, parentActionID)
// ... send childActions to stream ...

// Send sentinel to signal end of initial snapshot
// ... send sentinel ...

for {
    select {
    case <-ctx.Done():
        return nil
    case update, ok := <-updateCh:
        // ... send live update to stream ...
    }
}

Event Sharding and Ordering

The ActionsClient maintains a worker pool to process Kubernetes watch events. To ensure that events for the same TaskAction are processed in the correct order, Flyte shards events across workers using an FNV-32a hash of the resource name.

In actions/k8s/client.go, the dispatchEvent method routes events to specific worker channels:

func (c *ActionsClient) dispatchEvent(taskAction *executorv1.TaskAction, eventType watch.EventType) {
    h := fnv.New32a()
    _, _ = h.Write([]byte(taskAction.Name))
    shard := h.Sum32() % uint32(c.numWorkers)
    c.workerChs[shard] <- watch.Event{Type: eventType, Object: taskAction.DeepCopy()}
}

Each worker goroutine drains its assigned channel and notifies subscribers, ensuring that a single resource's updates are never processed concurrently by different workers.

State Synchronization and Deduplication

The Actions Service is responsible for synchronizing the state of Kubernetes TaskAction resources back to Flyte's internal RunService. This ensures the database reflects the current status of the execution.

Bloom Filter Deduplication

When a TaskAction is first created (an ADDED event), the ActionsClient calls RecordAction on the InternalRunServiceClient. To prevent duplicate database records during watch reconnects or service restarts, Flyte uses a Bloom filter (recordedFilter) to track which actions have already been recorded.

// From actions/k8s/client.go notifyRunService
if eventType == watch.Added {
    actionKey := []byte(buildTaskActionName(update.ActionID))
    isDuplicate := c.recordedFilter != nil && c.recordedFilter.Contains(ctx, actionKey)
    if !isDuplicate {
        // ... call c.runClient.RecordAction ...
        if c.recordedFilter != nil {
            c.recordedFilter.Add(ctx, actionKey)
        }
    }
}

Terminal Status Recording

Once an action reaches a terminal phase (Succeeded, Failed, or Aborted), the service updates the RunService and then patches the Kubernetes CR with a label flyte.org/terminal-status-recorded: "true". The ActionsClient uses this label to skip processing for resources that have already been finalized in the database, reducing unnecessary overhead.

Aborting Actions

If you need to stop a running task, the Abort gRPC method triggers a cancellation flow. The ActionsClient.AbortAction method updates the TaskAction CR to a terminal state. Because of the OwnerReferences established during enqueuing, Kubernetes handles the cascading deletion of any sub-tasks or pods associated with that action.

func (c *ActionsClient) AbortAction(ctx context.Context, actionID *common.ActionIdentifier, reason *string) error {
    taskAction, err := c.GetTaskAction(ctx, actionID)
    if err != nil {
        return err
    }
    // ... logic to update status to ABORTED or delete the CR ...
    return c.k8sClient.Delete(ctx, taskAction)
}

When the CR is deleted, the ActionsClient receives a Deleted watch event, which it then propagates as an ACTION_PHASE_ABORTED update to all active subscribers.