runctx

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: 16 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	// StreamTypeStdout indicates stdout stream
	StreamTypeStdout = 1
	// StreamTypeStderr indicates stderr stream
	StreamTypeStderr = 2
)

Stream type constants for LogWriterFactory.NewStepWriter

Variables

This section is empty.

Functions

func NewContext 

func NewContext(
	ctx context.Context,
	dag *ir.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 (WithDAGLoader, WithParams, etc.)

func WithContext 

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 

type Context struct {
	DAGRunID             string
	RootDAGRun           ir.DAGRunRef
	RetryPath            dagrun.RetryPath
	IncludeDownstream    bool
	AttemptID            string
	WorkerID             string
	TriggerType          ir.TriggerType
	TriggerActor         string
	RunStartedAt         string
	ScheduleTime         string
	DAG                  *ir.DAG
	DAGLoader            DAGLoader
	BaseEnv              *config.BaseEnv
	EnvScope             *cmnvalue.EnvScope // Unified environment scope for runtime variables
	CoordinatorCli       dispatch.Dispatcher
	RunStateStore        runstate.Store
	StateStore           dagrun.StateStore
	MaterializationStore build.MaterializationStore
	DAGRunLogDir         string
	DAGRunArtifactDir    string
	ProfileName          string
	ProfileResolvedAt    string
	ProfileEntries       []ir.RuntimeProfileEntry
	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")
	LogWriterFactory     LogWriterFactory     // For remote log streaming (nil = use local files)
	DefaultExecMode      config.ExecutionMode // Server-level default execution mode (local or distributed)
	NoReuse              bool
}

Context contains the execution metadata for a dag-run.

func GetContext 

func GetContext(ctx context.Context) Context

GetContext retrieves the DAGContext from the context.

func LookupContext 

func LookupContext(ctx context.Context) (Context, bool)

LookupContext returns the DAGContext when one is present in ctx.

func MustContext 

func MustContext(ctx context.Context) Context

MustContext returns the DAGContext from the context and panics when one is not present. Use it where a fully initialized dag-run context is a caller invariant; use LookupContext where absence is valid.

func (Context) AllEnvs 

func (e Context) AllEnvs() []string

AllEnvs returns every environment variable as "key=value" strings. Uses EnvScope as the single source of truth for all env vars.

func (Context) DAGRunRef 

func (e Context) DAGRunRef() ir.DAGRunRef

DAGRunRef returns the DAG-run reference for the current DAG context.

func (Context) UserEnvsMap 

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.

type ContextOption 

type ContextOption func(*contextOptions)

ContextOption configures optional parameters for NewContext.

func WithArtifactDir 

func WithArtifactDir(dir string) ContextOption

WithArtifactDir sets the per-DAG-run artifacts directory path.

func WithAttemptID 

func WithAttemptID(attemptID string) ContextOption

WithAttemptID sets the DAG-run attempt identifier for value resolution.

func WithCoordinator 

func WithCoordinator(cli dispatch.Dispatcher) ContextOption

WithCoordinator sets the coordinator dispatcher for distributed execution.

func WithDAGLoader added in v2.14.0 

func WithDAGLoader(loader DAGLoader) ContextOption

WithDAGLoader sets the DAG loader.

func WithDAGRunArtifactDir 

func WithDAGRunArtifactDir(dir string) ContextOption

WithDAGRunArtifactDir sets the base artifact directory for newly persisted DAG runs.

func WithDAGRunLogDir 

func WithDAGRunLogDir(dir string) ContextOption

WithDAGRunLogDir sets the base log directory for newly persisted DAG runs.

func WithDefaultEnvVars 

func WithDefaultEnvVars(envs ...string) ContextOption

WithDefaultEnvVars sets low-precedence inherited environment variables.

func WithDefaultExecMode 

func WithDefaultExecMode(mode config.ExecutionMode) ContextOption

WithDefaultExecMode sets the server-level default execution mode.

func WithDefaultSecrets 

func WithDefaultSecrets(secrets []string) ContextOption

WithDefaultSecrets sets low-precedence inherited secret environment variables.

func WithEnvVars 

func WithEnvVars(envs ...string) ContextOption

WithEnvVars sets additional execution-scoped environment variables.

func WithIncludeDownstream added in v2.15.0 

func WithIncludeDownstream(enabled bool) ContextOption

WithIncludeDownstream records that a targeted step retry should also reset reachable descendants in the DAG that contains the selected step.

func WithLogEncoding 

func WithLogEncoding(charset string) ContextOption

WithLogEncoding sets the log file character encoding.

func WithLogWriterFactory 

func WithLogWriterFactory(factory LogWriterFactory) ContextOption

WithLogWriterFactory sets the log writer factory for remote log streaming. When set, logs are streamed to the coordinator instead of written to local files.

func WithMaterializationStore 

func WithMaterializationStore(store build.MaterializationStore) ContextOption

WithMaterializationStore sets the build materialization store.

func WithNoReuse 

func WithNoReuse(disabled bool) ContextOption

WithNoReuse records that manifest hits are disabled for the run.

func WithParams 

func WithParams(params []string) ContextOption

WithParams sets runtime parameters.

func WithRetryPath 

func WithRetryPath(path dagrun.RetryPath) ContextOption

WithRetryPath sets the persisted child DAG path for a targeted retry.

func WithRootDAGRun 

func WithRootDAGRun(ref ir.DAGRunRef) ContextOption

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

func WithRunStartedAt 

func WithRunStartedAt(startedAt string) ContextOption

WithRunStartedAt sets the recorded DAG-run start timestamp for value resolution.

func WithRunStateStore added in v2.14.0 

func WithRunStateStore(store runstate.Store) ContextOption

WithRunStateStore sets the execution-state store.

func WithRuntimeProfile 

func WithRuntimeProfile(name, resolvedAt string, entries []ir.RuntimeProfileEntry) ContextOption

WithRuntimeProfile sets the selected profile metadata for this run context.

func WithScheduleTime 

func WithScheduleTime(scheduleTime string) ContextOption

WithScheduleTime sets the logical schedule time for value resolution.

func WithSecrets 

func WithSecrets(secrets []string) ContextOption

WithSecrets sets secret environment variables.

func WithStateStore 

func WithStateStore(store dagrun.StateStore) ContextOption

WithStateStore sets the persistent DAG state store for state actions.

func WithTriggerActor 

func WithTriggerActor(actor string) ContextOption

WithTriggerActor sets the attributable trigger actor for value resolution.

func WithTriggerType 

func WithTriggerType(triggerType ir.TriggerType) ContextOption

WithTriggerType sets the DAG-run trigger type for value resolution.

func WithWorkDir 

func WithWorkDir(dir string) ContextOption

WithWorkDir sets the per-DAG-run working directory path.

func WithWorkerID added in v2.14.0 

func WithWorkerID(workerID string) ContextOption

WithWorkerID records the execution host that owns the run.

type DAGLoader added in v2.14.0 

type DAGLoader interface {
	GetDAG(ctx context.Context, name string) (*ir.DAG, error)
}

DAGLoader loads DAG definitions needed during execution.

type LogWriterFactory 

type LogWriterFactory interface {
	// NewStepWriter creates a writer for a step's log output.
	// stepName identifies the step, streamType should be StreamTypeStdout or StreamTypeStderr.
	NewStepWriter(ctx context.Context, stepName string, streamType int) io.WriteCloser
}

LogWriterFactory creates log writers for step stdout/stderr. It abstracts where logs are written, allowing for: - Local file-based storage (default) - Remote streaming to coordinator

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.