actions

package
v0.5.0 Latest Latest
Warning

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

 Go to latest Published: Aug 3, 2026 License: Apache-2.0 Imports: 27 Imported by: 0

Documentation

Overview

Package actions provides remote action execution and hook management for plexd mesh nodes.

Index

Constants

View Source 
const DefaultHooksDir = "/etc/plexd/hooks"

DefaultHooksDir is the default directory for hook scripts.

View Source 
const DefaultMaxActionTimeout = 10 * time.Minute

DefaultMaxActionTimeout is the default maximum duration for a single action.

View Source 
const DefaultMaxConcurrent = 5

DefaultMaxConcurrent is the default maximum number of concurrent actions.

View Source 
const DefaultMaxOutputBytes = 1 << 20

DefaultMaxOutputBytes is the default maximum output size per action (1 MiB).

View Source 
const MaxSnapshotLines = 10000

MaxSnapshotLines is the maximum number of lines that logs.snapshot will return.

Variables

View Source 
var ErrDispatchDeferred = errors.New("actions: dispatch deferred")

ErrDispatchDeferred reports that a dispatch has not been settled: local backpressure prevented it — shutdown, a run already in flight under the same id, or a saturated concurrency slot — or a transient control-plane failure cut a callback sequence short before it resolved the execution. It is not a failure of the execution: the pull's executions block redelivers the entry, so the caller must retry it on a later cycle instead of suppressing it.

Functions

func DiscoverHooks 

func DiscoverHooks(hooksDir string, logger *slog.Logger) ([]api.HookInfo, error)

DiscoverHooks scans hooksDir for executable files and returns their metadata. Returns an empty slice (not nil) and no error if the directory does not exist. Individual file errors (hash failures, unreadable sidecars) are logged at warn level but do not prevent discovery of other hooks.

Types

type ActionReporter 

type ActionReporter interface {
	ExecutionCallback(ctx context.Context, nodeID, executionID string, req api.ExecutionCallbackRequest) (*api.ExecutionCallbackResponse, error)
	UploadExecutionOutput(ctx context.Context, uploadURL string, output []byte) error
}

ActionReporter abstracts control plane communication for testability.

type BuiltinFunc 

type BuiltinFunc func(ctx context.Context, params map[string]string) (stdout string, stderr string, exitCode int, err error)

BuiltinFunc is the signature for built-in action implementations. It receives a context (with timeout deadline) and parameters, and returns stdout, stderr, an exit code, and an optional error.

func ConfigDump 

func ConfigDump(provider ConfigProvider) BuiltinFunc

ConfigDump returns a BuiltinFunc that outputs the sanitized configuration.

func DiagnosticsCollect 

func DiagnosticsCollect() BuiltinFunc

DiagnosticsCollect returns a BuiltinFunc that collects system diagnostics and returns them as JSON. Optional parameters: "include_network" (default "true"), "include_processes" (default "true").

func DiagnosticsTraceroutePeer 

func DiagnosticsTraceroutePeer(info NodeInfoProvider) BuiltinFunc

DiagnosticsTraceroutePeer returns a BuiltinFunc that runs traceroute to a mesh peer. Requires a "peer_id" parameter (mesh IP). Optional "max_hops" parameter (default 15).

func GatherInfo 

func GatherInfo(info NodeInfoProvider) BuiltinFunc

GatherInfo returns a BuiltinFunc that collects system information and returns it as JSON. The output includes: hostname, os, arch, go_version, mesh_ip, peer_count, node_id.

func HealthCheck 

func HealthCheck(health HealthProvider) BuiltinFunc

HealthCheck returns a BuiltinFunc that reports the node's health status. Optional parameter: "include_peers" (default "true") — include per-peer status. Status is "healthy" if tunnel_count > 0, otherwise "degraded".

func LogsSnapshot 

func LogsSnapshot(provider LogProvider) BuiltinFunc

LogsSnapshot returns a BuiltinFunc that retrieves recent log lines. Accepts optional parameters:

  • "lines": number of lines to return (default 100, max 10000)
  • "since": duration string (e.g. "5m", "1h") to filter lines by age

func MeshReconnect 

func MeshReconnect(reconnector MeshReconnector) BuiltinFunc

MeshReconnect returns a BuiltinFunc that triggers mesh reconnection. On failure, returns exit code 1 with error details but no system error.

func PingPeer 

func PingPeer(info NodeInfoProvider) BuiltinFunc

PingPeer returns a BuiltinFunc that pings a mesh peer and reports latency. Requires a "peer_id" parameter (mesh IP). Optional "count" parameter (default 1).

func ServiceReloadConfig 

func ServiceReloadConfig() BuiltinFunc

ServiceReloadConfig returns a BuiltinFunc that sends SIGHUP to the current process to trigger a configuration reload.

func ServiceRestart 

func ServiceRestart() BuiltinFunc

ServiceRestart returns a BuiltinFunc that restarts the plexd service via systemctl.

func ServiceUpgrade 

