cmd

package
v2.11.2 Latest Latest
Warning

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

 Go to latest Published: Jul 31, 2026 License: GPL-3.0 Imports: 84 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var (
	// ErrDAGRunIDRequired is returned when a sub dag-run is attempted without providing a dag-run ID
	ErrDAGRunIDRequired = errors.New("dag-run ID must be provided for sub dag-runs")
)

Errors for start command

Functions

func Cleanup 

func Cleanup() *cobra.Command

Cleanup creates and returns a cobra command for removing old DAG run history.

func CmdCoordinator 

func CmdCoordinator() *cobra.Command

func CmdWorker 

func CmdWorker() *cobra.Command

func Config 

func Config() *cobra.Command

func ContextCommand 

func ContextCommand() *cobra.Command

func Dequeue 

func Dequeue() *cobra.Command

func Dry 

func Dry() *cobra.Command

Dry returns the cobra command for dry-run simulation.

func Enqueue 

func Enqueue() *cobra.Command

Enqueue returns the cobra command for queueing a DAG-run.

func Example 

func Example() *cobra.Command

Example creates the 'example' CLI command that displays example DAG definitions.

func ExampleCount 

func ExampleCount() int

ExampleCount returns the number of available examples.

func Exec 

func Exec() *cobra.Command

Exec returns the cobra command for executing inline commands without a DAG spec.

func ExecuteAgent 

func ExecuteAgent(ctx *Context, agentInstance *agent.Agent, dag *core.DAG, dagRunID string, logFile *os.File) error

ExecuteAgent runs an agent with optional progress display and handles common execution logic

func History 

func History() *cobra.Command

func HumanTask 

func HumanTask() *cobra.Command

HumanTask returns the command for managing human tasks.

func License 

func License() *cobra.Command

License returns the parent command for license management.

func NewCommand 

func NewCommand(cmd *cobra.Command, flags []commandLineFlag, runFunc func(cmd *Context, args []string) error) *cobra.Command

NewCommand creates a new command instance with the given cobra command and run function.

func Profile 

func Profile() *cobra.Command

func Restart 

func Restart() *cobra.Command

func Retry 

func Retry() *cobra.Command

func Scheduler 

func Scheduler() *cobra.Command

func Schema 

func Schema() *cobra.Command

Schema creates the 'schema' CLI command that displays JSON schema documentation for DAG definitions or configuration.

func Server 

func Server(serverOpts ...frontend.ServerOption) *cobra.Command

func Start 

func Start() *cobra.Command

Start creates and returns a cobra command for starting a dag-run

func StartAll 

func StartAll() *cobra.Command

func Status 

func Status() *cobra.Command

func Stop 

func Stop() *cobra.Command

func Sync 

func Sync() *cobra.Command

Sync returns the sync command with subcommands for Git sync operations.

func Upgrade 

func Upgrade() *cobra.Command

Upgrade returns the upgrade command.

func Validate 

func Validate() *cobra.Command

Validate creates the 'validate' CLI command that checks a DAG spec for errors.

It follows the same validation logic used by the API's UpdateDAGSpec handler: - Load the YAML without evaluation - Run DAG.Validate()

The command prints validation results and any errors found. Unlike other commands, this does NOT use NewCommand wrapper to allow proper error handling in tests without requiring subprocess patterns.

func Version 

func Version() *cobra.Command

Types

type Context 

type Context struct {
	context.Context

	Command *cobra.Command
	Flags   []commandLineFlag
	Config  *config.Config
	Quiet   bool
	Scope   commandScope

	EventService              *eventstore.Service
	EventSourceInstance       string
	DAGRunStore               exec.DAGRunStore
	DAGRunMgr                 runtime.Manager
	ProcStore                 exec.ProcStore
	QueueStore                exec.QueueStore
	StateStore                dagstate.Store
	ServiceRegistry           exec.ServiceRegistry
	DispatchTaskStore         exec.DispatchTaskStore
	WorkerHeartbeatStore      exec.WorkerHeartbeatStore
	DAGRunLeaseStore          exec.DAGRunLeaseStore
	ActiveDistributedRunStore exec.ActiveDistributedRunStore

	DAGStore       exec.DAGStore
	Proc           exec.ProcHandle
	LicenseManager *license.Manager
	ContextStore   *clicontext.Store
	CLIContext     *clicontext.Context
	ContextName    string
	Remote         *remoteClient
}

Context holds the configuration for a command.

func NewContext 

func NewContext(cmd *cobra.Command, flags []commandLineFlag) (*Context, error)

