agent

package
v2.12.0 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Aug 6, 2026 License: GPL-3.0 Imports: 54 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type Agent

type Agent struct {
	// contains filtered or unexported fields
}

Agent is responsible for running the DAG and handling communication via the unix socket. The agent performs the following tasks: 1. Start the DAG. 2. Propagate a signal to the running processes. 3. Handle the HTTP request via the unix socket. 4. Write the log and status to the data store.

func New

func New(
	dagRunID string,
	dag *core.DAG,
	logDir string,
	logFile string,
	drm runtime.Manager,
	ds exec.DAGStore,
	opts Options,
) *Agent

New creates a new Agent.

func (*Agent) HandleHTTP

func (a *Agent) HandleHTTP(ctx context.Context) sock.HTTPHandlerFunc

HandleHTTP handles HTTP requests via unix socket.

func (*Agent) PrintSummary

func (a *Agent) PrintSummary(ctx context.Context)

func (*Agent) Run

func (a *Agent) Run(ctx context.Context) error

Run setups the runner and runs the DAG.

func (*Agent) Signal

func (a *Agent) Signal(ctx context.Context, sig os.Signal)

Signal requests that running child processes stop.

func (*Agent) Status

func (a *Agent) Status(ctx context.Context) exec.DAGRunStatus

Status collects the current running status of the DAG and returns it.

type ArtifactFinalizer

type ArtifactFinalizer = runtime.ArtifactFinalizer

ArtifactFinalizer uploads or persists artifacts before the final terminal status is written.

type ControllerProgressDisplay added in v2.11.3

type ControllerProgressDisplay struct {
	// contains filtered or unexported fields
}

ControllerProgressDisplay renders a controller run as its decision timeline: one line per decision as it settles, a live line for what the controller is doing now, and a final line with the outcome.

Percent progress does not apply to a controller run: the number of turns is unknown in advance, an action may repeat, and a step the controller never picks was never pending work.

func NewControllerProgressDisplay added in v2.11.3

func NewControllerProgressDisplay(dag *core.DAG) *ControllerProgressDisplay

NewControllerProgressDisplay creates a progress display for a controller DAG.

func (*ControllerProgressDisplay) SetDAGRunInfo added in v2.11.3

func (p *ControllerProgressDisplay) SetDAGRunInfo(dagRunID, params string)

SetDAGRunInfo sets the DAG run ID and parameters.

func (*ControllerProgressDisplay) Start added in v2.11.3

func (p *ControllerProgressDisplay) Start()

Start begins the progress display.

func (*ControllerProgressDisplay) Stop added in v2.11.3

func (p *ControllerProgressDisplay) Stop()

Stop stops the progress display. Safe to call multiple times.

func (*ControllerProgressDisplay) UpdateNode added in v2.11.3

func (p *ControllerProgressDisplay) UpdateNode(node *exec.Node)

UpdateNode consumes node updates. The controller node carries the decision timeline; every other node marks which action is in flight.

func (*ControllerProgressDisplay) UpdateStatus added in v2.11.3

func (p *ControllerProgressDisplay) UpdateStatus(status *exec.DAGRunStatus)

UpdateStatus updates the overall DAG status.

type Options

