execution

package
v1.30.3 Latest Latest
Warning

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

 Go to latest Published: Jan 4, 2026 License: GPL-3.0 Imports: 14 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	// EnvKeyDAGName holds the name of the currently executing DAG.
	EnvKeyDAGName = "DAG_NAME"

	// EnvKeyDAGRunID holds the unique identifier for the current DAG run.
	EnvKeyDAGRunID = "DAG_RUN_ID"

	// EnvKeyDAGRunLogFile holds the path to the main log file for the DAG run.
	EnvKeyDAGRunLogFile = "DAG_RUN_LOG_FILE"

	// EnvKeyDAGRunStepName holds the name of the currently executing step.
	EnvKeyDAGRunStepName = "DAG_RUN_STEP_NAME"

	// EnvKeyDAGRunStepStdoutFile holds the path to the stdout log file for the current step.
	EnvKeyDAGRunStepStdoutFile = "DAG_RUN_STEP_STDOUT_FILE"

	// EnvKeyDAGRunStepStderrFile holds the path to the stderr log file for the current step.
	EnvKeyDAGRunStepStderrFile = "DAG_RUN_STEP_STDERR_FILE"

	// EnvKeyDAGRunStatus holds the current status of the DAG run (e.g., "running", "success", "failed").
	EnvKeyDAGRunStatus = "DAG_RUN_STATUS"

	// EnvKeyDAGParamsJSON exposes the resolved parameters encoded as JSON.
	// When params were provided as JSON, the original payload is preserved.
	EnvKeyDAGParamsJSON = "DAGU_PARAMS_JSON"
)

Environment variable keys that are automatically set by Dagu during execution.

View Source 
const (
	RoleSystem    = core.LLMRoleSystem
	RoleUser      = core.LLMRoleUser
	RoleAssistant = core.LLMRoleAssistant
	RoleTool      = core.LLMRoleTool
)

LLM message role constants - aliases for core package constants.

Variables

View Source 
var (
	ErrDAGAlreadyExists = errors.New("DAG already exists")
	ErrDAGNotFound      = errors.New("DAG is not found")
)

Errors for DAG file operations

View Source 
var (
	ErrDAGRunIDNotFound    = errors.New("dag-run ID not found")
	ErrNoStatusData        = errors.New("no status data")
	ErrCorruptedStatusFile = errors.New("corrupted status file") // Status file exists but contains no valid data or is corrupted
)

Errors related to dag-run management

View Source 
var (
	ErrQueueEmpty        = errors.New("queue is empty")
	ErrQueueItemNotFound = errors.New("queue item not found")
)

Errors for the queue

View Source 
var (
	ErrInvalidRunRefFormat = errors.New("invalid dag-run reference format")
)

Errors for RunRef parsing

Functions

func FormatTime 

func FormatTime(val time.Time) string

FormatTime formats a time.Time or returns empty string if it's the zero value

func NewContext added in v1.26.0 

func NewContext(
	ctx context.Context,
	dag *core.DAG,
	dagRunID string,
	logFile string,
	opts ...ContextOption,
) context.Context

NewContext creates a new context with DAG execution metadata. Required: ctx, dag, dagRunID, logFile Optional: use ContextOption functions (WithDatabase, WithParams, etc.)

func WithContext added in v1.26.0 

func WithContext(ctx context.Context, rCtx Context) context.Context

WithContext returns a new context with the given DAGContext. This is useful for tests that need to set up a DAGContext directly.

Types

type Context added in v1.26.0 

type Context struct {
	DAGRunID           string
	RootDAGRun         DAGRunRef
	DAG                *core.DAG
	DB                 Database
	BaseEnv            *config.BaseEnv
	Envs               map[string]string
	SecretEnvs         map[string]string // Secret environment variables (highest priority)
	CoordinatorCli     Dispatcher
	Shell              string // Default shell for this DAG (from DAG.Shell)
	LogEncodingCharset string // Character encoding for log files (e.g., "utf-8", "shift_jis", "euc-jp")
}

Context contains the execution metadata for a dag-run.

func GetContext added in v1.26.0 

func GetContext(ctx context.Context) Context

GetContext retrieves the DAGContext from the context.

func (Context) AllEnvs added in v1.26.0 

func (e Context) AllEnvs() []string

AllEnvs returns every environment variable as "key=value" with precedence: BaseEnv < DAG.Env < e.Envs < SecretEnvs < runtime metadata (e.g., DAGU_PARAMS_JSON).

