dagrun

package
v2.15.2 Latest Latest
Warning

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

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

Documentation

Index

Constants

View Source 
const (
	// SubDAGRunsDir is the name of the directory where status files for sub dag-runs are stored.
	SubDAGRunsDir = "sub"

	// LegacySubDAGRunsDir is the previous directory where status files for sub dag-runs were stored.
	LegacySubDAGRunsDir = "children"

	// LegacySubDAGRunDirPrefix is the previous prefix for sub dag-run directories.
	LegacySubDAGRunDirPrefix = "child_"

	// DAGRunDirPrefix is the prefix for dag-run directories.
	DAGRunDirPrefix = "dag-run_"

	// AttemptDirPrefix is the prefix for attempt directories.
	AttemptDirPrefix = "a_"

	// LegacyAttemptDirPrefix is the previous prefix for attempt directories.
	LegacyAttemptDirPrefix = "attempt_"

	// SubDAGWorkDirPrefix is the prefix for sub dag-run working directories.
	SubDAGWorkDirPrefix = "w_"
)
View Source 
const CancelRequestedFlag = "CANCEL_REQUESTED"

CancelRequestedFlag is a special flag used to indicate that a cancel request has been made.

View Source 
const DAGDefinition = "dag.json"

DAGDefinition is the name of the file where the DAG definition is stored.

View Source 
const JSONLStatusFile = "status.jsonl"

JSONLStatusFile is the name of the status file for each dag-run. It contains the status of the dag-run in JSON Lines format. While running the dag-run, new lines are appended to this file on each status update. After finishing the run, this file will be compacted into a single JSON line file.

View Source 
const MessagesDir = "messages"

MessagesDir is the directory where per-step LLM messages are stored.

View Source 
const OutputsFile = "outputs.json"

OutputsFile is the name of the file where collected step outputs are stored.

Variables

View Source 
var (
	ErrStatusFileOpen    = errors.New("status file already open")
	ErrStatusFileNotOpen = errors.New("status file not open")
	ErrReadFailed        = errors.New("failed to read status file")
	ErrWriteFailed       = errors.New("failed to write to status file")
	ErrCompactFailed     = errors.New("failed to compact status file")
	ErrContextCanceled   = errors.New("operation canceled by context")
)

Error definitions for common issues

View Source 
var (
	ErrInvalidDAGRunsDir = errors.New("invalid dag-runs directory name")
)

Error definitions for directory structure validation

View Source 
var (
	ErrWriterNotOpen = errors.New("writer is not open")
)

Functions

func IsAttemptDirName 

func IsAttemptDirName(name string) bool

func ParseStatusFile 

func ParseStatusFile(file string) (*ir.DAGRunStatus, error)

ParseStatusFile reads the status file and returns the last valid status. The bufferSize parameter controls the size of the read buffer.

Types

type Attempt 

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

Attempt manages an append-only status file with read, write, and compaction capabilities. It provides thread-safe operations and supports metrics collection.

func NewAttempt 

func NewAttempt(file string, cache *fileutil.Cache[*ir.DAGRunStatus]) (*Attempt, error)

NewAttempt creates a new Run for the specified file.

func (*Attempt) Abort 

func (att *Attempt) Abort(ctx context.Context) error

Abort implements models.Attempt. It creates a flag to indicate that the attempt should be canceled.

func (*Attempt) Close 

func (att *Attempt) Close(ctx context.Context) error

Close properly closes the status file, performs compaction, and invalidates the cache. It's safe to call Close multiple times. The context can be used to cancel the operation.

func (*Attempt) Compact 

func (att *Attempt) Compact(ctx context.Context) error

Compact performs file compaction to optimize storage and read performance. It's safe to call while the file is open or closed. The context can be used to cancel the operation.

func (*Attempt) Exists 

func (att *Attempt) Exists() bool

Exists returns true if the status file exists.

func (*Attempt) Hidden 

func (att *Attempt) Hidden() bool

Hidden returns true if the attempt is hidden from normal operations.

func (*Attempt) Hide 

func (att *Attempt) Hide(ctx context.Context) error

Hide renames the attempt directory to hide it from normal operations. It prefixes the directory name with a dot to make it hidden.

func (*Attempt) ID 

func (att *Attempt) ID() string

ID implements models.Attempt.

func (*Attempt) IsAborting 

func (att *Attempt) IsAborting(ctx context.Context) (bool, error)

IsAborting checks if a cancel request has been made for this attempt.

