Documentation
¶
Overview ¶
Package worker runs predefined shell commands dispatched by the primary over the hub transport (see hub.go/hub_agent.go).
A factum-worker instance activates every entry in ConfigWorker.Commands, keyed by name - there's no separate addressing by node name, and any instance with a given command name will run it if the primary dispatches to that name. If the name is also a valid sync target (worker.IsValidSyncTarget), the same predefined command also runs whenever someone clicks that target's "Sync" button on the web UI's sync overview page - web.ApiSyncTrigger and "factum-worker run" both go through the exact same RemoteManager.SendCommand/RunAndWait dispatch path on the primary, so there's no separate mechanism to keep in sync.
The primary (web.RemoteManager, hub.go) dials out to this instance's hub listener (ConfigWorker.Listen) rather than the other way round - see AGENTS.md's "Worker / hub transport" section for why. "Which instances are up and what they handle" is answered by the primary's own live connection state (web.ApiWorkerStatus), not a broadcast the instance has to participate in.
An agent never executes a command line received from the primary directly - only the name is used to look up a predefined command, and an instance only ever runs commands from its own ConfigWorker.Commands allowlist, so a forged or replayed message can at most trigger one of the commands the operator already defined for that instance.
Index ¶
- Constants
- Variables
- func EnabledSyncTargets(s *models.Settings) []string
- func IsValidSyncTarget(target string) bool
- func LogToSlog(msg LogMsg)
- func RunRemote(ctx context.Context, cfg *util.ConfigFactum, command string, args []string, ...) (exitCode int, err error)
- func SequencedSyncAllTargets(targets []string) []string
- type CommandMsg
- type Envelope
- type EnvelopeType
- type EventMsg
- type HelloMsg
- type LogMsg
- type NodeStatus
- type RemoteManager
- func (m *RemoteManager) Run(ctx context.Context)
- func (m *RemoteManager) RunAndWait(ctx context.Context, role string, args []string, onLine func(LogMsg)) (exitCode int, err error)
- func (m *RemoteManager) SendCommand(role string, args []string) (matched int, id string, err error)
- func (m *RemoteManager) StartJob(jobType, triggeredBy string, targets []string) (job models.Job, results []TaskResult, err error)
- func (m *RemoteManager) StatusAll() map[string]NodeStatus
- type TaskResult
- type Worker
Constants ¶
const ( StreamStdout = "stdout" StreamStderr = "stderr" StreamExit = "exit" )
const HubPath = "/hub"
HubPath is the single route the agent-side listener (runHubListener) exposes, and the path RemoteManager dials.
Variables ¶
var ErrSyncAlreadyRunning = errors.New("sync already running for this target")
ErrSyncAlreadyRunning is returned by StartJob (single-target) or recorded against a conflicted task (batch) when target already has an undispatched-exit task in flight.
var SyncTargets = []string{"becs", "netbox", "lime", "librenms", "icinga", "oxidized", "dns", "device-sync"}
SyncTargets lists the systems that can be synced from the web UI. Each name doubles as the role/command name a factum-worker instance activates to handle it - each one matches a corresponding "factum-<name> sync" CLI command.
Functions ¶
func EnabledSyncTargets ¶
EnabledSyncTargets filters SyncTargets down to the ones activated in Settings (the same *_enabled switches the admin Settings page edits). A nil pointer (the zero value until an admin explicitly flips the switch) means "not enabled".
func IsValidSyncTarget ¶
func LogToSlog ¶
func LogToSlog(msg LogMsg)
LogToSlog converts an agent's LogMsg into the equivalent slog call - called directly by connectOnce for every inbound log envelope, which is the entire integration into the web GUI's log window: hubHandler (web/logstream.go) already tees every slog record into the LogHub the frontend subscribes to.
func RunRemote ¶
func RunRemote(ctx context.Context, cfg *util.ConfigFactum, command string, args []string, onLine func(LogMsg)) (exitCode int, err error)
RunRemote asks the primary (cfg.URL, authenticated with cfg.Token - same fields internal/factum.FactumClient uses) to run command via RemoteManager.RunAndWait, calling onLine for each streamed LogMsg. Returns the exit code from the command's final StreamExit line, or an error if the request failed outright or the stream ended without ever seeing one (e.g. the connection dropped mid-run). ctx cancellation (e.g. Ctrl-C) aborts the wait - it does not cancel the already-dispatched remote command, matching the old rabbitmq-based CLI's behavior.
func SequencedSyncAllTargets ¶
SequencedSyncAllTargets orders targets (expected to already be filtered to enabled ones via EnabledSyncTargets) sources-first, destinations- second, preserving each group's relative order - the dispatch order StartJob's sequential batch path uses for "Sync all", so a destination sync never races a source sync that's still refreshing the data it reads.
Types ¶
type CommandMsg ¶
type CommandMsg struct {
ID string `json:"id"`
Command string `json:"command"`
Args []string `json:"args,omitempty"`
}
CommandMsg is sent by the primary (RemoteManager) to run a predefined command, routed by the agent's activated role name (CommandMsg.Command).
type Envelope ¶
type Envelope struct {
Type EnvelopeType `json:"type"`
Payload json.RawMessage `json:"payload,omitempty"`
}
Envelope wraps every message exchanged over a hub connection.
type EnvelopeType ¶
type EnvelopeType string
const ( EnvelopeHello EnvelopeType = "hello" EnvelopeCommand EnvelopeType = "command" EnvelopeLog EnvelopeType = "log" EnvelopeEvent EnvelopeType = "event" )
type EventMsg ¶
type EventMsg struct {
ID string `json:"id"`
Target string `json:"target"`
Level string `json:"level"`
Message string `json:"message"`
}
EventMsg is a structured info/warning/error line reported by a sync tool (internal/jobevent.Reporter) via its predefined command's stdout, when invoked with --job. ID/Target mirror LogMsg's ID/Command and are filled in by the agent (hub_agent.go's streamOutput) - the subprocess itself only ever knows its own Level/Message, not its dispatch ID.
type HelloMsg ¶
HelloMsg is sent by the agent immediately after the connection is established - the agent is always the one who knows its own hostname/roles, regardless of which side dialed.
type LogMsg ¶
type LogMsg struct {
ID string `json:"id"`
Command string `json:"command"`
Stream string `json:"stream"` // StreamStdout, StreamStderr or StreamExit
Data string `json:"data,omitempty"`
ExitCode int `json:"exit_code,omitempty"`
Err string `json:"err,omitempty"`
}
LogMsg is sent by the agent while (and after) it runs a command.
type NodeStatus ¶
type NodeStatus struct {
Connected bool
Hostname string
Roles []string
LastSeen time.Time
LastError string
}
NodeStatus is a WorkerNode's live connection state.
type RemoteManager ¶
type RemoteManager struct {
// contains filtered or unexported fields
}
RemoteManager dials out to every enabled models.WorkerNode and keeps one supervised connection per node alive, reconnecting with backoff - the primary-side half of the hub transport. internal/worker.Worker's runHubListener (hub_agent.go) is the agent-side half.
func NewRemoteManager ¶
func NewRemoteManager(db *gorm.DB) *RemoteManager
func (*RemoteManager) Run ¶
func (m *RemoteManager) Run(ctx context.Context)
Run reconciles the configured WorkerNode set against running dial-loops every reconcileInterval, until ctx is cancelled. Matches the lifecycle of web.GUI()'s other background loops: launched once with a long-lived context and no explicit shutdown path.
func (*RemoteManager) RunAndWait ¶
func (m *RemoteManager) RunAndWait(ctx context.Context, role string, args []string, onLine func(LogMsg)) (exitCode int, err error)
RunAndWait sends role/args and blocks until the first matching StreamExit line arrives, ctx is cancelled, or zero nodes matched - mirroring the old Worker.RunAndWait's "wait for the first responder" semantics (multiple agents can be bound to the same role).
The waiter is registered *before* the command is actually sent - not after - since a fast agent response could otherwise arrive before anything is listening for it, and deliverToWaiter silently drops a message with no registered waiter.
func (*RemoteManager) SendCommand ¶
SendCommand dispatches a fire-and-forget command to every connected node activated for role - used by sync-trigger, which doesn't wait for a result. See RunAndWait for the wait-for-completion variant.
func (*RemoteManager) StartJob ¶
func (m *RemoteManager) StartJob(jobType, triggeredBy string, targets []string) (job models.Job, results []TaskResult, err error)
StartJob creates a Job row and dispatches one JobTask per target - the single entry point both a single-target trigger (len(targets)==1) and "sync all" (every enabled target) go through, per the "every Job is a main job with one or more subjobs" model.
len(targets)==1 preserves exact single-target semantics: a busy target returns ErrSyncAlreadyRunning and nothing is persisted at all, not even the parent Job row. len(targets)>1 always creates the parent Job row first and lets individual targets fail/conflict independently (each still gets a visible, failed JobTask row, see startOneTask) rather than aborting the whole batch - but unlike a single-target call, targets are dispatched one at a time, each waited on to finish before the next starts (dispatchRemainingSequentially), so a batch job finishes in sum(subjob durations) rather than max(subjob durations). This deliberately restores the old client-side "sync all" loop's await-each- in-turn behavior, just moved server-side (see web.ApiSyncTriggerAll and worker.SequencedSyncAllTargets, which orders targets sources-first so a destination sync doesn't run - sequentially or otherwise - ahead of the source sync that's supposed to feed it) - two targets actually running their sync at the same time was the thing this restores protection against, not merely the order they're triggered in.
The first target is dispatched synchronously, so a caller gets an immediate TaskResult/error for it exactly as before; the remaining targets (if any) are dispatched from a background goroutine the caller doesn't wait on, since a batch can take much longer than an HTTP request should block for.
func (*RemoteManager) StatusAll ¶
func (m *RemoteManager) StatusAll() map[string]NodeStatus
StatusAll returns a snapshot of every node's last-known status, keyed by WorkerNode.Name - used by web.ApiWorkerStatus alongside the DB's Address/Enabled to build the sync-status page's response.
type TaskResult ¶
TaskResult is one target's outcome within a StartJob call - Err is set if that target's dispatch failed or conflicted with an already-running task for the same target (batch jobs record this against the target's JobTask row and keep going, rather than aborting the whole request - see startOneTask). Matched is 0 if dispatch found no connected node for it (also recorded against the JobTask row, via createAndDispatchTask).
Source Files