func (Context) DAGRunRef added in v1.26.0 

func (e Context) DAGRunRef() DAGRunRef

DAGRunRef returns the DAGRunRef for the current DAG context.

func (Context) UserEnvsMap added in v1.26.0 

func (e Context) UserEnvsMap() map[string]string

UserEnvsMap returns only user-defined environment variables as a map, excluding OS environment (BaseEnv). Use this for isolated execution environments. Precedence: SecretEnvs > Envs > DAG.Env

type ContextOption added in v1.26.0 

type ContextOption func(*contextOptions)

ContextOption configures optional parameters for NewContext.

func WithCoordinator added in v1.26.0 

func WithCoordinator(cli Dispatcher) ContextOption

WithCoordinator sets the coordinator dispatcher for distributed execution.

func WithDatabase added in v1.26.0 

func WithDatabase(db Database) ContextOption

WithDatabase sets the database interface.

func WithLogEncoding added in v1.26.0 

func WithLogEncoding(charset string) ContextOption

WithLogEncoding sets the log file character encoding.

func WithParams added in v1.26.0 

func WithParams(params []string) ContextOption

WithParams sets runtime parameters.

func WithRootDAGRun added in v1.26.0 

func WithRootDAGRun(ref DAGRunRef) ContextOption

WithRootDAGRun sets the root DAG run reference for sub-DAG execution.

func WithSecrets added in v1.26.0 

func WithSecrets(secrets []string) ContextOption

WithSecrets sets secret environment variables.

type DAGRunAttempt 

type DAGRunAttempt interface {
	// ID returns the identifier for the attempt that is unique within the dag-run.
	ID() string
	// Open prepares the attempt for writing status updates
	Open(ctx context.Context) error
	// Write updates the status of the attempt
	Write(ctx context.Context, status DAGRunStatus) error
	// Close finalizes writing to the attempt
	Close(ctx context.Context) error
	// ReadStatus retrieves the current status of the attempt
	ReadStatus(ctx context.Context) (*DAGRunStatus, error)
	// ReadDAG reads the DAG associated with this run attempt
	ReadDAG(ctx context.Context) (*core.DAG, error)
	// Abort requests aborting the attempt
	Abort(ctx context.Context) error
	// IsAborting checks if an abort has been requested for the attempt
	IsAborting(ctx context.Context) (bool, error)
	// Hide marks the attempt as hidden from normal operations.
	// This is useful for preserving previous state visibility when dequeuing.
	Hide(ctx context.Context) error
	// Hidden returns true if the attempt is hidden from normal operations.
	Hidden() bool
	// WriteOutputs writes the collected step outputs for the dag-run.
	// Does nothing if outputs is nil or has no output entries.
	WriteOutputs(ctx context.Context, outputs *DAGRunOutputs) error
	// ReadOutputs reads the collected step outputs for the dag-run.
	// Returns nil if no outputs file exists or if the file is in v1 format.
	ReadOutputs(ctx context.Context) (*DAGRunOutputs, error)
	// WriteStepMessages writes LLM messages for a single step.
	WriteStepMessages(ctx context.Context, stepName string, messages []LLMMessage) error
	// ReadStepMessages reads LLM messages for a single step.
	// Returns nil if no messages exist for the step.
	ReadStepMessages(ctx context.Context, stepName string) ([]LLMMessage, error)
}

DAGRunAttempt represents a single execution of a dag-run to record the status and execution details.

type DAGRunOutputs added in v1.29.0 

type DAGRunOutputs struct {
	Metadata OutputsMetadata   `json:"metadata"`
	Outputs  map[string]string `json:"outputs"`
}

DAGRunOutputs represents the full outputs file structure with metadata.

type DAGRunRef 

type DAGRunRef struct {
	Name string `json:"name,omitempty"`
	ID   string `json:"id,omitempty"`
}

DAGRunRef represents a reference to a dag-run

func NewDAGRunRef 

func NewDAGRunRef(name, runID string) DAGRunRef

NewDAGRunRef creates a new reference to dag-run with the given DAG name and run ID. It is used to identify a specific dag-run.

func ParseDAGRunRef 

func ParseDAGRunRef(s string) (DAGRunRef, error)

ParseDAGRunRef parses a string into a DAGRunRef. The expected format is "name:runId". If the format is invalid, it returns an error.

func (DAGRunRef) String 

func (e DAGRunRef) String() string

String returns a string representation of the dag-run reference.

func (DAGRunRef) Zero 

func (e DAGRunRef) Zero() bool