type Options struct {
	// Dry is a dry-run mode. It does not execute the actual command.
	// Dry run does not create runstore data.
	Dry bool
	// RetryTarget is the target status (runstore of execution) to retry.
	// If it's specified the agent will execute the DAG with the same
	// configuration as the specified history.
	RetryTarget *exec.DAGRunStatus
	// ParentDAGRun is the dag-run reference of the parent dag-run.
	// It is required for sub dag-runs to identify the parent dag-run.
	ParentDAGRun exec.DAGRunRef
	// ProgressDisplay indicates if the progress display should be shown.
	// This is typically enabled for CLI execution in a TTY environment.
	ProgressDisplay bool
	// ExtraEnvs are additional execution-scoped env vars injected into the DAG run context.
	ExtraEnvs []string
	// StepRetry is the name of the step to retry, if specified.
	StepRetry string
	// RetryPath identifies a persisted child DAG step retry.
	RetryPath exec.RetryPath
	// WorkerID is the identifier of the worker executing this DAG run.
	// For distributed execution, this is set to the worker's ID.
	// For local execution, this defaults to "local".
	WorkerID string
	// StatusPusher is used to push status updates to a remote coordinator.
	// When nil, status is written to local filesystem via the run-state attempt.
	StatusPusher StatusPusher
	// SubWorkflowRunnerFactory creates a runner for child workflows.
	SubWorkflowRunnerFactory SubWorkflowRunnerFactory
	// LogWriterFactory is used to create log writers for step output.
	// When nil, logs are written to local filesystem.
	LogWriterFactory exec.LogWriterFactory
	// QueuedRun indicates this execution is from a queued item.
	// When true, the agent will find the existing dag-run (created by enqueue)
	// instead of creating a new one. This is used for distributed execution
	// where the dag-run directory was already created by the scheduler.
	QueuedRun bool
	// AttemptID is the attempt ID from the coordinator.
	// When set, the agent creates an attempt with this ID instead of generating a new one.
	AttemptID string
	// PreparedAttempt is an exact attempt that was created or reopened before proc acquisition.
	// This is used for local execution so the proc heartbeat can include the final attempt ID.
	PreparedAttempt exec.DAGRunAttempt
	// RunStateStore records execution state for this DAG run.
	RunStateStore runstate.Store
	// DAGRunStore is the store for dag-run data. Nil for remote worker execution.
	DAGRunStore exec.DAGRunStore
	// QueueStore is the store for queued dag-run items. Nil when queues are unavailable.
	QueueStore exec.QueueStore
	// StateStore is the persistent state store shared across DAG runs.
	StateStore dagstate.Store
	// SecretStore resolves local registry refs and runtime profile secrets.
	SecretStore secretpkg.Store
	// SecretReferenceResolver resolves DAG-level registry refs.
	// When nil, SecretStore supplies the local resolver.
	SecretReferenceResolver secrets.ReferenceResolver
	// ProfileStore resolves named runtime profiles.
	ProfileStore profilepkg.Store
	// ProfileName selects the runtime profile for this DAG run.
	ProfileName string
	// ServiceRegistry is the registry for service discovery.
	ServiceRegistry exec.ServiceRegistry
	// RootDAGRun is the root dag-run reference for sub-DAG runs.
	RootDAGRun exec.DAGRunRef
	// PeerConfig is the configuration for peer communication.
	PeerConfig config.Peer
	// TriggerType indicates how this DAG run was initiated.
	TriggerType core.TriggerType
	// TriggerActor identifies the attributable actor that initiated the DAG run.
	TriggerActor string
	// DefaultExecMode is the server-level default execution mode.
	DefaultExecMode config.ExecutionMode
	// ScheduleTime is the RFC 3339 timestamp of when this run was scheduled.
	// Set by the scheduler for cron-triggered runs; empty for manual runs.
	ScheduleTime string
	// ArtifactDir is the per-run artifact directory when artifact storage is enabled.
	ArtifactDir string
	// DAGRunLogDir is the base log directory used for child DAG runs created by executors.
	DAGRunLogDir string
	// DAGRunArtifactDir is the base artifact directory used for child DAG runs created by executors.
	DAGRunArtifactDir string
	// ArtifactFinalizer persists artifacts before the final terminal status is written.
	ArtifactFinalizer ArtifactFinalizer
	// RemoteDAGLoader loads a DAG from a remote source when the local DAG store misses.
	// When nil, no remote fallback is attempted.
	RemoteDAGLoader RemoteDAGLoader
	// SocketServerFactory creates the local status/control transport.
	// When nil, the default Unix socket transport is used.
	SocketServerFactory SocketServerFactory
}

Options is the configuration for the Agent.

type ProgressReporter

type ProgressReporter interface {
	// Start begins the progress display
	Start()

	// Stop stops the progress display
	Stop()

	// UpdateNode updates the progress for a specific node
	UpdateNode(node *exec.Node)

	// UpdateStatus updates the overall DAG status
	UpdateStatus(status *exec.DAGRunStatus)

	// SetDAGRunInfo sets the DAG run ID and parameters
	SetDAGRunInfo(dagRunID, params string)
}

ProgressReporter is the interface for progress display implementations

type RemoteDAGLoader

type RemoteDAGLoader func(ctx context.Context, name string) (*core.DAG, error)

RemoteDAGLoader loads a DAG definition from a remote source. Returns nil, nil when the remote source does not have the DAG.

type Sender

type Sender interface {
	Send(ctx context.Context, from string, to []string, subject, body string, attachments []string) error
}

Sender is a mailer interface.

type SenderFn

type SenderFn func(ctx context.Context, from string, to []string, subject, body string, attachments []string) error

SenderFn is a function type for sending reports.

type SimpleProgressDisplay

type SimpleProgressDisplay struct {
	// contains filtered or unexported fields
}

SimpleProgressDisplay provides a minimal inline progress display.

func NewSimpleProgressDisplay

func NewSimpleProgressDisplay(dag *core.DAG) *SimpleProgressDisplay

NewSimpleProgressDisplay creates a new simple progress display.

func (*SimpleProgressDisplay) SetDAGRunInfo

func (p *SimpleProgressDisplay) SetDAGRunInfo(dagRunID, params string)

SetDAGRunInfo sets the DAG run ID and parameters.

func (*SimpleProgressDisplay) Start

func (p *SimpleProgressDisplay) Start()

Start begins the progress display.

func (*SimpleProgressDisplay) Stop

func (p *SimpleProgressDisplay) Stop()

Stop stops the progress display. Safe to call multiple times.

func (*SimpleProgressDisplay) UpdateNode

func (p *SimpleProgressDisplay) UpdateNode(node *exec.Node)

UpdateNode updates the progress for a specific node.

func (*SimpleProgressDisplay) UpdateStatus

func (p *SimpleProgressDisplay) UpdateStatus(status *exec.DAGRunStatus)

UpdateStatus updates the overall DAG status.

type SocketServer

type SocketServer interface {
	Serve(ctx context.Context, listen chan error) error
	Shutdown(ctx context.Context) error
}

SocketServer handles local status/control requests for a running DAG.

type SocketServerFactory

type SocketServerFactory func(addr string, handlerFunc sock.HTTPHandlerFunc) (SocketServer, error)

SocketServerFactory creates a local status/control transport.

type StatusPusher

type StatusPusher = runtime.StatusPusher

StatusPusher reports DAG run status outside the current execution process.

type SubWorkflowRunnerFactory

type SubWorkflowRunnerFactory func(ctx context.Context) (runtimeexec.SubWorkflowRunner, error)

SubWorkflowRunnerFactory creates a runner for child workflows.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL