Documentation
¶
Overview ¶
Package nodeagent is the Mercator node agent: the process that runs on a reusable machine and executes successive workloads there.
It connects outbound and never listens. It keeps enough local state to refuse a command it already applied, which is what stops a redelivered launch from starting a second container. It spools the facts it owes the control plane while disconnected, so a container that exited during a network partition is still reported rather than inferred.
Index ¶
- Constants
- type Agent
- type DockerRuntime
- func (docker *DockerRuntime) Facts(ctx context.Context) (capability.NodeFacts, error)
- func (docker *DockerRuntime) LaunchWorkload(ctx context.Context, command capability.LaunchWorkloadCommand) error
- func (docker *DockerRuntime) Observe(ctx context.Context) ([]capability.WorkloadObservation, error)
- func (docker *DockerRuntime) PrepareArtifact(ctx context.Context, command capability.PrepareArtifactCommand) error
- func (docker *DockerRuntime) PrepareImage(ctx context.Context, command capability.PrepareImageCommand) error
- func (docker *DockerRuntime) StopWorkload(ctx context.Context, command capability.StopWorkloadCommand) error
- type HTTPTransport
- func (transport *HTTPTransport) Enroll(ctx context.Context, request capability.EnrollmentRequest) (capability.Enrollment, error)
- func (transport *HTTPTransport) RenewSession(ctx context.Context, nodeID, sessionToken string) (capability.SessionRenewal, error)
- func (transport *HTTPTransport) SendEvents(ctx context.Context, nodeID, sessionToken string, events []node.Event) error
- func (transport *HTTPTransport) SendResult(ctx context.Context, nodeID, sessionToken string, result node.Result) error
- func (transport *HTTPTransport) Session(ctx context.Context, nodeID, sessionToken string, commands chan<- node.Command) error
- type Identity
- type Option
- type Runtime
- type RuntimeOption
- type State
- func (state *State) Applied(operationID string) bool
- func (state *State) ClearSpool() error
- func (state *State) Enrolled(enrollment capability.Enrollment) error
- func (state *State) FencingToken() uint64
- func (state *State) MarkApplied(operationID string) error
- func (state *State) NextEventID() string
- func (state *State) RecordWorkload(observation capability.WorkloadObservation)
- func (state *State) Renewed(renewal capability.SessionRenewal) error
- func (state *State) Session() (string, time.Time, bool)
- func (state *State) Spool(event node.Event) error
- func (state *State) Spooled() []node.Event
- func (state *State) WorkloadChanged(observation capability.WorkloadObservation) bool
- type Transport
Constants ¶
const ArtifactCopySource = "node_artifact_copy"
ArtifactCopySource names what a node measured when it timed itself reading Artifact content, in the words of the process that did it. It is the whole copy: the bytes crossing the path, landing on this disk, and being hashed on the way past, because one read does all three jobs here and the node holds one duration over all of them.
The name says so rather than claiming the link alone, and the difference is not pedantry. A machine on a ten gigabit path whose Artifact root is a slow disk delivers content at the disk's rate, and that rate is the honest answer to both questions anything asks this fact: how long the next read of forty gigabytes takes here, and whether a Run that states a floor on reading its dataset can be served by this machine. Timing the socket alone would answer neither, and would be wrong in the direction that wins placements the machine cannot honour.
const MeasuredLinkConfidence = 0.9
MeasuredLinkConfidence is how much a node stands behind its own reading of a path. It is more than domain.AssumedLinkConfidence, which is what Mercator's fleet-wide guess about an unmeasured path is worth, and deliberately short of certainty: what is published is the slowest transfer this node has seen, which is a floor over a handful of readings rather than a distribution.
The figure itself is stated rather than derived, and that is the honest description of it. Deriving it from the sample count would be an estimator this slice has not measured, and predicted-versus-actual for the fetch stage is what a calibration slice will replace it with.
const MeasuredLinkValidity = time.Hour
MeasuredLinkValidity is how long one reading of a path stands. A measurement is evidence about the path as it was, and a node that has moved nothing for an hour is not evidence about the path now: it keeps reporting its liveness, its containers, and its disk, so without an expiry its oldest reading would be published as a current fact forever.
It is the life of one reading, and every reading serves it alone. That is what retires the slow one: the machine that read a hundred megabits while a container shared its link stops standing behind that reading an hour later however busy it has been since, because no later transfer can extend a window it was not measured in.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Agent ¶
type Agent struct {
// contains filtered or unexported fields
}
Agent runs one machine's side of the node protocol.
type DockerRuntime ¶
type DockerRuntime struct {
// contains filtered or unexported fields
}
DockerRuntime executes workloads through the local Docker daemon. It is the first Runtime implementation, and it stays behind the narrow Runtime interface so containerd or another OCI runtime can replace it without the control-plane contract moving.
The daemon is reached over its local socket by this process only. Nothing exposes it to the network, and the control plane never talks to it.
func NewDockerRuntime ¶
func NewDockerRuntime(binary string, options ...RuntimeOption) *DockerRuntime
func (*DockerRuntime) Facts ¶
func (docker *DockerRuntime) Facts(ctx context.Context) (capability.NodeFacts, error)
func (*DockerRuntime) LaunchWorkload ¶
func (docker *DockerRuntime) LaunchWorkload(ctx context.Context, command capability.LaunchWorkloadCommand) error
LaunchWorkload starts one container and returns once it is running. The container's name is derived from the run and attempt, so the daemon itself refuses a second container for the same attempt even if the agent's own memory were lost.
func (*DockerRuntime) Observe ¶
func (docker *DockerRuntime) Observe(ctx context.Context) ([]capability.WorkloadObservation, error)
Observe reports every Mercator container this daemon knows about, running or exited, and when each of them really started. It is the node's own authority: the control plane learns that a process began and ended here, whatever the application did or did not report.
The start moment is a second read, because `docker ps` does not carry it in any format: the daemon reports `State.StartedAt` only on inspect. It is one inspect for the whole list rather than one per container, and a container the daemon will not describe is reported with no start moment rather than with a guess. A moment the daemon does state and this runtime cannot read fails the whole read instead, which is the same answer the Docker adapter gives to the same daemon: reporting it absent would publish no start for any container on this machine and degrade every start-latency row on the only reusable lane there is, with nothing anywhere saying that is what happened.
func (*DockerRuntime) PrepareArtifact ¶
func (docker *DockerRuntime) PrepareArtifact(ctx context.Context, command capability.PrepareArtifactCommand) error
PrepareArtifact replicates one immutable version onto this machine. It reads through the location the control plane minted for this one fetch and nowhere else: the object store's own key stays in the control plane, so a node holds no standing read of anything and a compromised machine can fetch exactly the content it was told to, until the read expires.
A command with no minted read is refused rather than attempted. The durable location the catalog states is a name for content, and a node that went to the network with it would fail in the object store's vocabulary instead of in Mercator's, which is an operator reading a 404 for a missing configuration.
func (*DockerRuntime) PrepareImage ¶
func (docker *DockerRuntime) PrepareImage(ctx context.Context, command capability.PrepareImageCommand) error
PrepareImage pulls an image by its exact manifest digest. A tag is never image identity, so the reference the control plane sends is already pinned.
It goes to the daemon's own API rather than to the CLI, for one reason: the credential. The CLI reads registry material out of a config file, so pulling a private image through it means writing the material onto the machine and remembering to remove it, and a pull that is killed halfway leaves it there. The API takes it as a header, so the material this node was handed exists in this process's memory for the length of one request and nowhere else. That is what "the machine forgets it afterwards" has to mean on a host an operator rents by the hour.
func (*DockerRuntime) StopWorkload ¶
func (docker *DockerRuntime) StopWorkload(ctx context.Context, command capability.StopWorkloadCommand) error
type HTTPTransport ¶
type HTTPTransport struct {
// contains filtered or unexported fields
}
HTTPTransport reaches the control plane over the node protocol. Every call is outbound: the agent opens the session and posts what it owes, and nothing ever connects to the machine.
func NewHTTPTransport ¶
func NewHTTPTransport(baseURL string, client *http.Client) *HTTPTransport
func (*HTTPTransport) Enroll ¶
func (transport *HTTPTransport) Enroll(ctx context.Context, request capability.EnrollmentRequest) (capability.Enrollment, error)
func (*HTTPTransport) RenewSession ¶
func (transport *HTTPTransport) RenewSession(ctx context.Context, nodeID, sessionToken string) (capability.SessionRenewal, error)
RenewSession takes a later credential using the one this agent still holds. It is the only exchange here authenticated by a credential the agent is about to stop using, and it is what keeps a machine reachable past the life of the invitation it joined with.
func (*HTTPTransport) SendEvents ¶
func (*HTTPTransport) SendResult ¶
type Identity ¶
type Identity struct {
ControlPlaneURL string
NodeID string
RentalID string
Generation uint64
EnrollmentToken string
AgentVersion string
}
Identity is the immutable material a machine was bootstrapped with. The agent never invents any of it.
type Option ¶
type Option func(*Agent)
Option configures an Agent.
func WithHeartbeat ¶
WithHeartbeat sets how often the agent reports its facts. It must stay well inside the control plane's lease, or a healthy machine will be declared lost.
func WithLogger ¶
WithLogger replaces the agent's logger.
func WithReconnectBackoff ¶
WithReconnectBackoff sets how long the agent waits before reopening a dropped session.
type Runtime ¶
type Runtime interface {
// Facts reports the host, accelerators, driver, runtime, disk, and exact
// content inventory this machine holds.
Facts(ctx context.Context) (capability.NodeFacts, error)
PrepareImage(ctx context.Context, command capability.PrepareImageCommand) error
PrepareArtifact(ctx context.Context, command capability.PrepareArtifactCommand) error
// LaunchWorkload starts the container and returns once it is created. The
// agent observes its lifecycle separately, so a launch that returns does
// not claim the workload succeeded.
LaunchWorkload(ctx context.Context, command capability.LaunchWorkloadCommand) error
StopWorkload(ctx context.Context, command capability.StopWorkloadCommand) error
// Observe reports every workload this runtime currently knows about,
// including recently exited ones the agent has not yet reported.
Observe(ctx context.Context) ([]capability.WorkloadObservation, error)
}
Runtime is the container runtime the agent drives. It is deliberately narrow: Docker is the first implementation, and nothing above this line knows which one is running, so containerd or another OCI runtime can be added without touching the control-plane contract.
type RuntimeOption ¶
type RuntimeOption func(*DockerRuntime)
RuntimeOption configures the Docker runtime.
func WithAcceleratorTool ¶
func WithAcceleratorTool(binary string) RuntimeOption
WithAcceleratorTool points this node at the vendor tool it asks about its own cards, by absolute path where the bare name will not resolve.
It is operator configuration and not only a test seam. A unit file with a trimmed Environment=PATH, or a distribution that installs the tool somewhere the service PATH does not reach, is a machine whose agent cannot find nvidia-smi by name; on an 8xH100 box that is the difference between a fleet that runs GPU work and one that reports it has none. It is the second reason this exists, the first being that a test can stand in a machine with four A100s or no driver at all, which is the only way the reports this node publishes about hardware can be exercised anywhere but on the hardware.
func WithArtifactRoot ¶
func WithArtifactRoot(root string) RuntimeOption
WithArtifactRoot gives this node somewhere to keep immutable Artifact copies. It is the agent's own directory, stated by whoever started the agent, because only they know which filesystem has the room: a default under the daemon's storage would be a directory this process usually cannot write to, and one under a temporary directory would lose every copy on reboot.
func WithKernelReports ¶
func WithKernelReports(root string) RuntimeOption
WithKernelReports points this node at the kernel's own reports about itself, which is /proc everywhere but in a case that has to stand in a machine whose NVIDIA module is loaded or absent.
type State ¶
type State struct {
// contains filtered or unexported fields
}
State is the agent's local durable memory. It exists for one reason: an agent that restarts must be able to say "I already did that" about a command it is handed again. Without it, every redelivered launch is a second container.
It also holds the session credential, so a restarted agent resumes rather than re-enrolling, and the spool of events it owes the control plane.
func (*State) ClearSpool ¶
func (*State) Enrolled ¶
func (state *State) Enrolled(enrollment capability.Enrollment) error
Enrolled records a fresh enrollment. A new fencing token means the control plane has superseded whatever this agent was doing, so the applied-operation memory stays and the session is replaced.
func (*State) FencingToken ¶
func (*State) MarkApplied ¶
MarkApplied records an operation as performed, durably, before its result is reported. A crash between the two costs a duplicate acknowledgement, never a duplicate container.
func (*State) NextEventID ¶
NextEventID mints a stable, monotonic identity for one event, so the same fact carries the same ID however many times it is delivered.
func (*State) RecordWorkload ¶
func (state *State) RecordWorkload(observation capability.WorkloadObservation)
RecordWorkload remembers what the control plane was told, so an unchanged phase is not repeated on every tick.
func (*State) Renewed ¶
func (state *State) Renewed(renewal capability.SessionRenewal) error
Renewed records a later credential for the session this agent already has. The fencing token comes back unchanged from a renewal, and it is written from the answer rather than left alone for the same reason every other field is: what this agent believes about itself is what the control plane last told it.
func (*State) Session ¶
Session returns the credential this agent holds and the moment it lapses, and whether it holds one at all.
Whether the credential is still worth using is the agent's judgment rather than this file's, which is why the expiry is answered instead of compared. An agent has to renew before the lapse, because the material it would need to enroll again is spent, so the interesting question is never "has this expired" but "will it still be valid when the request I am about to make lands".
func (*State) Spool ¶
Spool holds an event the control plane could not be told about. It keeps its ID, so replaying it after a reconnection changes nothing.
func (*State) WorkloadChanged ¶
func (state *State) WorkloadChanged(observation capability.WorkloadObservation) bool
WorkloadChanged reports whether this observation says something the control plane has not already been told.
type Transport ¶
type Transport interface {
Enroll(ctx context.Context, request capability.EnrollmentRequest) (capability.Enrollment, error)
// RenewSession takes a later credential using the one the agent holds. It is
// separate from Enroll because the material is separate: enrolling spends the
// single-use invitation the machine was bootstrapped with, and renewing
// spends nothing.
RenewSession(ctx context.Context, nodeID, sessionToken string) (capability.SessionRenewal, error)
// Session opens the command stream. It returns when the connection ends;
// the agent reconnects.
Session(ctx context.Context, nodeID, sessionToken string, commands chan<- node.Command) error
SendEvents(ctx context.Context, nodeID, sessionToken string, events []node.Event) error
SendResult(ctx context.Context, nodeID, sessionToken string, result node.Result) error
}
Transport is the agent's outbound connection to the control plane.
Source Files