worker

package
v1.1.1 Latest Latest
Warning

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

 Go to latest Published: Sep 20, 2026 License: AGPL-3.0 Imports: 35 Imported by: 0

Documentation

Overview

Package worker runs predefined shell commands dispatched by the primary over the hub transport (see hub.go/hub_agent.go).

A factum2-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 "factum2-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

View Source 
const (
	StreamStdout = "stdout"
	StreamStderr = "stderr"
	StreamExit   = "exit"
)
View Source 
const DNSRole = "dns"

DNSRole is the worker.commands key on the DNS dest (BIND/Kea). The hub uses CallRole this name for dest-local reads such as DHCP leases.

View Source 
const HousekeepingTarget = "housekeeping"

HousekeepingTarget is the in-process job that trims persisted job history (internal/housekeeping). It is a valid StartJob / scheduler target, but not a SyncTargets entry: it is never dispatched to a worker, never included in "Sync all", and is not gated by a Settings enable switch. Operators schedule it themselves.

View Source 
const HubPath = "/hub"

HubPath is the single route the agent-side listener (runHubListener) exposes, and the path RemoteManager dials.

View Source 
const StorageRole = "storage"

StorageRole is the worker.commands key a node activates to host factum2-storage. GUI file ops CallRole this name; copy uses CommandMsg extra args on the same command.

Variables

View Source 
var ErrHubDisconnected = errors.New("hub disconnected")

ErrHubDisconnected is returned by DoHubRequest when no hub session is current. The unix server maps it (and every other transport error) to 502.

View Source 
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.

View Source 
var SyncTargets = []string{"becs", "netbox", "lime", "librenms", "icinga", "oxidized", "prometheus", "dns", "certs", "device-sync"}

SyncTargets lists the systems that can be synced from the web UI. Each name doubles as the role/command name a factum2-worker instance activates to handle it - each one matches a corresponding "factum2-<name> sync" CLI command.

Functions

func AllowHubAPI added in v1.0.2 

func AllowHubAPI(method, path string) bool

AllowHubAPI reports whether method+path (query already stripped, path already normalized) is permitted over the hub.

func EnabledSyncTargets 

func EnabledSyncTargets(s *models.Settings) []string

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 EventToSlog added in v1.0.6 

func EventToSlog(msg EventMsg)

EventToSlog mirrors LogToSlog for structured job events (info/warning/ error lines from a sync tool). Without this, those lines only land in JobTaskEvent (the job-detail modal) and the live log window never sees them tagged with their target.

func IsHubAuth added in v1.0.2 

func IsHubAuth(ctx context.Context) bool

func IsValidJobTarget added in v1.0.4 

func IsValidJobTarget(target string) bool

IsValidJobTarget is the set StartJob / ApiSyncTrigger / the scheduler will accept: every sync target, plus housekeeping.

func IsValidSyncTarget 

func IsValidSyncTarget(target string) bool

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. "command" is the sync target (librenms, netbox, …) the log window uses as the line's source.

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 

func SequencedSyncAllTargets(targets []string) []string

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.

func WithHubAuth added in v1.0.2 

func WithHubAuth(ctx context.Context) context.Context

WithHubAuth marks ctx as a hub-authenticated in-process request. Network clients cannot set this; it is equivalent to a service token, not an admin session.

Types

type CallMsg added in v1.1.1 

type CallMsg struct {
	ID     string            `json:"id"`
	Method string            `json:"method"`
	Path   string            `json:"path"`
	Header map[string]string `json:"header,omitempty"`
	Body   []byte            `json:"body,omitempty"`
}

CallMsg is a primary→agent HTTP-subset proxy. Paths other than /dhcp/leases are forwarded to the storage unix socket.

type CallResultMsg added in v1.1.1 

type CallResultMsg struct {
	ID     string            `json:"id"`
	Status int               `json:"status"`
	Header map[string]string `json:"header,omitempty"`
	Body   []byte            `json:"body,omitempty"`
	Error  string            `json:"error,omitempty"`
}

CallResultMsg is the matching reply.

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"`
	// contains filtered or unexported fields
}

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"
	EnvelopeRequest    EnvelopeType = "request"     // agent → primary
	EnvelopeResponse   EnvelopeType = "response"    // primary → agent
	EnvelopeCall       EnvelopeType = "call"        // primary → agent (unix HTTP proxy)
	EnvelopeCallResult EnvelopeType = "call_result" // agent → primary
)

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 

type HelloMsg struct {
	Hostname string   `json:"hostname"`
	Roles    []string `json:"roles"`
	Version  string   `json:"version"`
	Commit   string   `json:"commit"`
}

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. Version/Commit are buildinfo identity; the primary refuses to register the node unless they match its own process (see checkHubVersion). Unstamped `go run` builds and APP_ENV=development skip that check.

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
	Version   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) CallRole added in v1.1.1 

func (m *RemoteManager) CallRole(ctx context.Context, role, method, path string, header map[string]string, body []byte) (CallResultMsg, error)

CallRole sends one HTTP-subset call to a single connected node whose hello roles include role (typically StorageRole), and waits for the result. The agent forwards it to its local factum2-storage unix socket.

func (*RemoteManager) HandleHubRequest added in v1.0.2 

func (m *RemoteManager) HandleHubRequest(ctx context.Context, node string, outbox chan Envelope, req RequestMsg)

HandleHubRequest runs one hub RPC against the attached API handler and writes the response envelope to outbox. outbox must be this connection's: looking up m.conns again could send the reply on a newer conn.

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 

func (m *RemoteManager) SendCommand(role string, args []string) (matched int, id string, err error)

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) SetAPIHandler added in v1.0.2 

func (m *RemoteManager) SetAPIHandler(h http.Handler)

SetAPIHandler attaches the primary's Echo router for in-process hub RPC. Must be called after routes are registered and before Run, so a connected worker never observes a nil handler.

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 RequestMsg added in v1.0.2 

type RequestMsg struct {
	ID     string          `json:"id"`
	Method string          `json:"method"`
	Path   string          `json:"path"`
	Body   json.RawMessage `json:"body,omitempty"`
}

RequestMsg is an HTTP-subset RPC. Path is RequestURI (path + query), e.g. "/api/device?include=interfaces" or "/api/device/name/core-sw1".

type ResponseMsg added in v1.0.2 

type ResponseMsg struct {
	ID     string          `json:"id"`
	Status int             `json:"status"`
	Body   json.RawMessage `json:"body,omitempty"`
	Error  string          `json:"error,omitempty"`
}

ResponseMsg is the matching RPC reply. Error is transport-level (unix 502); HTTP 4xx/5xx use Status+Body with Error empty.

type TaskResult 

type TaskResult struct {
	Target  string
	TaskID  string
	Matched int
	Err     error
}

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).

type Worker 

type Worker struct {
	// contains filtered or unexported fields
}

func New 

func New(cfg *util.ConfigWorker) *Worker

func (*Worker) DoHubRequest added in v1.0.2 

func (w *Worker) DoHubRequest(ctx context.Context, method, path string, body []byte) (status int, respBody []byte, err error)

DoHubRequest sends one HTTP-subset RPC over the current hub session and waits for the matching response. ResponseMsg.Error is always returned as err (unix 502), never as HTTP 200 with an error string.

func (*Worker) SetStorageSocket added in v1.1.1 

func (w *Worker) SetStorageSocket(socket string)

func (*Worker) Start 

func (w *Worker) Start(ctx context.Context) error

Start validates this instance's command allowlist and runs the hub listener and unix API until ctx is cancelled or either fails.

Source Files

View all Source files

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.