Zero checks if the DAGRunRef is a zero value.

type DAGRunStatus 

type DAGRunStatus struct {
	Root          DAGRunRef         `json:"root,omitzero"`
	Parent        DAGRunRef         `json:"parent,omitzero"`
	Name          string            `json:"name"`
	DAGRunID      string            `json:"dagRunId"`
	AttemptID     string            `json:"attemptId"`
	Status        core.Status       `json:"status"`
	WorkerID      string            `json:"workerId,omitempty"`
	PID           PID               `json:"pid,omitempty"`
	Nodes         []*Node           `json:"nodes,omitempty"`
	OnInit        *Node             `json:"onInit,omitempty"`
	OnExit        *Node             `json:"onExit,omitempty"`
	OnSuccess     *Node             `json:"onSuccess,omitempty"`
	OnFailure     *Node             `json:"onFailure,omitempty"`
	OnCancel      *Node             `json:"onCancel,omitempty"`
	OnWait        *Node             `json:"onWait,omitempty"`
	CreatedAt     int64             `json:"createdAt,omitempty"`
	QueuedAt      string            `json:"queuedAt,omitempty"`
	StartedAt     string            `json:"startedAt,omitempty"`
	FinishedAt    string            `json:"finishedAt,omitempty"`
	Log           string            `json:"log,omitempty"`
	Error         string            `json:"error,omitempty"`
	Params        string            `json:"params,omitempty"`
	ParamsList    []string          `json:"paramsList,omitempty"`
	Preconditions []*core.Condition `json:"preconditions,omitempty"`
}

DAGRunStatus represents the complete execution state of a dag-run.

func InitialStatus 

func InitialStatus(dag *core.DAG) DAGRunStatus

InitialStatus creates an initial Status object for the given DAG

func StatusFromJSON 

func StatusFromJSON(s string) (*DAGRunStatus, error)

StatusFromJSON deserializes a JSON string into a Status object

func (*DAGRunStatus) DAGRun 

func (st *DAGRunStatus) DAGRun() DAGRunRef

DAGRun returns a reference to the dag-run associated with this status

func (*DAGRunStatus) Errors 

func (st *DAGRunStatus) Errors() []error

Errors returns a slice of errors for the current status

func (*DAGRunStatus) NodeByName 

func (st *DAGRunStatus) NodeByName(name string) (*Node, error)

NodesByName returns a slice of nodes with the specified name

type DAGRunStore 

type DAGRunStore interface {
	// CreateAttempt creates a new execution record for a dag-run.
	CreateAttempt(ctx context.Context, dag *core.DAG, ts time.Time, dagRunID string, opts NewDAGRunAttemptOptions) (DAGRunAttempt, error)
	// RecentAttempts returns the most recent dag-run's attempt for the DAG name, limited by itemLimit
	RecentAttempts(ctx context.Context, name string, itemLimit int) []DAGRunAttempt
	// LatestAttempt returns the most recent dag-run's attempt for the DAG name.
	LatestAttempt(ctx context.Context, name string) (DAGRunAttempt, error)
	// ListStatuses returns a list of statuses.
	ListStatuses(ctx context.Context, opts ...ListDAGRunStatusesOption) ([]*DAGRunStatus, error)
	// FindAttempt finds the latest attempt for the dag-run.
	FindAttempt(ctx context.Context, dagRun DAGRunRef) (DAGRunAttempt, error)
	// FindSubAttempt finds a sub dag-run record by dag-run ID.
	FindSubAttempt(ctx context.Context, dagRun DAGRunRef, subDAGRunID string) (DAGRunAttempt, error)
	// RemoveOldDAGRuns deletes dag-run records older than retentionDays.
	// If retentionDays is negative, it won't delete any records.
	// If retentionDays is zero, it will delete all records for the DAG name.
	// But it will not delete the records with non-final statuses (e.g., running, queued).
	// Returns a list of dag-run IDs that were removed (or would be removed in dry-run mode).
	RemoveOldDAGRuns(ctx context.Context, name string, retentionDays int, opts ...RemoveOldDAGRunsOption) ([]string, error)
	// RenameDAGRuns renames all run data from oldName to newName
	// The name means the DAG name, renaming it will allow user to manage those runs
	// with the new DAG name.
	RenameDAGRuns(ctx context.Context, oldName, newName string) error
	// RemoveDAGRun removes a dag-run record by its reference.
	RemoveDAGRun(ctx context.Context, dagRun DAGRunRef) error
}

