supervised

package
v0.4.0 Latest Latest
Warning

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

 Go to latest Published: Nov 1, 2026 License: MIT Imports: 16 Imported by: 0

Documentation

Overview

Package supervised runs a command inside a container under a Python supervisor that bounds its run time and output, and that a canceled caller can stop without leaving the command behind. The container needs python3.

Index

Constants

View Source 
const (
	// ShellMarkerDir holds the markers of ShellScript invocations.
	ShellMarkerDir = "/tmp/tc-supervised-sh"
	// ShellStopGrace is how long, in seconds, ShellScript waits after SIGTERM
	// before it sends SIGKILL.
	ShellStopGrace = 2

	// ShellEndPrefix starts the last line of a ShellScript's standard error,
	// which carries the command's exit status; RunShell removes it.
	ShellEndPrefix = "tc-supervised-sh: end "
	// ShellDeadlineLine is the line of standard error before the end line of
	// a ShellScript that stopped its command at its limit; RunShell removes
	// it and reports Stopped.
	ShellDeadlineLine = "tc-supervised-sh: deadline"
)
View Source 
const (
	// MarkerVariable passes the supervisor its invocation's marker path, below
	// MarkerDir; see CancelScript.
	MarkerVariable = "TC_SUPERVISED"
	MarkerDir      = "/tmp/tc-supervised"
	// InputVariable gives the supervisor its input limit in bytes and makes
	// it pass its own standard input to the command; see Script.
	InputVariable = "TC_SUPERVISED_INPUT"
	// CancelTimeout bounds one Cancel.
	CancelTimeout = 10 * time.Second
	// ExecCushion is the default reporting allowance for short control reads.
	// Native command execution supplies its separate startup/report allowance.
	ExecCushion = 250 * time.Millisecond
	// LaunchFailedExit is the supervisor's exit code when it could not start
	// the command because of its program: it does not exist, is not
	// executable or is not a program (ENOENT, EACCES, ENOEXEC and the like).
	LaunchFailedExit = 120
	// LaunchUnavailableExit is the supervisor's exit code when it could not
	// start the command for another reason, such as no free process or
	// memory, or a security policy that refused the exec (EPERM).
	LaunchUnavailableExit = 122
	// CleanupFailedExit is the supervisor's exit code when it could not kill
	// and reap the command's group; it then keeps its marker, so that no
	// cancellation is confirmed and the command may still run.
	CleanupFailedExit = 121
)
View Source 
const CancelScript = `import os,sys,time
` + instanceFunction + `marker=sys.argv[1]; cancel_marker=marker+".cancel"; deadline=time.monotonic()+float(sys.argv[2])
os.makedirs(os.path.dirname(marker),mode=0o700,exist_ok=True)
try:
 fd=os.open(marker,os.O_CREAT|os.O_EXCL|os.O_WRONLY,0o600)
 os.write(fd,b"claim"); os.close(fd)
 sys.exit(0)
except FileExistsError: pass
try:
 with open(marker,"rb") as f: owner=f.read()
except FileNotFoundError: sys.exit(0)
current=instance()
recorded=owner[4:-1] if owner.startswith(b"run ") and owner.endswith(b"\n") else b""
if owner==b"claim" or (recorded and current and recorded!=current): sys.exit(0)
os.close(os.open(cancel_marker,os.O_CREAT|os.O_WRONLY,0o600))
while os.path.exists(marker):
 if time.monotonic()>deadline: sys.exit(3)
 time.sleep(0.01)
try: os.unlink(cancel_marker)
except FileNotFoundError: pass
`

CancelScript cancels one invocation through its marker and signals no process itself. If it claims the marker first, the supervisor has not started, and if Docker starts it later it finds the claim, removes it and starts nothing. A marker that holds its own claim, or that a supervisor of an earlier container instance wrote, needs nothing more: that supervisor and its command ended with their instance, and the file still blocks a late start. Otherwise it writes the cancel marker and waits until the supervisor removes the marker, which it does only after killing and reaping its child's group, or after its child ended on its own. A claim made after the supervisor finished stays as a small file in MarkerDir.

