job

package
v0.0.0-...-443440a Latest Latest
Warning

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

 Go to latest Published: Aug 16, 2026 License: MIT Imports: 13 Imported by: 0

Documentation

Overview

Package job provides NATS subject hierarchy for distributed job routing.

Subject Format: [namespace.]jobs.{type}.{routing_type}.{value...}

Routing Patterns:

  • Direct: jobs.query.host.<machineID> (specific host by machine ID)
  • Any: jobs.query._any (load-balanced across available agents)
  • Broadcast: jobs.modify._all (all agents receive)
  • Label: jobs.query.label.group.web (broadcast to label group)
  • Hierarchical: jobs.query.label.group.web.dev.us-east (prefix matching)

Agents subscribe to:

  • Their machine ID: jobs.*.host.<machineID> (permanent, never changes)
  • Load-balanced work: jobs.*._any (with queue group)
  • Broadcast messages: jobs.*._all
  • Label prefixes: jobs.*.label.group.web, jobs.*.label.group.web.dev, etc.

The controller resolves hostname targets to machine IDs before building subjects. This ensures direct routing uses the permanent machine ID, not the mutable hostname.

When a namespace is configured via Init(), all subjects are prefixed: 

Init("osapi") -> osapi.jobs.query._any, osapi.jobs.*.host.<machineID>, etc.

Index

Constants

View Source 
const (
	// AllHosts is a wildcard for targeting all hosts.
	AllHosts = "*"
	// AnyHost is load-balanced across available hosts.
	AnyHost = client.TargetAny
	// LocalHost targets the API server's host.
	LocalHost = "_local"
	// BroadcastHost broadcasts to all hosts (no queue group).
	BroadcastHost = client.TargetAll
)
View Source 
const (
	SubjectCategoryNode    = "node"
	SubjectCategoryNetwork = "network"
	SubjectCategoryDocker  = "docker"
)

Subject categories for different operations

View Source 
const (
	NodeOperationHostname = "hostname"
	NodeOperationStatus   = "status"
)

Node operation types

View Source 
const (
	NetworkOperationDNS  = "dns"
	NetworkOperationPing = "ping"
)

Network operation types

View Source 
const (
	TypeQuery  = client.JobTypeQuery
	TypeModify = client.JobTypeModify
)

Job type constants re-exported from the SDK.

View Source 
const (
	StatusSubmitted      = client.JobStatusSubmitted
	StatusAcknowledged   = client.JobStatusAcknowledged
	StatusStarted        = client.JobStatusStarted
	StatusPending        = client.JobStatusPending
	StatusProcessing     = client.JobStatusProcessing
	StatusCompleted      = client.JobStatusCompleted
	StatusFailed         = client.JobStatusFailed
	StatusSkipped        = client.JobStatusSkipped
	StatusPartialFailure = client.JobStatusPartialFailure
	StatusRetried        = client.JobStatusRetried
)

Job status constants re-exported from the SDK.

View Source 
const (
	OperationNodeHostnameGet    = client.OpNodeHostnameGet
	OperationNodeHostnameUpdate = client.OpNodeHostnameUpdate
	OperationNodeStatusGet      = client.OpNodeStatusGet
	OperationNodeUptimeGet      = client.OpNodeUptimeGet
	OperationNodeLoadGet        = client.OpNodeLoadGet
	OperationNodeMemoryGet      = client.OpNodeMemoryGet
	OperationNodeDiskGet        = client.OpNodeDiskGet
	OperationNodeOSGet          = client.OpNodeOSGet
)

Node operations — read-only operations that query node state.

View Source 
const (
	OperationNetworkDNSGet    = client.OpNetworkDNSGet
	OperationNetworkDNSUpdate = client.OpNetworkDNSUpdate
	OperationNetworkDNSDelete = client.OpNetworkDNSDelete
	OperationNetworkPingDo    = client.OpNetworkPingDo
)

Network operations.

View Source 
const (
	OperationCommandExecExecute  = client.OpCommandExec
	OperationCommandShellExecute = client.OpCommandShell
)

Command operations — execute arbitrary commands on agents.

View Source 
const (
	OperationFileDeployExecute   = client.OpFileDeploy
	OperationFileUndeployExecute = client.OpFileUndeploy
	OperationFileStatusGet       = client.OpFileStatusGet
)

File operations — manage file deployments and status.

View Source 
const (
	OperationDockerCreate      = client.OpDockerCreate
	OperationDockerStart       = client.OpDockerStart
	OperationDockerStop        = client.OpDockerStop
	OperationDockerRemove      = client.OpDockerRemove
	OperationDockerList        = client.OpDockerList
	OperationDockerInspect     = client.OpDockerInspect
	OperationDockerExec        = client.OpDockerExec
	OperationDockerPull        = client.OpDockerPull
	OperationDockerImageRemove = client.OpDockerImageRemove
)

