process

package
v0.3.0-alpha.1 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Aug 26, 2026 License: MIT Imports: 12 Imported by: 0

README

gokit/process

Subprocess execution with context cancellation, signal handling, and provider integration.

Overview

The process package provides a structured way to run external commands from Go services. It wraps os/exec with context-aware cancellation, process group management, graceful shutdown (SIGTERM → SIGKILL), and automatic output capture. Results include stdout, stderr, exit code, and duration.

For long-running or unreliable subprocesses, the package integrates with gokit's provider and resilience frameworks — adding retry, circuit breaker, and generic I/O adapters.

Installation

go get github.com/kbukum/gokit

process is part of the core module — no separate go get needed.

Quick Start

package main

import (
	"context"
	"fmt"

	"github.com/kbukum/gokit/process"
)

func main() {
	ctx := context.Background()

	result, err := process.Run(ctx, process.Command{
		Binary: "echo",
		Args:   []string{"hello", "world"},
	})
	if err != nil {
		panic(err)
	}

	fmt.Println(string(result.Stdout)) // "hello world\n"
	fmt.Println(result.ExitCode)       // 0
	fmt.Println(result.Duration)       // e.g. 2.1ms
}

API Reference

Types
Type Description
Command Subprocess configuration: binary, args, dir, env, stdin, grace period
Result Execution output: stdout, stderr, exit code, duration
Config Adapter-level defaults for name, grace period, and timeout
Adapter Wraps Run as a provider.RequestResponse[Command, *Result]
Runner Wraps Run with persistent resilience state (circuit breaker, retry)
SubprocessProvider[I, O] Generic provider that builds a command from input and parses output
Core Function
func Run(ctx context.Context, cmd Command) (*Result, error)

Executes the command, captures output, and returns a *Result (always populated, even on error).

Advanced Usage

Context Cancellation

When the context is cancelled, Run sends SIGTERM to the entire process group. If the process doesn't exit within GracePeriod (default 5s), it escalates to SIGKILL.

ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()

result, err := process.Run(ctx, process.Command{
	Binary:      "long-running-task",
	GracePeriod: 2 * time.Second,
})
// err wraps context.DeadlineExceeded; result.Stdout has partial output
Environment and Working Directory
result, _ := process.Run(ctx, process.Command{
	Binary: "python",
	Args:   []string{"script.py"},
	Dir:    "/opt/scripts",
	Env:    []string{"MODEL=large", "GPU=true"},
})

Extra environment variables are merged with the parent process environment.

Resilient Execution

Use Runner for subprocesses that may fail transiently. Circuit breaker state persists across calls.

runner := process.NewRunner(provider.ResilienceConfig{
	CircuitBreaker: &resilience.CircuitBreakerConfig{
		MaxFailures: 3, Timeout: 30 * time.Second,
	},
})

result, err := runner.Run(ctx, process.Command{Binary: "flaky-tool"})
Generic Provider

Convert any command-line tool into a typed provider:

p := process.NewSubprocessProvider[string, []Segment](
	"diarizer",
	func(audioPath string) process.Command {
		return process.Command{Binary: "python", Args: []string{"diarize.py", audioPath}}
	},
	func(result *process.Result) ([]Segment, error) {
		var segments []Segment
		return segments, json.Unmarshal(result.Stdout, &segments)
	},
)

segments, err := p.Execute(ctx, "audio.wav")

Testing

cd process
go test -race ./...

Contributing

Please refer to the root CONTRIBUTING.md for guidelines.

Documentation

Overview

Package process provides subprocess execution with context cancellation, signal handling, and structured output capture.

Run executes a command and waits for it, capturing stdout and stderr into bounded buffers; Stream observes output live through a callback while the process runs. Both classify a failure to spawn via SpawnError (a missing executable becomes NotFound, a permission failure Forbidden, anything else Internal) and record timeout or cancellation on the returned Result, whose Check reports the outcome as a typed error.

IO modes select how standard streams are wired: captured (the default), observed (Stream), or inherited from the parent terminal, with a stdin policy of closed, provided bytes, or inherited. LifecyclePolicy governs process-group isolation and graceful-termination escalation for every spawn.

Supervisor tracks live children and tears them all down on Shutdown, escalating to a force kill after the grace period. StartPersistent runs a long-lived subprocess with readiness detection (immediate, on an output marker, or after a delay) and graceful shutdown; a startup failure carries a machine-readable classification retrievable via StartErrorKind. The InterruptGroup, TerminateGroup, and KillGroup helpers signal a command's process group.

Pseudoterminal (PTY) execution is intentionally not provided here: it is a heavy, Unix-only capability that would pull a platform dependency into the root module.

Index

Constants

View Source
const DefaultGracePeriod = 5 * time.Second

DefaultGracePeriod is how long Run, Stream, and supervised shutdown wait after graceful termination before escalating to SIGKILL.

Variables

This section is empty.

Functions

func ConfigureSysProcAttr

func ConfigureSysProcAttr(c *exec.Cmd)

ConfigureSysProcAttr places the child in its own process group so we can signal the entire tree on cancellation. No-op on platforms (such as Windows) that do not support process groups.

func InterruptGroup

func InterruptGroup(c *exec.Cmd) error

InterruptGroup asks the command's process to stop gracefully. On Unix it sends SIGINT to the child's process group so descendants are signaled too; on platforms without process groups it falls back to signaling the immediate process. It is a no-op when the process has not started.

func KillGroup

func KillGroup(c *exec.Cmd) error

KillGroup force-kills the process. On Unix it sends SIGKILL to the child's process group; on other platforms it falls back to Process.Kill. It is a no-op when the process has not started.

func SpawnError

func SpawnError(context string, err error) error

SpawnError converts a subprocess spawn failure into a typed AppError.

The OS detail is kept as the visible "<context>: <error>" message, while the error code is classified from the underlying failure: a missing executable becomes a NotFound error, a permission-denied failure becomes Forbidden, and anything else becomes Internal. Callers can then tell "not installed" apart from other spawn failures instead of seeing every failure collapse into a generic error. The original error is preserved as the cause so the underlying chain survives errors.Is/As.

func TerminateGracefully

func TerminateGracefully(c *exec.Cmd) error

TerminateGracefully sends SIGTERM to the child's process group so any grandchildren are also signaled. Callers should set cmd.WaitDelay so the runtime escalates to SIGKILL if the child does not exit in time. On Windows this falls back to os.Process.Kill.

func TerminateGroup

func TerminateGroup(c *exec.Cmd) error

TerminateGroup requests graceful termination. On Unix it sends SIGTERM to the child's process group; on other platforms it falls back to Process.Kill or returns an honest error. It is a no-op when the process has not started.

Types

type Adapter added in v0.1.4

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

Adapter wraps subprocess execution as a provider.RequestResponse.

func NewAdapter added in v0.1.4

func NewAdapter(cfg Config) *Adapter

NewAdapter creates a new process adapter.

func (*Adapter) Execute added in v0.1.4

func (a *Adapter) Execute(ctx context.Context, cmd Command) (*Result, error)

Execute runs a command (implements provider.RequestResponse[Command, *Result]).

func (*Adapter) IsAvailable added in v0.1.4

func (a *Adapter) IsAvailable(_ context.Context) bool

IsAvailable always returns true for process adapters (implements provider.Provider).

func (*Adapter) Name added in v0.1.4

func (a *Adapter) Name() string

Name returns the adapter name (implements provider.Provider).

func (*Adapter) Run added in v0.1.4

func (a *Adapter) Run(ctx context.Context, cmd Command) (*Result, error)

Run executes a command, applying adapter-level defaults.

type Command