DAGRunStore provides an interface for interacting with the underlying database for storing and retrieving dag-run data. It abstracts the details of the storage mechanism, allowing for different implementations (e.g., file-based, in-memory, etc.) to be used interchangeably.

type DAGStore 

type DAGStore interface {
	// Create stores a new DAG definition with the given name and returns its file name
	Create(ctx context.Context, fileName string, spec []byte) error
	// Delete removes a DAG definition by name
	Delete(ctx context.Context, fileName string) error
	// List returns a paginated list of DAG definitions with filtering options
	List(ctx context.Context, params ListDAGsOptions) (PaginatedResult[*core.DAG], []string, error)
	// GetMetadata retrieves only the metadata of a DAG definition (faster than full load)
	GetMetadata(ctx context.Context, fileName string) (*core.DAG, error)
	// GetDetails retrieves the complete DAG definition including all fields
	GetDetails(ctx context.Context, fileName string, opts ...spec.LoadOption) (*core.DAG, error)
	// Grep searches for a pattern in all DAG definitions and returns matching results
	Grep(ctx context.Context, pattern string) (ret []*GrepDAGsResult, errs []string, err error)
	// Rename changes a DAG's identifier from oldID to newID
	Rename(ctx context.Context, oldID, newID string) error
	// GetSpec retrieves the raw YAML specification of a DAG
	GetSpec(ctx context.Context, fileName string) (string, error)
	// UpdateSpec modifies the specification of an existing DAG
	UpdateSpec(ctx context.Context, fileName string, spec []byte) error
	// LoadSpec loads a DAG from a YAML file and returns the DAG object
	LoadSpec(ctx context.Context, spec []byte, opts ...spec.LoadOption) (*core.DAG, error)
	// TagList returns all unique tags across all DAGs with any errors encountered
	TagList(ctx context.Context) ([]string, []string, error)
	// ToggleSuspend changes the suspension state of a DAG by ID
	ToggleSuspend(ctx context.Context, fileName string, suspend bool) error
	// IsSuspended checks if a DAG is currently suspended
	IsSuspended(ctx context.Context, fileName string) bool
}

DAGStore is an interface for interacting with underlying DAG storage systems. It allows for different implementations (e.g., local file system, database) to be used interchangeably.

type Database 

type Database interface {
	// GetDAG retrieves a DAG by its name.
	GetDAG(ctx context.Context, name string) (*core.DAG, error)
	// GetSubDAGRunStatus retrieves the status of a sub dag-run by its ID and the root dag-run reference.
	GetSubDAGRunStatus(ctx context.Context, dagRunID string, rootDAGRun DAGRunRef) (*RunStatus, error)
	// IsSubDAGRunCompleted checks if a sub dag-run has completed.
	IsSubDAGRunCompleted(ctx context.Context, dagRunID string, rootDAGRun DAGRunRef) (bool, error)
	// RequestChildCancel requests cancellation of a sub dag-run.
	RequestChildCancel(ctx context.Context, dagRunID string, rootDAGRun DAGRunRef) error
}

Database is the interface for accessing the database to retrieve DAGs and dag-run statuses. This interface abstracts the underlying storage mechanism, allowing for different implementations (e.g., SQL, NoSQL, in-memory).

type Dispatcher 

type Dispatcher interface {
	// Dispatch sends a task to the coordinator
	Dispatch(ctx context.Context, task *coordinatorv1.Task) error

	// Cleanup cleans up any resources used by the coordinator client
	Cleanup(ctx context.Context) error
}

Dispatcher defines the interface for coordinator operations

type GrepDAGsResult 

type GrepDAGsResult struct {
	Name    string    // Name of the DAG
	DAG     *core.DAG // The DAG object
	Matches []*Match  // Matching lines and their context
}

GrepDAGsResult represents the result of a pattern search within a DAG definition

type HostInfo 

type HostInfo struct {
	// ID is a unique identifier for the host
	ID string
	// Host is the hostname or IP address
	Host string
	// Port is the port number (0 if not applicable)
	Port int
	// Status is the operational status of the service instance
	Status ServiceStatus
	// StartedAt is when the service instance was started
	StartedAt time.Time
}

HostInfo contains information about a host in the service registry system

type LLMMessage added in v1.30.0 

type LLMMessage struct {
	// Role is the message role (system, user, assistant).
	Role core.LLMRole `json:"role"`
	// Content is the message content.
	Content string `json:"content"`
	// Metadata contains API call metadata (only set for assistant responses).
	Metadata *LLMMessageMetadata `json:"metadata,omitempty"`
}