Docker operations.

View Source 
const (
	OperationCronList   = client.OpCronList
	OperationCronGet    = client.OpCronGet
	OperationCronCreate = client.OpCronCreate
	OperationCronUpdate = client.OpCronUpdate
	OperationCronDelete = client.OpCronDelete
)

Schedule/Cron operations.

View Source 
const (
	OperationSysctlList   = client.OpSysctlList
	OperationSysctlGet    = client.OpSysctlGet
	OperationSysctlCreate = client.OpSysctlCreate
	OperationSysctlUpdate = client.OpSysctlUpdate
	OperationSysctlDelete = client.OpSysctlDelete
)

Sysctl operations.

View Source 
const (
	OperationNtpGet    = client.OpNtpGet
	OperationNtpCreate = client.OpNtpCreate
	OperationNtpUpdate = client.OpNtpUpdate
	OperationNtpDelete = client.OpNtpDelete
)

NTP operations.

View Source 
const (
	OperationTimezoneGet    = client.OpTimezoneGet
	OperationTimezoneUpdate = client.OpTimezoneUpdate
)

Timezone operations.

View Source 
const (
	OperationPowerReboot   = client.OpPowerReboot
	OperationPowerShutdown = client.OpPowerShutdown
)

Power operations.

View Source 
const (
	OperationProcessList   = client.OpProcessList
	OperationProcessGet    = client.OpProcessGet
	OperationProcessSignal = client.OpProcessSignal
)

Process operations.

View Source 
const (
	OperationUserList           = client.OpUserList
	OperationUserGet            = client.OpUserGet
	OperationUserCreate         = client.OpUserCreate
	OperationUserUpdate         = client.OpUserUpdate
	OperationUserDelete         = client.OpUserDelete
	OperationUserChangePassword = client.OpUserChangePassword
)

User operations.

View Source 
const (
	OperationGroupList   = client.OpGroupList
	OperationGroupGet    = client.OpGroupGet
	OperationGroupCreate = client.OpGroupCreate
	OperationGroupUpdate = client.OpGroupUpdate
	OperationGroupDelete = client.OpGroupDelete
)

Group operations.

View Source 
const (
	OperationSSHKeyList   = client.OpSSHKeyList
	OperationSSHKeyAdd    = client.OpSSHKeyAdd
	OperationSSHKeyRemove = client.OpSSHKeyRemove
)

SSH Key operations.

View Source 
const (
	OperationPackageList        = client.OpPackageList
	OperationPackageGet         = client.OpPackageGet
	OperationPackageInstall     = client.OpPackageInstall
	OperationPackageRemove      = client.OpPackageRemove
	OperationPackageUpdate      = client.OpPackageUpdate
	OperationPackageListUpdates = client.OpPackageListUpdates
)

Package operations.

View Source 
const (
	OperationLogQuery     = client.OpLogQuery
	OperationLogQueryUnit = client.OpLogQueryUnit
	OperationLogSources   = client.OpLogSources
)

Log operations.

View Source 
const (
	OperationCertificateCAList   = client.OpCertificateCAList
	OperationCertificateCACreate = client.OpCertificateCACreate
	OperationCertificateCAUpdate = client.OpCertificateCAUpdate
	OperationCertificateCADelete = client.OpCertificateCADelete
)

Certificate operations.

View Source 
const (
	OperationNetworkInterfaceList   = client.OpNetworkInterfaceList
	OperationNetworkInterfaceGet    = client.OpNetworkInterfaceGet
	OperationNetworkInterfaceCreate = client.OpNetworkInterfaceCreate
	OperationNetworkInterfaceUpdate = client.OpNetworkInterfaceUpdate
	OperationNetworkInterfaceDelete = client.OpNetworkInterfaceDelete
)

Network interface operations.

View Source 
const (
	OperationNetworkRouteList   = client.OpNetworkRouteList
	OperationNetworkRouteGet    = client.OpNetworkRouteGet
	OperationNetworkRouteCreate = client.OpNetworkRouteCreate
	OperationNetworkRouteUpdate = client.OpNetworkRouteUpdate
	OperationNetworkRouteDelete = client.OpNetworkRouteDelete
)

Network route operations.

View Source 
const (
	OperationServiceList    = client.OpServiceList
	OperationServiceGet     = client.OpServiceGet
	OperationServiceCreate  = client.OpServiceCreate
	OperationServiceUpdate  = client.OpServiceUpdate
	OperationServiceDelete  = client.OpServiceDelete
	OperationServiceStart   = client.OpServiceStart
	OperationServiceStop    = client.OpServiceStop
	OperationServiceRestart = client.OpServiceRestart
	OperationServiceEnable  = client.OpServiceEnable
	OperationServiceDisable = client.OpServiceDisable
)