type Command struct {
	// Binary is the executable path or name (resolved via PATH).
	Binary string
	// Args are the command-line arguments.
	Args []string
	// Dir is the working directory. If empty, uses the current directory.
	Dir string
	// Env is additional environment variables (key=value).
	// By default these are merged with the parent environment.
	Env []string
	// ScrubEnv starts from an empty environment instead of inheriting the parent.
	ScrubEnv bool
	// Stdin provides input to the process. May be nil. When set it takes precedence
	// over Input and is fed to the child, then closed.
	Stdin io.Reader
	// Input selects the stdin policy when Stdin is nil: closed (default) or inherited
	// from the parent process.
	Input InputPolicy
	// IO selects how stdout and stderr are wired. The zero value, IOCaptured, pipes both
	// into bounded buffers on Result. IOInherited forwards them to the parent's terminal
	// without capture. Stream always observes output live regardless of this field.
	IO IOMode
	// MaxOutputBytes bounds captured stdout and stderr independently. Zero
	// or negative means unlimited capture.
	MaxOutputBytes int
	// GracePeriod is how long to wait after SIGTERM before SIGKILL. Defaults to 5 seconds if zero.
	// When Lifecycle is set its GracePeriod takes precedence.
	GracePeriod time.Duration
	// Lifecycle configures process-group isolation and shutdown escalation. When nil,
	// Run and Stream apply DefaultLifecyclePolicy with GracePeriod honored as an override.
	Lifecycle *LifecyclePolicy
}

Command configures a subprocess to execute.

type Config added in v0.1.4

type Config struct {
	// Name identifies this adapter instance (used by provider.Provider interface).
	Name string `yaml:"name,omitempty" mapstructure:"name"`
	// GracePeriod is the default grace period for SIGTERM→SIGKILL.
	GracePeriod time.Duration `yaml:"grace_period,omitempty" mapstructure:"grace_period"`
	// Timeout is the default execution timeout. Zero means no timeout.
	Timeout time.Duration `yaml:"timeout,omitempty" mapstructure:"timeout"`
}

Config configures a process adapter.

type IOMode

type IOMode int

IOMode selects how a subprocess's standard streams are wired.

const (
	// IOCaptured pipes stdout and stderr into bounded in-memory buffers exposed on Result.
	// This is the default and matches Run's historical behavior.
	IOCaptured IOMode = iota
	// IOObserved streams stdout and stderr live through a callback while it runs.
	// This mode is realized by Stream; Run treats it as IOCaptured.
	IOObserved
	// IOInherited passes the parent's os.Stdout and os.Stderr straight to the child,
	// so its output goes to the terminal and nothing is captured on Result.
	IOInherited
)

type InputPolicy

type InputPolicy int

InputPolicy selects how a subprocess's standard input is wired.

When Command.Stdin is non-nil it always takes precedence (the Bytes policy in rskit terms): the reader is fed to the child and then closed. Otherwise the policy chooses between a closed stdin and inheriting the parent's stdin.

const (
	// InputClosed leaves the child without stdin. This is the default.
	InputClosed InputPolicy = iota
	// InputInherit passes the parent's os.Stdin to the child.
	InputInherit
)

type LifecyclePolicy

type LifecyclePolicy struct {
	// GracePeriod is how long to wait after graceful termination before kill escalation.
	GracePeriod time.Duration
	// IsolateProcessGroup places the child in a new process group where supported.
	IsolateProcessGroup bool
	// TerminateDescendants targets the whole process group rather than only the immediate child.
	TerminateDescendants bool
	// KillAfterGrace escalates to a force kill after GracePeriod expires. When false, no
	// escalation occurs: Run/Stream leave WaitDelay unset, so a child that ignores graceful
	// termination can block the call until it exits on its own or the context is canceled.
	KillAfterGrace bool
}

LifecyclePolicy governs a spawned child's isolation and shutdown escalation.

When IsolateProcessGroup is set the child is placed in its own process group so termination can target the whole group. When TerminateDescendants is set the group (not just the immediate child) is signaled. KillAfterGrace escalates to SIGKILL once GracePeriod elapses.