func ServiceUpgrade(fetcher ReleaseFetcher, verifier BundleVerifier) BuiltinFunc

ServiceUpgrade returns a BuiltinFunc that performs an in-place binary upgrade from the GitHub release channel.

It downloads the target release binary, verifies its SHA-256 checksum against the dispatched value, then downloads and verifies the release's Sigstore bundle against the fetched binary's digest. Only after both checks pass is the current binary made executable, atomically replaced, and a systemd restart triggered. A release without a bundle asset (the fetch fails) or one whose bundle fails verification is refused, and the on-disk binary is left untouched.

Required parameters:

  • version: target version string (e.g. "1.5.0")
  • checksum: expected SHA-256 checksum of the new binary (hex-encoded, with or without "sha256:" prefix)

type BundleVerifier added in v0.2.0 

type BundleVerifier interface {
	Verify(bundleJSON []byte, sha256Hex string) error
}

BundleVerifier verifies a Sigstore bundle against an artifact digest.

type Config 

type Config struct {
	// Enabled controls whether action execution is active.
	// nil means use default (true); explicit false disables execution.
	//
	// The tri-state is what makes the switch usable: with a plain bool an
	// operator's `enabled: false` is indistinguishable from an omitted key,
	// and defaulting turns it back on. Enabled is the only switch that stops
	// the control plane from running actions and hooks on the node, so it has
	// to survive ApplyDefaults exactly as written.
	Enabled *bool `yaml:"enabled"`

	// HooksDir is the directory containing hook scripts.
	// Default: /etc/plexd/hooks
	HooksDir string `yaml:"hooks_dir"`

	// MaxConcurrent is the maximum number of actions that can run concurrently.
	// Must be at least 1 when enabled. Default: 5.
	MaxConcurrent int `yaml:"max_concurrent"`

	// MaxActionTimeout is the maximum duration for a single action.
	// Must be at least 10s when enabled. Default: 10m.
	MaxActionTimeout time.Duration `yaml:"max_action_timeout"`

	// MaxOutputBytes is the maximum output size per action in bytes.
	// Must be at least 1024 when enabled. Default: 1 MiB.
	MaxOutputBytes int64 `yaml:"max_output_bytes"`
}

Config holds the configuration for remote action execution.

func (*Config) ApplyDefaults 

func (c *Config) ApplyDefaults()

ApplyDefaults sets default values for zero-valued fields.

func (*Config) IsEnabled added in v0.3.0 

func (c *Config) IsEnabled() bool

IsEnabled returns the effective Enabled setting: true unless explicitly set to false.

func (Config) MarshalYAML added in v0.3.0 

func (c Config) MarshalYAML() (any, error)

MarshalYAML renders the effective Enabled value so a dump of the live config never reports the switch that gates remote execution as `enabled: null`. config.dump is what an operator reads to audit which nodes accept control-plane-driven execution, and a null there reads as "unset" — which for this field reads as off, while it means on.

func (*Config) Validate 

func (c *Config) Validate() error

Validate checks that configuration values are within acceptable ranges.

type ConfigProvider 

type ConfigProvider interface {
	DumpConfig() string
}

ConfigProvider supplies sanitized configuration for dumping.

type Dispatcher added in v0.3.0 

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

Dispatcher consumes the executions block of the reconciliation pull and turns each entry into an execution on the Executor. The block is a delivery queue, not desired state: an entry keeps reappearing on every pull until its execution reaches a terminal status through the callback, so suppressing the re-observations is the node's job.

A Dispatcher is not safe for concurrent use. Handle is invoked only from the reconcile goroutine, one cycle at a time, so handled needs no mutex.

func NewDispatcher added in v0.3.0 

func NewDispatcher(executor *Executor, nodeID string, logger *slog.Logger) *Dispatcher

NewDispatcher creates a Dispatcher that dispatches the pull's executions block through executor on behalf of nodeID.

func (*Dispatcher) Handle added in v0.3.0 

func (d *Dispatcher) Handle(ctx context.Context, desired *api.NodeStateSnapshot)

Handle dispatches every entry of the snapshot's executions block, in block order. Its signature matches reconcile.DispatchHandler.

type Executor 

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

Executor orchestrates action execution, concurrency control, and result reporting.

func NewExecutor 

func NewExecutor(cfg Config, reporter ActionReporter, verifier HookVerifier, logger *slog.Logger) *Executor

NewExecutor creates an Executor with the given configuration, reporter, verifier, and logger.

func (*Executor) ActiveCount 

func (e *Executor) ActiveCount() int

ActiveCount returns the number of currently running actions.

func (*Executor) Capabilities 

func (e *Executor) Capabilities() ([]api.ActionInfo, []api.HookInfo)

Capabilities returns builtin action metadata and hooks for capability reporting.

func (*Executor) Execute 

func (e *Executor) Execute(ctx context.Context, nodeID string, entry api.NodeStateExecution) error