Service operations.

View Source 
const (
	ConditionMemoryPressure = client.ConditionMemoryPressure
	ConditionHighLoad       = client.ConditionHighLoad
	ConditionDiskPressure   = client.ConditionDiskPressure
)

Condition type constants re-exported from the SDK.

View Source 
const (
	AgentStateReady    = client.AgentReady
	AgentStateDraining = client.AgentDraining
	AgentStateCordoned = client.AgentCordoned
	AgentStatePending  = client.AgentPending
)

Agent state constants re-exported from the SDK.

Variables

View Source 
var (
	// JobsQueryPrefix is the subject hierarchy prefix for query operations.
	JobsQueryPrefix = "jobs.query"
	// JobsModifyPrefix is the subject hierarchy prefix for modify operations.
	JobsModifyPrefix = "jobs.modify"
)
View Source 
var StatusPriority = map[string]int{
	string(StatusSubmitted):    0,
	string(StatusAcknowledged): 1,
	string(StatusStarted):      2,
	string(StatusFailed):       3,
	string(StatusSkipped):      3,
	string(StatusCompleted):    4,
	string(StatusRetried):      4,
}

StatusPriority maps each job status to a numeric priority used when computing the overall status from append-only events. Higher values take precedence — a "completed" event (4) is never overwritten by a "started" event (2) that happens to sort later alphabetically in KV.

Functions

func ApplyNamespaceToInfraName 

func ApplyNamespaceToInfraName(
	namespace, name string,
) string

ApplyNamespaceToInfraName prefixes an infrastructure name (stream, KV bucket) with the namespace. Returns the name unchanged if namespace is empty. 

ApplyNamespaceToInfraName("", "JOBS")      -> "JOBS"
ApplyNamespaceToInfraName("osapi", "JOBS")  -> "osapi-JOBS"

func ApplyNamespaceToSubjects 

func ApplyNamespaceToSubjects(
	namespace, subjects string,
) string

ApplyNamespaceToSubjects prefixes a subject filter with the namespace. Returns the subject unchanged if namespace is empty. 

ApplyNamespaceToSubjects("", "jobs.>")      -> "jobs.>"
ApplyNamespaceToSubjects("osapi", "jobs.>")  -> "osapi.jobs.>"

func BuildAgentQueueGroup 

func BuildAgentQueueGroup(
	category string,
) string

BuildAgentQueueGroup returns the queue group name for load-balanced subscriptions. This ensures only one agent processes each "_any" message.

func BuildAgentSubscriptionPattern 

func BuildAgentSubscriptionPattern(
	hostname string,
	labels map[string]string,
) []string

BuildAgentSubscriptionPattern creates subscription patterns for agents. Agents typically subscribe to their own hostname and special routing patterns. If labels are provided, hierarchical prefix subscriptions are included for each label. For example, a label "group: web.dev.us-east" generates subscriptions at every prefix level (group:web, group:web.dev, group:web.dev.us-east).

func BuildLabelSubjects 

func BuildLabelSubjects(
	key, value string,
) []string

BuildLabelSubjects builds subscription subjects for a label with hierarchical prefix matching. For a label "group: web.dev.us-east", it returns subjects for every prefix level: 

jobs.*.label.group.web
jobs.*.label.group.web.dev
jobs.*.label.group.web.dev.us-east

This enables targeting at any level of the hierarchy: --target group:web matches all agents whose group label starts with "web".

func BuildModifySubject 

func BuildModifySubject(
	hostname string,
) string

BuildModifySubject creates a subject for modify operations. Example: jobs.modify.hostname

func BuildModifySubjectForAllHosts 

func BuildModifySubjectForAllHosts() string

BuildModifySubjectForAllHosts creates a modify subject targeting all hosts. Example: jobs.modify.*

func BuildQuerySubject 

func BuildQuerySubject(
	hostname string,
) string

BuildQuerySubject creates a subject for query operations. Example: jobs.query.hostname

func BuildQuerySubjectForAllHosts 

func BuildQuerySubjectForAllHosts() string

BuildQuerySubjectForAllHosts creates a query subject targeting all hosts. Example: jobs.query.*

func BuildSubjectFromTarget 

func BuildSubjectFromTarget(
	prefix, target string,
) string

BuildSubjectFromTarget builds the full NATS subject for any target value. For label targets with hierarchical values (e.g., "group:web.dev"), each dot-separated segment becomes a subject token: jobs.query.label.group.web.dev

func CountExpectedAgents 

func CountExpectedAgents(
	agents []AgentInfo,
	target string,
) int

CountExpectedAgents returns the number of agents expected to respond to a broadcast target. For _all it returns len(agents). For label targets it filters to agents whose label value equals or is a prefix of the target value (mirroring hierarchical NATS subject matching).