func DefaultLifecyclePolicy

func DefaultLifecyclePolicy() LifecyclePolicy

DefaultLifecyclePolicy returns the standard policy: a 5s grace period, process-group isolation, descendant termination, and kill-after-grace escalation all enabled.

type PersistentConfig

type PersistentConfig struct {
	// Readiness selects the readiness strategy. Defaults to ReadyImmediate.
	Readiness PersistentReadiness
	// OutputMarker is the substring awaited when Readiness is ReadyOnOutput.
	OutputMarker string
	// ReadyDelay is the wait applied when Readiness is ReadyAfterDelay.
	ReadyDelay time.Duration
	// ReadinessTimeout bounds how long StartPersistent waits for readiness. Defaults to 30s.
	ReadinessTimeout time.Duration
	// ShutdownGracePeriod is the wait after graceful termination before a force kill. Defaults to 5s.
	ShutdownGracePeriod time.Duration
	// MaxCaptureBytes bounds retained startup output per stream. Zero or negative means unbounded.
	MaxCaptureBytes int
	// Lifecycle configures process-group isolation and shutdown escalation.
	Lifecycle LifecyclePolicy
}

PersistentConfig configures a persistent (long-lived) subprocess.

func DefaultPersistentConfig

func DefaultPersistentConfig() PersistentConfig

DefaultPersistentConfig returns a config that is ready immediately with a 30s readiness timeout, a 5s shutdown grace period, and the default lifecycle policy.

type PersistentProcess

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

PersistentProcess is a running long-lived subprocess with graceful shutdown.

func (*PersistentProcess) Pid

func (p *PersistentProcess) Pid() int

Pid returns the process id, or -1 if the process is not running.

func (*PersistentProcess) Shutdown

Shutdown gracefully stops the persistent process, escalating to a force kill after the grace period, and returns the completed result. It reports AlreadyExited when the process had already ended. The context bounds the graceful wait before escalation.

func (*PersistentProcess) Wait

func (p *PersistentProcess) Wait() error

Wait blocks until the persistent process exits on its own and returns its result.

type PersistentReadiness

type PersistentReadiness int

PersistentReadiness selects how StartPersistent decides a long-lived process is ready.

const (
	// ReadyImmediate treats the process as ready as soon as it is spawned.
	ReadyImmediate PersistentReadiness = iota
	// ReadyOnOutput waits until either output stream contains OutputMarker.
	ReadyOnOutput
	// ReadyAfterDelay waits ReadyDelay after spawn before declaring readiness.
	ReadyAfterDelay
)

type PersistentRun

type PersistentRun struct {
	// Startup is the output captured up to the moment readiness completed.
	Startup PersistentStartup
	// Process is the running persistent process handle.
	Process *PersistentProcess
}

PersistentRun is the result of starting a persistent process: the startup output snapshot and the running process handle.

func StartPersistent

func StartPersistent(ctx context.Context, cmd Command, cfg PersistentConfig) (*PersistentRun, error)

StartPersistent spawns a long-lived subprocess and waits for it to become ready per cfg. On success it returns the startup output snapshot and a handle for waiting or shutting the process down. On failure it tears the process down and returns a classified AppError whose kind is retrievable via StartErrorKind.

type PersistentStartErrorKind

type PersistentStartErrorKind string

PersistentStartErrorKind is a machine-readable classification of why a persistent process failed to start and become ready.

const (
	// PersistentStartSpawnFailed indicates the persistent process could not be spawned.
	PersistentStartSpawnFailed PersistentStartErrorKind = "spawn_failed"
	// PersistentStartReadinessTimedOut indicates the process did not become ready before
	// the readiness timeout elapsed.
	PersistentStartReadinessTimedOut PersistentStartErrorKind = "readiness_timed_out"
	// PersistentStartExitedBeforeReadiness indicates the process exited before it became ready.
	PersistentStartExitedBeforeReadiness PersistentStartErrorKind = "exited_before_readiness"
	// PersistentStartOutputEndedBeforeReadiness indicates the output streams ended before
	// output readiness was observed.
	PersistentStartOutputEndedBeforeReadiness PersistentStartErrorKind = "output_ended_before_readiness"
)