func (*Attempt) ModTime 

func (att *Attempt) ModTime() (time.Time, error)

ModTime returns the last modification time of the status file. This is used to determine when the file was last updated.

func (*Attempt) Open 

func (att *Attempt) Open(ctx context.Context) error

Open initializes the status file for writing. It returns an error if the file is already open. The context can be used to cancel the operation.

func (*Attempt) ReadDAG 

func (att *Attempt) ReadDAG(_ context.Context) (*ir.DAG, error)

ReadDAG implements models.Attempt.

func (*Attempt) ReadOutputs 

func (att *Attempt) ReadOutputs(_ context.Context) (*ir.DAGRunOutputs, error)

ReadOutputs reads the collected step outputs from outputs.json. Returns nil if the file does not exist or if the file is in old format (no metadata field).

func (*Attempt) ReadStatus 

func (att *Attempt) ReadStatus(ctx context.Context) (*ir.DAGRunStatus, error)

ReadStatus reads the latest status from the file, using cache if available. The context can be used to cancel the operation.

func (*Attempt) ReadStepMessages 

func (att *Attempt) ReadStepMessages(_ context.Context, stepName string) ([]ir.LLMMessage, error)

ReadStepMessages reads LLM messages for a single step. Messages are stored at the dag-run level in a messages/ directory for retry persistence. Returns nil if no messages exist for the step.

func (*Attempt) SetDAG 

func (att *Attempt) SetDAG(dag *ir.DAG)

SetDAG sets the DAG for this attempt. Must be called before Open for DAG to be persisted.

func (*Attempt) Write 

func (att *Attempt) Write(ctx context.Context, status ir.DAGRunStatus) error

Write adds a new status to the file. It returns an error if the file is not open or is currently being closed. The context can be used to cancel the operation.

func (*Attempt) WriteOutputs 

func (att *Attempt) WriteOutputs(_ context.Context, outputs *ir.DAGRunOutputs) error

WriteOutputs writes the collected step outputs to outputs.json. If outputs is nil or has no output entries, no file is created.

func (*Attempt) WriteStepMessages 

func (att *Attempt) WriteStepMessages(_ context.Context, stepName string, messages []ir.LLMMessage) error

WriteStepMessages writes LLM messages for a single step. Messages are stored at the dag-run level in a messages/ directory for retry persistence.

type DAGRun 

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

DAGRun represents a dag-run with its associated timestamp and run ID.

func NewDAGRun 

func NewDAGRun(dir string) (*DAGRun, error)

NewDAGRun creates a new Run instance from a directory path. It parses the directory name to extract the timestamp and dag-run ID.

func (DAGRun) AttemptByDir 

func (dr DAGRun) AttemptByDir(attemptDir string, cache *fileutil.Cache[*ir.DAGRunStatus]) (*Attempt, error)

AttemptByDir constructs an Attempt directly from a known attempt directory name, skipping the directory listing and sorting done by LatestAttempt.

func (DAGRun) CreateAttempt 

func (dr DAGRun) CreateAttempt(_ context.Context, ts persis.TimeInUTC, cache *fileutil.Cache[*ir.DAGRunStatus], attemptID string) (*Attempt, error)

CreateAttempt creates a new Attempt for the dag-run with the given timestamp. It creates a new Attempt directory and initializes a record within it. If attemptID is provided, it uses that ID instead of generating a new one.

func (DAGRun) CreateSubDAGRun 

func (dr DAGRun) CreateSubDAGRun(_ context.Context, dagRunID string) (*DAGRun, error)

CreateSubDAGRun creates a new sub dag-run with the given timestamp and dag-run ID.

func (DAGRun) FindSubDAGRun 

func (dr DAGRun) FindSubDAGRun(_ context.Context, dagRunID string) (*DAGRun, error)

FindSubDAGRun searches for a sub dag-run by its run ID.

func (DAGRun) LatestAttempt 

func (dr DAGRun) LatestAttempt(ctx context.Context, cache *fileutil.Cache[*ir.DAGRunStatus]) (*Attempt, error)

LatestAttempt returns the most recent Attempt for the dag-run. It searches through all run directories and returns the first valid Attempt found. It skips hidden attempts (dequeued ones).

func (DAGRun) ListSubDAGRuns 

func (dr DAGRun) ListSubDAGRuns(ctx context.Context) ([]*DAGRun, error)

func (DAGRun) Remove 

func (dr DAGRun) Remove(ctx context.Context) error