View Source 
const Script = `import base64,errno,json,os,select,signal,subprocess,sys,time
` + instanceFunction + `limits=[int(sys.argv[1]),int(sys.argv[2])]; deadline=time.monotonic()+float(sys.argv[3])
marker=os.environ["TC_SUPERVISED"]; cancel_marker=marker+".cancel"
input_limit=int(os.environ.get("TC_SUPERVISED_INPUT","-1")); source_open=input_limit>=0; pending=bytearray(); received=0; delivered=0; input_eof=False; child_in=None
os.makedirs(os.path.dirname(marker),mode=0o700,exist_ok=True)
try: fd=os.open(marker,os.O_CREAT|os.O_EXCL|os.O_WRONLY,0o600)
except FileExistsError:
 os.unlink(marker)
 sys.exit(0)
os.write(fd,b"run "+instance()+b"\n"); os.close(fd)
try:
 import ctypes
 ctypes.CDLL(None).prctl(36,1,0,0,0)
except Exception: pass
def reap_descendants(group):
 until=time.monotonic()+0.2
 while True:
  try:
   while os.waitpid(-1,os.WNOHANG)[0]: pass
  except ChildProcessError: return
  try: os.killpg(group,0)
  except OSError: return
  if time.monotonic()>=until: return
  time.sleep(0.005)
output=[bytearray(),bytearray()]; clipped=[False,False]; timed_out=False; group_signaled=False; launched=False; cleaned=False
def close_input(eof):
 global child_in,input_eof
 if child_in is None: return
 try: child_in.close()
 except OSError: pass
 child_in=None; input_eof=eof; pending.clear()
def read_input():
 global source_open,received
 data=os.read(0,32768)
 if not data: source_open=False
 else:
  received+=len(data)
  if child_in is not None: pending.extend(data)
 if not source_open and not pending: close_input(True)
def write_input():
 global delivered
 try: written=os.write(child_in.fileno(),pending)
 except BlockingIOError: return
 except BrokenPipeError:
  close_input(False)
  return
 delivered+=written; del pending[:written]
 if not source_open and not pending: close_input(True)
def pump(streams,timeout):
 readable=list(streams)+([0] if source_open else [])
 ready,writable,_=select.select(readable,[child_in] if child_in is not None and pending else [],[],timeout)
 if writable: write_input()
 if 0 in ready: read_input()
 return [stream for stream in ready if stream!=0]
def kill_owned_group():
 global group_signaled
 if group_signaled: return
 try: os.killpg(p.pid,signal.SIGKILL)
 except ProcessLookupError: pass
 group_signaled=True
try:
 try: p=subprocess.Popen(sys.argv[4:],stdin=subprocess.PIPE if source_open else None,stdout=subprocess.PIPE,stderr=subprocess.PIPE,start_new_session=True)
 except OSError as e: sys.exit(120 if e.errno in (errno.ENOENT,errno.EACCES,errno.ENOEXEC,errno.ENOTDIR,errno.EISDIR,errno.ELOOP,errno.ENAMETOOLONG) else 122)
 launched=True
 if source_open:
  child_in=p.stdin; os.set_blocking(child_in.fileno(),False)
 streams={p.stdout:0,p.stderr:1}
 try:
  while streams:
   remaining=deadline-time.monotonic()
   if remaining<=0:
    timed_out=True
    kill_owned_group()
    break
   if os.path.exists(cancel_marker):
    kill_owned_group()
    break
   for stream in (pump(streams,min(remaining,0.1)) if input_limit>=0 else select.select(list(streams),[],[],min(remaining,0.1))[0]):
    data=os.read(stream.fileno(),32768)
    if not data:
     del streams[stream]
     continue
    channel=streams[stream]; room=max(0,limits[channel]-len(output[channel]))
    output[channel].extend(data[:room]); clipped[channel]=clipped[channel] or len(data)>room
  # Observe exit without releasing the child's PID/PGID identity. Reaping or
  # Popen.poll() before group cleanup could signal a subsequently reused group.
  while not group_signaled:
   remaining=deadline-time.monotonic()
   if remaining<=0:
    timed_out=True
    break
   if os.path.exists(cancel_marker):
    break
   if os.waitid(os.P_PID,p.pid,os.WEXITED|os.WNOHANG|os.WNOWAIT) is not None:
    break
   if input_limit>=0: pump((),min(remaining,0.01))
   else: time.sleep(min(remaining,0.01))
  kill_owned_group()
  code=p.wait()
 finally:
  kill_owned_group()
  p.wait()
  reap_descendants(p.pid)
  cleaned=True
  close_input(False)
  p.stdout.close()
  p.stderr.close()
finally:
 if launched and not cleaned: os._exit(121)
 for path in (marker,cancel_marker):
  try: os.unlink(path)
  except FileNotFoundError: pass
report={'schema_version':1,'stdout':base64.b64encode(output[0]).decode(),'stderr':base64.b64encode(output[1]).decode(),'exit_code':code,'timed_out':timed_out,'stdout_truncated':clipped[0],'stderr_truncated':clipped[1]}
if input_limit>=0: report.update(input_received=received,input_delivered=delivered,input_eof=input_eof)
print(json.dumps(report))
`