func StartErrorKind

func StartErrorKind(err error) (PersistentStartErrorKind, bool)

StartErrorKind returns the persistent startup failure classification attached to an error, reporting false when the error carries no such classification.

type PersistentStartup

type PersistentStartup struct {
	// Stdout is the stdout captured at the moment readiness completed.
	Stdout []byte
	// StdoutTruncated reports whether startup stdout exceeded MaxCaptureBytes.
	StdoutTruncated bool
	// Stderr is the stderr captured at the moment readiness completed.
	Stderr []byte
	// StderrTruncated reports whether startup stderr exceeded MaxCaptureBytes.
	StderrTruncated bool
	// Duration is the time from spawn until readiness completed.
	Duration time.Duration
}

PersistentStartup holds the output captured while waiting for readiness.

type Result

type Result struct {
	// Stdout is the captured standard output.
	Stdout []byte
	// StdoutTruncated reports whether stdout exceeded MaxOutputBytes.
	StdoutTruncated bool
	// Stderr is the captured standard error.
	Stderr []byte
	// StderrTruncated reports whether stderr exceeded MaxOutputBytes.
	StderrTruncated bool
	// ExitCode is the process exit code. -1 if the process was killed.
	ExitCode int
	// Duration is how long the process ran.
	Duration time.Duration
	// TimedOut reports whether the process was killed because the context deadline was exceeded.
	TimedOut bool
	// Canceled reports whether the process was killed because the context was canceled by the caller.
	Canceled bool
}

Result holds the output and status of a completed subprocess.

func Run

func Run(ctx context.Context, cmd Command) (*Result, error)

Run executes a subprocess and waits for it to complete. If the context is canceled, the process group receives SIGTERM on Unix (or the process is killed on Windows), then the runtime escalates to SIGKILL after the grace period via WaitDelay. When cmd.IO is IOInherited the child's stdout/stderr are wired to the parent terminal and nothing is captured; otherwise stdout/stderr are captured into Result.

func RunWithResilience

func RunWithResilience(ctx context.Context, cmd Command, runner *Runner) (*Result, error)

RunWithResilience is a convenience for one-shot subprocess execution with resilience. For repeated calls where circuit breaker state should persist, use NewRunner instead.

func Stream

func Stream(ctx context.Context, cmd Command, emit func(StreamChunk)) (*Result, error)

Stream executes a subprocess and emits stdout/stderr chunks while it runs. When emit is non-nil, Stream invokes it sequentially from an internal goroutine. The callback should return promptly; a slow callback can still apply backpressure to subprocess pipe reads after the internal buffer fills.

func (*Result) Check

func (r *Result) Check() error

Check verifies the process completed successfully and returns a typed error otherwise. Cancellation and timeout take precedence over the exit code, then a non-zero exit code (or a killed process, reported as exit code -1) yields an internal AppError describing the outcome. It returns nil when the process succeeded.

func (*Result) Success

func (r *Result) Success() bool

Success reports whether the process exited cleanly with exit code 0.

type Runner

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

Runner wraps subprocess execution with persistent resilience state. Use NewRunner to create one, then call Run repeatedly. The circuit breaker state persists across calls — repeated crashes trip the breaker.

func NewRunner

func NewRunner(cfg provider.ResilienceConfig) *Runner

NewRunner creates a Runner with the given resilience config. Nil config fields are skipped. Empty config means Run() calls process.Run directly.

func (*Runner) Run

func (r *Runner) Run(ctx context.Context, cmd Command) (*Result, error)

Run executes a subprocess through the resilience chain.

type ShutdownOutcome

type ShutdownOutcome struct {
	// AlreadyExited reports whether the process had already exited before shutdown was requested.
	AlreadyExited bool
	// Result is the completed process result.
	Result *Result
}