Remove deletes the entire dag-run directory and all its contents.

type DAGRunSummary 

type DAGRunSummary struct {
	LatestAttemptDir     string
	Status               ir.Status
	StartedAtUnix        int64
	FinishedAtUnix       int64
	Labels               []string
	Name                 string
	DagRunID             string
	WorkerID             string
	LeaseAt              int64
	Params               string
	QueuedAt             string
	ScheduleTime         string
	TriggerType          ir.TriggerType
	TriggerActor         string
	CreatedAt            int64
	AttemptID            string
	AutoRetryCount       int
	ParentName           string
	ParentID             string
	AutoRetryLimit       int
	AutoRetryInterval    time.Duration
	AutoRetryBackoff     float64
	AutoRetryMaxInterval time.Duration
	ProcGroup            string
	DefinitionID         string
	ArchiveDir           string
}

DAGRunSummary holds pre-loaded summary data from a day index. When non-nil, it allows filtering and constructing list responses without reading status.jsonl.

type DataRoot 

type DataRoot struct {
	dirlock.DirLock // Directory lock for concurrent access
	// contains filtered or unexported fields
}

DataRoot manages the directory structure for run history data. It handles the organization of run data in a hierarchical structure based on year, month, and day.

func NewDataRoot 

func NewDataRoot(baseDir, dagName string) DataRoot

NewDataRoot creates a new DataRoot instance for managing a DAG's run history. It sanitizes the DAG name to create a safe directory structure and applies any provided options.

Parameters:

  • baseDir: The base directory where all DAG data is stored
  • dagName: The name of the DAG (can be a path to a YAML file)

Returns:

  • A configured DataRoot instance

func NewDataRootWithArtifactDir 

func NewDataRootWithArtifactDir(baseDir, dagName, artifactDir string) DataRoot

NewDataRootWithArtifactDir creates a new DataRoot with an explicit trusted artifact root.

func (*DataRoot) CreateDAGRun 

func (dr *DataRoot) CreateDAGRun(ts persis.TimeInUTC, dagRunID string) (*DAGRun, error)

CreateDAGRun creates a new dag-run directory with the specified timestamp and ID. The directory structure follows the pattern: year/month/day/run-YYYYMMDD_HHMMSS_dagRunID

func (DataRoot) Exists 

func (dr DataRoot) Exists() bool

Exists checks if the dag-runs directory exists in the file system.

func (*DataRoot) FindByDAGRunID 

func (dr *DataRoot) FindByDAGRunID(ctx context.Context, dagRunID string) (*DAGRun, error)

FindByDAGRunID locates a dag-run by its ID. It scans the year/month/day hierarchy in reverse chronological order and returns the first exact match, which preserves the "newest run wins" behavior without materializing and sorting all matches across the full history tree.

func (DataRoot) IsEmpty 

func (dr DataRoot) IsEmpty() bool

IsEmpty checks if the dag-runs directory exists and contains no dag-run directories. Returns true if the directory doesn't exist or contains no dag-runs.

func (*DataRoot) Latest 

func (dr *DataRoot) Latest(ctx context.Context, itemLimit int) []*DAGRun

Latest returns the most recent dag-runs up to the specified limit. It searches through the dag-run directories and returns them sorted by timestamp (newest first).

func (*DataRoot) LatestAfter 

func (dr *DataRoot) LatestAfter(ctx context.Context, cutoff persis.TimeInUTC) (*DAGRun, error)

LatestAfter returns the most recent dag-run that occurred after the specified cutoff time. Returns ErrNoStatusData if no dag-run is found or if the latest run is before the cutoff.

func (DataRoot) Remove 

func (dr DataRoot) Remove() error

Remove completely removes the dag-runs directory and all its contents. This operation cannot be undone.

func (DataRoot) RemoveOldByRuns 

func (dr DataRoot) RemoveOldByRuns(ctx context.Context, retentionRuns int, dryRun bool) ([]string, error)

RemoveOldByRuns removes dag-runs beyond the most recent retentionRuns. Active runs are preserved even when they fall outside the retained window.

type Store 

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

Store manages DAG run status files on the local filesystem.

func NewStore added in v2.14.0 

func NewStore(baseDir string, opts ...StoreOption) *Store

NewStore creates filesystem DAG-run storage.

func (*Store) CompareAndSwapLatestAttemptStatus 

func (store *Store) CompareAndSwapLatestAttemptStatus(
	ctx context.Context,
	req persis.DAGRunCompareAndSwapStatusRequest,
) (*ir.DAGRunStatus, bool, error)