Script is the supervisor. It runs its arguments after the three limits (stdout bytes, stderr bytes, seconds) as a child in a new process group, bounds each output stream, kills only that group at the deadline, and prints one JSON envelope. Native stdout stays separate from Ceph's independently nonblocking stderr pipe. When it cannot start the command it exits LaunchFailedExit or LaunchUnavailableExit without an envelope, and when it cannot kill or reap the started command's group it exits CleanupFailedExit and keeps its marker.

The supervisor claims its invocation's marker before it starts the child, writing the container instance into it, and stops at once if CancelScript claimed it first. While the child runs, a cancel marker makes the supervisor kill the child's group, which it still owns because the child is not yet reaped. Removing the marker after the reap acknowledges the cancellation.

On Linux the supervisor is a child subreaper: members of the killed group whose parent died become its children, and it reaps them before it reports, for at most 200 ms while members remain. It does not wait for a descendant that left the group.

The marker record ends with a newline, so that a canceler never takes a partly written record for one of another instance.

With InputVariable set to a byte limit, the supervisor gives the command a pipe as its standard input. It reads its own standard input as soon as data arrives, so that a writer never waits for a command that does not read, keeps what the command has not taken yet, and writes it into the pipe as the command reads. At the end of its own input it closes the pipe once everything is written. When the command closes the pipe or ends, the rest is discarded. The envelope then also reports the bytes it received, the bytes it wrote into the pipe and whether the command got the end of input.

View Source 
const ShellCancelScript = `` /* 682-byte string literal not displayed */

ShellCancelScript cancels one ShellScript invocation through its marker, as CancelScript does for Script: it claims a marker that does not exist, confirms one of its own claim or of an earlier container instance, and otherwise writes the cancel marker and waits until the supervisor removes the marker. It exits 3 when that does not happen within its wait.

View Source 
const ShellScript = `marker=$1 limit=$2 grace=$3
shift 3
mkdir -p "${marker%/*}" 2>/dev/null
instance=$(cat /proc/sys/kernel/random/boot_id 2>/dev/null)/$(awk '{sub(/.*\) /, ""); print $20}' /proc/1/stat 2>/dev/null)
if ! (set -C; echo "run $instance" > "$marker") 2>/dev/null; then
    rm -f "$marker"
    exit 0
fi
"$@" &
pid=$!
(
    waited=0
    while kill -0 "$pid" 2>/dev/null; do
        [ -e "$marker.cancel" ] && break
        if [ "$limit" -gt 0 ] && [ "$waited" -ge "$limit" ]; then
            : > "$marker.deadline"
            break
        fi
        sleep 1
        waited=$((waited + 1))
    done
    if kill -TERM "$pid" 2>/dev/null; then
        waited=0
        while [ "$waited" -lt "$grace" ] && kill -0 "$pid" 2>/dev/null; do
            sleep 1
            waited=$((waited + 1))
        done
        kill -KILL "$pid" 2>/dev/null
    fi
) </dev/null >/dev/null 2>&1 &
watcher=$!
wait "$pid" 2>/dev/null
code=$?
kill "$watcher" 2>/dev/null
wait "$watcher" 2>/dev/null
stopped=
[ -e "$marker.deadline" ] && stopped=1
rm -f "$marker.deadline" "$marker"
[ -n "$stopped" ] && echo "` + ShellDeadlineLine + `" >&2
echo "` + ShellEndPrefix + `$code" >&2
exit "$code"
`

ShellScript is a supervisor for containers that have a POSIX shell but no python3, such as the RGW role image: it needs only sh, its kill builtin, sleep, cat, awk, mkdir and rm, and a writable /tmp. Its arguments are the invocation's marker, a limit in seconds (0 for none), the stop grace in seconds and the command.

It follows Script's marker protocol. It claims the marker before it starts the command, writing the container instance into it, and starts nothing if ShellCancelScript claimed it first. A watcher stops the command, with SIGTERM and SIGKILL after the grace, when a cancel marker appears or the limit passes; it holds none of the exec's streams, so a command that ends returns at once. Removing the marker after the command was reaped acknowledges a cancellation. The command is stopped by its PID: dash gives a background job no process group of its own without a terminal, so a command that starts processes of its own may leave them behind.