LLMMessage represents a single message in the conversation.

func DeduplicateSystemMessages added in v1.30.0 

func DeduplicateSystemMessages(messages []LLMMessage) []LLMMessage

DeduplicateSystemMessages keeps only the first system message.

type LLMMessageMetadata added in v1.30.0 

type LLMMessageMetadata struct {
	// Provider is the LLM provider used (openai, anthropic, etc.).
	Provider string `json:"provider,omitempty"`
	// Model is the model identifier used.
	Model string `json:"model,omitempty"`
	// PromptTokens is the number of tokens in the prompt.
	PromptTokens int `json:"promptTokens,omitempty"`
	// CompletionTokens is the number of tokens in the completion.
	CompletionTokens int `json:"completionTokens,omitempty"`
	// TotalTokens is the sum of prompt and completion tokens.
	TotalTokens int `json:"totalTokens,omitempty"`
}

LLMMessageMetadata contains metadata about an LLM API call.

type ListDAGRunStatusesOption 

type ListDAGRunStatusesOption func(*ListDAGRunStatusesOptions)

ListRunsOption is a functional option for configuring ListRunsOptions

func WithDAGRunID 

func WithDAGRunID(dagRunID string) ListDAGRunStatusesOption

WithDAGRunID sets the dag-run ID for listing dag-runs

func WithExactName 

func WithExactName(name string) ListDAGRunStatusesOption

WithExactName sets the name for listing dag-runs

func WithFrom 

func WithFrom(from TimeInUTC) ListDAGRunStatusesOption

WithFrom sets the start time for listing dag-runs

func WithName 

func WithName(name string) ListDAGRunStatusesOption

WithName sets the name for listing dag-runs

func WithStatuses 

func WithStatuses(statuses []core.Status) ListDAGRunStatusesOption

WithStatuses sets the statuses for listing dag-runs

func WithTo 

func WithTo(to TimeInUTC) ListDAGRunStatusesOption

WithTo sets the end time for listing dag-runs

type ListDAGRunStatusesOptions 

type ListDAGRunStatusesOptions struct {
	DAGRunID  string
	Name      string
	ExactName string
	From      TimeInUTC
	To        TimeInUTC
	Statuses  []core.Status
	Limit     int
}

ListDAGRunStatusesOptions contains options for listing runs

type ListDAGsOptions 

type ListDAGsOptions struct {
	Paginator *Paginator
	Name      string     // Optional name filter
	Tag       string     // Optional tag filter
	Sort      string     // Optional sort field (name, updated_at, created_at)
	Order     string     // Optional sort order (asc, desc)
	Time      *time.Time // Optional time for next run calculations (defaults to time.Now())
}

ListDAGsOptions contains parameters for paginated DAG listing

type ListDAGsResult 

type ListDAGsResult struct {
	DAGs   []*core.DAG // The list of DAGs for the current page
	Count  int         // Total count of DAGs matching the filter
	Errors []string    // Any errors encountered during listing
}

ListDAGsResult contains the result of a paginated DAG listing operation

type Match 

type Match struct {
	Line       string
	LineNumber int
	StartLine  int
}

Match contains matched line number and line content.

type MockDAGRunAttempt added in v1.30.0 

type MockDAGRunAttempt struct {
	mock.Mock
	// Status can be set for tests that need to return a specific status without mock setup
	Status *DAGRunStatus
}

MockDAGRunAttempt is a mock implementation of DAGRunAttempt for testing.

func (*MockDAGRunAttempt) Abort added in v1.30.0 

func (m *MockDAGRunAttempt) Abort(ctx context.Context) error

func (*MockDAGRunAttempt) Close added in v1.30.0 

func (m *MockDAGRunAttempt) Close(ctx context.Context) error

func (*MockDAGRunAttempt) Hidden added in v1.30.0 

func (m *MockDAGRunAttempt) Hidden() bool

func (*MockDAGRunAttempt) Hide added in v1.30.0 

func (m *MockDAGRunAttempt) Hide(ctx context.Context) error

func (*MockDAGRunAttempt) ID added in v1.30.0 

func (m *MockDAGRunAttempt) ID() string

func (*MockDAGRunAttempt) IsAborting added in v1.30.0 

func (m *MockDAGRunAttempt) IsAborting(ctx context.Context) (bool, error)

func (*MockDAGRunAttempt) Open added in v1.30.0 

func (m *MockDAGRunAttempt) Open(ctx context.Context) error