func ExpectedAgentHostnames 

func ExpectedAgentHostnames(
	agents []AgentInfo,
	target string,
) []string

ExpectedAgentHostnames returns the hostnames of agents expected to respond to a broadcast target. Uses the same filtering logic as CountExpectedAgents.

func GetAgentHostname 

func GetAgentHostname(
	configuredHostname string,
) (string, error)

GetAgentHostname returns the hostname that should be used by agents. It first checks the configured hostname, then falls back to system hostname using gopsutil. This function respects configuration while using gopsutil for system detection.

func GetAgentHostnameWithProvider 

func GetAgentHostnameWithProvider(
	configuredHostname string,
	provider HostnameProvider,
) (string, error)

GetAgentHostnameWithProvider returns the hostname using the provided provider. This allows for testing with mock providers.

func GetJobsConsumerConfig 

func GetJobsConsumerConfig(
	consumerConfig *config.AgentConsumer,
	streamSubjects string,
) jetstream.ConsumerConfig

GetJobsConsumerConfig returns the consumer configuration for processing job requests.

func GetJobsStreamConfig 

func GetJobsStreamConfig(
	streamConfig *config.NATSStream,
) *jetstream.StreamConfig

GetJobsStreamConfig returns the stream configuration for job processing. This creates a stream that accepts all job-related subjects (job.>).

func GetKVBucketConfig 

func GetKVBucketConfig(
	kvConfig *config.NATSKV,
) jetstream.KeyValueConfig

GetKVBucketConfig returns the KeyValue bucket configuration for storing job responses.

func GetLocalHostname 

func GetLocalHostname() (string, error)

GetLocalHostname returns the current system hostname using gopsutil. This is for backward compatibility with existing code.

func GetLocalHostnameWithProvider 

func GetLocalHostnameWithProvider(
	provider HostnameProvider,
) (string, error)

GetLocalHostnameWithProvider returns the hostname using the provided provider. This allows for testing with mock providers.

func Init 

func Init(
	namespace string,
)

Init configures the subject namespace. An empty namespace keeps the default "jobs.*" hierarchy. A non-empty namespace prepends it: 

Init("")      -> jobs.query, jobs.modify
Init("osapi") -> osapi.jobs.query, osapi.jobs.modify

func IsBroadcastTarget 

func IsBroadcastTarget(
	target string,
) bool

IsBroadcastTarget returns true if the target requires publishAndCollect (broadcast) semantics: _all or any key:value label target.

func IsSpecialHostname 

func IsSpecialHostname(
	hostname string,
) bool

IsSpecialHostname checks if a hostname is a special routing directive.

func ParseSubject 

func ParseSubject(
	subject string,
) (prefix, hostname string, err error)

ParseSubject extracts the prefix and routing target from a job subject. Supported formats (with optional namespace prefix):

  • [ns.]jobs.{type}._any
  • [ns.]jobs.{type}._all
  • [ns.]jobs.{type}.host.{hostname}
  • [ns.]jobs.{type}.label.{key}.{value...}

func ParseTarget 

func ParseTarget(
	target string,
) (routingType, key, value string)

ParseTarget parses a --target value into routing components. Returns routingType ("host", "label", AnyHost, or BroadcastHost), key, and value. Label values may contain dots for hierarchical targeting (e.g., "group:web.dev.us-east").

func SanitizeHostname 

func SanitizeHostname(
	hostname string,
) string

SanitizeHostname converts a hostname to a valid NATS consumer/routing name. NATS consumer names and routing must be alphanumeric with underscores only.

func ValidateLabel 

func ValidateLabel(
	key, value string,
) error

ValidateLabel checks that a label key and value are valid for use in NATS subjects. Keys must be a single segment matching [a-zA-Z0-9_-]+. Values may be hierarchical (dot-separated), where each segment matches [a-zA-Z0-9_-]+.

Types

type AgentInfo 

