Documentation
¶
Overview ¶
Package nodes owns node identity and the node stream: enrollment, the in-memory session registry, heartbeat bookkeeping and the event batches nodes push back.
Index ¶
- Constants
- Variables
- type Ingestor
- type Registry
- func (r *Registry) CloseAll()
- func (r *Registry) Drain(nodeID string, draining bool) bool
- func (r *Registry) Forget(nodeID, taskID string)
- func (r *Registry) Get(nodeID string) (*Session, bool)
- func (r *Registry) Release(taskID string)
- func (r *Registry) SetLabels(nodeID string, labels []string) bool
- func (r *Registry) Snapshot() []Snapshot
- func (r *Registry) SnapshotOf(nodeID string) (Snapshot, bool)
- type Service
- func (s *Service) Assign(ctx context.Context, nodeID string, a *podiumv1.Assign, cost TaskCost) error
- func (s *Service) Cancel(ctx context.Context, nodeID, taskID, reason string) error
- func (s *Service) Candidates() []Snapshot
- func (s *Service) Enroll(ctx context.Context, req *connect.Request[podiumv1.EnrollRequest]) (*connect.Response[podiumv1.EnrollResponse], error)
- func (s *Service) Inject(ctx context.Context, nodeID, taskID, text string) error
- func (s *Service) LoseTask(ctx context.Context, task store.Task, node store.Node, reason string)
- func (s *Service) Registry() *Registry
- func (s *Service) Release(taskID string)
- func (s *Service) RunWatchdog(ctx context.Context) error
- func (s *Service) SetAllowUntaggedNodes(v bool)
- func (s *Service) SetArtifacts(a *artifacts.Service)
- func (s *Service) SetDrain(ctx context.Context, nodeID string, draining bool) error
- func (s *Service) SetSlots(ctx context.Context, nodeID string, maxTasks int32) error
- func (s *Service) SetWatchdog(w Watchdog)
- func (s *Service) Stream(ctx context.Context, ...) error
- func (s *Service) UploadArtifact(ctx context.Context, ...) (*connect.Response[podiumv1.UploadArtifactResponse], error)
- type Session
- type Snapshot
- type TaskCost
- type Watchdog
Constants ¶
const CancelReasonNotOurs = "reconciliation: the control plane no longer holds this task on this node"
CancelReasonNotOurs is what a node is told when it reports a container the control plane no longer holds. It is deliberately explicit: the operator reading the node's log should not have to guess why a container they can see was torn down.
Variables ¶
var ErrNoSession = errors.New("nodes: node has no live stream")
ErrNoSession is returned when a node has no live stream at all.
var ErrSessionClosed = errors.New("nodes: session closed")
ErrSessionClosed is returned when a message is sent to a node whose stream has ended.
Functions ¶
This section is empty.
Types ¶
type Ingestor ¶
type Ingestor interface {
Ingest(ctx context.Context, taskID string, batch []*podiumv1.TaskEvent) (uint64, error)
// Note records something the control plane observed about a task rather than
// something the node reported — losing the node being the case that matters.
Note(ctx context.Context, taskID, message string) error
}
Ingestor is the logs service, as much of it as the node stream needs.
type Registry ¶
type Registry struct {
// contains filtered or unexported fields
}
Registry holds the live node sessions of this server process. It is deliberately in memory and single-process: a node's stream terminates on one server, so that server is the only one that can reach it.
func (*Registry) CloseAll ¶
func (r *Registry) CloseAll()
CloseAll ends every session. Graceful shutdown calls it so node stream handlers return before the HTTP server waits on them; the nodes reconnect to whoever comes back up.
func (*Registry) Drain ¶
Drain marks a node's live session draining (or undraining) and reports whether it had one.
func (*Registry) Forget ¶
Forget drops a task from whichever session claims it, without crediting a slot back. It is what reconciliation does with a container the control plane no longer owns.
func (*Registry) Release ¶
Release frees the slot whichever session is holding taskID booked for it. A task no session holds is a no-op: replayed terminal events are routine, and so is a node that reconnected and rebuilt its running set from Hello in the meantime.
func (*Registry) SetLabels ¶
SetLabels replaces a node's live labels and reports whether it had a session. A node that is not connected needs nothing: the next stream reads the labels off its row.
type Service ¶
type Service struct {
// contains filtered or unexported fields
}
Service implements podium.v1.NodeService.
func NewService ¶
NewService returns the node-facing service and its (empty) session registry.
func (*Service) Assign ¶
func (s *Service) Assign(ctx context.Context, nodeID string, a *podiumv1.Assign, cost TaskCost) error
Assign hands a task to a connected node and charges the task's cost against the node's budget, so the scheduler's next tick sees the capacity go rather than waiting for a heartbeat up to ten seconds away. Every log statement that touches an Assign goes through podiumv1.RedactForLog; nothing else may log one.
func (*Service) Cancel ¶
Cancel asks a node to stop a task. It does not wait: the container only dies once the node has run its SIGTERM grace period, and the terminal status lands with the exited/finished events.
func (*Service) Candidates ¶
Candidates is what the scheduler matches against.
func (*Service) Enroll ¶
func (s *Service) Enroll( ctx context.Context, req *connect.Request[podiumv1.EnrollRequest], ) (*connect.Response[podiumv1.EnrollResponse], error)
Enroll exchanges a single-use enrollment token for a node identity. The node key is 32 random bytes returned exactly once; only its SHA-256 reaches Postgres.
func (*Service) Inject ¶
Inject delivers one human message into a running task on a node. It does not wait: the node writes the text to the task's inbox, and whatever is blocked on a read — an interactive playbook's ask tool — picks it up.
func (*Service) LoseTask ¶
LoseTask applies the node-loss policy to one task: a new attempt when the spec says the task may be re-run and the budget allows, otherwise `lost` with an explanation.
`lost` is not `failed`, and the distinction is the point of it: the task did not do anything wrong, the machine it was on went away, and an operator deciding whether to re-run it needs to be told which of those happened. Every path through here first writes a synthetic event, so the task's own log says so too.
It goes through TransitionTask with an explicit `from`, so a heartbeat expiring at the same moment a late `finished` event lands is a rejected transition — a no-op — rather than a task being dragged out of a terminal state.
func (*Service) Registry ¶
Registry is the live session registry, which the scheduler and the admin API read.
func (*Service) Release ¶
Release gives back the slot and the resources a node was holding for a task, on whichever session claims it. The event path already does this for a task that reaches a terminal status; this is for the ones that never get there — an assignment revoked before the node accepted it, or a cancelled task the server wrote off without its node. Without it the session keeps a phantom entry until the node reconnects, and reports both a running task that is not running and less free CPU and memory than it has.
func (*Service) RunWatchdog ¶
RunWatchdog sweeps node health until ctx is cancelled: a node that has stopped heartbeating becomes unreachable, then offline, and an offline node's tasks are requeued or lost.
It is deliberately driven by the stored heartbeat rather than by the session registry. A session ending already marks the node unreachable, but a node whose TCP connection is wedged open holds a session it is not using, and only the heartbeat clock notices that.
func (*Service) SetAllowUntaggedNodes ¶
SetAllowUntaggedNodes lets an untagged tailnet device enroll. It is a setter because the server reads the flag from its own configuration, which nodes must not import.
func (*Service) SetArtifacts ¶
SetArtifacts wires in the object store. It is a setter because server.New builds the node service before it knows whether an object store is configured.
func (*Service) SetDrain ¶
SetDrain tells a connected node to stop accepting work, or to start again. The stored nodes.draining column is the durable half and is the caller's job; this is what makes the live session act on it immediately.
A node with no live session is not an error: draining an offline node is a legitimate thing to do before it comes back, and the column is what it reads when it does.
func (*Service) SetSlots ¶
SetSlots tells a connected node how many tasks it may run at once, 0 for "back to your own max_tasks". The durable half is nodes.max_tasks_override, written by the caller: this is only the live half, and a node with no session picks the number up from its next HelloAck.
Both halves are needed and neither is enough. The node is the only thing that can enforce a budget — it rejects an assignment it has no slot for — so a raise that never reaches it would leave the scheduler assigning work the node refuses. The session is what the scheduler reads, so a cut that only reached the node would keep over-assigning until the next heartbeat.
func (*Service) SetWatchdog ¶
SetWatchdog configures the health sweep. It is a setter because the timings belong to the scheduler's Timing struct, which imports this package.
func (*Service) Stream ¶
func (s *Service) Stream( ctx context.Context, stream *connect.BidiStream[podiumv1.NodeMessage, podiumv1.ServerMessage], ) error
Stream is the steady-state node stream. The first message must be a Hello carrying a valid node key within five seconds; after that the node heartbeats and pushes task events, and the server pushes assignments, acks and cancels.
func (*Service) UploadArtifact ¶
func (s *Service) UploadArtifact( ctx context.Context, stream *connect.ClientStream[podiumv1.UploadArtifactRequest], ) (*connect.Response[podiumv1.UploadArtifactResponse], error)
UploadArtifact takes one file a task produced and puts it in the object store.
The node is the only party that can reach the container, and the server is the only party that can reach S3, so the bytes cross the control plane. That is deliberate: a presigned PUT straight from the node would be one fewer hop and would also mean every worker needs a route and a credential to the object store, which is exactly the invariant the rest of the networking design is built to avoid.
The first message must carry the metadata, with the same node_id and node_key Hello presents — an upload is its own HTTP request and carries no session. Every message after it is a chunk, and the body is streamed straight through to S3: nothing here buffers a whole artifact.
type Session ¶
type Session struct {
// contains filtered or unexported fields
}
Session is one live node stream. It is created by Stream, lives in the Registry for exactly as long as that stream, and is the only way to push work at a node.
func (*Session) Done ¶
func (s *Session) Done() <-chan struct{}
Done is closed when the session ends, whether the node disconnected or the server replaced or shut it down.
type Snapshot ¶
type Snapshot struct {
NodeID string
Labels []string
FreeSlots int32
RunningTasks int32
LastSeen time.Time
Draining bool
Capacity store.NodeCapacity
// FreeCPU and FreeMemoryMB are what the node advertised minus what it is already
// committed to. A capacity of zero means "unmeasured", not "none", and is treated as
// no constraint: refusing every task because gopsutil could not count the cores would
// be a worse failure than over-committing one.
FreeCPU float64
FreeMemoryMB int64
LastAssignedAt time.Time
}
Snapshot is what the scheduler and ListNodes read off a live session. It is a value, so a scheduler tick can decrement its own copy while it places a batch without holding a lock on the session or racing the node's next heartbeat.
type TaskCost ¶
TaskCost is what one task takes out of a node's budget: the task container's limits plus every sidecar's, because a pod's sidecars run on the same machine as the task.
type Watchdog ¶
type Watchdog struct {
// Interval is how often node health is swept.
Interval time.Duration
// UnreachableAfter is how long without a heartbeat before a node is unreachable. Its
// tasks keep running: a node that cannot talk has not necessarily stopped working, and
// the design's whole point is that a brief network loss costs nothing.
UnreachableAfter time.Duration
// OfflineAfter is how long without a heartbeat before a node is offline and its leases
// expire.
OfflineAfter time.Duration
}
Watchdog is how node health is decided. The numbers come from the scheduler's Timing so there is one policy in the tree; they are passed in rather than imported because the scheduler already depends on this package and the cycle would be worse than the setter.
Source Files