func (*MockDAGRunAttempt) ReadDAG added in v1.30.0 

func (m *MockDAGRunAttempt) ReadDAG(ctx context.Context) (*core.DAG, error)

func (*MockDAGRunAttempt) ReadOutputs added in v1.30.0 

func (m *MockDAGRunAttempt) ReadOutputs(ctx context.Context) (*DAGRunOutputs, error)

func (*MockDAGRunAttempt) ReadStatus added in v1.30.0 

func (m *MockDAGRunAttempt) ReadStatus(ctx context.Context) (*DAGRunStatus, error)

func (*MockDAGRunAttempt) ReadStepMessages added in v1.30.0 

func (m *MockDAGRunAttempt) ReadStepMessages(ctx context.Context, stepName string) ([]LLMMessage, error)

func (*MockDAGRunAttempt) Write added in v1.30.0 

func (m *MockDAGRunAttempt) Write(ctx context.Context, status DAGRunStatus) error

func (*MockDAGRunAttempt) WriteOutputs added in v1.30.0 

func (m *MockDAGRunAttempt) WriteOutputs(ctx context.Context, outputs *DAGRunOutputs) error

func (*MockDAGRunAttempt) WriteStepMessages added in v1.30.0 

func (m *MockDAGRunAttempt) WriteStepMessages(ctx context.Context, stepName string, messages []LLMMessage) error

type NewDAGRunAttemptOptions 

type NewDAGRunAttemptOptions struct {
	// RootDAGRun is the root dag-run reference for this attempt.
	RootDAGRun *DAGRunRef
	// Retry indicates whether this is a retry of a previous run.
	Retry bool
}

NewDAGRunAttemptOptions contains options for creating a new run record

type Node 

type Node struct {
	Step            core.Step            `json:"step,omitzero"`
	Stdout          string               `json:"stdout"` // standard output log file path
	Stderr          string               `json:"stderr"` // standard error log file path
	StartedAt       string               `json:"startedAt"`
	FinishedAt      string               `json:"finishedAt"`
	Status          core.NodeStatus      `json:"status"`
	RetriedAt       string               `json:"retriedAt,omitempty"`
	RetryCount      int                  `json:"retryCount,omitempty"`
	DoneCount       int                  `json:"doneCount,omitempty"`
	Repeated        bool                 `json:"repeated,omitempty"` // indicates if the node has been repeated
	Error           string               `json:"error,omitempty"`
	SubRuns         []SubDAGRun          `json:"children,omitempty"`
	SubRunsRepeated []SubDAGRun          `json:"childrenRepeated,omitempty"` // repeated sub DAG runs
	OutputVariables *collections.SyncMap `json:"outputVariables,omitempty"`
	// ApprovedAt records when this wait step was approved (HITL)
	ApprovedAt string `json:"approvedAt,omitempty"`
	// ApprovalInputs stores key-value parameters provided during approval
	ApprovalInputs map[string]string `json:"approvalInputs,omitempty"`
	// ApprovedBy records who approved this wait step (username)
	ApprovedBy string `json:"approvedBy,omitempty"`
	// RejectedAt records when this wait step was rejected (HITL)
	RejectedAt string `json:"rejectedAt,omitempty"`
	// RejectedBy records who rejected this wait step (username)
	RejectedBy string `json:"rejectedBy,omitempty"`
	// RejectionReason stores the optional reason for rejection
	RejectionReason string `json:"rejectionReason,omitempty"`
}

Node represents a DAG step with its execution state for persistence

func NewNodeFromStep 

func NewNodeFromStep(step core.Step) *Node

NewNodeFromStep creates a new Node with default status values for the given step.

func NewNodeOrNil 

func NewNodeOrNil(s *core.Step) *Node

NewNodeOrNil creates a Node from a Step or returns nil if the step is nil.

func NewNodesFromSteps added in v1.24.5 

func NewNodesFromSteps(steps []core.Step) []*Node

NewNodesFromSteps converts a list of DAG steps to persistence Node objects.

type OutputsMetadata added in v1.29.0 

type OutputsMetadata struct {
	DAGName     string `json:"dagName"`
	DAGRunID    string `json:"dagRunId"`
	AttemptID   string `json:"attemptId"`
	Status      string `json:"status"`
	CompletedAt string `json:"completedAt"`
	Params      string `json:"params,omitempty"` // JSON-serialized parameters
}

OutputsMetadata contains execution context for the outputs.

type PID 

type PID int

