Documentation
¶
Overview ¶
Package openjd implements the OpenJD progress-line interceptor for sqi-worker.
An Interceptor wraps a downstream output handler and inspects each process stdout line for recognized OpenJD directive prefixes:
- openjd_progress: <0-100> — updates the task's in-memory progress percentage - openjd_status: <text> — publishes a live "running" status update via NATS - openjd_fail: <text> — cancels the per-task context and stores a failure reason
All other lines — including unrecognized openjd_* prefixes and all stderr lines — are forwarded to the downstream handler unmodified.
Integration with executor.Executor ¶
Interceptor implements [executor.TaskLifecycleHook], which the executor calls to register per-task metadata (task ID, job ID, session ID, and a context.CancelFunc) before the process starts and to retrieve any stored failure reason after the process exits.
The executor creates a per-task derived context so that openjd_fail cancels only the affected task rather than the worker-wide context.
Index ¶
- type EnvOp
- type EnvOpKind
- type Interceptor
- func (i *Interceptor) Deregister(attemptID string)
- func (i *Interceptor) FlushLogs(ctx context.Context, taskID, attemptID string) error
- func (i *Interceptor) HandleLine(ctx context.Context, taskID, attemptID, sessionID, stream, line string)
- func (i *Interceptor) LastProgress(attemptID string) *int
- func (i *Interceptor) RegisterTask(attemptID, taskID, jobID, sessionID string, cancel context.CancelFunc)
- func (i *Interceptor) TakeFailReason(attemptID string) (string, bool)
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type EnvOp ¶
EnvOp is a single parsed environment-mutation directive.
- Kind reports whether the variable is being set or unset.
- Name is the (validated) environment variable name.
- Value is the variable value (set only; empty for unset).
- Redacted is true when the value originated from openjd_redacted_env and therefore must never be written to logs in cleartext.
func ParseEnvDirective ¶
ParseEnvDirective classifies a single stdout line as one of the three OpenJD environment directives — openjd_env, openjd_redacted_env, openjd_unset_env — and returns the parsed mutation. ok is false when the line is not a valid environment directive (unrecognized prefix, malformed payload, or an invalid environment variable name); callers should log and ignore such lines.
Surrounding whitespace is trimmed. For set directives the payload is split on the FIRST '=' so values may themselves contain '='. For unset the payload is the bare variable name. Names must be non-empty, start with a non-digit, and contain only ASCII letters, digits, and underscores (OpenJD's env-name rule).
type EnvOpKind ¶
type EnvOpKind int
EnvOpKind classifies an EnvOp as either a set (assign/override) or an unset (remove) operation on the session environment.
type Interceptor ¶
type Interceptor struct {
// contains filtered or unexported fields
}
Interceptor is an output handler that intercepts recognized OpenJD directive lines from process stdout before forwarding the rest to the downstream handler.
Interceptor also implements executor.TaskLifecycleHook so the executor can register per-task metadata (including a cancel function for openjd_fail) and retrieve any stored failure reason after the process exits.
Interceptor is safe for concurrent use from multiple goroutines.
func New ¶
func New(downstream outputHandler, nc natsPublisher, workerID string, logger *slog.Logger) *Interceptor
New returns a ready-to-use Interceptor.
downstream receives all non-directive stdout lines and all stderr lines. nc is used to publish openjd_status messages to the task.status NATS subject. workerID is embedded in every status message published by the interceptor.
func (*Interceptor) Deregister ¶
func (i *Interceptor) Deregister(attemptID string)
Deregister removes the per-task state for attemptID. Safe to call multiple times.
func (*Interceptor) FlushLogs ¶
func (i *Interceptor) FlushLogs(ctx context.Context, taskID, attemptID string) error
FlushLogs forwards a flush request to the downstream handler if it implements the optional FlushLogs method. This allows the executor to call FlushLogs on the Interceptor (which it holds as the configured OutputHandler) and have the call reach the logstreamer that wraps underneath.
A local interface is used instead of importing executor.LogFlusher to avoid a circular dependency between the openjd and executor packages.
func (*Interceptor) HandleLine ¶
func (i *Interceptor) HandleLine(ctx context.Context, taskID, attemptID, sessionID, stream, line string)
HandleLine intercepts recognized OpenJD directive lines from stdout and forwards everything else to the downstream handler unmodified.
Stderr lines are always forwarded without inspection because tasks are expected to emit directives only to stdout. Recognized directive lines (openjd_progress, openjd_status, openjd_fail) are consumed without forwarding; all other lines — including unrecognized openjd_* prefixes — are passed through.
func (*Interceptor) LastProgress ¶
func (i *Interceptor) LastProgress(attemptID string) *int
LastProgress returns the most recent progress percentage (0–100) seen via openjd_progress for the given attemptID, or nil if none has been seen.
This is not part of executor.TaskLifecycleHook; it is provided for callers that include the last-known progress in status messages.
func (*Interceptor) RegisterTask ¶
func (i *Interceptor) RegisterTask(attemptID, taskID, jobID, sessionID string, cancel context.CancelFunc)
RegisterTask associates the given task metadata and cancel function with attemptID. The cancel function is called when an openjd_fail directive is seen, so the executor can terminate the specific task process without canceling the worker context.
RegisterTask must be called before the first HandleLine invocation for the given attemptID.
func (*Interceptor) TakeFailReason ¶
func (i *Interceptor) TakeFailReason(attemptID string) (string, bool)
TakeFailReason returns the failure reason stored by the first openjd_fail directive seen for attemptID, and true if one was present. Returns ("", false) if no openjd_fail was seen.
The stored reason is not cleared by this call; Deregister cleans it up. Repeated calls return a consistent value between RegisterTask and Deregister. An empty reason string with ok=true means openjd_fail was seen with no text.
Source Files