type AgentInfo struct {
	// MachineID is the permanent host identifier.
	MachineID string `json:"machine_id"`
	// Hostname is the hostname of the agent.
	Hostname string `json:"hostname"`
	// Labels are the key-value labels configured on the agent.
	Labels map[string]string `json:"labels,omitempty"`
	// RegisteredAt is the timestamp when the agent last registered (heartbeat).
	RegisteredAt time.Time `json:"registered_at"`
	// StartedAt is the timestamp when the agent process started.
	StartedAt time.Time `json:"started_at"`
	// OSInfo contains operating system information.
	OSInfo *host.Result `json:"os_info,omitempty"`
	// Uptime is the system uptime.
	Uptime time.Duration `json:"uptime,omitempty"`
	// LoadAverages contains the system load averages.
	LoadAverages *load.Result `json:"load_averages,omitempty"`
	// MemoryStats contains memory usage information.
	MemoryStats *mem.Result `json:"memory_stats,omitempty"`
	// AgentVersion is the version of the agent binary.
	AgentVersion string `json:"agent_version,omitempty"`
	// Architecture is the CPU architecture (e.g., x86_64, aarch64).
	Architecture string `json:"architecture,omitempty"`
	// KernelVersion is the kernel version string.
	KernelVersion string `json:"kernel_version,omitempty"`
	// CPUCount is the number of logical CPUs.
	CPUCount int `json:"cpu_count,omitempty"`
	// FQDN is the fully qualified domain name.
	FQDN string `json:"fqdn,omitempty"`
	// ServiceMgr is the init/service manager (e.g., systemd).
	ServiceMgr string `json:"service_mgr,omitempty"`
	// PackageMgr is the package manager (e.g., apt, yum).
	PackageMgr string `json:"package_mgr,omitempty"`
	// Interfaces contains network interface information.
	Interfaces []NetworkInterface `json:"interfaces,omitempty"`
	// PrimaryInterface is the name of the interface used for the default route.
	PrimaryInterface string `json:"primary_interface,omitempty"`
	// Routes contains the network routing table.
	Routes []Route `json:"routes,omitempty"`
	// Facts contains arbitrary key-value facts collected by the agent.
	Facts map[string]any `json:"facts,omitempty"`
	// Conditions contains the evaluated node conditions.
	Conditions []Condition `json:"conditions,omitempty"`
	// State is the agent's scheduling state (Ready, Draining, Cordoned, Pending).
	State string `json:"state,omitempty"`
	// Fingerprint is the SHA256 fingerprint of the agent's PKI public key.
	// Empty when PKI is disabled.
	Fingerprint string `json:"fingerprint,omitempty"`
	// Timeline contains the chronological sequence of state transition events.
	Timeline []TimelineEvent `json:"timeline,omitempty"`
}

AgentInfo represents information about an active agent.

type AgentRegistration 

type AgentRegistration struct {
	// MachineID is the permanent host identifier.
	MachineID string `json:"machine_id"`
	// Hostname is the hostname of the agent.
	Hostname string `json:"hostname"`
	// Labels are the key-value labels configured on the agent.
	Labels map[string]string `json:"labels,omitempty"`
	// RegisteredAt is the timestamp when the agent last registered.
	RegisteredAt time.Time `json:"registered_at"`
	// StartedAt is the timestamp when the agent process started.
	StartedAt time.Time `json:"started_at"`
	// OSInfo contains operating system information.
	OSInfo *host.Result `json:"os_info,omitempty"`
	// Uptime is the system uptime.
	Uptime time.Duration `json:"uptime,omitempty"`
	// LoadAverages contains the system load averages.
	LoadAverages *load.Result `json:"load_averages,omitempty"`
	// MemoryStats contains memory usage information.
	MemoryStats *mem.Result `json:"memory_stats,omitempty"`
	// AgentVersion is the version of the agent binary.
	AgentVersion string `json:"agent_version,omitempty"`
	// Process holds process-level resource usage.
	Process *ProcessMetrics `json:"process,omitempty"`
	// Conditions contains the evaluated node conditions.
	Conditions []Condition `json:"conditions,omitempty"`
	// State is the agent's scheduling state (Ready, Draining, Cordoned, Pending).
	State string `json:"state,omitempty"`
	// Fingerprint is the SHA256 fingerprint of the agent's PKI public key.
	// Empty when PKI is disabled.
	Fingerprint string `json:"fingerprint,omitempty"`
	// SubComponents reports the status of internal services.
	SubComponents map[string]SubComponentInfo `json:"sub_components,omitempty"`
}

AgentRegistration represents an agent's registration entry in the KV registry.

type AgentState 

type AgentState struct {
	Status    string    `json:"status"`
	Error     string    `json:"error,omitempty"`
	Duration  string    `json:"duration,omitempty"`
	StartTime time.Time `json:"start_time,omitempty"`
	EndTime   time.Time `json:"end_time,omitempty"`
}

AgentState represents the state of a specific agent processing a job

type CommandExecData 

type CommandExecData struct {
	// Command is the executable name or path
	Command string `json:"command"`
	// Args are the command arguments
	Args []string `json:"args,omitempty"`
	// Cwd is the optional working directory
	Cwd string `json:"cwd,omitempty"`
	// Timeout is the timeout in seconds
	Timeout int `json:"timeout,omitempty"`
}

CommandExecData represents data for direct command execution

type CommandShellData 

type CommandShellData struct {
	// Command is the full shell command string
	Command string `json:"command"`
	// Cwd is the optional working directory
	Cwd string `json:"cwd,omitempty"`
	// Timeout is the timeout in seconds
	Timeout int `json:"timeout,omitempty"`
}

CommandShellData represents data for shell command execution

