Help us shape the future of Go! Take the 2026 Go Developer Survey

execstream

package
v0.5.0 Latest Latest
Warning

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

Go to latest
Published: Oct 18, 2026 License: MIT Imports: 14 Imported by: 0

Documentation

Overview

Package execstream is the framed protobuf stream between the host process of ermite-runner and the executor running inside a sandbox.

Wire format: each frame is a 4-byte big-endian length followed by that many bytes of a serialized execpb.Frame (see exec.proto). A frame is at most MaxFrame bytes; longer output is split by the sender. The stream is transport-agnostic: vsock in a micro-VM, a Unix socket in a container.

Index

Constants

View Source
const ChunkSize = 32 * 1024

ChunkSize is the largest output payload placed in one Output frame.

View Source
const MaxFrame = 2 << 20

MaxFrame is the largest accepted serialized frame. It leaves room for the largest job result: outputs of 1 MiB in total (the limit of ermite) plus the overhead of the encoding.

View Source
const MaxLogLine = 64 * 1024

MaxLogLine is the largest piece of a log line that fits an AppendLog line (64 KiB); longer lines are cut by SplitLine.

Variables

View Source
var ErrFrameTooLarge = errors.New("execstream: frame too large")

ErrFrameTooLarge is returned for a frame above MaxFrame, on either side.

Functions

func Drive

func Drive(rw io.ReadWriter, run *execpb.Run, cancel <-chan *execpb.Cancel, onOutput func(*execpb.Output)) (*execpb.Hello, *execpb.Exit, error)

Drive is the host side of a session: it waits for the executor's Hello, sends run, passes every Output frame to onOutput, and returns the Exit frame.

cancel, when non-nil, asks the executor to stop the script: the first value received is sent as one Cancel frame (the executor escalates SIGINT, SIGTERM, SIGKILL by itself); closing the channel without a value sends nothing. It is deliberately not a context: cancelling the session's context would also abort the reads that deliver the executor's Exit. Drive keeps waiting for Exit after a Cancel; the caller bounds that wait by stopping the sandbox, which closes the stream and makes Drive return an error.

func DriveJob

func DriveJob(rw io.ReadWriter, run *execpb.Run, cancel <-chan *execpb.Cancel, h JobHandler) (*execpb.Hello, *execpb.Exit, error)

DriveJob is Drive for a session that carries a job: besides Output frames it delivers the step, log and result frames to h.

func SplitLine

func SplitLine(text string, max int) []string

SplitLine cuts text into pieces of at most max bytes at UTF-8 character boundaries. The result has one piece for a short line, and always at least one (an empty line is one empty piece). Invalid UTF-8 in text is replaced with U+FFFD first, because the protocol only carries valid text.

Types

type Conn

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

Conn reads and writes frames on a byte stream. Send is safe for concurrent use; Recv must be called from one goroutine.

func NewConn

func NewConn(rw io.ReadWriter) *Conn

NewConn wraps a byte stream.

func (*Conn) Recv

func (c *Conn) Recv() (*execpb.Frame, error)

Recv reads one frame. It returns io.EOF only on a clean end of stream between frames; a stream cut inside a frame yields io.ErrUnexpectedEOF.

func (*Conn) Send

func (c *Conn) Send(f *execpb.Frame) error

Send writes one frame.

type JobHandler

type JobHandler struct {
	Hello         func(*execpb.Hello)
	Output        func(*execpb.Output)
	StepStarted   func(*execpb.StepStarted)
	StepCompleted func(*execpb.StepCompleted)
	Log           func(*execpb.Log)
	JobResult     func(*execpb.JobResult)
	// Resolve answers the executor's ResolveRequest frames, one at a time and in
	// order, from a single worker (the call goes to the server); the answer is
	// sent as a ResolveResponse frame carrying the id of the request. ctx ends
	// with the session: a call still running is cancelled and DriveJob waits for
	// it before it returns. At most resolveQueue requests wait; the next ones are
	// answered with an error at once, without a call.
	Resolve func(ctx context.Context, r *execpb.ResolveRequest) *execpb.ResolveResponse
	// StepAnnotations and StepSummary come before the StepCompleted of their
	// step.
	StepAnnotations func(*execpb.StepAnnotations)
	StepSummary     func(*execpb.StepSummary)
	// Secrets and Vars (Job.secrets, Job.vars) are sent before the Run frame,
	// one NamedValue frame each, so that their total size is not bound by the
	// size of one frame (MaxFrame). A single value must fit in a frame.
	Secrets, Vars map[string]string
}

JobHandler receives the frames of a session. Nil functions ignore their frames. The functions are called from the goroutine of DriveJob, in the order of the stream.

Directories

Path Synopsis

Jump to

Keyboard shortcuts

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