ShutdownOutcome describes how a persistent process ended.

type StreamChunk

type StreamChunk struct {
	Stream StreamName
	Data   []byte
}

StreamChunk is one chunk of subprocess output.

type StreamName

type StreamName string

StreamName identifies a subprocess output stream.

const (
	// StreamStdout identifies standard output chunks.
	StreamStdout StreamName = "stdout"
	// StreamStderr identifies standard error chunks.
	StreamStderr StreamName = "stderr"
)

type SubprocessProvider

type SubprocessProvider[I, O any] struct {
	// contains filtered or unexported fields
}

SubprocessProvider wraps a Command as a provider.RequestResponse. The input function builds a Command from the input, and the output function parses the Result into the desired output type.

func NewSubprocessProvider

func NewSubprocessProvider[I, O any](
	name string,
	buildCmd func(I) Command,
	parseOut func(*Result) (O, error),
) *SubprocessProvider[I, O]

NewSubprocessProvider creates a RequestResponse provider backed by subprocess execution.

func (*SubprocessProvider[I, O]) Execute

func (p *SubprocessProvider[I, O]) Execute(ctx context.Context, input I) (O, error)

func (*SubprocessProvider[I, O]) IsAvailable

func (p *SubprocessProvider[I, O]) IsAvailable(ctx context.Context) bool

func (*SubprocessProvider[I, O]) Name

func (p *SubprocessProvider[I, O]) Name() string

func (*SubprocessProvider[I, O]) WithAvailabilityCheck

func (p *SubprocessProvider[I, O]) WithAvailabilityCheck(fn func(context.Context) bool) *SubprocessProvider[I, O]

WithAvailabilityCheck sets a custom availability check for the provider.

type Supervisor

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

Supervisor tracks live child processes and tears them all down on Shutdown.

A caller registers a started *exec.Cmd with Track (handing reaping to the supervisor) or a bare pid with TrackPid (best-effort, signal-only). Shutdown gracefully terminates every still-tracked child, waits the policy grace period, escalates to SIGKILL when enabled, and drains each to completion. It is safe for concurrent use and idempotent: double cleanup is a no-op. On non-Unix platforms a pid-only child that cannot be signaled yields an honest error from Shutdown rather than a silent success.

func NewSupervisor

func NewSupervisor(policy LifecyclePolicy) *Supervisor

NewSupervisor creates a Supervisor governed by the given lifecycle policy. A zero-value policy is replaced with DefaultLifecyclePolicy.

func (*Supervisor) Len

func (s *Supervisor) Len() int

Len reports the number of children currently tracked.

func (*Supervisor) Release

func (s *Supervisor) Release(handle TrackHandle)

Release removes a tracked child from supervision, for use when the caller has already reaped it. It is a no-op for an unknown or zero handle.

func (*Supervisor) Shutdown

func (s *Supervisor) Shutdown(ctx context.Context, reason string) error

Shutdown terminates every still-tracked child and drains each to completion. It is idempotent and returns the joined error of any child that could not be torn down. The context bounds the whole operation; when it is done, remaining children are force-killed.

func (*Supervisor) Track

func (s *Supervisor) Track(cmd *exec.Cmd) TrackHandle

Track registers a started command for supervised shutdown and hands its reaping to the supervisor. The returned handle can be passed to Release once the caller has observed the child exit on its own. Track returns a zero handle when cmd has not started.

func (*Supervisor) TrackPid

func (s *Supervisor) TrackPid(pid int) TrackHandle

TrackPid registers a bare pid for best-effort supervised shutdown. The supervisor can signal the pid but cannot reap it, so the owner remains responsible for wait/reap.

type TrackHandle

type TrackHandle int

TrackHandle identifies a tracked child so it can be released after normal completion.

Directories

Path Synopsis
Package testutil provides process test fixtures for gokit process tests and downstream users.
Package testutil provides process test fixtures for gokit process tests and downstream users.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL