Documentation
¶
Index ¶
- Variables
- func MarkSeedFailed(ctx context.Context, repo *persis.DAGRunRepository, status *ir.DAGRunStatus, ...)
- func SeedRun(ctx context.Context, req SeedRequest) (dagrun.Attempt, *ir.DAGRunStatus, error)
- func SelectedStepNodes(dag *ir.DAG, steps []string, source *ir.DAGRunStatus, ...) ([]runtime.NodeData, error)
- func WebhookHeaders(forward []string, header func(name string) []string) (string, error)
- func WebhookRuntimeParams(payload, headers string) string
- type LocalAttemptBuilder
- type LocalPreparation
- type LocalProcRepository
- type LocalRequest
- type QueueRequest
- type QueuedRun
- type SeedRequest
Constants ¶
This section is empty.
Variables ¶
var ( ErrLocalExecutionAlreadyExists = errors.New("local execution already exists") ErrProcAcquisitionFailed = errors.New("failed to acquire process handle") )
Functions ¶
func MarkSeedFailed ¶ added in v2.17.2
func MarkSeedFailed(ctx context.Context, repo *persis.DAGRunRepository, status *ir.DAGRunStatus, cause error)
MarkSeedFailed records cause on a seeded run that never left the queued state, so a launch failure does not leave the run queued forever.
func SeedRun ¶ added in v2.17.2
func SeedRun(ctx context.Context, req SeedRequest) (dagrun.Attempt, *ir.DAGRunStatus, error)
SeedRun creates the queued attempt described by req and returns it with the status it recorded. The attempt is removed again if any later step fails.
func SelectedStepNodes ¶ added in v2.17.2
func SelectedStepNodes(dag *ir.DAG, steps []string, source *ir.DAGRunStatus, outputs map[string]map[string]string) ([]runtime.NodeData, error)
SelectedStepNodes returns seed nodes for a run of dag that executes only steps, each given by step name or ID. Every other step is recorded as skipped; when source is set, it carries the outputs source recorded for it. outputs sets outputs of skipped steps, keyed by step name or ID and then by output name, and takes precedence over outputs carried from source.
func WebhookHeaders ¶
WebhookHeaders encodes the request headers a webhook's run receives: those named in forward, under lowercase names, each with the values header gives for it. Names the run must never see, such as Authorization, are left out.
func WebhookRuntimeParams ¶
WebhookRuntimeParams are the params a webhook's run receives: the request body as WEBHOOK_PAYLOAD and the forwarded headers, encoded by WebhookHeaders, as WEBHOOK_HEADERS.
Types ¶
type LocalAttemptBuilder ¶
LocalAttemptBuilder creates or resolves the attempt that a local execution will own.
type LocalPreparation ¶
type LocalPreparation struct {
Attempt dagrun.Attempt
Proc proc.ProcHandle
}
LocalPreparation is the successfully prepared local execution ownership.
func PrepareLocalExecution ¶
func PrepareLocalExecution(ctx context.Context, req LocalRequest) (*LocalPreparation, error)
PrepareLocalExecution creates or resolves the execution attempt, acquires the local process heartbeat, and records a failed status if heartbeat acquisition fails after an attempt was prepared.
type LocalProcRepository ¶ added in v2.14.0
type LocalProcRepository interface {
WithLock(ctx context.Context, groupName string, fn func() error) error
Acquire(ctx context.Context, groupName string, meta proc.ProcMeta) (proc.ProcHandle, error)
}
LocalProcRepository is the process-repository surface needed to claim local execution ownership.
type LocalRequest ¶
type LocalRequest struct {
ProcRepository LocalProcRepository
DAG *ir.DAG
DAGRunID string
DefinitionID string
Root ir.DAGRunRef
Parent ir.DAGRunRef
TriggerType ir.TriggerType
TriggerActor string
ScheduleTime string
ProfileName string
LogBaseDir string
ArtifactBaseDir string
BuildAttempt LocalAttemptBuilder
}
LocalRequest describes local DAG-run intake before execution starts.
type QueueRequest ¶
type QueueRequest struct {
DAGRunRepository *persis.DAGRunRepository
QueueStore queue.QueueStore
DAG *ir.DAG
DAGRunID string
QueueName string
LogBaseDir string
ArtifactBaseDir string
Root ir.DAGRunRef
Parent ir.DAGRunRef
TriggerType ir.TriggerType
TriggerActor string
ParallelItem string
ScheduleTime string
ProfileName string
DefinitionID string
NoReuse bool
AttemptOptions persis.DAGRunCreateAttemptOptions
// ProceedOnStatusCloseErr preserves legacy CLI enqueue behavior: publish
// the queue item after best-effort close so readers can see the queued status.
ProceedOnStatusCloseErr bool
Now func() time.Time
}
QueueRequest describes a DAG-run intake operation that persists a queued attempt before publishing the queue item.
type QueuedRun ¶
type QueuedRun struct {
DAGRun ir.DAGRunRef
Attempt dagrun.Attempt
Status ir.DAGRunStatus
QueueName string
LogFile string
ArtifactDir string
// StatusCloseErr is set only when ProceedOnStatusCloseErr allowed intake
// to continue after the status attempt close failed.
StatusCloseErr error
}
QueuedRun is the result of successful DAG-run queue intake.
func EnqueueRun ¶
func EnqueueRun(ctx context.Context, req QueueRequest) (*QueuedRun, error)
EnqueueRun creates a queued DAG-run attempt, writes the queued status, closes the attempt, then publishes the queue item. If any post-create step fails, the created attempt is rolled back.
type SeedRequest ¶ added in v2.17.2
type SeedRequest struct {
DAGRunRepository *persis.DAGRunRepository
DAG *ir.DAG
DAGRunID string
Nodes []runtime.NodeData
// Source is the run whose recorded outputs the skipped nodes carry. When
// set, its work directory is copied into the new run and the carried
// output variables are rewritten to point at the copy.
Source *ir.DAGRunStatus
Params string
TriggerType ir.TriggerType
TriggerActor string
ProfileName string
DefinitionID string
NoReuse bool
LogBaseDir string
ArtifactBaseDir string
}
SeedRequest describes a new DAG-run whose first attempt is persisted as queued with fixed node states, to be executed through the retry path.
Source Files