func (*Store) CreateAttempt 

func (store *Store) CreateAttempt(ctx context.Context, req persis.DAGRunCreateAttemptRequest) (dagrun.Attempt, error)

CreateAttempt creates an attempt within a root or child DAG run.

func (*Store) FindAttempt 

func (store *Store) FindAttempt(ctx context.Context, ref ir.DAGRunRef) (dagrun.Attempt, error)

FindAttempt finds the latest attempt by DAG-run ID.

func (*Store) FindSubAttempt 

func (store *Store) FindSubAttempt(ctx context.Context, ref ir.DAGRunRef, subDAGRunID string) (dagrun.Attempt, error)

FindSubAttempt finds a sub dag-run by its ID. It returns the latest attempt for the specified sub DAG-run ID.

func (*Store) LatestAttempt 

func (store *Store) LatestAttempt(ctx context.Context, query persis.DAGRunLatestAttemptQuery) (dagrun.Attempt, error)

LatestAttempt returns the newest visible attempt matching the query.

func (*Store) ListRetryCandidates 

func (store *Store) ListRetryCandidates(ctx context.Context, from persis.TimeInUTC) ([]*ir.DAGRunStatus, error)

func (*Store) QueryStatuses added in v2.14.0 

func (store *Store) QueryStatuses(ctx context.Context, query persis.DAGRunStatusQuery) (persis.DAGRunStatusPage, error)

QueryStatuses executes a normalized status query.

func (*Store) RecentStatuses added in v2.14.0 

func (store *Store) RecentStatuses(ctx context.Context, dagName string, itemLimit int) ([]ir.DAGRunStatus, error)

RecentStatuses returns the newest readable status for recent DAG runs.

func (*Store) RemoveDAGRun 

func (store *Store) RemoveDAGRun(ctx context.Context, req persis.DAGRunRemoveRequest) error

RemoveDAGRun removes a DAG run and all of its attempts.

func (*Store) RemoveOldDAGRuns 

func (store *Store) RemoveOldDAGRuns(ctx context.Context, req persis.DAGRunRetentionRequest) ([]ir.DAGRunRef, error)

RemoveOldDAGRuns removes final runs outside a normalized retention policy.

type StoreOption added in v2.14.0 

type StoreOption func(*options)

StoreOption configures filesystem DAG-run storage.

func WithArtifactDir 

func WithArtifactDir(dir string) StoreOption

WithArtifactDir sets the trusted root for artifact cleanup operations.

func WithHistoryFileCache 

func WithHistoryFileCache(cache *fileutil.Cache[*ir.DAGRunStatus]) StoreOption

WithHistoryFileCache sets the file cache for Store.

type WorkDirStore added in v2.14.0 

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

WorkDirStore manages file-backed DAG-run work directories.

func NewWorkDirStore added in v2.14.0 

func NewWorkDirStore(rootDir, historyDir string) *WorkDirStore

NewWorkDirStore creates a work-directory store rooted at rootDir. Existing work directories nested under historyDir remain accessible.

func (*WorkDirStore) Materialize added in v2.14.0 

func (s *WorkDirStore) Materialize(ctx context.Context, ref dagrun.WorkDirRef) (string, error)

func (*WorkDirStore) Remove added in v2.14.0 

func (s *WorkDirStore) Remove(_ context.Context, ref dagrun.WorkDirRef) error

func (*WorkDirStore) Snapshot added in v2.14.0 

func (*WorkDirStore) Snapshot(context.Context, dagrun.WorkDirRef, string) error

type Writer 

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

func NewWriter 

func NewWriter(target string, opts ...WriterOption) *Writer

NewWriter creates a new Writer instance for the specified target file path.

func (*Writer) Close 

func (w *Writer) Close(ctx context.Context) error

Close flushes any buffered data and closes the underlying file. It's safe to call close multiple times.

func (*Writer) IsOpen 

func (w *Writer) IsOpen() bool

IsOpen returns true if the writer is currently open.

func (*Writer) Open 

func (w *Writer) Open() error

Open prepares the writer for writing by creating necessary directories and opening the target file.

func (*Writer) Write 

func (w *Writer) Write(ctx context.Context, st ir.DAGRunStatus) error

Write serializes the status to JSON and appends it to the file. It automatically flushes data to ensure durability.

type WriterOption 

type WriterOption func(*Writer)

WriterOption defines functional options for configuring a Writer.

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.