Documentation
¶
Overview ¶
Package executor implements the bare-metal task executor for sqi-worker.
An Executor starts OS processes via os/exec for each assigned task, captures their stdout and stderr line-by-line, manages per-task execution timeouts and cancellation per the OpenJD Action.cancelation method (TERMINATE = immediate SIGKILL, the spec default; NOTIFY_THEN_TERMINATE = SIGTERM then SIGKILL after the notify period), and publishes task-status messages back to sqi-server via NATS.
Interfaces ¶
Executor implements both [pull.TaskDispatcher] and [pull.StateSource] so it can be wired directly into the pull loop. It also satisfies [heartbeat.StateSource] by exposing Executor.ActiveTaskCount, Executor.ActiveTaskIDs, and Executor.LastAssignmentAt.
Concurrency ¶
The server gates how many tasks a worker receives via CPU-core accounting; the worker runs whatever it is leased without an additional local cap. Multiple leases are dispatched in separate goroutines and run concurrently.
Output handling ¶
The OutputHandler interface abstracts what happens with each line of process output. The log-chunk publisher and OpenJD progress line parser will implement this interface; until then the LogOutput implementation forwards lines to the structured logger.
Root-user check ¶
CheckRootUser should be called before creating an Executor. It returns an error if the worker process is running as root on Linux/macOS and Config.AllowRoot is false (see docs/worker-configuration.md, "worker.allow_root").
Index ¶
- func CheckRootUser(allowRoot bool, logger *slog.Logger) error
- type CancelRegistrar
- type Config
- type DiscardOutput
- type Executor
- func (e *Executor) ActiveTaskCount() int
- func (e *Executor) ActiveTaskIDs() []string
- func (e *Executor) Cancel(taskID string) bool
- func (e *Executor) Dispatch(ctx context.Context, msg *protocol.AssignMsg) error
- func (e *Executor) DrainAndShutdown(gracePeriod time.Duration) (completed, killed int)
- func (e *Executor) FlushShutdownStatuses()
- func (e *Executor) LastAssignmentAt() *time.Time
- func (e *Executor) SetCancelRegistrar(cr CancelRegistrar)
- type LogFlusher
- type LogOutput
- type OutputHandler
- type TaskLifecycleHook
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func CheckRootUser ¶
CheckRootUser warns and returns an error if the worker is running as the root user on Linux/macOS and allowRoot is false (see docs/worker-configuration.md, "worker.allow_root").
Call this at worker startup, before constructing an Executor. If the function returns a non-nil error the caller should treat it as fatal and exit with a non-zero code.
Types ¶
type CancelRegistrar ¶
type CancelRegistrar interface {
// Register subscribes to cancel messages for the given taskID.
// Errors are logged as warnings and do not abort task execution.
Register(taskID string) error
// Deregister unsubscribes from cancel messages for the given taskID.
Deregister(taskID string)
}
CancelRegistrar is an optional hook that the executor calls to subscribe to and unsubscribe from per-task cancel NATS messages.
Register is called after the per-task context is set up, immediately before the task process starts. Deregister is called when the task goroutine exits.
Implementations must be safe for concurrent use from multiple goroutines (multiple tasks may register/deregister simultaneously).
type Config ¶
type Config struct {
// KillGracePeriod is the time the executor waits after sending SIGTERM
// before escalating to SIGKILL when forcibly terminating a process during
// worker force-shutdown (DrainAndShutdown). Defaults to 10 s when zero or
// negative.
//
// Note: per-task timeout and per-task cancellation no longer use this value;
// they follow the assignment's OpenJD Action.cancelation method (TERMINATE =
// immediate SIGKILL; NOTIFY_THEN_TERMINATE = SIGTERM then the action's notify
// period). KillGracePeriod governs only the worker-shutdown grace window.
KillGracePeriod time.Duration
// AllowRoot, when true, allows the worker to run as the root user on
// Linux/macOS. [CheckRootUser] returns an error if this is false and the
// process UID is 0.
AllowRoot bool
}
Config holds the tunable parameters for an Executor.
type DiscardOutput ¶
type DiscardOutput struct{}
DiscardOutput is an OutputHandler that silently drops every line. Use it in tests or when output logging is intentionally disabled.
func (DiscardOutput) HandleLine ¶
func (DiscardOutput) HandleLine(_ context.Context, _, _, _, _, _ string)
HandleLine discards the output line.
type Executor ¶
type Executor struct {
// contains filtered or unexported fields
}
Executor starts and manages bare-metal OS processes for assigned tasks.
Create an instance with New and wire it into the pull loop and heartbeat publisher. Call CheckRootUser before creating an Executor to verify the process is not running as root. After construction, call [SetCancelRegistrar] to wire in the NATS cancel subscription handler.
func New ¶
func New( statusPub *status.Publisher, sessionMgr *session.Manager, m *metrics.Metrics, outputHandler OutputHandler, cfg Config, logger *slog.Logger, ) *Executor
New creates a ready-to-use Executor.
statusPub is the typed status publisher used for all task-status messages; it must not be nil.
outputHandler may be nil; if so LogOutput is used to forward process output to the structured logger.
cfg.KillGracePeriod defaults to 10 s if zero or negative.
func (*Executor) ActiveTaskCount ¶
ActiveTaskCount returns the number of task processes currently executing. Implements both [pull.StateSource] and [heartbeat.StateSource].
func (*Executor) ActiveTaskIDs ¶
ActiveTaskIDs returns a snapshot of the IDs of all currently running tasks. Implements [heartbeat.StateSource].
func (*Executor) Cancel ¶
Cancel requests cancellation of the task identified by taskID.
If the task is actively executing its per-task context is canceled, triggering termination per the assignment's OpenJD Action.cancelation method (TERMINATE by default → immediate SIGKILL; NOTIFY_THEN_TERMINATE → SIGTERM then SIGKILL after the notify period). If the task is known but its process has not yet started (race between Dispatch and goroutine scheduling), the cancel is noted and takes effect as soon as runTask initializes its context.
Returns true if taskID was found in the active-tasks map (regardless of whether cancellation completed), false if the task was not found (already finished or never dispatched).
func (*Executor) Dispatch ¶
Dispatch implements [pull.TaskDispatcher].
It creates a session and launches a goroutine that executes the task process. Dispatch itself returns quickly; the task goroutine runs independently. The server gates concurrency by CPU cores; the worker runs whatever it is leased.
func (*Executor) DrainAndShutdown ¶
DrainAndShutdown implements the worker graceful-shutdown sequence (tasks 86–87).
It first waits up to gracePeriod for all in-flight tasks to complete naturally. If tasks are still running when the grace period expires, it sets the shuttingDown flag (so goroutines publish "failed"/"worker_shutdown" rather than "canceled") and cancels the shared execCtx, triggering SIGTERM → SIGKILL escalation in each runTask goroutine. It then blocks until all goroutines have published their terminal statuses and exited.
Returns (completed, killed): the number of tasks that finished naturally during the grace period vs the number that were force-terminated.
A gracePeriod of zero skips the wait phase and force-kills immediately.
func (*Executor) FlushShutdownStatuses ¶
func (e *Executor) FlushShutdownStatuses()
FlushShutdownStatuses publishes a "failed"/"worker_shutdown" status for every task currently in the active-tasks map.
This must be called before draining the NATS connection during worker shutdown so the server learns about task failures immediately rather than waiting for the heartbeat-timeout sweep. Each task goroutine will also publish its own terminal status as it exits; the server must handle duplicate terminal statuses gracefully (accept the first, ignore subsequent).
The total flush is bounded by a 5-second deadline so that a saturated NATS connection does not stall the shutdown sequence indefinitely.
func (*Executor) LastAssignmentAt ¶
LastAssignmentAt returns the time of the most recent task assignment dispatched to this executor, or nil if no task has been dispatched yet. Implements [heartbeat.StateSource].
func (*Executor) SetCancelRegistrar ¶
func (e *Executor) SetCancelRegistrar(cr CancelRegistrar)
SetCancelRegistrar wires a CancelRegistrar into the executor so that per-task NATS cancel subscriptions are managed automatically.
Must be called before any task is dispatched; it is safe to call concurrently with other methods but races with active Dispatch calls are not supported.
type LogFlusher ¶
type LogFlusher interface {
// FlushLogs drains all remaining buffered lines for the given
// (taskID, attemptID) pair and blocks until the drain is complete.
// It is idempotent: calling it multiple times is safe.
FlushLogs(ctx context.Context, taskID, attemptID string) error
}
LogFlusher is an optional interface that an OutputHandler may implement to flush remaining buffered log output for a specific task attempt before the terminal status message is published.
If the configured OutputHandler also implements LogFlusher, the executor calls FlushLogs immediately after the task process exits and before publishing the terminal "succeeded", "failed", or "canceled" status. This guarantees complete log delivery to the server before the task transitions to a terminal state.
type LogOutput ¶
LogOutput is an OutputHandler that forwards each output line to the structured logger at debug level. This is the default handler used when nil is passed to New.
type OutputHandler ¶
type OutputHandler interface {
// HandleLine is called for each line read from the task process stdout or
// stderr. stream is "stdout" or "stderr". The line does not include a
// trailing newline.
HandleLine(ctx context.Context, taskID, attemptID, sessionID, stream, line string)
}
OutputHandler processes lines of output emitted by a running task process. Implementations include the log-chunk publisher and the OpenJD progress-line parser. An OutputHandler must be safe for concurrent use from multiple goroutines (multiple tasks may emit output simultaneously).
type TaskLifecycleHook ¶
type TaskLifecycleHook interface {
// RegisterTask associates task metadata and a cancel function with
// attemptID. The cancel function is called when an openjd_fail directive
// is seen; termination then honors the action's OpenJD cancelation method
// (TERMINATE by default → immediate SIGKILL; NOTIFY_THEN_TERMINATE →
// SIGTERM then SIGKILL after the notify period).
RegisterTask(attemptID, taskID, jobID, sessionID string, cancel context.CancelFunc)
// Deregister removes the per-task state for attemptID. Safe to call
// multiple times.
Deregister(attemptID string)
// TakeFailReason returns the failure reason stored by an openjd_fail
// directive for attemptID and true if one is present. Returns ("", false)
// if no openjd_fail was seen.
TakeFailReason(attemptID string) (string, bool)
}
TaskLifecycleHook is an optional interface that an OutputHandler may implement to participate in the per-task execution lifecycle.
The executor calls [RegisterTask] before starting each task process and [Deregister] after the task completes, giving the handler a per-task cancel function it can call when an OpenJD fail directive is seen. After the process exits, the executor calls [TakeFailReason] to retrieve any stored failure reason before publishing the terminal status message.
The OpenJD interceptor (internal/worker/openjd) implements this interface so that openjd_fail directives can terminate a specific task without canceling the worker-wide context.
Source Files