Both waits discard the shell's own standard error: bash, the sh of the official image, reports a job that a signal ended there.

Its last line of standard error is ShellEndPrefix and the command's exit status, after ShellDeadlineLine when the limit stopped it. A run without that line did not finish: a signal killed the supervisor.

Variables

View Source 
var ErrInputEnded = errors.New("the command's input has ended")

ErrInputEnded reports a write that the command's input did not take: the command ended, or its input stream failed. Part of the data may have been written before.

View Source 
var ErrInputLimit = errors.New("the write would exceed the input limit")

ErrInputLimit refuses a write that would take the input past its limit; nothing of it is written, and the input stays open.

Functions

func Cancel 

func Cancel(ctx context.Context, ctr testcontainers.Container, label, marker string) error

Cancel stops the invocation that owns marker in ctr and returns once it can no longer run its command. It uses its own bounded context, because the caller's may have ended.

func CancelShell added in v0.4.0 

func CancelShell(ctx context.Context, ctr testcontainers.Container, label, marker string) error

CancelShell stops the ShellScript invocation that owns marker in ctr and returns once it can no longer run its command. It uses its own bounded context, because the caller's may have ended.

func Unconfirmed 

func Unconfirmed(err error) (string, bool)

Unconfirmed returns the marker of the invocation whose end err leaves unconfirmed: Cancel could not stop it, or a signal killed its supervisor before its cleanup. Cancel with the marker confirms it later.

Types

type Attacher added in v0.3.0 

type Attacher interface {
	AttachExec(ctx context.Context, cmd, env []string) (Attachment, error)
}

Attacher starts an exec with its standard input attached. A container that implements it runs RunInput's supervisor itself; for any other the Docker engine runs it, in the container with the same ID.

type Attachment added in v0.3.0 

type Attachment struct {
	// Input writes to the exec's standard input; CloseWrite ends it, and
	// Close closes the whole connection, which unblocks Input and Output.
	Input interface {
		io.Writer
		CloseWrite() error
		Close() error
	}
	// Output is the exec's stdout and stderr in Docker's multiplexed stream
	// format; it ends when the exec has ended.
	Output io.Reader
	// ExitCode returns the exit code of the exec after its output ended.
	ExitCode func(context.Context) (int, error)
	// Release frees what the attachment holds, after Close.
	Release func()
}

Attachment is one started exec whose standard streams the caller holds.

type BackendError 

type BackendError struct {
	Op  string
	Err error
}

BackendError is an error of Docker or of a container implementation during Op. Its text holds the cause's text, which this package does not control.

func (*BackendError) Error 

func (e *BackendError) Error() string

func (*BackendError) Unwrap 

func (e *BackendError) Unwrap() error

type ExitError 

type ExitError struct {
	Label     string
	Code      int
	Oversized bool
	Marker    string
}

ExitError reports a supervisor that exited without a report. Code 126 or 127 means that the container could not run python3, and LaunchFailedExit or LaunchUnavailableExit that the supervisor could not start the command, because of its program or for another reason. A code above 128 means that a signal killed the supervisor before its cleanup, and CleanupFailedExit that its cleanup failed; then its command may still run, and Cancel with Marker can confirm that only after the container restarted.

func (*ExitError) Error 

func (e *ExitError) Error() string

type Input added in v0.3.0 

type Input struct {
	// Limit bounds the bytes written to the command in all.
	Limit int64
	// Writes carries the data to write, in order. Each write is taken whole
	// or refused whole with ErrInputLimit.
	Writes <-chan InputWrite
	// Closed ends the input when it is closed: the command gets the end of
	// input after everything written before.
	Closed <-chan struct{}
	// Finished is closed once RunInput takes no more writes.
	Finished chan<- struct{}
}

Input is the standard input of one RunInput call.

type InputWrite added in v0.3.0 

type InputWrite struct {
	Data []byte
	Done chan<- error
}

InputWrite is one write to an Input. RunInput owns Data once it took the write, and sends its outcome to Done, which must have room for it.

type NotStoppedError 

type NotStoppedError struct {
	Marker string
	Err    error
}

NotStoppedError reports an invocation that Command could not confirm as stopped after Exec ended early: its command may still run in the container. A later Cancel with Marker that succeeds confirms that it no longer can.

func (*NotStoppedError) Error 