Execute is the main entry point for action execution. It returns ErrDispatchDeferred when the execution is left unresolved — local backpressure, or a transient control-plane failure during the claim handshake or a rejection walk — and nil once the run has been accepted or the execution settled with a callback.

func (*Executor) FailOrphan added in v0.3.0 

func (e *Executor) FailOrphan(ctx context.Context, nodeID, executionID string) error

FailOrphan reports an execution the control plane still holds at started but whose run this agent no longer owns — the process restarted mid-run. Actions are not idempotent, so the run is not repeated; started → failed is a legal edge, so the single terminal callback settles the execution. The report runs under its own deadline so an unreachable control plane cannot pin a reconciliation cycle open.

It returns ErrDispatchDeferred when the report was not delivered. That leaves the execution exactly where it was — at started, with no terminal recorded — so the caller must let the next pull redeliver the entry rather than treat it as settled.

func (*Executor) IsActive added in v0.3.0 

func (e *Executor) IsActive(executionID string) bool

IsActive reports whether this executor is running the given execution right now. It is the authoritative answer to "did this agent lose that run?", which a caller tracking dispatches by id cannot answer on its own.

func (*Executor) RegisterBuiltin 

func (e *Executor) RegisterBuiltin(name, description string, params []api.ActionParam, fn BuiltinFunc)

RegisterBuiltin stores a builtin action for execution.

func (*Executor) RunLocal 

func (e *Executor) RunLocal(ctx context.Context, action string, params map[string]string) (string, string, int, error)

RunLocal executes a built-in action synchronously and returns the output. This is used by the local node API for CLI-triggered action execution. Only built-in actions are supported: hooks are arbitrary operator scripts, and dispatch from the pull's executions block is the only path the control plane authorizes server-side.

func (*Executor) SetHooks 

func (e *Executor) SetHooks(hooks []api.HookInfo)

SetHooks sets the discovered hooks snapshot and pins the integrity anchor of every hook this process has not seen before.

The snapshot itself is refreshed by HookWatcher, which re-hashes a hook on every write, so verifying an execution against the snapshot digest would compare a file with a hash of itself and pass for any bytes an attacker with write access to the hooks directory puts there. The pin is therefore recorded once, at first discovery — the digest this node also reports to the control plane — and never updated: a hook whose bytes change afterwards fails verification and stays unrunnable until the agent restarts and re-attests it.

func (*Executor) Shutdown 

func (e *Executor) Shutdown(_ context.Context)

Shutdown cancels all running actions, prevents new ones from starting, and waits for all in-flight goroutines to drain.

type HealthProvider 

type HealthProvider interface {
	TunnelCount() int
	ConnectedPeers() int
	Uptime() time.Duration
	LastHeartbeat() time.Time
	LastReconcile() time.Time
}

HealthProvider supplies health status information to built-in actions.

type HookChangeCallback 

type HookChangeCallback func(hooks []api.HookInfo)

HookChangeCallback is called when hooks change. Receives the full current hooks list.

type HookEvent 

type HookEvent struct {
	Type string // "add", "update", "remove"
	Hook api.HookInfo
}

HookEvent describes what happened to a hook.

type HookVerifier 

type HookVerifier interface {
	VerifyHook(ctx context.Context, nodeID, hookPath, expectedChecksum string) (bool, error)
}

HookVerifier abstracts hook integrity verification for testability.

type HookWatcher 

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

HookWatcher monitors a hooks directory for changes using fsnotify.

func NewHookWatcher 

func NewHookWatcher(hooksDir string, onChange HookChangeCallback, onIntegrity IntegrityAlertCallback, logger *slog.Logger) *HookWatcher

NewHookWatcher creates a new HookWatcher.

func (*HookWatcher) Hooks 

func (w *HookWatcher) Hooks() []api.HookInfo

Hooks returns a sorted snapshot of the current hooks.

func (*HookWatcher) Watch 

func (w *HookWatcher) Watch(ctx context.Context) error

Watch monitors the hooks directory for changes. It blocks until ctx is cancelled. Returns nil on clean shutdown via context cancellation.

type IntegrityAlertCallback 

type IntegrityAlertCallback func(hookName, oldChecksum, newChecksum string)

IntegrityAlertCallback is called when a hook file's checksum changes unexpectedly.

type LogProvider 

type LogProvider interface {
	RecentLines(n int) []string
}

LogProvider supplies recent log lines.

type MeshReconnector 

type MeshReconnector interface {
	Reconnect(ctx context.Context) error
}

MeshReconnector triggers mesh reconnection.

type NodeInfoProvider 

type NodeInfoProvider interface {
	NodeID() string
	MeshIP() string
	PeerCount() int
}

NodeInfoProvider supplies mesh node information to built-in actions.

type ReleaseFetcher added in v0.2.0 

type ReleaseFetcher interface {
	FetchBinary(ctx context.Context, version string) (io.ReadCloser, error)
	FetchBundle(ctx context.Context, version string) ([]byte, error)
}

ReleaseFetcher downloads plexd release assets for a version.

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.