PID represents a process ID for a running dag-run

func (PID) String 

func (p PID) String() string

String returns the string representation of the PID, or an empty string if 0

type PageRange 

type PageRange struct {
	Range     []int
	SkipFirst bool
	SkipLast  bool
}

type PaginatedResult 

type PaginatedResult[T any] struct {
	Items       []T
	CurrentPage int
	TotalPages  int
	TotalCount  int
	Offset      int
	HasNextPage bool
	HasPrevPage bool
	NextPage    int
	PrevPage    int
}

func NewPaginatedResult 

func NewPaginatedResult[T any](items []T, total int, pg Paginator) PaginatedResult[T]

func (PaginatedResult[T]) Data 

func (r PaginatedResult[T]) Data() []T

func (PaginatedResult[T]) PageRange 

func (r PaginatedResult[T]) PageRange(size int) PageRange

func (PaginatedResult[T]) RangeEnd 

func (r PaginatedResult[T]) RangeEnd() int

func (PaginatedResult[T]) RangeStart 

func (r PaginatedResult[T]) RangeStart() int

type Paginator 

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

func DefaultPaginator 

func DefaultPaginator() Paginator

func NewPaginator 

func NewPaginator(page, perPage int) Paginator

func (*Paginator) Limit 

func (pg *Paginator) Limit() int

func (*Paginator) Offset 

func (pg *Paginator) Offset() int

type ProcHandle 

type ProcHandle interface {
	// Stop stops the heartbeat for the process.
	Stop(ctx context.Context) error
	// GetMeta retrieves the metadata for the process.
	GetMeta() ProcMeta
}

ProcHandle represents a process that is associated with a dag-run.

type ProcMeta 

type ProcMeta struct {
	StartedAt int64
	Name      string
	DAGRunID  string
}

ProcMeta is a struct that holds metadata for a process.

type ProcStore 

type ProcStore interface {
	// Lock try to lock process group return error if it's held by another process
	Lock(ctx context.Context, groupName string) error
	// UnLock unlocks process group
	Unlock(ctx context.Context, groupName string)
	// Acquire creates a new process for a given group name and DAG-run reference.
	// It automatically starts the heartbeat for the process.
	Acquire(ctx context.Context, groupName string, dagRun DAGRunRef) (ProcHandle, error)
	// CountAlive retrieves the number of processes associated with a group name.
	CountAlive(ctx context.Context, groupName string) (int, error)
	// CountAlive retrieves the number of processes associated with a group name.
	CountAliveByDAGName(ctx context.Context, groupName, dagName string) (int, error)
	// IsRunAlive checks if a specific DAG run is currently alive.
	IsRunAlive(ctx context.Context, groupName string, dagRun DAGRunRef) (bool, error)
	// ListAlive returns list of running DAG runs by the group name.
	ListAlive(ctx context.Context, groupName string) ([]DAGRunRef, error)
	// ListAllAlive returns all running DAG runs across all groups.
	// Returns a map where key is the group name and value is list of DAG runs.
	ListAllAlive(ctx context.Context) (map[string][]DAGRunRef, error)
}

ProcStore is an interface for managing process storage.

type QueuePriority 

type QueuePriority int

QueuePriority represents the priority of a queued item 

const (
	QueuePriorityHigh QueuePriority = iota
	QueuePriorityLow
)

type QueueStore 

type QueueStore interface {
	// Enqueue adds an item to the queue
	Enqueue(ctx context.Context, name string, priority QueuePriority, dagRun DAGRunRef) error
	// DequeueByName retrieves an item from the queue and removes it
	DequeueByName(ctx context.Context, name string) (QueuedItemData, error)
	// DequeueByDAGRunID retrieves items from the queue by dag-run reference and removes them
	DequeueByDAGRunID(ctx context.Context, name string, dagRun DAGRunRef) ([]QueuedItemData, error)
	// Len returns the number of items in the queue
	Len(ctx context.Context, name string) (int, error)
	// List returns all items in the queue with the given name
	List(ctx context.Context, name string) ([]QueuedItemData, error)
	// All returns all items in the queue
	All(ctx context.Context) ([]QueuedItemData, error)
	// ListByDAGName returns all items that has a specific DAG name
	ListByDAGName(ctx context.Context, name, dagName string) ([]QueuedItemData, error)
	// QueueList lists all queue names that have at least one item in the queue
	QueueList(ctx context.Context) ([]string, error)
	// Watcher returns a QueueWatcher for the queue data
	QueueWatcher(ctx context.Context) QueueWatcher
}