type ComponentRegistration 

type ComponentRegistration struct {
	// Type is the component type: "controller" or "nats".
	Type string `json:"type"`
	// Hostname is the hostname of the component.
	Hostname string `json:"hostname"`
	// StartedAt is the timestamp when the component process started.
	StartedAt time.Time `json:"started_at"`
	// RegisteredAt is the timestamp of the last heartbeat.
	RegisteredAt time.Time `json:"registered_at"`
	// Process holds process-level resource usage.
	Process *ProcessMetrics `json:"process,omitempty"`
	// Conditions contains evaluated process conditions.
	Conditions []Condition `json:"conditions,omitempty"`
	// Version is the component binary version.
	Version string `json:"version,omitempty"`
	// SubComponents reports the status of internal services.
	SubComponents map[string]SubComponentInfo `json:"sub_components,omitempty"`
}

ComponentRegistration represents a non-agent component's heartbeat entry in the KV registry. Used by API server and NATS server.

type Condition 

type Condition struct {
	Type               string    `json:"type"`
	Status             bool      `json:"status"`
	Reason             string    `json:"reason,omitempty"`
	LastTransitionTime time.Time `json:"last_transition_time"`
}

Condition represents a node condition evaluated agent-side.

type DockerCreateData 

type DockerCreateData struct {
	Image     string            `json:"image"`
	Name      string            `json:"name,omitempty"`
	Hostname  string            `json:"hostname,omitempty"`
	DNS       []string          `json:"dns,omitempty"`
	Command   []string          `json:"command,omitempty"`
	Env       map[string]string `json:"env,omitempty"`
	Ports     []PortMapping     `json:"ports,omitempty"`
	Volumes   []VolumeMapping   `json:"volumes,omitempty"`
	AutoStart bool              `json:"auto_start,omitempty"`
}

DockerCreateData represents data for docker container creation.

type DockerExecData 

type DockerExecData struct {
	Command    []string          `json:"command"`
	Env        map[string]string `json:"env,omitempty"`
	WorkingDir string            `json:"working_dir,omitempty"`
}

DockerExecData represents data for executing a command in a docker container.

type DockerImageRemoveData 

type DockerImageRemoveData struct {
	Image string `json:"image"`
	Force bool   `json:"force,omitempty"`
}

DockerImageRemoveData represents data for removing a docker image.

type DockerListData 

type DockerListData struct {
	State string `json:"state,omitempty"`
	Limit int    `json:"limit,omitempty"`
}

DockerListData represents data for listing docker containers.

type DockerPullData 

type DockerPullData struct {
	Image string `json:"image"`
}

DockerPullData represents data for pulling a docker image.

type DockerRemoveData 

type DockerRemoveData struct {
	Force bool `json:"force,omitempty"`
}

DockerRemoveData represents data for removing a docker container.

type DockerStopData 

type DockerStopData struct {
	Timeout *int `json:"timeout,omitempty"`
}

DockerStopData represents data for stopping a docker container.

type FactsRegistration 

type FactsRegistration struct {
	Architecture     string             `json:"architecture,omitempty"`
	KernelVersion    string             `json:"kernel_version,omitempty"`
	CPUCount         int                `json:"cpu_count,omitempty"`
	FQDN             string             `json:"fqdn,omitempty"`
	ServiceMgr       string             `json:"service_mgr,omitempty"`
	PackageMgr       string             `json:"package_mgr,omitempty"`
	Containerized    bool               `json:"containerized"`
	Interfaces       []NetworkInterface `json:"interfaces,omitempty"`
	PrimaryInterface string             `json:"primary_interface,omitempty"`
	Routes           []Route            `json:"routes,omitempty"`
	Facts            map[string]any     `json:"facts,omitempty"`
}

FactsRegistration represents an agent's facts entry in the facts KV bucket.

type FileState 

type FileState struct {
	ObjectName   string            `json:"object_name"`
	Path         string            `json:"path"`
	SHA256       string            `json:"sha256"`
	Mode         string            `json:"mode,omitempty"`
	Owner        string            `json:"owner,omitempty"`
	Group        string            `json:"group,omitempty"`
	DeployedAt   string            `json:"deployed_at"`
	ContentType  string            `json:"content_type"`
	UndeployedAt string            `json:"undeployed_at,omitempty"`
	Metadata     map[string]string `json:"metadata,omitempty"`
}

FileState represents a deployed file's state in the file-state KV. Keyed by <hostname>.<sha256-of-path>.

type HostnameProvider 

type HostnameProvider interface {
	Hostname() (string, error)
}

HostnameProvider defines the interface for getting hostname

type NetworkDNSUpdateData 

