Documentation
¶
Overview ¶
claimer_loop.go contains the claim loop orchestration and backoff logic.
This file owns the Start method that continuously polls for work (jobs) with exponential backoff. Backoff increases when no work is available or on errors, and resets when work is successfully claimed. Nodes claim from a single unified jobs queue (FIFO by next_id); there is no separate Build Gate queue or claim path. Isolating loop mechanics from claim/execution details simplifies backoff testing.
container_job.go contains mig job implementations, the shared container job executor, and workspace lifecycle helpers.
Package nodeagent contains the ployd node execution agent.
Responsibilities:
- Accept run requests from the control plane and orchestrate execution.
- Hydrate workspaces from Git, run Build Gate validation, execute mig containers, and collect/upload artifacts, diffs, and terminal status.
- Execute discrete job types from the unified queue (gate, mig).
Key files:
- execution.go — high level run lifecycle and runtime factories.
- gate_job.go — gate job execution and failure context persistence.
- container_job.go — mig job execution and shared container lifecycle.
- job_reporting.go — centralized diff/status upload helpers.
- workspace.go — workspace/file utilities.
- manifest.go — request→manifest translation helpers.
- job.go — job status types, image name persistence.
- http.go — base HTTP client, URL builders, compression helpers.
execution.go contains the high-level run lifecycle orchestration.
This file owns executeRun, the main entry point for executing a single run. It coordinates runtime initialization and dispatches to specialized job handlers based on job type. Job implementations live in:
- container_job.go — mig jobs + standard executor
- gate_job.go — gate validation jobs
job_reporting.go contains upload, status reporting, diff generation, and artifact helpers used by job executors.
run_options.go defines typed option structs for nodeagent execution. These types replace untyped map[string]any lookups with type-safe accessors.
Index ¶
- Constants
- Variables
- func BuildURL(base, p string) (string, error)
- func MustBuildURL(base, p string) string
- type Agent
- type ArtifactBundleEntry
- type BuildGateOptions
- type ClaimManager
- type ClaimResponse
- type Config
- type ContainerSpec
- type HTTPConfig
- type HeartbeatConfig
- type HeartbeatManager
- type HeartbeatPayload
- type LogHook
- type LogStreamer
- type RunController
- type RunOptions
- type Server
- type ServerMetadataOptions
- type StartRunRequest
- type StartRunResponse
- type StepOptions
- type StopRunRequest
- type StopRunResponse
- type TLSConfig
Constants ¶
const ( MaxUploadSize = 10 << 20 // 10 MiB // SoftUploadSize is the threshold at which log chunks are flushed. The 64-byte // margin accounts for gzip footer overhead when finalizing a chunk, ensuring // the closed stream stays under MaxUploadSize. SoftUploadSize = MaxUploadSize - 64 )
const MaxConcurrency = 64
Variables ¶
var ErrPayloadTooLarge = errors.New("payload exceeds size cap")
ErrPayloadTooLarge is returned when compressed data exceeds MaxUploadSize.
Functions ¶
func MustBuildURL ¶
MustBuildURL is like BuildURL but panics on error.
Types ¶
type Agent ¶
type Agent struct {
// contains filtered or unexported fields
}
Agent coordinates the node agent's HTTP server, heartbeat manager, and claim loop.
type ArtifactBundleEntry ¶
ArtifactBundleEntry defines one filesystem source and its archive path inside an uploaded artifact bundle.
type BuildGateOptions ¶
type BuildGateOptions struct {
Disabled bool
Images []contracts.BuildGateImageRule
Pre *contracts.BuildGatePhaseConfig
Post *contracts.BuildGatePhaseConfig
}
BuildGateOptions configures pre-mig build gate validation.
type ClaimManager ¶
type ClaimManager struct {
// contains filtered or unexported fields
}
ClaimManager periodically polls the server for work and executes claimed jobs. Nodes claim from a single unified jobs queue (FIFO by next_id); there is no separate Build Gate queue or claim path. Contains configuration, HTTP client, run controller, and backoff state for polling intervals when no work is available.
func NewClaimManager ¶
func NewClaimManager(cfg Config, controller RunController) (*ClaimManager, error)
NewClaimManager constructs a claim manager for the unified jobs queue. Nodes claim jobs from a single queue (FIFO by next_id); there is no separate Build Gate queue. Initializes backoff parameters for the claim loop polling interval.
func (*ClaimManager) Start ¶
func (c *ClaimManager) Start(ctx context.Context) error
Start begins the claim loop. Continuously polls for work (jobs) using a ticker with exponential backoff. Backoff increases when no work is available or on errors, resets when work is successfully claimed. Nodes claim from a single unified jobs queue (FIFO by next_id).
type ClaimResponse ¶
type ClaimResponse struct {
RunID types.RunID `json:"id"` // Run ID (KSUID identifying the parent run)
Name *string `json:"name,omitempty"`
RepoID types.MigRepoID `json:"repo_id"` // Repo ID (NanoID identifying the repo execution)
JobID types.JobID `json:"job_id"` // Claimed job ID
JobName string `json:"job_name"` // Job name (e.g., "pre-gate", "mig-0")
JobType types.JobType `json:"job_type"` // Job phase: pre_gate, mig, post_gate
JobImage string `json:"job_image"` // Container image for mig jobs
NextID *types.JobID `json:"next_id"`
RepoURL types.RepoURL `json:"repo_url"`
Status string `json:"status"`
NodeID types.NodeID `json:"node_id"`
BaseRef types.GitRef `json:"base_ref"`
CommitSha *types.CommitSHA `json:"commit_sha,omitempty"`
RepoShaIn *types.CommitSHA `json:"repo_sha_in,omitempty"`
StartedAt string `json:"started_at"`
CreatedAt string `json:"created_at"`
Spec json.RawMessage `json:"spec,omitempty"`
MigContext *contracts.MigClaimContext `json:"mig_context,omitempty"`
GateContext *contracts.GateClaimContext `json:"gate_context,omitempty"`
DetectedStack *contracts.StackExpectation `json:"detected_stack,omitempty"`
}
ClaimResponse represents the response from POST /v1/nodes/{id}/claim. Returned by the server when a job is successfully claimed and assigned to this node. Contains the run metadata plus the claimed job's ID and name. Note: The RunID field uses json:"id" to maintain wire compatibility with the existing API schema while providing type clarity in Go code.
type Config ¶
type Config struct {
// HTTP configuration for the node agent API server.
HTTP HTTPConfig `yaml:"http"`
// Server URL for the control-plane server.
ServerURL string `yaml:"server_url"`
// NodeID identifies this node (NanoID-backed).
NodeID domaintypes.NodeID `yaml:"node_id"`
// Concurrency defines the maximum number of concurrent runs.
Concurrency int `yaml:"concurrency"`
// Heartbeat configuration.
Heartbeat HeartbeatConfig `yaml:"heartbeat"`
}
Config holds node agent configuration. Uses domain types for type-safe node identification.
func LoadConfig ¶
LoadConfig reads and parses the YAML configuration file.
type ContainerSpec ¶
type ContainerSpec struct {
Image contracts.JobImage
Command contracts.CommandSpec
Env map[string]string
Options map[string]any
// Hydra resource entries for staged materialization and mount planning.
In []string // canonical read-only input entries (shortHash:/in/dst)
Out []string // canonical read-write output entries (shortHash:/out/dst)
Home []string // canonical home-relative entries (shortHash:dst{:ro})
Tmp []string // canonical writable temporary entries (shortHash:/tmp/dst)
}
ContainerSpec describes a container's image, command, and env. Used for execution options and step entries.
type HTTPConfig ¶
type HTTPConfig struct {
// Listen address (e.g., ":8444").
Listen string `yaml:"listen"`
// TLS configuration.
TLS TLSConfig `yaml:"tls"`
// Timeouts.
ReadTimeout time.Duration `yaml:"read_timeout"`
WriteTimeout time.Duration `yaml:"write_timeout"`
IdleTimeout time.Duration `yaml:"idle_timeout"`
}
HTTPConfig specifies HTTP listener and TLS settings for the node agent.
type HeartbeatConfig ¶
type HeartbeatConfig struct {
// Interval between heartbeats.
Interval time.Duration `yaml:"interval"`
// Timeout for heartbeat requests.
Timeout time.Duration `yaml:"timeout"`
}
HeartbeatConfig specifies heartbeat interval and timeout.
type HeartbeatManager ¶
type HeartbeatManager struct {
// contains filtered or unexported fields
}
HeartbeatManager periodically sends resource snapshots to the server.
func NewHeartbeatManager ¶
func NewHeartbeatManager(cfg Config) (*HeartbeatManager, error)
NewHeartbeatManager constructs a heartbeat manager.
type HeartbeatPayload ¶
type HeartbeatPayload struct {
CPUFreeMillis int32 `json:"cpu_free_millis"`
CPUTotalMillis int32 `json:"cpu_total_millis"`
MemFreeBytes int64 `json:"mem_free_bytes"`
MemTotalBytes int64 `json:"mem_total_bytes"`
DiskFreeBytes int64 `json:"disk_free_bytes"`
DiskTotalBytes int64 `json:"disk_total_bytes"`
Version string `json:"version,omitempty"`
}
HeartbeatPayload contains resource snapshot data sent to the server.
type LogHook ¶
LogHook is a function that processes log data before it is compressed and sent to the server. A nil LogHook is treated as a no-op.
type LogStreamer ¶
type LogStreamer struct {
// contains filtered or unexported fields
}
LogStreamer buffers logs and streams them as gzipped chunks to the server.
func NewLogStreamer ¶
func NewLogStreamer(cfg Config, runID types.RunID, jobID types.JobID, client *http.Client) (*LogStreamer, error)
NewLogStreamer creates a new log streamer for a specific run and (optionally) job. When client is non-nil it is reused; otherwise a new HTTP client is created. Returns an error if HTTP client creation fails (e.g., missing bearer token).
func (*LogStreamer) Close ¶
func (ls *LogStreamer) Close() error
Close flushes any remaining logs and stops the streamer. Returns all errors encountered during close using errors.Join.
func (*LogStreamer) SetHook ¶
func (ls *LogStreamer) SetHook(hook LogHook)
SetHook sets the log processing hook. Must be called before any writes. This method is not safe for concurrent use with Write.
func (*LogStreamer) StderrWriter ¶
func (ls *LogStreamer) StderrWriter() io.Writer
StderrWriter returns an io.Writer that tags all writes as stderr.
func (*LogStreamer) StdoutWriter ¶
func (ls *LogStreamer) StdoutWriter() io.Writer
StdoutWriter returns an io.Writer that tags all writes as stdout.
type RunController ¶
type RunController interface {
StartRun(ctx context.Context, req StartRunRequest) error
StopRun(ctx context.Context, req StopRunRequest) error
// AcquireSlot blocks until a concurrency slot is available or the context
// is canceled. Returns nil when a slot is acquired, or ctx.Err() if the
// context was canceled while waiting.
//
// Slot ownership:
// - On StartRun success (nil error), the controller is responsible for
// releasing the slot when the job completes.
// - On StartRun failure (non-nil error), the caller must ReleaseSlot()
// before returning.
AcquireSlot(ctx context.Context) error
// ReleaseSlot frees a previously acquired concurrency slot.
// Must be called exactly once for each successful AcquireSlot call.
ReleaseSlot()
}
RunController manages run lifecycle on the node.
type RunOptions ¶
type RunOptions struct {
BuildGate BuildGateOptions
Execution ContainerSpec
ServerMetadata ServerMetadataOptions
Steps []StepOptions
StackGate *contracts.StepGateStackSpec
// BundleMap maps content hashes to spec bundle download identifiers.
// Populated from MigSpec.BundleMap during spec-to-run-options conversion.
BundleMap map[string]string
}
RunOptions holds all typed configuration options for a run execution.
type Server ¶
type Server struct {
// contains filtered or unexported fields
}
Server exposes the node agent API over HTTPS with mTLS.
func NewServer ¶
func NewServer(cfg Config, controller RunController) (*Server, error)
NewServer constructs a new node agent HTTP server.
type ServerMetadataOptions ¶
type ServerMetadataOptions struct {
JobID domaintypes.JobID
}
ServerMetadataOptions holds server-injected metadata for uploads and tracking.
type StartRunRequest ¶
type StartRunRequest struct {
RunID types.RunID `json:"run_id,omitempty"`
JobID types.JobID `json:"job_id,omitempty"` // Job ID for artifact/diff uploads
RepoID types.MigRepoID `json:"repo_id,omitempty"` // Repo ID (NanoID) for run artifacts (diffs/logs)
RepoURL types.RepoURL `json:"repo_url,omitempty"` // Repository URL for this run
// Name is an optional human-friendly run name provided by the control plane.
// When set (e.g., for runs), it can be used for branch naming.
Name string `json:"name,omitempty"`
BaseRef types.GitRef `json:"base_ref,omitempty"`
CommitSHA types.CommitSHA `json:"commit_sha,omitempty"`
RepoSHAIn types.CommitSHA `json:"repo_sha_in,omitempty"`
JobType types.JobType `json:"job_type,omitempty"` // Job type: pre_gate, mig, post_gate
JobImage string `json:"job_image,omitempty"` // Container image for this job
NextID *types.JobID `json:"next_id,omitempty"` // Linked successor in run chain
JobName string `json:"job_name,omitempty"` // Deprecated: kept for wire compatibility during context rollout.
// MigContext carries concrete mig step routing.
MigContext *contracts.MigClaimContext `json:"mig_context,omitempty"`
// GateContext carries concrete gate cycle routing.
GateContext *contracts.GateClaimContext `json:"gate_context,omitempty"`
// DetectedStack carries the canonical gate-detected stack tuple for this job.
DetectedStack *contracts.StackExpectation `json:"detected_stack,omitempty"`
// TypedOptions contains strongly-typed run configuration. This is the canonical
// source of truth for all option keys understood by the nodeagent. Execution,
// manifest building, and artifact upload paths all consume TypedOptions
// directly rather than parsing raw maps.
TypedOptions RunOptions `json:"-"` // Not serialized; populated by claimer_loop from parsed spec
Env map[string]string `json:"env"` // Environment variables merged from spec
ServerURL string `json:"server_url,omitempty"`
}
StartRunRequest describes a run start request from the server.
TypedOptions contains all run configuration options in strongly-typed form. This is the canonical source of truth for all option keys understood by the nodeagent. Callers must use TypedOptions fields instead of raw map[string]any access. The typed options include:
- BuildGate: enabled flag and image overrides for gate validation.
- Execution: container image, command, and retention settings.
- Artifacts: artifact name and workspace-relative paths to upload.
- ServerMetadata: server-injected job ID for upload correlation.
- Steps: multi-step migs array for sequential execution.
JobType field:
- Identifies the job type: "pre_gate", "mig", "post_gate".
- Used by orchestrator to dispatch to appropriate execution handler.
type StartRunResponse ¶
StartRunResponse is returned when a run is accepted.
type StepOptions ¶
type StepOptions struct {
ContainerSpec
Stack *contracts.StackGateSpec
}
StepOptions describes a single mig step in a multi-step run (steps[] array). Each step has its own container spec and optional Stack Gate validation.
type StopRunRequest ¶
StopRunRequest describes a run stop/cancel request.
type StopRunResponse ¶
StopRunResponse is returned when a stop request is processed.
type TLSConfig ¶
type TLSConfig struct {
// Enabled indicates whether mTLS is enabled.
Enabled bool `yaml:"enabled"`
// CertPath is the path to the node certificate.
CertPath string `yaml:"cert_path"`
// KeyPath is the path to the node private key.
KeyPath string `yaml:"key_path"`
// CAPath is the path to the CA certificate.
CAPath string `yaml:"ca_path"`
// BootstrapCAPath is the path to the CA certificate used to verify
// the server during bootstrap (before mTLS certificates are obtained).
// If empty, the CA at CAPath is used if it exists; otherwise
// system roots are used (for public PKI scenarios).
BootstrapCAPath string `yaml:"bootstrap_ca_path"`
}
TLSConfig specifies mTLS certificate and key paths.
Source Files
¶
- agent.go
- claimer.go
- claimer_loop.go
- config.go
- container_job.go
- controller.go
- controller_cancel_watch.go
- crash_reconcile.go
- crash_reconcile_running.go
- crash_reconcile_startup.go
- doc.go
- error_normalization.go
- execution.go
- execution_artifacts.go
- execution_paths.go
- gate_context.go
- gate_job.go
- gate_metadata_helpers.go
- gate_sbom.go
- handlers.go
- heartbeat.go
- http.go
- job.go
- job_metrics_helpers.go
- job_reporting.go
- logstreamer.go
- manifest.go
- node_events.go
- resource_materialization.go
- run_options.go
- server.go
- stack_runtime_env.go
- storage.go
- uploaders.go
- workspace.go
- workspace_hydration.go