QueueStore provides an interface for interacting with the underlying database for storing and retrieving queued dag-run items.

type QueueWatcher added in v1.24.8 

type QueueWatcher interface {
	// Start start swatching queue data and signal when a queue state changed
	Start(ctx context.Context) (<-chan struct{}, error)
	// Stop stops watching queue data
	Stop(ctx context.Context)
}

QueueWatcher watches the queue state

type QueuedItem 

type QueuedItem struct {
	QueuedItemData
}

QueuedItem is a wrapper for QueuedItemData

func NewQueuedItem 

func NewQueuedItem(data QueuedItemData) *QueuedItem

NewQueuedItem creates a new QueuedItem

type QueuedItemData 

type QueuedItemData interface {
	// ID returns the ID of the queued item
	ID() string
	// Data returns the data of the queued item
	Data() (*DAGRunRef, error)
}

QueuedItemData represents a dag-run reference that is queued for execution.

type RemoveOldDAGRunsOption added in v1.27.0 

type RemoveOldDAGRunsOption func(*RemoveOldDAGRunsOptions)

RemoveOldDAGRunsOption is a functional option for configuring RemoveOldDAGRunsOptions

func WithDryRun added in v1.27.0 

func WithDryRun() RemoveOldDAGRunsOption

WithDryRun sets the dry-run mode for removing old dag-runs

type RemoveOldDAGRunsOptions added in v1.27.0 

type RemoveOldDAGRunsOptions struct {
	// DryRun if true, only returns the paths that would be removed without actually deleting
	DryRun bool
}

RemoveOldDAGRunsOptions contains options for removing old dag-runs

type RunStatus 

type RunStatus struct {
	// Name represents the name of the executed DAG.
	Name string
	// DAGRunID is the ID of the dag-run.
	DAGRunID string
	// Params is the parameters of the DAG.
	Params string
	// Outputs is the outputs of the dag-run.
	Outputs map[string]string
	// Status is the execution status of the dag-run.
	Status core.Status
}

SubDAGRunStatus is an interface that represents the status of a sub dag-run.

func (*RunStatus) MarshalJSON 

func (r *RunStatus) MarshalJSON() ([]byte, error)

MarshalJSON implements the json.Marshaler interface for RunStatus.

type ServiceName 

type ServiceName string

ServiceName represents the name of a service in the service registry system 

const (
	// ServiceNameCoordinator is the name of the coordinator service
	ServiceNameCoordinator ServiceName = "coordinator"
	// ServiceNameScheduler is the name of the scheduler service
	ServiceNameScheduler ServiceName = "scheduler"
)

type ServiceRegistry 

type ServiceRegistry interface {
	// Register registers services for the given service name and host info.
	// It returns an error if the registry failed to start heartbeat.
	Register(ctx context.Context, serviceName ServiceName, hostInfo HostInfo) error

	// Unregister un-registers current service.
	Unregister(ctx context.Context)

	// GetServiceMembers returns the list of active hosts for the given service.
	// This method combines service resolution and member lookup.
	GetServiceMembers(ctx context.Context, serviceName ServiceName) ([]HostInfo, error)

	// UpdateStatus updates the status of the current registered instance
	UpdateStatus(ctx context.Context, serviceName ServiceName, status ServiceStatus) error
}

ServiceRegistry is responsible for registering and persisting running service information.

type ServiceStatus 

type ServiceStatus int

ServiceStatus represents the operational status of a service instance 

const (
	// ServiceStatusUnknown indicates unknown status
	ServiceStatusUnknown ServiceStatus = iota
	// ServiceStatusActive indicates the service is active (e.g., scheduler holds lock)
	ServiceStatusActive
	// ServiceStatusInactive indicates the service is inactive (e.g., scheduler waiting for lock)
	ServiceStatusInactive
)

func (ServiceStatus) String 

func (s ServiceStatus) String() string

String returns the string representation of the service status

type SubDAGRun added in v1.24.0 

type SubDAGRun struct {
	DAGRunID string `json:"dagRunId,omitempty"`
	Params   string `json:"params,omitempty"`
}

SubDAGRun represents a sub DAG run associated with a node

type TimeInUTC 

type TimeInUTC struct{ time.Time }

TimeInUTC is a wrapper for time.Time that ensures the time is in UTC.

func NewUTC 

func NewUTC(t time.Time) TimeInUTC

NewUTC creates a new timeInUTC from a time.Time.

Source Files

View all Source files

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.