type NetworkDNSUpdateData struct {
	// DNSServers is a list of DNS server IP addresses (IPv4 or IPv6)
	DNSServers []string `json:"dns_servers"`
	// SearchDomains is a list of search domains for DNS resolution
	SearchDomains []string `json:"search_domains"`
	// InterfaceName is the name of the network interface to apply DNS settings to
	InterfaceName string `json:"interface_name"`
}

NetworkDNSUpdateData represents data for DNS configuration changes

type NetworkInterface 

type NetworkInterface struct {
	Name   string `json:"name"`
	IPv4   string `json:"ipv4,omitempty"`
	IPv6   string `json:"ipv6,omitempty"`
	MAC    string `json:"mac,omitempty"`
	Family string `json:"family,omitempty"`
}

NetworkInterface represents a network interface with its address.

type NetworkPingExecuteData 

type NetworkPingExecuteData struct {
	// Target is the hostname or IP address to ping
	Target string `json:"target"`
	// Count is the number of ping packets to send (optional, default: 4)
	Count int `json:"count,omitempty"`
	// Timeout is the timeout duration in seconds (optional, default: 5)
	Timeout int `json:"timeout,omitempty"`
}

NetworkPingExecuteData represents data for ping operations

type NodeDiskResponse 

type NodeDiskResponse struct {
	Disks []disk.Result `json:"disks"`
}

NodeDiskResponse represents the response for node.disk.get operations.

type NodeHostnameGetData 

type NodeHostnameGetData struct {
}

NodeHostnameGetData represents data for hostname retrieval

type NodeShutdownData 

type NodeShutdownData struct {
	// Action specifies whether to reboot or shutdown the system
	Action string `json:"action"` // "reboot" or "shutdown"
	// DelaySeconds is an optional field to specify a delay in seconds before reboot/shutdown
	DelaySeconds int32 `json:"delay_seconds,omitempty"`
	// Message is an optional message to log or display before reboot/shutdown
	Message string `json:"message,omitempty"`
}

NodeShutdownData represents data for node shutdown/reboot operations

type NodeStatusResponse 

type NodeStatusResponse struct {
	// Hostname from the host provider
	Hostname string `json:"hostname"`
	// Uptime from the host provider
	Uptime time.Duration `json:"uptime"`
	// OSInfo from the host provider
	OSInfo *host.Result `json:"os_info"`
	// LoadAverages from the load provider
	LoadAverages *load.Result `json:"load_averages"`
	// MemoryStats from the memory provider
	MemoryStats *mem.Result `json:"memory_stats"`
	// DiskUsage from the disk provider
	DiskUsage []disk.Result `json:"disk_usage"`
}

NodeStatusResponse aggregates node status information from multiple providers. This represents the response for node.status.get operations in the job queue.

type NodeUptimeResponse 

type NodeUptimeResponse struct {
	UptimeSeconds float64 `json:"uptime_seconds"`
	Uptime        string  `json:"uptime"`
}

NodeUptimeResponse represents the response for node.uptime.get operations.

type Operation 

type Operation struct {
	// Type specifies the type of operation using hierarchical format
	// (e.g., "node.hostname.get", "network.dns.update")
	Type OperationType `json:"type"`
	// Data contains the operation-specific data as raw JSON
	Data json.RawMessage `json:"data"`
}

Operation represents an operation in the new hierarchical format

type OperationType 

type OperationType = client.JobOperation

OperationType is a type alias for client.JobOperation.

type PortMapping 

type PortMapping struct {
	Host      int `json:"host"`
	Container int `json:"container"`
}

PortMapping maps a host port to a container port (job layer). Intentionally duplicated from runtime.PortMapping to keep the job layer decoupled from the provider layer. Both have the same shape.

type ProcessMetrics 

type ProcessMetrics struct {
	// CPUPercent is the process CPU usage as a percentage.
	CPUPercent float64 `json:"cpu_percent"`
	// RSSBytes is the resident set size in bytes.
	RSSBytes int64 `json:"rss_bytes"`
	// Goroutines is the number of active goroutines.
	Goroutines int `json:"goroutines"`
}

ProcessMetrics holds process-level resource usage.

type QueueStats 

type QueueStats struct {
	TotalJobs    int            `json:"total_jobs"`
	StatusCounts map[string]int `json:"status_counts"`
	DLQCount     int            `json:"dlq_count"`
}

QueueStats represents statistics about the job queue.

type QueuedJob 

