execution

type ApiError ¶

type ApiError struct {
	Err    string
	Detail any
}

ApiError represents an error returned by the supervisor over the comm socket.

func (e *ApiError) Error() string

type BundleInfoMsg ¶

type BundleInfoMsg struct {
	Name    string
	Version string
}

BundleInfoMsg holds bundle identification from StartupDetails.

type ConnectionResult ¶

type ConnectionResult struct {
	ConnID   string
	ConnType string
	Host     string
	Schema   string
	Login    *string
	Password *string
	Port     int
	Extra    string
}

ConnectionResult is the response to GetConnection. Login and Password are nullable in the supervisor schema (None vs "" are distinct), so they are decoded as *string to preserve that distinction.

type CoordinatorClient ¶

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

CoordinatorClient implements sdk.Client by communicating with the Airflow supervisor over the comm socket using msgpack-framed IPC instead of HTTP.

func NewCoordinatorClient(comm *CoordinatorComm) *CoordinatorClient

func (c *CoordinatorClient) GetConnection(
	ctx context.Context,
	connID string,
) (sdk.Connection, error)

func (c *CoordinatorClient) GetVariable(ctx context.Context, key string) (string, error)

func (c *CoordinatorClient) GetXCom(
	ctx context.Context,
	dagId, runId, taskId string,
	mapIndex *int,
	key string,
	_ any,
) (any, error)

func (c *CoordinatorClient) PushXCom(
	ctx context.Context,
	ti api.TaskInstance,
	key string,
	value any,
) error

func (c *CoordinatorClient) UnmarshalJSONVariable(
	ctx context.Context,
	key string,
	pointer any,
) error

type CoordinatorComm ¶

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

CoordinatorComm manages bidirectional communication with the Airflow supervisor over a length-prefixed msgpack socket connection.

Reads are multiplexed by frame ID. A single background reader goroutine, lazily started on the first Communicate call, consumes inbound frames and dispatches each one to the Communicate caller whose request used that ID. This lets multiple goroutines call Communicate concurrently without serialising the full send-then-read round trip behind a single mutex, and guarantees the response a caller receives matches the request it sent.

The supervisor's initial StartupDetails frame arrives unsolicited, before any client request is in flight, and is read synchronously via ReadMessage; ReadMessage must not be called after the dispatcher has been started.

func NewCoordinatorComm(reader io.Reader, writer io.Writer, logger *slog.Logger) *CoordinatorComm

func (c *CoordinatorComm) Communicate(
	ctx context.Context,
	body map[string]any,
) (map[string]any, error)

func (c *CoordinatorComm) ReadMessage() (IncomingFrame, error)

func (c *CoordinatorComm) SendRequest(id int64, body map[string]any) error

type ErrorResponse ¶

type ErrorResponse struct {
	Error  string
	Detail any
}

ErrorResponse represents an error returned by the supervisor.

type GetConnectionMsg ¶

type GetConnectionMsg struct {
	ConnID string
}

GetConnectionMsg is sent to request a connection from the supervisor.

type GetVariableMsg ¶

type GetVariableMsg struct {
	Key string
}

GetVariableMsg is sent to request a variable from the supervisor.

type GetXComMsg ¶

type GetXComMsg struct {
	Key               string
	DagID             string
	TaskID            string
	RunID             string
	MapIndex          *int
	IncludePriorDates bool
}

GetXComMsg is sent to request an XCom value from the supervisor.

type IncomingFrame ¶

type IncomingFrame struct {
	ID   int64
	Body map[string]any
	Err  map[string]any // non-nil only for response frames (3-element arrays)
}

IncomingFrame represents a decoded frame received from the comm socket. ID is int64 to match the wire encoding and CoordinatorComm.nextID; narrowing to int would reintroduce wraparound on 32-bit GOARCH.

type MetadataFormat ¶

type MetadataFormat string

MetadataFormat selects the encoding DumpAirflowMetadata writes to stdout for the bundle binary's --airflow-metadata flag.