NewContext creates and initializes an application Context for the given Cobra command. It binds command flags, loads configuration scoped to the command, configures logging (respecting debug, quiet, and log format settings), logs any configuration warnings, and initializes history, DAG run, proc, queue, and service registry stores and managers. Returns an initialized Context or an error if flag retrieval, configuration loading, or other initialization steps fail.

func (*Context) GenArtifactDir 

func (c *Context) GenArtifactDir(dag *core.DAG, dagRunID string) (string, error)

GenArtifactDir generates an artifact directory path for the DAG run when artifacts are enabled.

func (*Context) GenLogFileName 

func (c *Context) GenLogFileName(dag *core.DAG, dagRunID string) (string, error)

GenLogFileName generates a log file name based on the DAG and dag-run ID.

func (*Context) IsRemote 

func (c *Context) IsRemote() bool

func (*Context) LogToFile 

func (c *Context) LogToFile(f *os.File)

LogToFile creates a new logger context with a file writer.

func (*Context) NewCoordinatorClient 

func (c *Context) NewCoordinatorClient() coordinator.Client

NewCoordinatorClient creates a new coordinator client using the global peer configuration. Returns nil when the coordinator is disabled via configuration.

func (*Context) NewScheduler 

func (c *Context) NewScheduler() (*scheduler.Scheduler, error)

NewScheduler creates a scheduler for this command context.

func (*Context) NewServer 

func (c *Context) NewServer(rs *resource.Service, opts ...frontend.ServerOption) (*frontend.Server, error)

NewServer creates and returns a new web UI server for this command context.

func (*Context) OpenLogFile 

func (c *Context) OpenLogFile(
	dag *core.DAG,
	dagRunID string,
) (*os.File, error)

OpenLogFile creates and opens a log file for a given dag-run. It evaluates the log directory, validates settings, creates the log directory, builds a filename using the current timestamp and dag-run ID, and then opens the file.

func (*Context) RecordEarlyFailure 

func (c *Context) RecordEarlyFailure(dag *core.DAG, dagRunID string, err error) error

RecordEarlyFailure records a failure in the execution history before the DAG has fully started. This is used for infrastructure errors like singleton conflicts or process acquisition failures.

func (*Context) StringParam 

func (c *Context) StringParam(name string) (string, error)

StringParam retrieves a string parameter from the command line flags. It checks if the parameter is wrapped in quotes and removes them if necessary.

func (*Context) SubWorkflowRunnerFactory 

func (c *Context) SubWorkflowRunnerFactory() func(context.Context) (runtimeexec.SubWorkflowRunner, error)

func (*Context) WithContext 

func (c *Context) WithContext(ctx context.Context) *Context

WithContext returns a new Context with a different underlying context.Context. This is useful for creating a signal-aware context for service operations.

func (*Context) WithEventSource 

func (c *Context) WithEventSource(service string) *Context

WithEventSource returns a shallow copy whose context carries the given event source. If the event store is not configured, the original context is preserved.

type ExecOptions 

type ExecOptions struct {
	Name          string
	CommandArgs   []string
	ShellOverride string
	WorkingDir    string
	Env           []string
	DotenvFiles   []string
	BaseConfig    string
	WorkerLabels  map[string]string
}

ExecOptions captures the inline configuration for building an ad-hoc DAG.

type LogConfig 

type LogConfig = logpath.Config

LogConfig defines configuration for log file creation.

type RemoteProgressDisplay 

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

RemoteProgressDisplay displays progress for distributed DAG execution. It polls the coordinator for status updates and renders a progress display similar to the local SimpleProgressDisplay.

func NewRemoteProgressDisplay 

func NewRemoteProgressDisplay(dag *core.DAG, dagRunID string) *RemoteProgressDisplay

NewRemoteProgressDisplay creates a new remote progress display.

func (*RemoteProgressDisplay) GetLastStatus 

func (p *RemoteProgressDisplay) GetLastStatus() *exec.DAGRunStatus

GetLastStatus returns the last known status.

func (*RemoteProgressDisplay) PrintSummary 

func (p *RemoteProgressDisplay) PrintSummary()

PrintSummary prints the final tree summary.

func (*RemoteProgressDisplay) SetCancelled 

func (p *RemoteProgressDisplay) SetCancelled()

SetCancelled updates the status to Aborted when cancellation is requested.

func (*RemoteProgressDisplay) Start 

func (p *RemoteProgressDisplay) Start()

Start prints the initial header and starts the animation loop.

func (*RemoteProgressDisplay) Stop 

func (p *RemoteProgressDisplay) Stop()

Stop stops the display and prints final status.

func (*RemoteProgressDisplay) Update 

func (p *RemoteProgressDisplay) Update(status *exec.DAGRunStatus)

Update updates the display with new status from coordinator.

Source Files

View all Source files

Directories

Path Synopsis

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.