type QueuedJob struct {
	// ID is the unique identifier for this job
	ID string `json:"id"`
	// Status tracks the current state of the job
	Status string `json:"status"` // "unprocessed", "processing", "completed", "failed"
	// Created is the timestamp when the job was created
	Created string `json:"created"`
	// Subject is the NATS subject for this job (optional)
	Subject string `json:"subject,omitempty"`
	// Operation contains the actual work to be performed (stored as flexible JSON)
	Operation map[string]interface{} `json:"operation"`
	// StatusHistory tracks status transitions (optional)
	StatusHistory []interface{} `json:"status_history,omitempty"`
	// Result contains the output when the job is completed (optional)
	Result json.RawMessage `json:"result,omitempty"`
	// Changed indicates whether the operation modified system state.
	// Nil for query operations; set for mutation operations.
	Changed *bool `json:"changed,omitempty"`
	// Error contains error details if the job failed (optional)
	Error string `json:"error,omitempty"`
	// Hostname identifies which agent processed this job (optional)
	Hostname string `json:"hostname,omitempty"`
	// UpdatedAt is the timestamp when the job was last updated (optional)
	UpdatedAt string `json:"updated_at,omitempty"`
	// AgentStates contains detailed state for each agent that processed this job
	AgentStates map[string]AgentState `json:"agent_states,omitempty"`
	// Timeline contains the chronological sequence of events for this job
	Timeline []TimelineEvent `json:"timeline,omitempty"`
	// Responses contains the actual response data from each agent
	Responses map[string]Response `json:"responses,omitempty"`
}

QueuedJob represents a job stored in the KV queue with metadata

type Request 

type Request struct {
	// JobID is a unique identifier for this job.
	JobID string `json:"job_id"`
	// Type specifies whether this is a query or modify operation.
	Type Type `json:"type"`
	// Category specifies the operation category (node, network, etc.).
	Category string `json:"category"`
	// Operation specifies the specific operation to perform.
	Operation OperationType `json:"operation"`
	// Data contains operation-specific parameters as raw JSON.
	Data json.RawMessage `json:"data,omitempty"`
	// Timestamp indicates when the request was created.
	Timestamp time.Time `json:"timestamp"`
}

Request represents a request to perform a job operation.

type Response 

type Response struct {
	// JobID matches the original job ID.
	JobID string `json:"job_id"`
	// Status indicates the job completion status.
	Status Status `json:"status"`
	// Data contains the operation results as raw JSON.
	Data json.RawMessage `json:"data,omitempty"`
	// Error contains error information if the job failed.
	Error string `json:"error,omitempty"`
	// Changed indicates whether the operation modified system state.
	// Nil for query operations; set for mutation operations.
	Changed *bool `json:"changed,omitempty"`
	// Hostname identifies which agent processed this job.
	Hostname string `json:"hostname"`
	// Timestamp indicates when the response was created.
	Timestamp time.Time `json:"timestamp"`
}

Response represents the response from a job operation.

type Route 

type Route struct {
	Destination string `json:"destination"`
	Gateway     string `json:"gateway"`
	Interface   string `json:"interface"`
	Mask        string `json:"mask,omitempty"`
	Metric      int    `json:"metric,omitempty"`
	Flags       string `json:"flags,omitempty"`
}

Route represents a network routing table entry.

type SignedEnvelope 

type SignedEnvelope struct {
	// Payload is the raw JSON payload (job data or response data).
	Payload []byte `json:"payload"`
	// Signature is the Ed25519 signature of the payload.
	Signature []byte `json:"signature"`
	// Fingerprint is the SHA256 fingerprint of the signer's public key.
	Fingerprint string `json:"fingerprint"`
}

SignedEnvelope wraps a job or response payload with an Ed25519 signature.

type Status 

type Status = client.JobStatus

Status represents the current status of a job. Status is a type alias for client.JobStatus so internal code and the SDK share the same type. All status constants are defined in pkg/sdk/client/.

type SubComponentInfo 

type SubComponentInfo struct {
	// Status is the sub-component status (e.g., "ok", "disabled").
	Status string `json:"status"`
	// Address is the optional network endpoint (e.g., "http://0.0.0.0:9090").
	Address string `json:"address,omitempty"`
}

SubComponentInfo holds the status and optional address of a sub-component.

type TimelineEvent 

type TimelineEvent struct {
	Timestamp time.Time `json:"timestamp"`
	Event     string    `json:"event"`
	Hostname  string    `json:"hostname"`
	Message   string    `json:"message"`
	Error     string    `json:"error,omitempty"`
}

TimelineEvent represents a single event in the job timeline

type Type 

type Type = client.JobType

Type is a type alias for client.JobType.

type VolumeMapping 

type VolumeMapping struct {
	Host      string `json:"host"`
	Container string `json:"container"`
}

VolumeMapping maps a host path to a container path (job layer). Intentionally duplicated from runtime.VolumeMapping for the same reason.

Source Files

View all Source files

Directories

Path Synopsis
Package client provides job client operations for NATS JetStream.
 Package client provides job client operations for NATS JetStream.
Package mocks provides mock implementations for testing.
 Package mocks provides mock implementations for testing.
hostname
Package hostname provides mock implementations for HostnameProvider testing.
 Package hostname provides mock implementations for HostnameProvider testing.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.