const (
	// MetadataFormatYAML is the default; it matches the airflow-metadata.yaml a
	// bundle embeds, so `mybundle --airflow-metadata > airflow-metadata.yaml`
	// yields a ready-to-use file.
	MetadataFormatYAML MetadataFormat = "yaml"
	// MetadataFormatJSON is opt-in via --format json.
	MetadataFormatJSON MetadataFormat = "json"
)

func ParseMetadataFormat(s string) (MetadataFormat, error)

type SetXComMsg ¶

type SetXComMsg struct {
	Key          string
	Value        any
	DagID        string
	TaskID       string
	RunID        string
	MapIndex     *int
	MappedLength *int
}

SetXComMsg is sent to set an XCom value. MapIndex mirrors Python's SetXCom.map_index (int | None): nil means "unmapped task", and is omitted from the wire payload rather than encoded as a -1 sentinel.

type SocketLogHandler ¶

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

SocketLogHandler is an slog.Handler that streams structured JSON log lines to the logs TCP socket. Each log entry is a single JSON object followed by a newline, matching the Airflow log streaming format.

Key mapping:

• "event" for the log message (not "msg")
• "level" in lowercase (not "INFO"/"ERROR")
• "timestamp" in RFC3339Nano format (not "time")
• Additional attributes are included as top-level fields

Groups are encoded as dotted key prefixes on a flat JSON object (`{"grp.key": "val"}`), not as nested objects. The Airflow supervisor's log-streaming format consumes flat top-level fields, so emitting nested objects here would not be parsed correctly. Do not change this without updating the supervisor side in lockstep.

func NewSocketLogHandler(writer io.Writer, level slog.Level) *SocketLogHandler

func (h *SocketLogHandler) Connect(w io.Writer)

func (h *SocketLogHandler) Enabled(_ context.Context, level slog.Level) bool

func (h *SocketLogHandler) Handle(_ context.Context, r slog.Record) error

func (h *SocketLogHandler) WithAttrs(attrs []slog.Attr) slog.Handler

func (h *SocketLogHandler) WithGroup(name string) slog.Handler

type StartupDetails ¶

type StartupDetails struct {
	TI                TaskInstanceInfo
	DagRelPath        string
	BundleInfo        BundleInfoMsg
	StartDate         time.Time
	TIContext         TIRunContext
	SentryIntegration string
}

StartupDetails is sent by the supervisor to initiate task execution.

type SucceedTaskMsg ¶

type SucceedTaskMsg struct {
	EndDate      time.Time
	TaskOutlets  []any
	OutletEvents []any
}

SucceedTaskMsg is sent as a terminal message when a task succeeds.

type TIRunContext ¶

type TIRunContext struct {
	LogicalDate       *time.Time
	DataIntervalStart *time.Time
	DataIntervalEnd   *time.Time
}

TIRunContext holds the runtime context for a task instance.

type TaskInstanceInfo ¶

type TaskInstanceInfo struct {
	ID             string
	TaskID         string
	DagID          string
	RunID          string
	TryNumber      int
	DagVersionID   string
	MapIndex       int
	ContextCarrier map[string]any
}

TaskInstanceInfo holds task instance details from StartupDetails.

type TaskState ¶

type TaskState string

TaskState is the terminal non-success state reported via TaskStateMsg. The wire values match Python's TaskInstanceState enum (and the generated api.TerminalStateNonSuccess); we define a local typed string so call sites get compile-time checking and don't have to import pkg/api just for the constants.

const (
	TaskStateFailed  TaskState = "failed"
	TaskStateRemoved TaskState = "removed"
	TaskStateSkipped TaskState = "skipped"
)

type TaskStateMsg ¶

type TaskStateMsg struct {
	State   TaskState
	EndDate time.Time
}

TaskStateMsg is sent as a terminal message for failed/removed/skipped tasks.

type VariableResult ¶

type VariableResult struct {
	Key   string
	Value any
}

VariableResult is the response to GetVariable.

type XComResult ¶

type XComResult struct {
	Key   string
	Value any
}

XComResult is the response to GetXCom.