Documentation
¶
Overview ¶
Package agent runs steps on behalf of an engine it does not trust.
The other half of internal/execution/remote. The engine hands over a command and streams the result back; this is what is on the host.
It was written second, on purpose ¶
The executor exists and its contract is pinned by thirteen tests against a fake. Writing the agent first would have made it the specification by accident -- the contract would then be whatever that program happened to do, and the engine would have grown to match its quirks. Everything here is written to satisfy internal/execution/remote/protocol.go, and the test that matters puts the real engine and the real agent on a socket together.
What it refuses to be ¶
Not an orchestrator. It knows nothing of workflows, dependencies, schedules or retries, exactly as the TaskExec comment says of every executor. It takes a command, runs it, and reports what happened.
Not a secret store. It RESOLVES secrets from one -- a directory of files, the same shape the kubelet mounts and Docker uses -- under an allowlist that belongs to this host and not to the engine's cluster. The engine never sends a value and never learns one.
Index ¶
- Variables
- type Agent
- func (a *Agent) Advertise(addr string)
- func (a *Agent) Cancel(execID string) error
- func (a *Agent) Drain()
- func (a *Agent) Handler(log *slog.Logger) http.Handler
- func (a *Agent) Restore(log *slog.Logger) error
- func (a *Agent) Resume(execID string, after int64) (io.ReadCloser, error)
- func (a *Agent) Serve(srv *http.Server, ln net.Listener, stop, force <-chan struct{}, ...) (int, error)
- func (a *Agent) Start(ctx context.Context, req remote.StartRequest) (io.ReadCloser, error)
- func (a *Agent) Wait(ctx context.Context) error
- type Options
Constants ¶
This section is empty.
Variables ¶
var ErrDraining = errors.New("this agent is draining: it finishes the steps it " +
"has and takes no new one")
ErrDraining is a start refused because this agent is on its way out (#136).
Typed, because it is the one refusal that says nothing about the request: the same step would run on another agent, or on this one's replacement. The handler answers it 503 rather than 400 for that reason.
Functions ¶
This section is empty.
Types ¶
type Agent ¶
type Agent struct {
// contains filtered or unexported fields
}
Agent holds the running executions.
func (*Agent) Advertise ¶
Advertise sets where the engine should come back to for this agent's executions, after construction.
After, because an address is not always known at construction: a test starts the server to learn its address, and a pod may be told by an init step. It is a plain setter under the same lock the runs map uses, so a handler serving a request while this is called reads one value or the other and never a torn one.
func (*Agent) Cancel ¶
Cancel stops a step.
The process GROUP is signalled, not the process: the shell spawned whatever the command is, and killing only the shell leaves it running.
func (*Agent) Drain ¶
func (a *Agent) Drain()
Drain stops this agent taking new steps.
It is the first half of a shutdown and only that: the steps already running go on, and so do their resumes and cancels -- an engine that restarted in the same rollout comes back for a step it handed over, and a drain that refused it would be waiting for a step nobody can read. /health turns 503, so a balancer in front of a pool stops sending starts here.
What it will wait for is fixed HERE, under the lock the reservation takes: every execution in `runs` at this instant -- one still being set up included -- and nothing after, because nothing after is let in.
Idempotent: a second signal must not undo or repeat the first.
func (*Agent) Handler ¶
Handler serves the three routes the engine calls, and one it does not.
Thin on purpose: everything that decides behavior is in the Agent, so the tests that matter drive it directly and this file is transport. The route names are remote.HTTPAgent's, and the two are the only pair that has to agree -- which is why a test puts them on a socket together rather than trusting that they do.
The fourth is for the kubelet. It is the only one on `free`'s list, and the only GET.
func (*Agent) Restore ¶
Restore reads back what the state directory holds, before the agent serves.
It also adopts log for what the agent has to say about that directory later, when a write fails. It returns an error only when the directory itself cannot be read; a file that is wrong is said and removed, never a reason not to start.
func (*Agent) Resume ¶
Resume serves the same execution from `after`.
A gap is answered rather than papered over: when the ring no longer reaches back that far the engine is told what it wanted and what is available, and it fails the step. Serving from the oldest line instead would hand back a stream with a hole in the middle and no way to know how big.
func (*Agent) Serve ¶
func (a *Agent) Serve(srv *http.Server, ln net.Listener, stop, force <-chan struct{}, drainTimeout time.Duration, log *slog.Logger) (int, error)
Serve runs srv on ln until stop closes, then drains, and returns when the listener is closed -- with how many steps it gave up on: canceled while running, or ended with an exit nobody read. main exits non-zero on any.
HERE and not in main, for the reason Options.Check is: a shutdown that only exists inside `func main` is a shutdown nothing tests, and the one main had logged "running steps keep their streams" and then exited with them (#136).
The server stays up through the drain. srv.Shutdown first would close the listener, and the resume the drain is waiting for -- from an engine that restarted in the same rollout -- would find nothing to connect to. It would also wait on the wrong thing: connections, while a step does not live on one.
The drain ends when every step running at stop has ended and been read (Wait). A drainTimeout of zero is no limit, the orchestrator's grace period being the real one; force closing -- a second signal -- ends it as the timeout would. Either way what is left is canceled, the whole process group as /cancel does: a step left behind by an agent that exited is out of anybody's reach.
func (*Agent) Start ¶
func (a *Agent) Start(ctx context.Context, req remote.StartRequest) (io.ReadCloser, error)
Start runs a step and returns a reader over its stream.
The request is validated BEFORE anything is spawned: a refused step must leave no process behind, and an unresolvable secret must be a refusal rather than a command running with a variable silently unset.
func (*Agent) Wait ¶
Wait returns when every step running at Drain has ended AND been read to its exit, or when ctx ends -- naming, then, what it was still waiting for.
Ended is not enough. A step that ends while the engine is away keeps its exit only in this process's memory; leaving then loses it, and the engine coming back is answered "unknown execution" and starts the step AGAIN under the same id (#125). So the drain waits for somebody to read it: the engine's resume, which is what the rollout that restarts it is about to send.
A step that was canceled, or abandoned before it spawned, is not waited for: whoever canceled already knows how it ends, and an abandoned start was answered with its refusal.
type Options ¶
type Options struct {
// Token authenticates the engine. Empty is REFUSED unless InsecureNoToken
// says so, because an agent with no token runs any command for any caller.
//
// One token for every engine, and the gap is stated rather than
// discovered: no revoking one without changing them all, and no per-engine
// identity in the audit trail. See the package's README for what a second
// version would need.
Token string
// InsecureNoToken allows an agent with no token at all.
//
// It exists so the dangerous choice is a DECLARATION and not an omission.
// Without a token this process runs any command anybody who reaches the
// port sends it, as whatever user it is -- not a weak password, no
// password. That used to be allowed with a warning at startup, which is
// the shape of a hole nobody sees: the warning scrolls past and the
// manifest that caused it says nothing at all.
//
// Now it has to be written down. It shows up in `ps`, in the manifest, and
// in a review, which is the whole of what this field buys.
InsecureNoToken bool
// SecretsDir is the root of the secret store: `<dir>/<name>/<key>` holds
// one value, which is how the kubelet mounts a Secret and how Docker mounts
// one. Empty means this host serves no secrets at all, and a step that asks
// for one is refused naming the coordinate.
SecretsDir string
// AllowedSecrets limits which secret NAMES a step may ask for.
//
// The same division the pod executor has, with the boundary moved: the
// installation says which secrets exist, the workflow says which step gets
// which. Empty denies everything -- a host somebody else administers should
// not hand over its store because an engine asked nicely.
AllowedSecrets []string
// Advertise is the address the engine should come back to for this
// agent's own executions: `https://dbt-runner-2.dbt.svc:9443`.
//
// It matters only behind a SHARED address. An agent is stateful per
// execution -- a ring and a process handle in this process's memory -- so
// a resume or a cancel that reaches a different replica is answered with
// "not running here", and the engine fails a step that is still running
// over here. See remote.HeaderInstance.
//
// TOLD, not discovered. This process can read its hostname; it cannot know
// which of its names the engine can route to, and a guess that resolves
// nowhere is worse than no answer -- the engine would address an instance
// that does not exist instead of falling back to the address it has.
//
// Empty is the ordinary case and changes nothing: one agent at one address.
Advertise string
// WorkDir is where a step runs when it names no directory of its own.
WorkDir string
// RingSize is how many lines are kept for a resumed connection.
//
// Bounded on purpose, and the consequence is stated rather than implied:
// this many lines of network outage survive, and beyond it the engine is
// told the gap is unrecoverable and fails the step. A step missing an
// unknown number of lines is worse than a step that failed.
RingSize int
// AliveEvery is how often a running step emits a heartbeat. Zero takes the
// protocol's default.
AliveEvery time.Duration
// StateDir is what survives this process restarting: the execID -> pid
// map, so cancel still reaches a running step, and the finished
// executions, so a resume is still answered (#137). Empty keeps both in
// memory only: cancel is then best-effort across a restart, and a
// finished execution is forgotten by one.
StateDir string
// Shell runs the command. Empty is `/bin/sh -c`, matching the local
// executor: `run:` in the YAML is a shell line, not an argv.
Shell []string
// KeepFinished is how long an execution stays resumable after its process
// exits, so a scheduler that was down while a step ended can still read
// how it ended (#125). Zero takes defaultKeepFinished.
KeepFinished time.Duration
// KeepFinishedMax bounds how many finished executions are kept, each with
// its ring; past it the oldest goes first. Zero takes
// defaultKeepFinishedMax.
KeepFinishedMax int
// Now is the clock retention is measured on. Nil is time.Now; a test sets
// it so an hour of retention takes no hour.
Now func() time.Time
}
Options configure one agent.
Source Files