func (e *NotStoppedError) Error() string

func (*NotStoppedError) Unwrap 

func (e *NotStoppedError) Unwrap() error

type Report 

type Report struct {
	Stdout, Stderr []byte
	ExitCode       int
	// TimedOut means that the supervisor killed the command's process group
	// at its deadline; the output is what the command wrote until then.
	TimedOut bool
	// StdoutTruncated and StderrTruncated mean that a stream exceeded its
	// limit; the output holds its first limit bytes.
	StdoutTruncated, StderrTruncated bool
	// InputWritten, InputDelivered and InputEOF describe the input of a
	// RunInput call: the bytes written to the supervisor, the bytes that it
	// wrote into the command's standard input, and whether the command got
	// the end of input after all of them.
	InputWritten, InputDelivered int64
	InputEOF                     bool
}

Report is the supervisor's complete report of one run.

func Run 

func Run(ctx context.Context, ctr testcontainers.Container, label string, processLimit time.Duration, stdoutLimit, stderrLimit int, argv ...string) (Report, error)

Run runs argv in ctr under Script, which kills the command's process group at its deadline and bounds each output stream. The deadline is processLimit, shortened to the caller's deadline. When Exec ends before the supervisor does, which includes a manual cancellation, Cancel stops this invocation inside the container before Run returns, whether its supervisor is running or has not started yet; when Cancel fails, the error holds a NotStoppedError. label names the command in errors. Run returns the supervisor's report, also for a run that timed out or was truncated, and an error only when it has no complete report.

func RunInput added in v0.3.0 

func RunInput(ctx context.Context, ctr testcontainers.Container, label string, processLimit time.Duration, stdoutLimit, stderrLimit int, input Input, argv ...string) (Report, error)

RunInput runs argv like Run, with input as its standard input. The supervisor reads what is written at once, up to input.Limit, and passes it on as the command reads; see Script. Writes end when the command ends or RunInput returns, and RunInput returns only after it stopped taking writes and its input and output work ended.

The exec runs through the target's Attacher, or the Docker engine. When the run ends before the supervisor does, RunInput closes the connection and Cancel stops the invocation inside the container, as in Run. A supervisor that a signal killed, or whose cleanup failed, leaves its invocation unconfirmed (see Unconfirmed) also when the run ended meanwhile.

func RunInputWithExecCushion added in v0.4.0 

func RunInputWithExecCushion(ctx context.Context, ctr testcontainers.Container, label string, processLimit, execCushion time.Duration, stdoutLimit, stderrLimit int, input Input, argv ...string) (Report, error)

RunInputWithExecCushion is RunInput with the startup/report allowance of RunWithExecCushion. The caller's deadline still bounds the whole attempt.

func RunWithExecCushion added in v0.4.0 

func RunWithExecCushion(ctx context.Context, ctr testcontainers.Container, label string, processLimit, execCushion time.Duration, stdoutLimit, stderrLimit int, argv ...string) (Report, error)

RunWithExecCushion runs like Run with a separate allowance for Docker exec startup, interpreter initialization and returning the timeout report. It does not lengthen processLimit or a shorter caller deadline. Control reads keep Run's small default so their enclosing retry budgets remain available.

type Result 

type Result struct {
	Stdout, Stderr []byte
	ExitCode       int
}

Result is the complete output of one supervised command.

func Command 

func Command(ctx context.Context, ctr testcontainers.Container, label string, processLimit time.Duration, stdoutLimit, stderrLimit int, argv ...string) (Result, error)

Command runs argv like Run and returns its result only for a run that completed within its deadline and output limits; a timed-out or truncated run is an error. A nonzero exit is returned in the result.

type ShellReport added in v0.4.0 

type ShellReport struct {
	Stdout, Stderr []byte
	ExitCode       int
	// Stopped means that the limit passed and the supervisor stopped the
	// command.
	Stopped bool
}

ShellReport is the outcome of one RunShell that the supervisor finished.

func RunShell added in v0.4.0 

func RunShell(ctx context.Context, ctr testcontainers.Container, label string, limit int, argv ...string) (ShellReport, error)

RunShell runs argv in ctr under ShellScript with a limit in seconds (0 for none). When the call ends before the supervisor reports, because ctx ended or the exec failed, CancelShell stops the invocation inside the container before RunShell returns, also when Docker starts it only later; when that fails, the error holds a NotStoppedError, and a later CancelShell with its marker can confirm the stop. label names the command in errors.

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.