mirror of
https://github.com/rcourtman/Pulse.git
synced 2026-09-10 10:35:51 +00:00
rcourtman
2643 lines
85 KiB
Go
2643 lines
85 KiB
Go
package hostagent
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"crypto/rand"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"net"
|
|
"net/http"
|
|
"net/netip"
|
|
"net/url"
|
|
"os"
|
|
"os/exec"
|
|
"path/filepath"
|
|
"runtime"
|
|
"sort"
|
|
"strconv"
|
|
"strings"
|
|
"sync"
|
|
"sync/atomic"
|
|
"time"
|
|
|
|
"github.com/rcourtman/pulse-go-rewrite/internal/agenttarget"
|
|
"github.com/rcourtman/pulse-go-rewrite/internal/agenttls"
|
|
"github.com/rcourtman/pulse-go-rewrite/internal/agentupdate"
|
|
"github.com/rcourtman/pulse-go-rewrite/internal/config"
|
|
"github.com/rcourtman/pulse-go-rewrite/internal/platformsupport"
|
|
"github.com/rcourtman/pulse-go-rewrite/internal/remoteconfig"
|
|
"github.com/rcourtman/pulse-go-rewrite/internal/securityutil"
|
|
"github.com/rcourtman/pulse-go-rewrite/internal/sensors"
|
|
"github.com/rcourtman/pulse-go-rewrite/internal/utils"
|
|
agentshost "github.com/rcourtman/pulse-go-rewrite/pkg/agents/host"
|
|
"github.com/rcourtman/pulse-go-rewrite/pkg/diskinventory"
|
|
"github.com/rs/zerolog"
|
|
gohost "github.com/shirou/gopsutil/v4/host"
|
|
)
|
|
|
|
// Config controls the behaviour of the runtime-side Unified Agent.
|
|
type Config struct {
|
|
PulseURL string
|
|
APIToken string
|
|
Interval time.Duration
|
|
HostnameOverride string
|
|
AgentID string
|
|
AgentType string // "unified" when running as part of pulse-agent, empty for standalone
|
|
AgentVersion string // Version to report; if empty, uses the Unified Agent binary version
|
|
Tags []string
|
|
InsecureSkipVerify bool
|
|
CACertPath string
|
|
ServerFingerprint string
|
|
CustomSensorsFile string
|
|
Observers []ObserverTarget
|
|
RunOnce bool
|
|
LogLevel zerolog.Level
|
|
Logger *zerolog.Logger
|
|
|
|
// Proxmox integration
|
|
EnableProxmox bool // If true, creates Proxmox API token and registers node on startup
|
|
ProxmoxType string // "pve", "pbs", or "" for auto-detect
|
|
|
|
// Security options
|
|
EnableCommands bool // If true, enables the command execution feature (AI auto-fix)
|
|
CommandAuthorityProfile CommandAuthorityProfile // Immutable local ceiling; empty preserves legacy update continuity.
|
|
|
|
// Enrollment
|
|
Enroll bool // If true, exchange bootstrap token for runtime token on startup
|
|
|
|
// Disk filtering
|
|
DiskExclude []string // Mount points or path prefixes to exclude from disk monitoring
|
|
DiskInclude []string // Devices or mount points to opt into monitoring despite automatic filtering
|
|
|
|
// State directory for persistent files (agent-id, proxmox registration, etc.)
|
|
StateDir string // Default: /var/lib/pulse-agent
|
|
|
|
// Deploy SSH configuration for peer-node install fan-out.
|
|
DeploySSHUser string // Default: root. Non-root users must support passwordless sudo for install steps.
|
|
|
|
// Network configuration
|
|
ReportIP string // IP address to report instead of auto-detected (for multi-NIC systems)
|
|
DisableCeph bool // If true, disables local Ceph status polling
|
|
|
|
// AvailabilityTargets are the externally probed availability checks the
|
|
// server assigned to this agent at startup. Later assignments arrive
|
|
// through ApplyRemoteConfig.
|
|
AvailabilityTargets []config.AvailabilityTarget
|
|
|
|
// AppliedConfig is the non-secret fingerprint of the managed config that
|
|
// was applied before this runtime started. UpdateStatus supplies live
|
|
// self-update state without coupling report collection to updater internals.
|
|
AppliedConfig *agentshost.ConfigFingerprint
|
|
UpdateStatus func() agentupdate.Status
|
|
ModuleStatus func() []agentshost.ModuleStatus
|
|
// PrivilegeHelperStatus is the shared typed-helper health recorder used by
|
|
// host and container collection. Nil creates a host-local recorder when
|
|
// PrivilegedTelemetry is configured.
|
|
PrivilegeHelperStatus *PrivilegeHelperStatus
|
|
|
|
// OnServerVersion is called with the server version carried on each
|
|
// authoritative report acknowledgement, so the updater can react to a
|
|
// server upgrade within one report cycle instead of its next hourly check.
|
|
// Nil disables the hook; acks from observer destinations never invoke it.
|
|
OnServerVersion func(version string)
|
|
// OnPrimaryReportAccepted is called only after the current process's newly
|
|
// collected authoritative report is accepted. Buffered reports and observer
|
|
// acknowledgements do not satisfy this startup-health signal.
|
|
OnPrimaryReportAccepted func()
|
|
// UpdatedFromVersion is supplied by the durable helper-update handoff and
|
|
// is carried by the first report built by the replacement process.
|
|
UpdatedFromVersion string
|
|
|
|
Collector SystemCollector // Optional: override default system information collector (for testing)
|
|
// PrivilegedTelemetry routes the exceptional SMART and Proxmox LXC
|
|
// filesystem snapshots through the typed no-network helper. When present,
|
|
// helper failure degrades those snapshots and never falls back to a broader
|
|
// local privilege path.
|
|
PrivilegedTelemetry PrivilegedTelemetry
|
|
|
|
// DockerContainerUpdater bridges typed container update operations to the
|
|
// Docker / Podman module when the unified agent runs both. Nil when the
|
|
// Docker module is disabled or not yet connected.
|
|
DockerContainerUpdater DockerContainerUpdater
|
|
|
|
// DockerContainerLifecycleOperator bridges typed start, stop, and restart
|
|
// operations to the already-connected Docker / Podman module. Containerized
|
|
// unified agents use this API-backed path because the agent image does not
|
|
// bundle an external docker or podman CLI.
|
|
DockerContainerLifecycleOperator DockerContainerLifecycleOperator
|
|
|
|
newCommandClientFn func(Config, string, string, string, string) *CommandClient
|
|
runCommandClientFn func(*CommandClient, context.Context) error
|
|
updatedFromVersionFn func() string
|
|
packageUpdates *packageUpdateManager
|
|
storageCleanup *storageCleanupManager
|
|
customSensorExecute customSensorExecution
|
|
}
|
|
|
|
// ObserverTarget is a report-only Pulse destination. It can receive the same
|
|
// collected host payload and an independently provisioned Proxmox source, but
|
|
// it never supplies remote config, commands, enrollment, or update authority.
|
|
type ObserverTarget struct {
|
|
Name string
|
|
ID string
|
|
PulseURL string
|
|
APIToken string
|
|
InsecureSkipVerify bool
|
|
AllowPlaintextHTTP bool
|
|
CACertPath string
|
|
ServerFingerprint string
|
|
ProvisionProxmox bool
|
|
}
|
|
|
|
type observerReporter struct {
|
|
target ObserverTarget
|
|
httpClient *http.Client
|
|
reportBuffer *utils.Queue[agentshost.Report]
|
|
lastAuthFailureLog time.Time
|
|
}
|
|
|
|
// Agent is responsible for collecting host metrics and shipping them to Pulse.
|
|
type Agent struct {
|
|
cfg Config
|
|
logger zerolog.Logger
|
|
httpClient *http.Client
|
|
|
|
hostInfo *gohost.InfoStat
|
|
hostname string
|
|
displayName string
|
|
platform string
|
|
osName string
|
|
osVersion string
|
|
kernelVersion string
|
|
architecture string
|
|
machineID string
|
|
agentID string
|
|
agentVersion string
|
|
updatedFrom string // Previous version if recently auto-updated (reported once)
|
|
reportIP string // User-specified IP to report (for multi-NIC systems)
|
|
interval time.Duration
|
|
stateDir string
|
|
trimmedPulseURL string
|
|
configMu sync.RWMutex
|
|
remoteConfigChanged chan struct{}
|
|
reportBuffer *utils.Queue[agentshost.Report]
|
|
observerReporters []*observerReporter
|
|
commandClient *CommandClient
|
|
commandClientMu sync.Mutex
|
|
commandClientRunCancel context.CancelFunc
|
|
commandClientParentCtx context.Context
|
|
// commandAuthorityProfile is the immutable local install-time ceiling for
|
|
// this process. Remote configuration may disable and later re-enable a
|
|
// command-capable or legacy install, but it cannot promote monitoring-only.
|
|
commandAuthorityProfile CommandAuthorityProfile
|
|
commandAuthorityWarningSeen bool
|
|
collector SystemCollector
|
|
privilegedTelemetry PrivilegedTelemetry
|
|
privilegeHelperHealth *PrivilegeHelperStatus
|
|
newCommandClient func(Config, string, string, string, string) *CommandClient
|
|
runCommandClient func(*CommandClient, context.Context) error
|
|
packageUpdates *packageUpdateManager
|
|
storageCleanup *storageCleanupManager
|
|
customSensors *customSensorRuntime
|
|
availability *availabilityProbeModule
|
|
reportStreamID string
|
|
reportSequence atomic.Uint64
|
|
// canonicalIDWarned holds the last server-acknowledged agent ID a
|
|
// mismatch warning was emitted for, so the warning fires once per
|
|
// resolved identity instead of on every report cycle. Only the
|
|
// report/enroll ack paths touch it, which run sequentially.
|
|
canonicalIDWarned string
|
|
// libreHardwareMonitorEndpoint is an unexported test seam. Production
|
|
// collection always uses the fixed loopback URL when this is empty.
|
|
libreHardwareMonitorEndpoint string
|
|
|
|
// lastAuthFailureLog throttles the actionable 401 error so a permanently
|
|
// rejected token does not spam the log every report interval. Only touched
|
|
// from the single-threaded report loop (process/flushBuffer).
|
|
lastAuthFailureLog time.Time
|
|
}
|
|
|
|
const defaultInterval = 30 * time.Second
|
|
const defaultStateDir = "/var/lib/pulse-agent"
|
|
const agentReportEndpoint = "/api/agents/agent/report"
|
|
|
|
// authFailureLogInterval bounds how often the agent re-emits the actionable
|
|
// "token rejected" error while the server keeps returning 401.
|
|
const authFailureLogInterval = 5 * time.Minute
|
|
|
|
type reportHTTPStatusError struct {
|
|
Endpoint string
|
|
Status string
|
|
StatusCode int
|
|
}
|
|
|
|
func (e *reportHTTPStatusError) Error() string {
|
|
return fmt.Sprintf("pulse responded with status %s", e.Status)
|
|
}
|
|
|
|
// New constructs a fully initialised runtime-side Unified Agent.
|
|
func New(cfg Config) (*Agent, error) {
|
|
commandAuthorityProfile, err := NormalizeCommandAuthorityProfile(string(cfg.CommandAuthorityProfile))
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
cfg.CommandAuthorityProfile = commandAuthorityProfile
|
|
if effectiveCommands, accepted := ResolveCommandAuthority(commandAuthorityProfile, cfg.EnableCommands); !accepted {
|
|
cfg.EnableCommands = effectiveCommands
|
|
}
|
|
|
|
if cfg.Interval <= 0 {
|
|
cfg.Interval = defaultInterval
|
|
}
|
|
|
|
if strings.TrimSpace(cfg.StateDir) == "" {
|
|
cfg.StateDir = defaultStateDir
|
|
}
|
|
|
|
if zerolog.GlobalLevel() == zerolog.DebugLevel && cfg.LogLevel != zerolog.DebugLevel {
|
|
zerolog.SetGlobalLevel(cfg.LogLevel)
|
|
}
|
|
|
|
if cfg.Logger == nil {
|
|
defaultLogger := zerolog.New(zerolog.NewConsoleWriter()).
|
|
Level(cfg.LogLevel).
|
|
With().
|
|
Timestamp().
|
|
Logger()
|
|
cfg.Logger = &defaultLogger
|
|
}
|
|
|
|
logger := cfg.Logger.Level(cfg.LogLevel).With().Str("component", "agent").Logger()
|
|
|
|
if cfg.Enroll && strings.TrimSpace(cfg.APIToken) == "" {
|
|
return nil, fmt.Errorf("api token is required when enrollment is enabled")
|
|
}
|
|
|
|
pulseURL := strings.TrimSpace(cfg.PulseURL)
|
|
if pulseURL == "" {
|
|
pulseURL = "http://localhost:7655"
|
|
}
|
|
pulseURL, err = normalizePulseURL(pulseURL)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("invalid pulse URL: %w", err)
|
|
}
|
|
cfg.PulseURL = pulseURL
|
|
|
|
cfg.ProxmoxType, err = normalizeProxmoxType(cfg.ProxmoxType)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
cfg.ReportIP, err = normalizeReportIP(cfg.ReportIP)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
cfg.DeploySSHUser, err = NormalizeDeploySSHUser(cfg.DeploySSHUser)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
|
defer cancel()
|
|
|
|
collector := cfg.Collector
|
|
if collector == nil {
|
|
collector = &defaultCollector{}
|
|
}
|
|
newCommandClientFn := cfg.newCommandClientFn
|
|
if newCommandClientFn == nil {
|
|
newCommandClientFn = NewCommandClient
|
|
}
|
|
runCommandClientFn := cfg.runCommandClientFn
|
|
if runCommandClientFn == nil {
|
|
runCommandClientFn = func(client *CommandClient, ctx context.Context) error {
|
|
return client.Run(ctx)
|
|
}
|
|
}
|
|
updatedFromVersionFn := cfg.updatedFromVersionFn
|
|
if updatedFromVersionFn == nil {
|
|
updatedFromVersionFn = agentupdate.GetUpdatedFromVersion
|
|
}
|
|
|
|
info, err := collector.HostInfo(ctx)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("fetch host info: %w", err)
|
|
}
|
|
|
|
hostname := strings.TrimSpace(cfg.HostnameOverride)
|
|
if hostname == "" {
|
|
hostname = strings.TrimSpace(info.Hostname)
|
|
}
|
|
if hostname == "" {
|
|
hostname = "unknown-host"
|
|
}
|
|
|
|
displayName := hostname
|
|
|
|
machineID := GetReliableMachineID(collector, info.HostID, logger)
|
|
|
|
agentID := strings.TrimSpace(cfg.AgentID)
|
|
if agentID == "" {
|
|
agentID = machineID
|
|
}
|
|
if agentID == "" {
|
|
agentID = hostname
|
|
}
|
|
|
|
platform := normaliseRuntimePlatform(collector.GOOS(), info.Platform)
|
|
// Use Platform (specific distro like "ubuntu") over PlatformFamily (distro family like "debian")
|
|
// This ensures Ubuntu shows as "ubuntu 24.04" not "debian 24.04" (refs #927)
|
|
osName := strings.TrimSpace(info.Platform)
|
|
if osName == "" {
|
|
osName = strings.TrimSpace(info.PlatformFamily)
|
|
}
|
|
osVersion := strings.TrimSpace(info.PlatformVersion)
|
|
osName, osVersion = resolveHostOSIdentity(collector, osName, osVersion)
|
|
if profile, ok := platformsupport.AgentHostProfileForIdentity(osName, platform); ok {
|
|
platform = platformsupport.NormalizeRuntimePlatformForAgentHostProfile(profile.ID, platform)
|
|
}
|
|
kernelVersion := strings.TrimSpace(info.KernelVersion)
|
|
arch := strings.TrimSpace(info.KernelArch)
|
|
if arch == "" {
|
|
arch = runtime.GOARCH
|
|
}
|
|
client, err := newAgentHTTPClient(cfg.CACertPath, cfg.InsecureSkipVerify, cfg.ServerFingerprint)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("invalid TLS configuration: %w", err)
|
|
}
|
|
|
|
observerReporters, err := newObserverReporters(cfg.Observers)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
customSensorDefinitions, err := loadCustomSensorDefinitions(cfg.CustomSensorsFile)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("load custom sensors: %w", err)
|
|
}
|
|
agenttarget.MarkConfigured("host", "primary", "primary")
|
|
for _, observer := range observerReporters {
|
|
agenttarget.MarkConfigured("host", observer.target.Name, "observer")
|
|
}
|
|
|
|
trimmedTags := make([]string, 0, len(cfg.Tags))
|
|
seenTags := make(map[string]struct{}, len(cfg.Tags))
|
|
for _, tag := range cfg.Tags {
|
|
tag = strings.TrimSpace(tag)
|
|
if tag == "" {
|
|
continue
|
|
}
|
|
if _, exists := seenTags[tag]; exists {
|
|
continue
|
|
}
|
|
seenTags[tag] = struct{}{}
|
|
trimmedTags = append(trimmedTags, tag)
|
|
}
|
|
cfg.Tags = trimmedTags
|
|
|
|
// Use configured version or fall back to package version
|
|
agentVersion := cfg.AgentVersion
|
|
if agentVersion == "" {
|
|
agentVersion = Version
|
|
}
|
|
|
|
reportStreamID, err := newReportStreamID()
|
|
if err != nil {
|
|
return nil, fmt.Errorf("create report stream ID: %w", err)
|
|
}
|
|
|
|
const bufferCapacity = 60
|
|
|
|
// Check if agent was recently auto-updated (only reported once per restart)
|
|
updatedFrom := strings.TrimSpace(cfg.UpdatedFromVersion)
|
|
if updatedFrom == "" {
|
|
updatedFrom = updatedFromVersionFn()
|
|
}
|
|
if updatedFrom != "" {
|
|
logger.Info().
|
|
Str("previousVersion", updatedFrom).
|
|
Str("currentVersion", agentVersion).
|
|
Msg("Agent was auto-updated")
|
|
}
|
|
|
|
packageUpdates := cfg.packageUpdates
|
|
storageCleanup := cfg.storageCleanup
|
|
packageUpdates, storageCleanup = configurePackageManagers(platform, packageUpdates, storageCleanup)
|
|
cfg.packageUpdates = packageUpdates
|
|
cfg.storageCleanup = storageCleanup
|
|
privilegeHelperHealth := cfg.PrivilegeHelperStatus
|
|
if privilegeHelperHealth == nil && cfg.PrivilegedTelemetry != nil {
|
|
privilegeHelperHealth = NewPrivilegeHelperStatus()
|
|
}
|
|
|
|
agent := &Agent{
|
|
cfg: cfg,
|
|
logger: logger,
|
|
httpClient: client,
|
|
hostInfo: info,
|
|
hostname: hostname,
|
|
displayName: displayName,
|
|
platform: platform,
|
|
osName: osName,
|
|
osVersion: osVersion,
|
|
kernelVersion: kernelVersion,
|
|
architecture: arch,
|
|
machineID: machineID,
|
|
agentID: agentID,
|
|
agentVersion: agentVersion,
|
|
updatedFrom: updatedFrom,
|
|
reportIP: cfg.ReportIP,
|
|
stateDir: cfg.StateDir,
|
|
interval: cfg.Interval,
|
|
trimmedPulseURL: pulseURL,
|
|
remoteConfigChanged: make(chan struct{}, 1),
|
|
reportBuffer: utils.New[agentshost.Report](bufferCapacity),
|
|
observerReporters: observerReporters,
|
|
collector: collector,
|
|
privilegedTelemetry: cfg.PrivilegedTelemetry,
|
|
privilegeHelperHealth: privilegeHelperHealth,
|
|
newCommandClient: newCommandClientFn,
|
|
runCommandClient: runCommandClientFn,
|
|
commandAuthorityProfile: commandAuthorityProfile,
|
|
packageUpdates: packageUpdates,
|
|
storageCleanup: storageCleanup,
|
|
customSensors: newCustomSensorRuntime(customSensorDefinitions, cfg.customSensorExecute),
|
|
availability: newAvailabilityProbeModule(logger, cfg.AvailabilityTargets),
|
|
reportStreamID: reportStreamID,
|
|
}
|
|
if len(customSensorDefinitions) > 0 {
|
|
logger.Info().
|
|
Int("sensorCount", len(customSensorDefinitions)).
|
|
Str("configPath", cfg.CustomSensorsFile).
|
|
Msg("Custom metric collection enabled from local configuration")
|
|
}
|
|
|
|
// Create command client for AI command execution (only if enabled)
|
|
if cfg.EnableCommands {
|
|
agent.commandClient = newCommandClientFn(cfg, agentID, hostname, platform, agentVersion)
|
|
cfg.Logger.Info().Msg("Command execution enabled via --enable-commands flag")
|
|
} else {
|
|
cfg.Logger.Info().Msg("Command execution disabled (use --enable-commands to enable)")
|
|
}
|
|
|
|
return agent, nil
|
|
}
|
|
|
|
func newReportStreamID() (string, error) {
|
|
var raw [16]byte
|
|
if _, err := rand.Read(raw[:]); err != nil {
|
|
return "", err
|
|
}
|
|
return hex.EncodeToString(raw[:]), nil
|
|
}
|
|
|
|
func (a *Agent) nextReportSequenceID() string {
|
|
if a == nil || strings.TrimSpace(a.reportStreamID) == "" {
|
|
return ""
|
|
}
|
|
return agentshost.FormatReportSequenceID(a.reportStreamID, a.reportSequence.Add(1))
|
|
}
|
|
|
|
func agentControlPlaneProxy(request *http.Request) (*url.URL, error) {
|
|
return resolveAgentControlPlaneProxy(request, http.ProxyFromEnvironment)
|
|
}
|
|
|
|
func resolveAgentControlPlaneProxy(request *http.Request, ambient func(*http.Request) (*url.URL, error)) (*url.URL, error) {
|
|
if request == nil || request.URL == nil || strings.EqualFold(request.URL.Scheme, "http") {
|
|
// Plaintext self-hosted control-plane traffic already carries its bearer
|
|
// in the clear on the explicitly selected local/private path. Never let
|
|
// ambient process state silently redirect that credential to a proxy.
|
|
return nil, nil
|
|
}
|
|
if ambient == nil {
|
|
return nil, nil
|
|
}
|
|
return ambient(request)
|
|
}
|
|
|
|
func newAgentHTTPClient(caCertPath string, insecureSkipVerify bool, serverFingerprint string) (*http.Client, error) {
|
|
tlsConfig, err := agenttls.NewClientTLSConfig(caCertPath, insecureSkipVerify, serverFingerprint)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return &http.Client{
|
|
Timeout: 15 * time.Second,
|
|
Transport: &http.Transport{
|
|
Proxy: agentControlPlaneProxy,
|
|
TLSClientConfig: tlsConfig,
|
|
},
|
|
// Redirects can rewrite report POSTs to GETs and must never cross an
|
|
// authority boundary implicitly.
|
|
CheckRedirect: func(req *http.Request, _ []*http.Request) error {
|
|
return fmt.Errorf("server returned redirect to %s - use the final Pulse URL explicitly", req.URL)
|
|
},
|
|
}, nil
|
|
}
|
|
|
|
func newObserverReporters(targets []ObserverTarget) ([]*observerReporter, error) {
|
|
const bufferCapacity = 60
|
|
result := make([]*observerReporter, 0, len(targets))
|
|
seen := make(map[string]struct{}, len(targets))
|
|
for _, target := range targets {
|
|
target.Name = strings.TrimSpace(target.Name)
|
|
target.ID = strings.TrimSpace(target.ID)
|
|
target.APIToken = strings.TrimSpace(target.APIToken)
|
|
if target.Name == "" || target.ID == "" || target.APIToken == "" {
|
|
return nil, errors.New("observer destination requires name, id, and API token")
|
|
}
|
|
url, err := agenttarget.NormalizePulseURL(target.PulseURL, false, target.AllowPlaintextHTTP)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("invalid observer %q Pulse URL: %w", target.Name, err)
|
|
}
|
|
target.PulseURL = url
|
|
if _, exists := seen[target.ID]; exists {
|
|
return nil, fmt.Errorf("duplicate observer destination id for %q", target.Name)
|
|
}
|
|
seen[target.ID] = struct{}{}
|
|
client, err := newAgentHTTPClient(target.CACertPath, target.InsecureSkipVerify, target.ServerFingerprint)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("configure observer %q TLS: %w", target.Name, err)
|
|
}
|
|
result = append(result, &observerReporter{
|
|
target: target,
|
|
httpClient: client,
|
|
reportBuffer: utils.New[agentshost.Report](bufferCapacity),
|
|
})
|
|
}
|
|
return result, nil
|
|
}
|
|
|
|
func normalizeProxmoxType(raw string) (string, error) {
|
|
value := strings.TrimSpace(strings.ToLower(raw))
|
|
switch value {
|
|
case "", "auto":
|
|
return "", nil
|
|
case "pve", "pbs":
|
|
return value, nil
|
|
default:
|
|
return "", fmt.Errorf("invalid proxmox type %q: must be pve, pbs, or auto", raw)
|
|
}
|
|
}
|
|
|
|
func normalizeReportIP(raw string) (string, error) {
|
|
value := strings.TrimSpace(raw)
|
|
if value == "" {
|
|
return "", nil
|
|
}
|
|
|
|
addr, err := netip.ParseAddr(value)
|
|
if err != nil {
|
|
return "", fmt.Errorf("invalid report IP %q: must be a valid IPv4 or IPv6 address", raw)
|
|
}
|
|
if addr.IsUnspecified() {
|
|
return "", fmt.Errorf("invalid report IP %q: unspecified addresses are not allowed", raw)
|
|
}
|
|
|
|
return addr.String(), nil
|
|
}
|
|
|
|
// Run executes the agent until the context is cancelled.
|
|
func (a *Agent) Run(ctx context.Context) error {
|
|
if a.cfg.RunOnce {
|
|
return a.runOnce(ctx)
|
|
}
|
|
|
|
a.commandClientMu.Lock()
|
|
a.commandClientParentCtx = ctx
|
|
commandClient := a.commandClient
|
|
a.commandClientMu.Unlock()
|
|
defer func() {
|
|
a.commandClientMu.Lock()
|
|
a.commandClientParentCtx = nil
|
|
a.commandClientMu.Unlock()
|
|
a.stopCommandClient(true)
|
|
}()
|
|
|
|
// Enrollment: exchange bootstrap token for runtime token (must run before
|
|
// Proxmox setup and reporting, which require the runtime token's scopes).
|
|
if a.cfg.Enroll {
|
|
if err := a.runEnrollmentLoop(ctx); err != nil {
|
|
return fmt.Errorf("enrollment failed: %w", err)
|
|
}
|
|
// Re-create command client with the new runtime token, since the
|
|
// original was built with the bootstrap token during New().
|
|
if a.cfg.EnableCommands {
|
|
a.commandClientMu.Lock()
|
|
a.commandClient = a.newCommandClient(a.cfg, a.agentID, a.hostname, a.platform, a.agentVersion)
|
|
commandClient = a.commandClient
|
|
a.commandClientMu.Unlock()
|
|
}
|
|
}
|
|
|
|
// Proxmox setup (if enabled)
|
|
if a.cfg.EnableProxmox {
|
|
a.runProxmoxSetup(ctx)
|
|
go a.runProxmoxHealthCheckLoop(ctx)
|
|
}
|
|
|
|
// Start command client in background for AI command execution
|
|
if commandClient != nil {
|
|
a.startCommandClient(commandClient)
|
|
}
|
|
|
|
// Externally probed availability checks run on their own schedule, not on
|
|
// the report interval, and queue their results for the next report.
|
|
go a.availability.Run(ctx)
|
|
|
|
// Load any reports buffered from a previous shutdown
|
|
a.loadPersistedBuffer()
|
|
a.loadPersistedObserverBuffers()
|
|
|
|
ticker := time.NewTicker(a.currentInterval())
|
|
defer ticker.Stop()
|
|
defer a.persistBuffer()
|
|
defer a.persistObserverBuffers()
|
|
|
|
if err := a.process(ctx); err != nil && !errors.Is(err, context.Canceled) {
|
|
a.logger.Error().
|
|
Err(err).
|
|
Str("hostname", a.hostname).
|
|
Msg("Initial host report failed")
|
|
}
|
|
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return ctx.Err()
|
|
case <-a.remoteConfigChanged:
|
|
ticker.Reset(a.currentInterval())
|
|
case <-ticker.C:
|
|
if err := a.process(ctx); err != nil {
|
|
if errors.Is(err, context.Canceled) {
|
|
return err
|
|
}
|
|
a.logger.Error().
|
|
Err(err).
|
|
Str("hostname", a.hostname).
|
|
Msg("Failed to process host report")
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
type runtimeConfigSnapshot struct {
|
|
agentType string
|
|
interval time.Duration
|
|
commandsEnabled bool
|
|
diskExclude []string
|
|
diskInclude []string
|
|
tags []string
|
|
reportIP string
|
|
disableCeph bool
|
|
appliedConfig *agentshost.ConfigFingerprint
|
|
}
|
|
|
|
func (a *Agent) currentInterval() time.Duration {
|
|
a.configMu.RLock()
|
|
defer a.configMu.RUnlock()
|
|
if a.interval <= 0 {
|
|
return time.Minute
|
|
}
|
|
return a.interval
|
|
}
|
|
|
|
func (a *Agent) currentReportIP() string {
|
|
a.configMu.RLock()
|
|
defer a.configMu.RUnlock()
|
|
return a.reportIP
|
|
}
|
|
|
|
func (a *Agent) runtimeConfigSnapshot() runtimeConfigSnapshot {
|
|
a.configMu.RLock()
|
|
defer a.configMu.RUnlock()
|
|
return runtimeConfigSnapshot{
|
|
agentType: a.cfg.AgentType,
|
|
interval: a.interval,
|
|
commandsEnabled: a.cfg.EnableCommands,
|
|
diskExclude: append([]string(nil), a.cfg.DiskExclude...),
|
|
diskInclude: append([]string(nil), a.cfg.DiskInclude...),
|
|
tags: append([]string(nil), a.cfg.Tags...),
|
|
reportIP: a.reportIP,
|
|
disableCeph: a.cfg.DisableCeph,
|
|
appliedConfig: cloneConfigFingerprint(a.cfg.AppliedConfig),
|
|
}
|
|
}
|
|
|
|
func cloneConfigFingerprint(value *agentshost.ConfigFingerprint) *agentshost.ConfigFingerprint {
|
|
if value == nil {
|
|
return nil
|
|
}
|
|
copy := *value
|
|
return ©
|
|
}
|
|
|
|
func (a *Agent) currentUpdateStatus() *agentshost.UpdateStatus {
|
|
if a.cfg.UpdateStatus == nil {
|
|
return nil
|
|
}
|
|
status := a.cfg.UpdateStatus()
|
|
return &agentshost.UpdateStatus{
|
|
State: status.State,
|
|
AutoUpdate: status.AutoUpdate,
|
|
AvailableVersion: status.AvailableVersion,
|
|
LastCheckedAt: status.LastCheckedAt,
|
|
LastAttemptAt: status.LastAttemptAt,
|
|
LastSuccessAt: status.LastSuccessAt,
|
|
LastError: status.LastError,
|
|
}
|
|
}
|
|
|
|
func (a *Agent) currentModuleStatus() []agentshost.ModuleStatus {
|
|
var statuses []agentshost.ModuleStatus
|
|
if a.cfg.ModuleStatus != nil {
|
|
statuses = append(statuses, a.cfg.ModuleStatus()...)
|
|
}
|
|
if a.privilegeHelperHealth != nil {
|
|
statuses = append(statuses, a.privilegeHelperHealth.ModuleStatus())
|
|
}
|
|
// The probe module is owned by the host agent rather than the runtime
|
|
// supervisor, so it reports itself and only while it has work assigned.
|
|
if status, ok := a.availability.moduleStatus(); ok {
|
|
statuses = append(statuses, status)
|
|
}
|
|
return statuses
|
|
}
|
|
|
|
func (a *Agent) signalRemoteConfigChanged() {
|
|
select {
|
|
case a.remoteConfigChanged <- struct{}{}:
|
|
default:
|
|
}
|
|
}
|
|
|
|
func (a *Agent) currentOperationReceiptVersion() int {
|
|
a.commandClientMu.Lock()
|
|
defer a.commandClientMu.Unlock()
|
|
if a.commandClient == nil {
|
|
return 0
|
|
}
|
|
return a.commandClient.operationReceiptVersion()
|
|
}
|
|
|
|
func (a *Agent) startCommandClient(client *CommandClient) bool {
|
|
if client == nil {
|
|
return false
|
|
}
|
|
|
|
a.commandClientMu.Lock()
|
|
if a.commandClientRunCancel != nil {
|
|
a.commandClientMu.Unlock()
|
|
return true
|
|
}
|
|
|
|
parentCtx := a.commandClientParentCtx
|
|
if parentCtx == nil {
|
|
a.commandClientMu.Unlock()
|
|
return false
|
|
}
|
|
runCtx, cancel := context.WithCancel(parentCtx)
|
|
a.commandClientRunCancel = cancel
|
|
a.commandClientMu.Unlock()
|
|
|
|
go func() {
|
|
if err := a.runCommandClient(client, runCtx); err != nil && !errors.Is(err, context.Canceled) {
|
|
a.logger.Error().Err(err).Msg("Command client stopped with error")
|
|
}
|
|
}()
|
|
return true
|
|
}
|
|
|
|
func (a *Agent) stopCommandClient(clearClient bool) {
|
|
a.commandClientMu.Lock()
|
|
client := a.commandClient
|
|
cancel := a.commandClientRunCancel
|
|
a.commandClientRunCancel = nil
|
|
if clearClient {
|
|
a.commandClient = nil
|
|
}
|
|
a.commandClientMu.Unlock()
|
|
|
|
if cancel != nil {
|
|
cancel()
|
|
}
|
|
if client != nil {
|
|
if err := client.Close(); err != nil {
|
|
a.logger.Debug().Err(err).Msg("Error closing command client connection")
|
|
}
|
|
}
|
|
}
|
|
|
|
func (a *Agent) runOnce(ctx context.Context) error {
|
|
return a.process(ctx)
|
|
}
|
|
|
|
func (a *Agent) process(ctx context.Context) error {
|
|
report, err := a.buildReport(ctx)
|
|
if err != nil {
|
|
return fmt.Errorf("build report: %w", err)
|
|
}
|
|
primaryErr := a.deliverPrimaryReport(ctx, report)
|
|
for _, observer := range a.observerReporters {
|
|
a.deliverObserverReport(ctx, observer, report)
|
|
}
|
|
return primaryErr
|
|
}
|
|
|
|
func (a *Agent) deliverPrimaryReport(ctx context.Context, report agentshost.Report) error {
|
|
// Preserve collection order across reconnects. Sending the newest report
|
|
// before the buffered FIFO can let an older long-running-task snapshot
|
|
// overwrite a newer cancellation or completion on the server.
|
|
if !a.reportBuffer.IsEmpty() {
|
|
a.flushBuffer(ctx)
|
|
if !a.reportBuffer.IsEmpty() {
|
|
a.bufferPrimaryReport(report)
|
|
return nil
|
|
}
|
|
}
|
|
|
|
if err := a.sendReport(ctx, report); err != nil {
|
|
agenttarget.MarkDelivery("host", "primary", "primary", false)
|
|
a.availability.discardInFlight()
|
|
var statusErr *reportHTTPStatusError
|
|
if errors.As(err, &statusErr) && statusErr.StatusCode == http.StatusForbidden {
|
|
a.logger.Error().
|
|
Err(err).
|
|
Str("endpoint", statusErr.Endpoint).
|
|
Int("status_code", statusErr.StatusCode).
|
|
Msg("Failed to send Unified Agent report (403 Forbidden). API token may lack 'Unified Agent reporting' scope. Set PULSE_ENABLE_HOST=false if host monitoring is not needed.")
|
|
return nil
|
|
}
|
|
if errors.As(err, &statusErr) && statusErr.StatusCode == http.StatusUnauthorized {
|
|
// 401 means the server does not recognise this agent's API token
|
|
// (for example after a Pulse restore or a v5 -> v6 upgrade that did
|
|
// not carry the token across). Retrying with the same token loops
|
|
// forever, so drop the report instead of buffering it and surface an
|
|
// actionable error the operator can act on. Throttle it so a
|
|
// permanently rejected token does not flood the log.
|
|
a.logAuthFailure(statusErr)
|
|
return nil
|
|
}
|
|
|
|
a.bufferPrimaryReport(report)
|
|
event := a.logger.Warn().
|
|
Err(err).
|
|
Str("endpoint", agentReportEndpoint).
|
|
Int("buffered_reports", a.reportBuffer.Len())
|
|
if errors.As(err, &statusErr) {
|
|
event.Str("endpoint", statusErr.Endpoint)
|
|
event.Int("status_code", statusErr.StatusCode)
|
|
}
|
|
event.Msg("Failed to send report, buffering")
|
|
return nil
|
|
}
|
|
agenttarget.MarkDelivery("host", "primary", "primary", true)
|
|
if a.cfg.OnPrimaryReportAccepted != nil {
|
|
a.cfg.OnPrimaryReportAccepted()
|
|
}
|
|
|
|
// The server has counted every availability result in this report, so they
|
|
// must never be offered to it again.
|
|
a.availability.commitDelivered()
|
|
|
|
// A successful report means the token is accepted again; reset the auth
|
|
// failure throttle so a later rejection is reported promptly.
|
|
a.lastAuthFailureLog = time.Time{}
|
|
|
|
// If successful, try to flush buffer
|
|
a.flushBuffer(ctx)
|
|
|
|
a.logger.Debug().
|
|
Str("hostname", report.Host.Hostname).
|
|
Str("platform", report.Host.Platform).
|
|
Msg("Unified Agent report sent")
|
|
return nil
|
|
}
|
|
|
|
// bufferPrimaryReport queues a report the primary destination has not accepted
|
|
// yet. Availability results are stripped from the buffered copy and left in the
|
|
// probe module's queue, so the retry carries them exactly once: a buffered copy
|
|
// that also held them could be delivered alongside a fresh report and make the
|
|
// server count the same observation twice.
|
|
func (a *Agent) bufferPrimaryReport(report agentshost.Report) {
|
|
a.availability.discardInFlight()
|
|
report.AvailabilityResults = nil
|
|
a.reportBuffer.Push(report)
|
|
}
|
|
|
|
func (a *Agent) deliverObserverReport(ctx context.Context, observer *observerReporter, report agentshost.Report) {
|
|
if !observer.reportBuffer.IsEmpty() {
|
|
a.flushObserverBuffer(ctx, observer)
|
|
if !observer.reportBuffer.IsEmpty() {
|
|
observer.reportBuffer.Push(report)
|
|
return
|
|
}
|
|
}
|
|
|
|
if err := a.sendReportToDestination(ctx, report, observer.target.PulseURL, observer.target.APIToken, observer.httpClient, false); err != nil {
|
|
agenttarget.MarkDelivery("host", observer.target.Name, "observer", false)
|
|
var statusErr *reportHTTPStatusError
|
|
if errors.As(err, &statusErr) && (statusErr.StatusCode == http.StatusUnauthorized || statusErr.StatusCode == http.StatusForbidden) {
|
|
if time.Since(observer.lastAuthFailureLog) >= authFailureLogInterval {
|
|
observer.lastAuthFailureLog = time.Now()
|
|
a.logger.Error().Err(err).
|
|
Str("destination", observer.target.Name).
|
|
Str("role", "observer").
|
|
Str("pulse_url", observer.target.PulseURL).
|
|
Msg("Pulse observer rejected its API token; reports for this destination are dropped until the token is replaced")
|
|
}
|
|
return
|
|
}
|
|
observer.reportBuffer.Push(report)
|
|
a.logger.Warn().Err(err).
|
|
Str("destination", observer.target.Name).
|
|
Str("role", "observer").
|
|
Int("buffered_reports", observer.reportBuffer.Len()).
|
|
Msg("Failed to send observer report, buffering only for this destination")
|
|
return
|
|
}
|
|
|
|
agenttarget.MarkDelivery("host", observer.target.Name, "observer", true)
|
|
observer.lastAuthFailureLog = time.Time{}
|
|
a.flushObserverBuffer(ctx, observer)
|
|
}
|
|
|
|
func (a *Agent) flushObserverBuffer(ctx context.Context, observer *observerReporter) {
|
|
for !observer.reportBuffer.IsEmpty() {
|
|
report, ok := observer.reportBuffer.Peek()
|
|
if !ok {
|
|
return
|
|
}
|
|
if err := a.sendReportToDestination(ctx, report, observer.target.PulseURL, observer.target.APIToken, observer.httpClient, false); err != nil {
|
|
var statusErr *reportHTTPStatusError
|
|
if errors.As(err, &statusErr) && (statusErr.StatusCode == http.StatusUnauthorized || statusErr.StatusCode == http.StatusForbidden) {
|
|
for {
|
|
if _, ok := observer.reportBuffer.Pop(); !ok {
|
|
break
|
|
}
|
|
}
|
|
}
|
|
return
|
|
}
|
|
observer.reportBuffer.Pop()
|
|
}
|
|
}
|
|
|
|
// logAuthFailure emits an actionable, throttled error when the server rejects
|
|
// the agent's API token. It is only called from the single-threaded report
|
|
// loop, so lastAuthFailureLog needs no additional synchronisation.
|
|
func (a *Agent) logAuthFailure(statusErr *reportHTTPStatusError) {
|
|
if time.Since(a.lastAuthFailureLog) < authFailureLogInterval {
|
|
return
|
|
}
|
|
a.lastAuthFailureLog = time.Now()
|
|
|
|
endpoint := agentReportEndpoint
|
|
statusCode := http.StatusUnauthorized
|
|
if statusErr != nil {
|
|
if statusErr.Endpoint != "" {
|
|
endpoint = statusErr.Endpoint
|
|
}
|
|
statusCode = statusErr.StatusCode
|
|
}
|
|
|
|
event := a.logger.Error().
|
|
Str("endpoint", endpoint).
|
|
Int("status_code", statusCode).
|
|
Str("pulse_url", a.trimmedPulseURL)
|
|
if statusCode == http.StatusForbidden {
|
|
event.Msg("Pulse rejected this agent's API token (403 Forbidden). The token may lack the Unified Agent reporting scope. Re-run the agent install command from the Pulse UI to mint a correctly scoped token. Reports are dropped until the token is replaced.")
|
|
return
|
|
}
|
|
event.Msg("Pulse rejected this agent's API token (401 Unauthorized). The token is no longer valid on the server, which usually means Pulse was restored/reinstalled or upgraded (for example v5 -> v6) without carrying the token across. Re-run the agent install command from the Pulse UI to mint a fresh token. Reports are dropped until the token is replaced.")
|
|
}
|
|
|
|
func (a *Agent) flushBuffer(ctx context.Context) {
|
|
if a.reportBuffer.IsEmpty() {
|
|
return
|
|
}
|
|
|
|
a.logger.Info().Int("count", a.reportBuffer.Len()).Msg("Flushing buffered reports")
|
|
|
|
for !a.reportBuffer.IsEmpty() {
|
|
// Peek first
|
|
report, ok := a.reportBuffer.Peek()
|
|
if !ok {
|
|
break
|
|
}
|
|
|
|
if err := a.sendReport(ctx, report); err != nil {
|
|
var statusErr *reportHTTPStatusError
|
|
if errors.As(err, &statusErr) &&
|
|
(statusErr.StatusCode == http.StatusUnauthorized || statusErr.StatusCode == http.StatusForbidden) {
|
|
// The token is rejected; buffered reports will never be
|
|
// accepted with it. Drop them so the buffer cannot grow
|
|
// unbounded across restarts, and surface the actionable error.
|
|
a.logAuthFailure(statusErr)
|
|
for {
|
|
if _, ok := a.reportBuffer.Pop(); !ok {
|
|
break
|
|
}
|
|
}
|
|
return
|
|
}
|
|
event := a.logger.Warn().
|
|
Err(err).
|
|
Str("endpoint", agentReportEndpoint).
|
|
Int("remaining_buffered_reports", a.reportBuffer.Len())
|
|
if errors.As(err, &statusErr) {
|
|
event.Str("endpoint", statusErr.Endpoint)
|
|
event.Int("status_code", statusErr.StatusCode)
|
|
}
|
|
event.Msg("Failed to flush buffered report, stopping flush")
|
|
return
|
|
}
|
|
|
|
// Pop only on success
|
|
a.reportBuffer.Pop()
|
|
}
|
|
}
|
|
|
|
const bufferFileName = "report-buffer.json"
|
|
|
|
func observerBufferFileName(id string) string {
|
|
return "report-buffer-observer-" + id + ".json"
|
|
}
|
|
|
|
// persistBuffer writes buffered reports to disk on shutdown so they can be
|
|
// retransmitted on the next startup. Uses atomic write (tmp + rename) to
|
|
// prevent corruption if the process is killed mid-write.
|
|
func (a *Agent) persistBuffer() {
|
|
a.persistReportQueue(a.reportBuffer, bufferFileName, "primary")
|
|
}
|
|
|
|
func (a *Agent) persistObserverBuffers() {
|
|
for _, observer := range a.observerReporters {
|
|
a.persistReportQueue(observer.reportBuffer, observerBufferFileName(observer.target.ID), observer.target.Name)
|
|
}
|
|
}
|
|
|
|
func (a *Agent) persistReportQueue(queue *utils.Queue[agentshost.Report], fileName, destination string) {
|
|
items := queue.Items()
|
|
if len(items) == 0 {
|
|
return
|
|
}
|
|
|
|
if a.stateDir == "" {
|
|
a.logger.Debug().Msg("No state dir configured, skipping buffer persistence")
|
|
return
|
|
}
|
|
|
|
data, err := json.Marshal(items)
|
|
if err != nil {
|
|
a.logger.Warn().Err(err).Msg("Failed to marshal report buffer for persistence")
|
|
return
|
|
}
|
|
|
|
if err := os.MkdirAll(a.stateDir, 0700); err != nil {
|
|
a.logger.Warn().Err(err).Str("dir", a.stateDir).Msg("Failed to create state dir for buffer persistence")
|
|
return
|
|
}
|
|
if err := os.Chmod(a.stateDir, 0700); err != nil {
|
|
a.logger.Warn().Err(err).Str("dir", a.stateDir).Msg("Failed to enforce state dir permissions for buffer persistence")
|
|
return
|
|
}
|
|
|
|
path := filepath.Join(a.stateDir, fileName)
|
|
tmpPath := path + ".tmp"
|
|
if err := os.WriteFile(tmpPath, data, 0600); err != nil {
|
|
a.logger.Warn().Err(err).Str("path", tmpPath).Msg("Failed to write report buffer temp file")
|
|
return
|
|
}
|
|
if err := os.Rename(tmpPath, path); err != nil {
|
|
a.logger.Warn().Err(err).Str("path", path).Msg("Failed to rename report buffer file")
|
|
_ = os.Remove(tmpPath)
|
|
return
|
|
}
|
|
|
|
a.logger.Info().Int("count", len(items)).Str("path", path).Str("destination", destination).Msg("Persisted report buffer to disk")
|
|
}
|
|
|
|
// loadPersistedBuffer loads buffered reports from a previous shutdown and
|
|
// attempts to flush them. The file is deleted after loading regardless of
|
|
// whether the flush succeeds (items are pushed back into the in-memory buffer).
|
|
func (a *Agent) loadPersistedBuffer() {
|
|
a.loadPersistedReportQueue(a.reportBuffer, bufferFileName, "primary")
|
|
}
|
|
|
|
func (a *Agent) loadPersistedObserverBuffers() {
|
|
for _, observer := range a.observerReporters {
|
|
a.loadPersistedReportQueue(observer.reportBuffer, observerBufferFileName(observer.target.ID), observer.target.Name)
|
|
}
|
|
}
|
|
|
|
func (a *Agent) loadPersistedReportQueue(queue *utils.Queue[agentshost.Report], fileName, destination string) {
|
|
if a.stateDir == "" {
|
|
return
|
|
}
|
|
|
|
path := filepath.Join(a.stateDir, fileName)
|
|
data, err := os.ReadFile(path)
|
|
if err != nil {
|
|
if !os.IsNotExist(err) {
|
|
a.logger.Warn().Err(err).Str("path", path).Msg("Failed to read persisted report buffer")
|
|
}
|
|
return
|
|
}
|
|
|
|
// Always delete the file after reading
|
|
_ = os.Remove(path)
|
|
|
|
var items []agentshost.Report
|
|
if err := json.Unmarshal(data, &items); err != nil {
|
|
a.logger.Warn().Err(err).Str("path", path).Msg("Corrupt report buffer file, discarding")
|
|
return
|
|
}
|
|
|
|
if len(items) == 0 {
|
|
return
|
|
}
|
|
|
|
for _, item := range items {
|
|
queue.Push(item)
|
|
}
|
|
|
|
a.logger.Info().Int("count", len(items)).Str("destination", destination).Msg("Loaded persisted report buffer from disk")
|
|
}
|
|
|
|
func (a *Agent) buildReport(ctx context.Context) (agentshost.Report, error) {
|
|
// Unraid's native inventory is the authority for array membership. Collect
|
|
// it before optional disk diagnostics and on an independent deadline so a
|
|
// slow or unsupported SMART target cannot starve mdcmd/disks.ini.
|
|
unraidCtx, cancelUnraid := context.WithTimeout(ctx, 5*time.Second)
|
|
unraidData := a.collectUnraidStorage(unraidCtx)
|
|
cancelUnraid()
|
|
|
|
collectCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
|
|
defer cancel()
|
|
|
|
runtimeConfig := a.runtimeConfigSnapshot()
|
|
uptime, err := a.collector.HostUptime(collectCtx)
|
|
if err != nil {
|
|
a.logger.Debug().Err(err).Msg("Failed to collect host uptime; defaulting to 0")
|
|
}
|
|
snapshot, err := collectMetricsWithDiskFilters(collectCtx, a.collector, runtimeConfig.diskExclude, runtimeConfig.diskInclude)
|
|
if err != nil {
|
|
return agentshost.Report{}, fmt.Errorf("collect metrics: %w", err)
|
|
}
|
|
|
|
// Collect temperature data (best effort - don't fail if unavailable)
|
|
sensorData := a.collectTemperatures(collectCtx)
|
|
sensorData.Custom = a.customSensors.collect(collectCtx)
|
|
|
|
// Collect RAID array data (best effort - don't fail if unavailable)
|
|
raidData := a.collectRAIDArrays(collectCtx)
|
|
|
|
// Collect Ceph cluster data (best effort - only on Ceph nodes)
|
|
cephData := a.collectCephStatus(collectCtx, runtimeConfig.disableCeph)
|
|
|
|
// Collect local libvirt domain inventory on KVM-capable Linux hosts.
|
|
// The collector uses a read-only connection and its own bounded deadline so
|
|
// an unavailable hypervisor cannot delay the rest of the host report.
|
|
libvirtData := a.collectLibvirtInventory(ctx)
|
|
|
|
// XCP-ng exposes pool inventory through the local, read-only xe CLI. It has
|
|
// an independent bounded deadline so an unavailable XAPI cannot delay the
|
|
// normal host report.
|
|
xcpngData := a.collectXCPNGInventory(ctx)
|
|
|
|
// Proxmox VE exposes per-mount LXC usage only through the node-local pct
|
|
// CLI. Query running containers on an independent bounded deadline.
|
|
proxmoxLXCData := a.collectProxmoxLXCFilesystemsForReport(ctx)
|
|
|
|
// Collect S.M.A.R.T. disk data after topology owners. Enumeration and each
|
|
// device probe have their own deadlines; a single shared 10-second budget
|
|
// caused later disks on larger arrays to be cancelled before their probe.
|
|
smartData := a.collectSMARTData(ctx, runtimeConfig.diskExclude, unraidData)
|
|
if len(smartData) > 0 {
|
|
annotateSMARTWithDiskIO(smartData, snapshot.DiskIO)
|
|
sensorData.SMART = smartData
|
|
}
|
|
|
|
// Collect temperature data from Proxmox cluster peers via SSH (best effort).
|
|
// Uses parent ctx, not collectCtx — cluster SSH has its own 15s budget that
|
|
// would be capped by collectCtx's 10s timeout.
|
|
clusterSensors := a.collectClusterSensors(ctx)
|
|
|
|
packageUpdateCtx, cancelPackageUpdates := context.WithTimeout(ctx, 20*time.Second)
|
|
packageUpdates := a.currentPackageUpdateStatus(packageUpdateCtx)
|
|
cancelPackageUpdates()
|
|
storageCleanupCtx, cancelStorageCleanup := context.WithTimeout(ctx, 20*time.Second)
|
|
storageCleanup := a.currentStorageCleanupStatus(storageCleanupCtx)
|
|
cancelStorageCleanup()
|
|
|
|
// Carry updated_from on the first freshly built v6 report only. If that
|
|
// report is buffered, the buffered copy still retains the field for retry.
|
|
updatedFrom := a.updatedFrom
|
|
if updatedFrom != "" {
|
|
a.updatedFrom = ""
|
|
}
|
|
moduleStatus := a.currentModuleStatus()
|
|
if libvirtData != nil {
|
|
moduleStatus = append(moduleStatus, agentshost.ModuleStatus{
|
|
Name: "libvirt",
|
|
Enabled: true,
|
|
State: "running",
|
|
UpdatedAt: libvirtData.CollectedAt,
|
|
})
|
|
}
|
|
if xcpngData != nil {
|
|
moduleStatus = append(moduleStatus, agentshost.ModuleStatus{
|
|
Name: "xcp-ng",
|
|
Enabled: true,
|
|
State: "running",
|
|
UpdatedAt: xcpngData.CollectedAt,
|
|
})
|
|
}
|
|
if proxmoxLXCData != nil {
|
|
moduleStatus = append(moduleStatus, agentshost.ModuleStatus{
|
|
Name: "proxmox-lxc-filesystems",
|
|
Enabled: true,
|
|
State: "running",
|
|
UpdatedAt: proxmoxLXCData.CollectedAt,
|
|
})
|
|
}
|
|
|
|
report := agentshost.Report{
|
|
Agent: agentshost.AgentInfo{
|
|
ID: a.agentID,
|
|
Version: a.agentVersion,
|
|
Type: runtimeConfig.agentType,
|
|
IntervalSeconds: int(runtimeConfig.interval / time.Second),
|
|
Hostname: a.hostname,
|
|
UpdatedFrom: updatedFrom,
|
|
CommandsEnabled: runtimeConfig.commandsEnabled,
|
|
OperationReceiptVersion: a.currentOperationReceiptVersion(),
|
|
DiskExclude: append([]string(nil), runtimeConfig.diskExclude...),
|
|
AppliedConfig: runtimeConfig.appliedConfig,
|
|
Update: a.currentUpdateStatus(),
|
|
Modules: moduleStatus,
|
|
Privilege: collectPrivilegeStatus(a.commandAuthorityProfile),
|
|
},
|
|
Host: agentshost.HostInfo{
|
|
ID: a.machineID,
|
|
Hostname: a.hostname,
|
|
DisplayName: a.displayName,
|
|
MachineID: a.machineID,
|
|
Platform: a.platform,
|
|
OSName: a.osName,
|
|
OSVersion: a.osVersion,
|
|
KernelVersion: a.kernelVersion,
|
|
Architecture: a.architecture,
|
|
CPUModel: "",
|
|
CPUCount: snapshot.CPUCount,
|
|
UptimeSeconds: int64(uptime),
|
|
LoadAverage: append([]float64(nil), snapshot.LoadAverage...),
|
|
ReportIP: runtimeConfig.reportIP,
|
|
PackageUpdates: packageUpdates,
|
|
StorageCleanup: storageCleanup,
|
|
},
|
|
Metrics: agentshost.Metrics{
|
|
CPUUsagePercent: snapshot.CPUUsagePercent,
|
|
Memory: snapshot.Memory,
|
|
},
|
|
Disks: append([]agentshost.Disk(nil), snapshot.Disks...),
|
|
DiskIO: append([]agentshost.DiskIO(nil), snapshot.DiskIO...),
|
|
Network: append([]agentshost.NetworkInterface(nil), snapshot.Network...),
|
|
Sensors: sensorData,
|
|
RAID: raidData,
|
|
Unraid: unraidData,
|
|
Ceph: cephData,
|
|
Libvirt: libvirtData,
|
|
XCPNG: xcpngData,
|
|
ProxmoxLXC: proxmoxLXCData,
|
|
ClusterSensors: clusterSensors,
|
|
// Results stay queued until the primary destination accepts this
|
|
// report, so a delivery failure retries them instead of losing them.
|
|
AvailabilityResults: a.availability.snapshotForReport(),
|
|
Tags: append([]string(nil), runtimeConfig.tags...),
|
|
Timestamp: a.collector.Now(),
|
|
SequenceID: a.nextReportSequenceID(),
|
|
}
|
|
|
|
return report, nil
|
|
}
|
|
|
|
func (a *Agent) currentPackageUpdateStatus(ctx context.Context) *agentshost.PackageUpdateStatus {
|
|
if a == nil || a.packageUpdates == nil {
|
|
return nil
|
|
}
|
|
snapshot := a.packageUpdates.Snapshot(ctx, false)
|
|
packages := make([]agentshost.PackageUpdate, len(snapshot.Packages))
|
|
for i, pkg := range snapshot.Packages {
|
|
packages[i] = agentshost.PackageUpdate{
|
|
Name: pkg.Name,
|
|
InstalledVersion: pkg.InstalledVersion,
|
|
AvailableVersion: pkg.AvailableVersion,
|
|
}
|
|
}
|
|
return &agentshost.PackageUpdateStatus{
|
|
Supported: snapshot.Supported,
|
|
Manager: snapshot.Manager,
|
|
InventoryHash: snapshot.InventoryHash,
|
|
PendingCount: snapshot.PendingCount,
|
|
Packages: packages,
|
|
CheckedAt: snapshot.CheckedAt,
|
|
RebootRequired: snapshot.RebootRequired,
|
|
Error: snapshot.Error,
|
|
}
|
|
}
|
|
|
|
func (a *Agent) currentStorageCleanupStatus(ctx context.Context) *agentshost.StorageCleanupStatus {
|
|
if a == nil || a.storageCleanup == nil {
|
|
return nil
|
|
}
|
|
snapshot := a.storageCleanup.Snapshot(ctx, false)
|
|
return &agentshost.StorageCleanupStatus{
|
|
Supported: snapshot.Supported,
|
|
Provider: snapshot.Provider,
|
|
Fingerprint: snapshot.Fingerprint,
|
|
ReclaimableBytes: snapshot.ReclaimableBytes,
|
|
CheckedAt: snapshot.CheckedAt,
|
|
Error: snapshot.Error,
|
|
}
|
|
}
|
|
|
|
func (a *Agent) sendReport(ctx context.Context, report agentshost.Report) error {
|
|
return a.sendReportToDestination(ctx, report, a.trimmedPulseURL, a.cfg.APIToken, a.httpClient, true)
|
|
}
|
|
|
|
func (a *Agent) sendReportToDestination(ctx context.Context, report agentshost.Report, pulseURL, token string, client *http.Client, authoritative bool) error {
|
|
payload, err := json.Marshal(report)
|
|
if err != nil {
|
|
return fmt.Errorf("marshal report: %w", err)
|
|
}
|
|
|
|
compressed, err := utils.CompressJSON(payload)
|
|
if err != nil {
|
|
return fmt.Errorf("compress report: %w", err)
|
|
}
|
|
|
|
endpoint := agentReportEndpoint
|
|
url := fmt.Sprintf("%s%s", strings.TrimRight(pulseURL, "/"), endpoint)
|
|
req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(compressed))
|
|
if err != nil {
|
|
return fmt.Errorf("create request: %w", err)
|
|
}
|
|
|
|
req.Header.Set("Content-Type", "application/json")
|
|
req.Header.Set("Content-Encoding", "gzip")
|
|
if token := strings.TrimSpace(token); token != "" {
|
|
req.Header.Set("Authorization", "Bearer "+token)
|
|
req.Header.Set("X-API-Token", token)
|
|
}
|
|
req.Header.Set("User-Agent", "pulse-agent/"+Version)
|
|
|
|
resp, err := client.Do(req)
|
|
if err != nil {
|
|
return fmt.Errorf("send request: %w", err)
|
|
}
|
|
defer func() {
|
|
if closeErr := resp.Body.Close(); closeErr != nil {
|
|
a.logger.Debug().Err(closeErr).Msg("Failed to close report response body")
|
|
}
|
|
}()
|
|
|
|
if resp.StatusCode >= 300 {
|
|
return &reportHTTPStatusError{
|
|
Endpoint: endpoint,
|
|
Status: resp.Status,
|
|
StatusCode: resp.StatusCode,
|
|
}
|
|
}
|
|
if !authoritative {
|
|
_, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, 64*1024))
|
|
return nil
|
|
}
|
|
|
|
// Parse response to check for server-side config overrides
|
|
var reportResp struct {
|
|
Success bool `json:"success"`
|
|
AgentID string `json:"agentId"`
|
|
ServerVersion string `json:"serverVersion"`
|
|
Config *struct {
|
|
CommandsEnabled *bool `json:"commandsEnabled"`
|
|
} `json:"config,omitempty"`
|
|
}
|
|
if err := json.NewDecoder(resp.Body).Decode(&reportResp); err != nil {
|
|
// Non-fatal: just log and continue
|
|
a.logger.Debug().Err(err).Msg("Failed to parse report response, ignoring config")
|
|
return nil
|
|
}
|
|
|
|
// Surface the server's version so the updater can converge promptly after
|
|
// a server upgrade rather than waiting out its hourly check interval.
|
|
if a.cfg.OnServerVersion != nil {
|
|
if serverVersion := strings.TrimSpace(reportResp.ServerVersion); serverVersion != "" {
|
|
a.cfg.OnServerVersion(serverVersion)
|
|
}
|
|
}
|
|
|
|
// Persist the server-acknowledged agent ID so uninstall can deregister.
|
|
canonicalAgentID := strings.TrimSpace(reportResp.AgentID)
|
|
if canonicalAgentID != "" {
|
|
a.warnCanonicalIDMismatch(canonicalAgentID)
|
|
a.persistAgentID(canonicalAgentID)
|
|
}
|
|
|
|
// Apply server config overrides
|
|
if reportResp.Config != nil && reportResp.Config.CommandsEnabled != nil {
|
|
a.applyRemoteConfig(*reportResp.Config.CommandsEnabled)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// warnCanonicalIDMismatch surfaces the case where the server resolved this
|
|
// agent to a different identity than the one it presented. Host identity
|
|
// continuity deliberately keeps a known machine on its enrolled ID even when
|
|
// the operator passes --agent-id or hand-edits the agent-id state file
|
|
// (#1739), and the acknowledged ID then overwrites that file for uninstall
|
|
// deregistration. Without this warning the override is ignored silently and
|
|
// the file rewrite looks like corruption. Warns once per resolved identity.
|
|
func (a *Agent) warnCanonicalIDMismatch(canonicalAgentID string) {
|
|
if canonicalAgentID == a.agentID || canonicalAgentID == a.canonicalIDWarned {
|
|
return
|
|
}
|
|
a.canonicalIDWarned = canonicalAgentID
|
|
a.logger.Warn().
|
|
Str("presented_agent_id", a.agentID).
|
|
Str("server_agent_id", canonicalAgentID).
|
|
Msg("Server resolved this agent to its enrolled identity and the presented agent ID was not adopted. Host identity continuity keeps a known machine on its existing ID. To enroll under a new ID, remove the host in Pulse first, then restart the agent.")
|
|
}
|
|
|
|
// persistAgentID writes the server-assigned agent ID to the state directory.
|
|
// This file is read by the uninstall script to deregister the agent from the server.
|
|
// Errors are debug-logged, never fatal — same resilience pattern as proxmox_setup.go.
|
|
func (a *Agent) persistAgentID(agentID string) {
|
|
if a.stateDir == "" {
|
|
return
|
|
}
|
|
if err := a.collector.MkdirAll(a.stateDir, 0700); err != nil {
|
|
a.logger.Debug().Err(err).Msg("Failed to create state directory for agent-id")
|
|
return
|
|
}
|
|
if err := a.collector.Chmod(a.stateDir, 0700); err != nil {
|
|
a.logger.Debug().Err(err).Msg("Failed to enforce state directory permissions for agent-id")
|
|
return
|
|
}
|
|
agentIDPath := filepath.Join(a.stateDir, "agent-id")
|
|
if err := a.collector.WriteFile(agentIDPath, []byte(agentID), 0600); err != nil {
|
|
a.logger.Debug().Err(err).Msg("Failed to persist agent-id")
|
|
return
|
|
}
|
|
if err := a.collector.Chmod(agentIDPath, 0600); err != nil {
|
|
a.logger.Debug().Err(err).Msg("Failed to enforce agent-id file permissions")
|
|
}
|
|
}
|
|
|
|
// ApplyRemoteConfig applies server-side configuration overrides that can be
|
|
// reconciled safely without restarting the agent process.
|
|
func (a *Agent) ApplyRemoteConfig(settings map[string]interface{}, commandsEnabled *bool) {
|
|
commandAuthorityApplied := true
|
|
if commandsEnabled != nil {
|
|
_, commandAuthorityApplied = a.applyRemoteConfig(*commandsEnabled)
|
|
}
|
|
if interval, ok := remoteDurationSetting(settings, "interval"); ok && interval > 0 {
|
|
intervalChanged := false
|
|
a.configMu.Lock()
|
|
if a.interval != interval {
|
|
intervalChanged = true
|
|
a.interval = interval
|
|
a.cfg.Interval = interval
|
|
}
|
|
a.configMu.Unlock()
|
|
if intervalChanged {
|
|
a.signalRemoteConfigChanged()
|
|
a.logger.Info().Dur("interval", interval).Msg("Applied remote host report interval")
|
|
}
|
|
}
|
|
if reportIP, ok := remoteStringSetting(settings, "report_ip"); ok {
|
|
trimmedReportIP := strings.TrimSpace(reportIP)
|
|
reportIPChanged := false
|
|
a.configMu.Lock()
|
|
if a.reportIP != trimmedReportIP {
|
|
reportIPChanged = true
|
|
a.reportIP = trimmedReportIP
|
|
a.cfg.ReportIP = trimmedReportIP
|
|
}
|
|
a.configMu.Unlock()
|
|
if reportIPChanged {
|
|
a.logger.Info().Str("report_ip", trimmedReportIP).Msg("Applied remote host report IP override")
|
|
}
|
|
}
|
|
if disableCeph, ok := remoteBoolSetting(settings, "disable_ceph"); ok {
|
|
disableCephChanged := false
|
|
a.configMu.Lock()
|
|
if a.cfg.DisableCeph != disableCeph {
|
|
disableCephChanged = true
|
|
a.cfg.DisableCeph = disableCeph
|
|
}
|
|
a.configMu.Unlock()
|
|
if disableCephChanged {
|
|
a.logger.Info().Bool("disable_ceph", disableCeph).Msg("Applied remote Ceph collection setting")
|
|
}
|
|
}
|
|
a.applyRemoteAvailabilityTargets(settings)
|
|
if commandAuthorityApplied && remoteConfigAppliedWithoutRestart(settings) && remoteconfig.HasAppliedDesiredConfig(commandsEnabled, settings) {
|
|
metadata, err := remoteconfig.BuildDesiredConfigMetadata(commandsEnabled, settings)
|
|
if err != nil {
|
|
a.logger.Warn().Err(err).Msg("Failed to derive refreshed managed configuration fingerprint")
|
|
return
|
|
}
|
|
a.configMu.Lock()
|
|
a.cfg.AppliedConfig = &agentshost.ConfigFingerprint{Version: metadata.Version, Hash: metadata.Hash}
|
|
a.configMu.Unlock()
|
|
}
|
|
}
|
|
|
|
// applyRemoteAvailabilityTargets reconciles the externally probed availability
|
|
// checks assigned to this agent. A payload the agent cannot decode leaves the
|
|
// current assignment in place: results for targets it no longer owns are
|
|
// rejected server-side, so keeping the schedule is safer than blindly
|
|
// abandoning coverage over a malformed field.
|
|
func (a *Agent) applyRemoteAvailabilityTargets(settings map[string]interface{}) {
|
|
targets, err := availabilityTargetsFromSettings(settings)
|
|
if err != nil {
|
|
a.logger.Warn().Err(err).Msg("Ignoring unreadable availability probe assignments")
|
|
return
|
|
}
|
|
if !a.availability.applyTargets(targets) {
|
|
return
|
|
}
|
|
a.logger.Info().Int("targets", len(targets)).Msg("Applied remote availability probe assignments")
|
|
}
|
|
|
|
func remoteConfigAppliedWithoutRestart(settings map[string]interface{}) bool {
|
|
for key := range settings {
|
|
switch key {
|
|
case "interval", "report_ip", "disable_ceph", availabilitySettingsKey:
|
|
continue
|
|
default:
|
|
return false
|
|
}
|
|
}
|
|
return true
|
|
}
|
|
|
|
// applyRemoteConfig applies server-side command execution overrides without
|
|
// exceeding the immutable authority provisioned when this process started.
|
|
// It returns the effective state and whether the requested state was accepted.
|
|
func (a *Agent) applyRemoteConfig(commandsEnabled bool) (bool, bool) {
|
|
var (
|
|
clientToStart *CommandClient
|
|
commandCfg Config
|
|
shouldStop bool
|
|
logRejected bool
|
|
effectiveState, authorityAccepted = ResolveCommandAuthority(a.commandAuthorityProfile, commandsEnabled)
|
|
)
|
|
|
|
a.configMu.Lock()
|
|
if !authorityAccepted && !a.commandAuthorityWarningSeen {
|
|
a.commandAuthorityWarningSeen = true
|
|
logRejected = true
|
|
}
|
|
if !commandsEnabled {
|
|
a.commandAuthorityWarningSeen = false
|
|
}
|
|
a.cfg.EnableCommands = effectiveState
|
|
commandCfg = a.cfg
|
|
a.configMu.Unlock()
|
|
|
|
a.commandClientMu.Lock()
|
|
currentlyEnabled := a.commandClient != nil
|
|
switch {
|
|
case effectiveState && !currentlyEnabled:
|
|
clientToStart = a.newCommandClient(commandCfg, a.agentID, a.hostname, a.platform, a.agentVersion)
|
|
a.commandClient = clientToStart
|
|
case !effectiveState && currentlyEnabled:
|
|
shouldStop = true
|
|
}
|
|
a.commandClientMu.Unlock()
|
|
|
|
if logRejected {
|
|
a.logger.Warn().Msg("Ignored remote command enablement because this monitoring-only runtime was not provisioned with command authority; reinstall with an explicit command profile")
|
|
} else if clientToStart != nil {
|
|
if a.startCommandClient(clientToStart) {
|
|
a.logger.Info().Msg("Server enabled command execution - starting command client")
|
|
} else {
|
|
a.logger.Info().Msg("Server enabled command execution - command client configured pending run context")
|
|
}
|
|
} else if shouldStop {
|
|
a.logger.Info().Msg("Server disabled command execution - stopping command client")
|
|
a.stopCommandClient(true)
|
|
}
|
|
|
|
return effectiveState, authorityAccepted
|
|
}
|
|
|
|
func remoteBoolSetting(settings map[string]interface{}, key string) (bool, bool) {
|
|
if settings == nil {
|
|
return false, false
|
|
}
|
|
value, ok := settings[key]
|
|
if !ok {
|
|
return false, false
|
|
}
|
|
parsed, ok := value.(bool)
|
|
return parsed, ok
|
|
}
|
|
|
|
func remoteStringSetting(settings map[string]interface{}, key string) (string, bool) {
|
|
if settings == nil {
|
|
return "", false
|
|
}
|
|
value, ok := settings[key]
|
|
if !ok {
|
|
return "", false
|
|
}
|
|
parsed, ok := value.(string)
|
|
return parsed, ok
|
|
}
|
|
|
|
func remoteDurationSetting(settings map[string]interface{}, key string) (time.Duration, bool) {
|
|
if settings == nil {
|
|
return 0, false
|
|
}
|
|
value, ok := settings[key]
|
|
if !ok {
|
|
return 0, false
|
|
}
|
|
switch typed := value.(type) {
|
|
case string:
|
|
parsed, err := time.ParseDuration(typed)
|
|
if err != nil {
|
|
return 0, false
|
|
}
|
|
return parsed, true
|
|
case float64:
|
|
return time.Duration(typed * float64(time.Second)), true
|
|
case int:
|
|
return time.Duration(typed) * time.Second, true
|
|
case int64:
|
|
return time.Duration(typed) * time.Second, true
|
|
default:
|
|
return 0, false
|
|
}
|
|
}
|
|
|
|
func normalisePlatform(platform string) string {
|
|
normalized := platformsupport.NormalizeAgentReportedPlatform(platform)
|
|
if runtimePlatform := platformsupport.RuntimePlatformForHostIdentityToken(normalized); runtimePlatform != "" {
|
|
return runtimePlatform
|
|
}
|
|
return normalized
|
|
}
|
|
|
|
func normaliseRuntimePlatform(goos, reportedPlatform string) string {
|
|
switch strings.ToLower(strings.TrimSpace(goos)) {
|
|
case "linux":
|
|
return platformsupport.RuntimePlatformLinux
|
|
case "darwin":
|
|
return platformsupport.RuntimePlatformMacOS
|
|
case "freebsd":
|
|
return platformsupport.RuntimePlatformFreeBSD
|
|
case "windows":
|
|
return platformsupport.RuntimePlatformWindows
|
|
default:
|
|
return normalisePlatform(reportedPlatform)
|
|
}
|
|
}
|
|
|
|
func normalizePulseURL(rawURL string) (string, error) {
|
|
parsed, err := securityutil.NormalizePulseHTTPBaseURLWithOptions(rawURL, securityutil.PulseURLValidationOptions{
|
|
AllowLocalNetworkHTTP: true,
|
|
})
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
|
|
return parsed.String(), nil
|
|
}
|
|
|
|
// collectTemperatures attempts to collect temperature data from the local system.
|
|
// Returns an empty Sensors struct if collection fails (best-effort).
|
|
func (a *Agent) collectTemperatures(ctx context.Context) agentshost.Sensors {
|
|
switch a.collector.GOOS() {
|
|
case "linux":
|
|
// Continue below with lm-sensors path
|
|
case "darwin":
|
|
return a.collectDarwinThermalState(ctx)
|
|
case "freebsd":
|
|
return a.collectFreeBSDTemperatures(ctx)
|
|
case "windows":
|
|
// Windows has no reliable built-in CPU or motherboard temperature API.
|
|
// Pulse accepts those readings only from a local driver-backed
|
|
// LibreHardwareMonitor instance. The native Storage module supplies
|
|
// physical-disk temperatures, and NVIDIA's driver supplies nvidia-smi.
|
|
return a.collectWindowsTemperatureSensors(ctx)
|
|
default:
|
|
return agentshost.Sensors{}
|
|
}
|
|
|
|
// Collect sensor JSON output
|
|
jsonOutput, err := a.collector.SensorsLocal(ctx)
|
|
if err != nil {
|
|
a.logger.Debug().Err(err).Msg("Failed to collect sensor data (lm-sensors may not be installed)")
|
|
return a.collectNVIDIATemperatureSensors(ctx)
|
|
}
|
|
|
|
// Parse the sensor output
|
|
tempData, err := a.collector.SensorsParse(jsonOutput)
|
|
if err != nil {
|
|
a.logger.Debug().Err(err).Msg("Failed to parse sensor data")
|
|
return a.collectNVIDIATemperatureSensors(ctx)
|
|
}
|
|
|
|
if !tempData.Available {
|
|
a.logger.Debug().Msg("No temperature sensors available on this system")
|
|
return a.collectNVIDIATemperatureSensors(ctx)
|
|
}
|
|
|
|
result := convertTemperatureDataToSensors(tempData)
|
|
a.mergeNVIDIATemperatures(ctx, &result)
|
|
|
|
// Collect power consumption data (Intel RAPL, etc.)
|
|
if powerData, err := a.collector.SensorsPower(ctx); err == nil && powerData != nil && powerData.Available {
|
|
result.PowerWatts = make(map[string]float64)
|
|
if powerData.PackageWatts > 0 {
|
|
result.PowerWatts["cpu_package"] = powerData.PackageWatts
|
|
}
|
|
if powerData.CoreWatts > 0 {
|
|
result.PowerWatts["cpu_core"] = powerData.CoreWatts
|
|
}
|
|
if powerData.DRAMWatts > 0 {
|
|
result.PowerWatts["dram"] = powerData.DRAMWatts
|
|
}
|
|
a.logger.Debug().
|
|
Float64("packageWatts", powerData.PackageWatts).
|
|
Str("source", powerData.Source).
|
|
Msg("Collected power data")
|
|
}
|
|
|
|
a.logger.Debug().
|
|
Int("temperatureCount", len(result.TemperatureCelsius)).
|
|
Int("fanCount", len(result.FanRPM)).
|
|
Int("powerCount", len(result.PowerWatts)).
|
|
Int("additionalCount", len(result.Additional)).
|
|
Msg("Collected sensor data")
|
|
|
|
return result
|
|
}
|
|
|
|
// convertTemperatureDataToSensors converts parsed sensor data into the agent report
|
|
// sensor format. This is shared between local collection and cluster peer collection.
|
|
func convertTemperatureDataToSensors(tempData *sensors.TemperatureData) agentshost.Sensors {
|
|
result := agentshost.Sensors{
|
|
TemperatureCelsius: make(map[string]float64),
|
|
}
|
|
|
|
// Add CPU package temperature
|
|
if tempData.CPUPackage > 0 {
|
|
result.TemperatureCelsius["cpu_package"] = tempData.CPUPackage
|
|
}
|
|
|
|
// Add individual core temperatures
|
|
for coreName, temp := range tempData.Cores {
|
|
normalizedName := strings.ToLower(strings.ReplaceAll(coreName, " ", "_"))
|
|
result.TemperatureCelsius["cpu_"+normalizedName] = temp
|
|
}
|
|
|
|
// Add NVMe temperatures
|
|
for nvmeName, temp := range tempData.NVMe {
|
|
result.TemperatureCelsius[nvmeName] = temp
|
|
}
|
|
|
|
// Add GPU temperatures
|
|
for gpuName, temp := range tempData.GPU {
|
|
result.TemperatureCelsius[gpuName] = temp
|
|
}
|
|
|
|
// Add fan speeds (RPM)
|
|
if len(tempData.Fans) > 0 {
|
|
result.FanRPM = make(map[string]float64)
|
|
for fanName, rpm := range tempData.Fans {
|
|
result.FanRPM[fanName] = rpm
|
|
}
|
|
}
|
|
|
|
// Add other temperatures (DDR5, motherboard, etc.)
|
|
if len(tempData.Other) > 0 {
|
|
result.Additional = make(map[string]float64)
|
|
for sensorName, temp := range tempData.Other {
|
|
result.Additional[sensorName] = temp
|
|
}
|
|
}
|
|
|
|
return result
|
|
}
|
|
|
|
// collectDarwinThermalState reads macOS thermal pressure signals. macOS does
|
|
// not expose Linux-style raw temperature sensors through a stable unprivileged
|
|
// interface, so this reports OS pressure state without inventing Celsius data.
|
|
func (a *Agent) collectDarwinThermalState(ctx context.Context) agentshost.Sensors {
|
|
cmdCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
|
|
defer cancel()
|
|
|
|
output, err := a.collector.CommandCombinedOutput(cmdCtx, "pmset", "-g", "therm")
|
|
if err != nil {
|
|
a.logger.Debug().Err(err).Msg("Failed to collect macOS thermal state via pmset")
|
|
return agentshost.Sensors{}
|
|
}
|
|
|
|
state := parseDarwinPMSetThermalState(output)
|
|
if state == nil {
|
|
a.logger.Debug().Msg("No macOS thermal state available from pmset")
|
|
return agentshost.Sensors{}
|
|
}
|
|
|
|
a.logger.Debug().
|
|
Str("pressure", state.Pressure).
|
|
Int("limitCount", len(state.LimitsPercent)).
|
|
Msg("Collected macOS thermal state")
|
|
|
|
return agentshost.Sensors{ThermalState: state}
|
|
}
|
|
|
|
func parseDarwinPMSetThermalState(output string) *agentshost.ThermalState {
|
|
output = strings.TrimSpace(output)
|
|
if output == "" {
|
|
return nil
|
|
}
|
|
|
|
state := &agentshost.ThermalState{
|
|
Source: "pmset",
|
|
Pressure: agentshost.ThermalPressureUnknown,
|
|
}
|
|
seen := false
|
|
constrained := false
|
|
limits := make(map[string]int)
|
|
|
|
for _, rawLine := range strings.Split(output, "\n") {
|
|
line := strings.TrimSpace(rawLine)
|
|
if line == "" {
|
|
continue
|
|
}
|
|
lower := strings.ToLower(line)
|
|
|
|
switch {
|
|
case strings.Contains(lower, "no thermal warning level"):
|
|
state.ThermalWarningLevel = intPtr(0)
|
|
seen = true
|
|
continue
|
|
case strings.Contains(lower, "no performance warning level"):
|
|
state.PerformanceWarningLevel = intPtr(0)
|
|
seen = true
|
|
continue
|
|
case strings.Contains(lower, "no cpu power status"):
|
|
state.CPUPowerStatus = intPtr(0)
|
|
seen = true
|
|
continue
|
|
}
|
|
|
|
if value, ok := parseIntegerAfterMarker(line, "thermal warning level"); ok {
|
|
state.ThermalWarningLevel = intPtr(value)
|
|
seen = true
|
|
if value > 0 {
|
|
constrained = true
|
|
}
|
|
continue
|
|
}
|
|
if value, ok := parseIntegerAfterMarker(line, "performance warning level"); ok {
|
|
state.PerformanceWarningLevel = intPtr(value)
|
|
seen = true
|
|
if value > 0 {
|
|
constrained = true
|
|
}
|
|
continue
|
|
}
|
|
if value, ok := parseIntegerAfterMarker(line, "cpu power status"); ok {
|
|
state.CPUPowerStatus = intPtr(value)
|
|
seen = true
|
|
if value > 0 {
|
|
constrained = true
|
|
}
|
|
continue
|
|
}
|
|
if key, value, ok := parsePMSetLimitPercent(line); ok {
|
|
limits[key] = value
|
|
seen = true
|
|
if value >= 0 && value < 100 {
|
|
constrained = true
|
|
}
|
|
}
|
|
}
|
|
|
|
if !seen {
|
|
return nil
|
|
}
|
|
if len(limits) > 0 {
|
|
state.LimitsPercent = limits
|
|
}
|
|
if constrained {
|
|
state.Pressure = agentshost.ThermalPressureConstrained
|
|
} else {
|
|
state.Pressure = agentshost.ThermalPressureNominal
|
|
}
|
|
return state
|
|
}
|
|
|
|
func parsePMSetLimitPercent(line string) (string, int, bool) {
|
|
parts := strings.SplitN(line, "=", 2)
|
|
if len(parts) != 2 {
|
|
return "", 0, false
|
|
}
|
|
key := strings.TrimSpace(parts[0])
|
|
if key == "" || !strings.Contains(strings.ToLower(key), "limit") {
|
|
return "", 0, false
|
|
}
|
|
value, ok := parseFirstInteger(parts[1])
|
|
if !ok {
|
|
return "", 0, false
|
|
}
|
|
return normalizePMSetLimitKey(key), value, true
|
|
}
|
|
|
|
func normalizePMSetLimitKey(key string) string {
|
|
key = strings.TrimSpace(strings.ToLower(key))
|
|
replacer := strings.NewReplacer(" ", "_", "-", "_")
|
|
key = replacer.Replace(key)
|
|
for strings.Contains(key, "__") {
|
|
key = strings.ReplaceAll(key, "__", "_")
|
|
}
|
|
return strings.Trim(key, "_")
|
|
}
|
|
|
|
func parseIntegerAfterMarker(line, marker string) (int, bool) {
|
|
lower := strings.ToLower(line)
|
|
idx := strings.Index(lower, strings.ToLower(marker))
|
|
if idx < 0 {
|
|
return 0, false
|
|
}
|
|
return parseFirstInteger(line[idx+len(marker):])
|
|
}
|
|
|
|
func parseFirstInteger(input string) (int, bool) {
|
|
for i := 0; i < len(input); i++ {
|
|
ch := input[i]
|
|
if (ch < '0' || ch > '9') && ch != '-' {
|
|
continue
|
|
}
|
|
j := i + 1
|
|
for j < len(input) && input[j] >= '0' && input[j] <= '9' {
|
|
j++
|
|
}
|
|
value, err := strconv.Atoi(input[i:j])
|
|
if err != nil {
|
|
continue
|
|
}
|
|
return value, true
|
|
}
|
|
return 0, false
|
|
}
|
|
|
|
func intPtr(value int) *int {
|
|
return &value
|
|
}
|
|
|
|
// collectFreeBSDTemperatures reads CPU and ACPI thermal zone temperatures
|
|
// from sysctl on FreeBSD. Returns an empty Sensors struct on any error.
|
|
func (a *Agent) collectFreeBSDTemperatures(ctx context.Context) agentshost.Sensors {
|
|
cmdCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
|
|
defer cancel()
|
|
|
|
cmd := exec.CommandContext(cmdCtx, "sysctl", "-a")
|
|
output, err := cmd.Output()
|
|
if err != nil {
|
|
a.logger.Debug().Err(err).Msg("Failed to collect FreeBSD temperature data via sysctl")
|
|
return agentshost.Sensors{}
|
|
}
|
|
|
|
result := parseFreeBSDSysctlTemperatures(string(output))
|
|
|
|
if len(result.TemperatureCelsius) > 0 {
|
|
a.logger.Debug().
|
|
Int("temperatureCount", len(result.TemperatureCelsius)).
|
|
Msg("Collected FreeBSD temperature data via sysctl")
|
|
}
|
|
|
|
return result
|
|
}
|
|
|
|
// parseFreeBSDSysctlTemperatures parses sysctl output for temperature readings.
|
|
// FreeBSD exposes CPU temps via dev.cpu.N.temperature and ACPI thermal zones via
|
|
// hw.acpi.thermal.tzN.temperature. Values are formatted as "45.0C".
|
|
func parseFreeBSDSysctlTemperatures(sysctlOutput string) agentshost.Sensors {
|
|
result := agentshost.Sensors{
|
|
TemperatureCelsius: make(map[string]float64),
|
|
}
|
|
|
|
for _, line := range strings.Split(sysctlOutput, "\n") {
|
|
if !strings.Contains(line, ".temperature:") {
|
|
continue
|
|
}
|
|
parts := strings.SplitN(line, ":", 2)
|
|
if len(parts) != 2 {
|
|
continue
|
|
}
|
|
key := strings.TrimSpace(parts[0])
|
|
valStr := strings.TrimSpace(parts[1])
|
|
valStr = strings.TrimSuffix(valStr, "C")
|
|
|
|
temp, err := strconv.ParseFloat(valStr, 64)
|
|
if err != nil || temp <= 0 {
|
|
continue
|
|
}
|
|
|
|
keyParts := strings.Split(key, ".")
|
|
switch {
|
|
case strings.HasPrefix(key, "dev.cpu.") && len(keyParts) >= 3:
|
|
result.TemperatureCelsius["cpu_core_"+keyParts[2]] = temp
|
|
case strings.HasPrefix(key, "hw.acpi.thermal.") && len(keyParts) >= 4:
|
|
result.TemperatureCelsius["acpi_"+keyParts[3]] = temp
|
|
default:
|
|
normalized := strings.ReplaceAll(key, ".", "_")
|
|
normalized = strings.TrimSuffix(normalized, "_temperature")
|
|
result.TemperatureCelsius[normalized] = temp
|
|
}
|
|
}
|
|
|
|
return result
|
|
}
|
|
|
|
// collectRAIDArrays attempts to collect mdadm RAID array information.
|
|
// Returns an empty slice if collection fails (best-effort).
|
|
func (a *Agent) collectRAIDArrays(ctx context.Context) []agentshost.RAIDArray {
|
|
// Only collect on Linux (mdadm is Linux-specific)
|
|
if a.collector.GOOS() != "linux" {
|
|
return nil
|
|
}
|
|
|
|
arrays, err := a.collector.RAIDArrays(ctx)
|
|
if err != nil {
|
|
a.logger.Debug().Err(err).Msg("Failed to collect RAID array data (mdadm may not be installed)")
|
|
return nil
|
|
}
|
|
|
|
if len(arrays) > 0 {
|
|
a.logger.Debug().
|
|
Int("arrayCount", len(arrays)).
|
|
Msg("Collected RAID array data")
|
|
}
|
|
|
|
return arrays
|
|
}
|
|
|
|
// collectUnraidStorage attempts to collect Unraid array topology.
|
|
// Returns nil when not running on Unraid or if collection fails.
|
|
func (a *Agent) collectUnraidStorage(ctx context.Context) *agentshost.UnraidStorage {
|
|
if a.collector.GOOS() != "linux" {
|
|
return nil
|
|
}
|
|
|
|
storage, err := a.collector.UnraidStorage(ctx)
|
|
if err != nil {
|
|
a.logger.Debug().Err(err).Msg("Failed to collect Unraid storage data")
|
|
return nil
|
|
}
|
|
if storage == nil {
|
|
return nil
|
|
}
|
|
|
|
a.logger.Debug().
|
|
Bool("arrayStarted", storage.ArrayStarted).
|
|
Int("diskCount", len(storage.Disks)).
|
|
Msg("Collected Unraid storage data")
|
|
|
|
return storage
|
|
}
|
|
|
|
// collectCephStatus attempts to collect Ceph cluster status.
|
|
// Returns nil if Ceph is not available or not configured on this host.
|
|
func (a *Agent) collectCephStatus(ctx context.Context, disableCeph bool) *agentshost.CephCluster {
|
|
if disableCeph {
|
|
return nil
|
|
}
|
|
// Only collect on Linux
|
|
if a.collector.GOOS() != "linux" {
|
|
return nil
|
|
}
|
|
|
|
status, err := a.collector.CephStatus(ctx)
|
|
if err != nil {
|
|
a.logger.Debug().Err(err).Msg("Failed to collect Ceph status")
|
|
return nil
|
|
}
|
|
if status == nil {
|
|
return nil
|
|
}
|
|
|
|
// Convert internal ceph types to agent report types
|
|
result := &agentshost.CephCluster{
|
|
FSID: status.FSID,
|
|
Health: agentshost.CephHealth{
|
|
Status: string(status.Health.Status),
|
|
Checks: make(map[string]agentshost.CephCheck),
|
|
},
|
|
MonMap: agentshost.CephMonitorMap{
|
|
Epoch: status.MonMap.Epoch,
|
|
NumMons: status.MonMap.NumMons,
|
|
},
|
|
MgrMap: agentshost.CephManagerMap{
|
|
Available: status.MgrMap.Available,
|
|
NumMgrs: status.MgrMap.NumMgrs,
|
|
ActiveMgr: status.MgrMap.ActiveMgr,
|
|
Standbys: status.MgrMap.Standbys,
|
|
},
|
|
OSDMap: agentshost.CephOSDMap{
|
|
Epoch: status.OSDMap.Epoch,
|
|
NumOSDs: status.OSDMap.NumOSDs,
|
|
NumUp: status.OSDMap.NumUp,
|
|
NumIn: status.OSDMap.NumIn,
|
|
NumDown: status.OSDMap.NumDown,
|
|
NumOut: status.OSDMap.NumOut,
|
|
},
|
|
PGMap: agentshost.CephPGMap{
|
|
NumPGs: status.PGMap.NumPGs,
|
|
BytesTotal: status.PGMap.BytesTotal,
|
|
BytesUsed: status.PGMap.BytesUsed,
|
|
BytesAvailable: status.PGMap.BytesAvailable,
|
|
DataBytes: status.PGMap.DataBytes,
|
|
UsagePercent: status.PGMap.UsagePercent,
|
|
DegradedRatio: status.PGMap.DegradedRatio,
|
|
MisplacedRatio: status.PGMap.MisplacedRatio,
|
|
ReadBytesPerSec: status.PGMap.ReadBytesPerSec,
|
|
WriteBytesPerSec: status.PGMap.WriteBytesPerSec,
|
|
ReadOpsPerSec: status.PGMap.ReadOpsPerSec,
|
|
WriteOpsPerSec: status.PGMap.WriteOpsPerSec,
|
|
},
|
|
CollectedAt: status.CollectedAt.Format(time.RFC3339),
|
|
}
|
|
|
|
// Convert monitors
|
|
for _, mon := range status.MonMap.Monitors {
|
|
result.MonMap.Monitors = append(result.MonMap.Monitors, agentshost.CephMonitor{
|
|
Name: mon.Name,
|
|
Rank: mon.Rank,
|
|
Addr: mon.Addr,
|
|
Status: mon.Status,
|
|
})
|
|
}
|
|
|
|
// Convert health checks
|
|
for name, check := range status.Health.Checks {
|
|
result.Health.Checks[name] = agentshost.CephCheck{
|
|
Severity: string(check.Severity),
|
|
Message: check.Message,
|
|
Detail: check.Detail,
|
|
}
|
|
}
|
|
|
|
// Convert health summary
|
|
for _, s := range status.Health.Summary {
|
|
result.Health.Summary = append(result.Health.Summary, agentshost.CephHealthSummary{
|
|
Severity: string(s.Severity),
|
|
Message: s.Message,
|
|
})
|
|
}
|
|
|
|
// Convert pools
|
|
for _, pool := range status.Pools {
|
|
result.Pools = append(result.Pools, agentshost.CephPool{
|
|
ID: pool.ID,
|
|
Name: pool.Name,
|
|
BytesUsed: pool.BytesUsed,
|
|
BytesAvailable: pool.BytesAvailable,
|
|
Objects: pool.Objects,
|
|
PercentUsed: pool.PercentUsed,
|
|
})
|
|
}
|
|
|
|
// Convert services
|
|
for _, svc := range status.Services {
|
|
result.Services = append(result.Services, agentshost.CephService{
|
|
Type: string(svc.Type),
|
|
Running: svc.Running,
|
|
Total: svc.Total,
|
|
Daemons: svc.Daemons,
|
|
})
|
|
}
|
|
|
|
a.logger.Debug().
|
|
Str("fsid", result.FSID).
|
|
Str("health", result.Health.Status).
|
|
Int("osds", result.OSDMap.NumOSDs).
|
|
Int("pools", len(result.Pools)).
|
|
Msg("Collected Ceph cluster status")
|
|
|
|
return result
|
|
}
|
|
|
|
// collectSMARTData collects S.M.A.R.T. data from local disks.
|
|
// Returns nil if smartctl is not available or no disks are found.
|
|
func (a *Agent) collectSMARTData(ctx context.Context, diskExclude []string, unraid *agentshost.UnraidStorage) []agentshost.DiskSMART {
|
|
goos := a.collector.GOOS()
|
|
if goos != "linux" && goos != "freebsd" {
|
|
return nil
|
|
}
|
|
|
|
var smartData []DiskSMART
|
|
var err error
|
|
if a.privilegedTelemetry != nil {
|
|
smartData, err = a.privilegedTelemetry.SMARTSnapshot(ctx)
|
|
a.recordPrivilegeHelperOperation(privilegeHelperOperationSMART, err)
|
|
if err == nil && len(diskExclude) > 0 {
|
|
filtered := smartData[:0]
|
|
for _, disk := range smartData {
|
|
deviceName := filepath.Base(disk.Device)
|
|
if matchesDeviceExclude(deviceName, disk.Device, diskExclude) {
|
|
continue
|
|
}
|
|
filtered = append(filtered, disk)
|
|
}
|
|
smartData = filtered
|
|
}
|
|
} else {
|
|
smartData, err = a.collector.SMARTLocal(ctx, diskExclude, unraid)
|
|
}
|
|
if err != nil {
|
|
a.logger.Debug().Err(err).Msg("Failed to collect S.M.A.R.T. data (smartctl may not be installed)")
|
|
return nil
|
|
}
|
|
|
|
if len(smartData) == 0 {
|
|
return nil
|
|
}
|
|
|
|
// Convert internal smartctl types to agent report types
|
|
result := make([]agentshost.DiskSMART, 0, len(smartData))
|
|
for _, disk := range smartData {
|
|
entry := agentshost.DiskSMART{
|
|
Device: disk.Device,
|
|
Model: disk.Model,
|
|
Serial: disk.Serial,
|
|
WWN: disk.WWN,
|
|
Type: disk.Type,
|
|
Controller: disk.Controller,
|
|
Target: disk.Target,
|
|
SizeBytes: disk.SizeBytes,
|
|
Temperature: disk.Temperature,
|
|
Health: disk.Health,
|
|
Standby: disk.Standby,
|
|
Collection: diskinventory.CloneStatus(disk.Collection),
|
|
}
|
|
if disk.Attributes != nil {
|
|
entry.Attributes = &agentshost.SMARTAttributes{
|
|
PowerOnHours: disk.Attributes.PowerOnHours,
|
|
PowerCycles: disk.Attributes.PowerCycles,
|
|
ReallocatedSectors: disk.Attributes.ReallocatedSectors,
|
|
PendingSectors: disk.Attributes.PendingSectors,
|
|
OfflineUncorrectable: disk.Attributes.OfflineUncorrectable,
|
|
UDMACRCErrors: disk.Attributes.UDMACRCErrors,
|
|
PercentageUsed: disk.Attributes.PercentageUsed,
|
|
AvailableSpare: disk.Attributes.AvailableSpare,
|
|
MediaErrors: disk.Attributes.MediaErrors,
|
|
UnsafeShutdowns: disk.Attributes.UnsafeShutdowns,
|
|
}
|
|
}
|
|
result = append(result, entry)
|
|
}
|
|
|
|
if pools, err := ZFSDiskPoolMap(ctx); err != nil {
|
|
for index := range result {
|
|
ensureAgentDiskCollection(&result[index]).Pool = diskinventory.Unavailable(
|
|
"zpool",
|
|
"ZFS pool membership could not be collected",
|
|
)
|
|
}
|
|
a.logger.Debug().Err(err).Msg("Failed to collect ZFS pool membership for SMART annotation")
|
|
} else {
|
|
if len(pools) > 0 {
|
|
annotateSMARTWithZFSPools(result, pools)
|
|
}
|
|
for index := range result {
|
|
ensureAgentDiskCollection(&result[index]).Pool = diskinventory.Available("zpool")
|
|
}
|
|
}
|
|
|
|
a.logger.Debug().
|
|
Int("diskCount", len(result)).
|
|
Msg("Collected S.M.A.R.T. disk data")
|
|
|
|
return result
|
|
}
|
|
|
|
func ensureAgentDiskCollection(disk *agentshost.DiskSMART) *diskinventory.CollectionStatus {
|
|
if disk.Collection == nil {
|
|
disk.Collection = &diskinventory.CollectionStatus{}
|
|
}
|
|
return disk.Collection
|
|
}
|
|
|
|
// annotateSMARTWithDiskIO binds cumulative kernel counters to a SMART
|
|
// inventory entry only when the block-device association is unambiguous.
|
|
// Controller-member targets share one logical kernel device, so assigning that
|
|
// aggregate counter to any member would fabricate per-disk I/O.
|
|
func annotateSMARTWithDiskIO(smart []agentshost.DiskSMART, diskIO []agentshost.DiskIO) {
|
|
if len(smart) == 0 {
|
|
return
|
|
}
|
|
|
|
ioByDevice := make(map[string]agentshost.DiskIO, len(diskIO))
|
|
for _, entry := range diskIO {
|
|
device := diskinventory.DeviceToken(entry.Device)
|
|
if device != "" {
|
|
ioByDevice[strings.ToLower(device)] = entry
|
|
}
|
|
}
|
|
smartCountByDevice := make(map[string]int, len(smart))
|
|
for _, disk := range smart {
|
|
device := strings.ToLower(diskinventory.DeviceToken(disk.Device))
|
|
if device != "" {
|
|
smartCountByDevice[device]++
|
|
}
|
|
}
|
|
|
|
for index := range smart {
|
|
disk := &smart[index]
|
|
collection := ensureAgentDiskCollection(disk)
|
|
device := strings.ToLower(diskinventory.DeviceToken(disk.Device))
|
|
if isMultiplexedDeviceType(disk.Target) || (device != "" && smartCountByDevice[device] > 1) {
|
|
collection.IO = diskinventory.Unsupported(
|
|
"kernel_diskstats",
|
|
"controller target does not expose member-level I/O counters",
|
|
)
|
|
continue
|
|
}
|
|
ioEntry, ok := ioByDevice[device]
|
|
if !ok {
|
|
if len(diskIO) == 0 {
|
|
collection.IO = diskinventory.Unavailable(
|
|
"kernel_diskstats",
|
|
"disk I/O counters could not be collected",
|
|
)
|
|
} else {
|
|
collection.IO = diskinventory.Missing(
|
|
"kernel_diskstats",
|
|
"disk was present but its I/O counters were not reported",
|
|
)
|
|
}
|
|
continue
|
|
}
|
|
ioCopy := ioEntry
|
|
disk.IO = &ioCopy
|
|
collection.IO = diskinventory.Available("kernel_diskstats")
|
|
}
|
|
}
|
|
|
|
// runProxmoxSetup performs one-time Proxmox API token setup and node registration.
|
|
// Supports hosts with multiple Proxmox products (e.g., PVE + PBS on same host).
|
|
func (a *Agent) runProxmoxSetup(ctx context.Context) {
|
|
a.logger.Info().Msg("Proxmox mode enabled, checking setup...")
|
|
for _, destination := range a.proxmoxDestinations() {
|
|
results, err := destination.setup.RunAll(ctx)
|
|
if err != nil {
|
|
a.logger.Error().Err(err).Str("destination", destination.name).Msg("Proxmox setup failed")
|
|
continue
|
|
}
|
|
for _, result := range results {
|
|
if result.Registered {
|
|
a.logger.Info().Str("destination", destination.name).
|
|
Str("type", result.ProxmoxType).
|
|
Str("host", result.NodeHost).
|
|
Str("token_id", result.TokenID).
|
|
Msg("Proxmox node registered successfully")
|
|
} else {
|
|
a.logger.Warn().Str("destination", destination.name).
|
|
Str("type", result.ProxmoxType).
|
|
Str("host", result.NodeHost).
|
|
Msg("Proxmox token created but registration failed")
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
type proxmoxDestination struct {
|
|
name string
|
|
setup *ProxmoxSetup
|
|
}
|
|
|
|
func (a *Agent) proxmoxDestinations() []proxmoxDestination {
|
|
destinations := []proxmoxDestination{{
|
|
name: "primary",
|
|
setup: NewProxmoxSetup(a.logger, a.httpClient, a.collector, a.trimmedPulseURL, a.cfg.APIToken,
|
|
a.cfg.ProxmoxType, a.hostname, a.currentReportIP(), a.stateDir, a.cfg.InsecureSkipVerify),
|
|
}}
|
|
for _, observer := range a.observerReporters {
|
|
if !observer.target.ProvisionProxmox {
|
|
continue
|
|
}
|
|
stateDir := filepath.Join(a.stateDir, "observers", observer.target.ID)
|
|
tokenName := "pulse-" + proxmoxTokenScope(a.hostname, a.trimmedPulseURL) + "-" + observer.target.ID
|
|
setup := NewProxmoxSetup(a.logger, observer.httpClient, a.collector, observer.target.PulseURL,
|
|
observer.target.APIToken, a.cfg.ProxmoxType, a.hostname, a.currentReportIP(), stateDir,
|
|
observer.target.InsecureSkipVerify).
|
|
WithObserverPlaintextPolicy(observer.target.AllowPlaintextHTTP).
|
|
WithMonitorTokenName(tokenName)
|
|
destinations = append(destinations, proxmoxDestination{name: observer.target.Name, setup: setup})
|
|
}
|
|
return destinations
|
|
}
|
|
|
|
const (
|
|
proxmoxHealthCheckInitialDelay = 2 * time.Minute
|
|
proxmoxHealthCheckInterval = 5 * time.Minute
|
|
)
|
|
|
|
// runProxmoxHealthCheckLoop periodically verifies that Proxmox nodes registered
|
|
// at startup are still connected to Pulse. This closes the cold-startup race
|
|
// where the monitor may not have connection data at the time of the initial
|
|
// registration check, causing a stale-token node to be silently skipped.
|
|
func (a *Agent) runProxmoxHealthCheckLoop(ctx context.Context) {
|
|
// Wait for the monitor to build initial connection-health state before
|
|
// the first check so that a stale token is detectable.
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-time.After(proxmoxHealthCheckInitialDelay):
|
|
}
|
|
|
|
ticker := time.NewTicker(proxmoxHealthCheckInterval)
|
|
defer ticker.Stop()
|
|
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-ticker.C:
|
|
for _, destination := range a.proxmoxDestinations() {
|
|
results, err := destination.setup.RunHealthCheck(ctx)
|
|
if err != nil {
|
|
a.logger.Warn().Err(err).Str("destination", destination.name).Msg("Proxmox health check failed")
|
|
continue
|
|
}
|
|
for _, result := range results {
|
|
event := a.logger.Info()
|
|
if !result.Registered {
|
|
event = a.logger.Warn()
|
|
}
|
|
event.Str("destination", destination.name).
|
|
Str("type", result.ProxmoxType).
|
|
Str("host", result.NodeHost).
|
|
Bool("registered", result.Registered).
|
|
Msg("Proxmox destination health repair completed")
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// isLXCContainer detects if we're running inside an LXC container.
|
|
// LXC containers share the host's /sys/class/dmi/id/product_uuid, which causes
|
|
// gopsutil to return identical HostIDs for all LXC containers on the same host.
|
|
func isLXCContainer(c SystemCollector) bool {
|
|
// Check systemd-detect-virt if available
|
|
if data, err := c.ReadFile("/run/systemd/container"); err == nil {
|
|
container := strings.TrimSpace(string(data))
|
|
if strings.Contains(container, "lxc") {
|
|
return true
|
|
}
|
|
}
|
|
|
|
// Check /proc/1/environ for container=lxc
|
|
if data, err := c.ReadFile("/proc/1/environ"); err == nil {
|
|
if strings.Contains(string(data), "container=lxc") {
|
|
return true
|
|
}
|
|
}
|
|
|
|
// Check /proc/1/cgroup for lxc markers
|
|
if data, err := c.ReadFile("/proc/1/cgroup"); err == nil {
|
|
text := string(data)
|
|
if strings.Contains(text, "/lxc/") || strings.Contains(text, "lxc.payload") {
|
|
return true
|
|
}
|
|
}
|
|
|
|
return false
|
|
}
|
|
|
|
// getReliableMachineID returns a machine ID that's unique per container/host.
|
|
// On Linux, /etc/machine-id is always preferred over gopsutil's HostID because:
|
|
// - LXC containers share the host's /sys/class/dmi/id/product_uuid
|
|
// - Cloned VMs/hosts may share the same DMI product UUID
|
|
// - Proxmox cluster nodes with identical hardware may have the same UUID
|
|
// The /etc/machine-id file is guaranteed unique per installation.
|
|
// GetReliableMachineID attempts to find a stable machine ID.
|
|
func GetReliableMachineID(c SystemCollector, gopsutilHostID string, logger zerolog.Logger) string {
|
|
gopsutilID := strings.TrimSpace(gopsutilHostID)
|
|
|
|
// On Linux, prefer /etc/machine-id when available.
|
|
// This avoids ID collisions from:
|
|
// - LXC containers sharing host's DMI product UUID
|
|
// - Cloned VMs with identical hardware UUIDs
|
|
// - Proxmox cluster nodes with same hardware configuration
|
|
if c.GOOS() == "linux" {
|
|
if data, err := c.ReadFile("/etc/machine-id"); err == nil {
|
|
machineID := strings.TrimSpace(string(data))
|
|
if machineID != "" && len(machineID) >= 8 {
|
|
// Format as UUID if it's a 32-char hex string (like machine-id typically is).
|
|
if len(machineID) == 32 && utils.IsHexString(machineID) {
|
|
machineID = fmt.Sprintf("%s-%s-%s-%s-%s",
|
|
machineID[0:8], machineID[8:12], machineID[12:16],
|
|
machineID[16:20], machineID[20:32])
|
|
}
|
|
if isLXCContainer(c) {
|
|
logger.Debug().
|
|
Str("machineID", machineID).
|
|
Msg("LXC container detected, using /etc/machine-id for unique identification")
|
|
} else {
|
|
logger.Debug().
|
|
Str("machineID", machineID).
|
|
Msg("Linux host detected, using /etc/machine-id for unique identification")
|
|
}
|
|
return machineID
|
|
}
|
|
}
|
|
|
|
if macID := getPrimaryMACIdentifier(c); macID != "" {
|
|
logger.Debug().
|
|
Str("machineID", macID).
|
|
Msg("Linux host missing usable /etc/machine-id, using MAC address for unique identification")
|
|
return macID
|
|
}
|
|
}
|
|
|
|
return gopsutilID
|
|
}
|
|
|
|
func getPrimaryMACIdentifier(c SystemCollector) string {
|
|
interfaces, err := c.NetInterfaces()
|
|
if err != nil {
|
|
return ""
|
|
}
|
|
|
|
sort.Slice(interfaces, func(i, j int) bool {
|
|
return interfaces[i].Name < interfaces[j].Name
|
|
})
|
|
|
|
// Prefer a stable-looking interface name first to avoid selecting docker bridges
|
|
// or other virtual interfaces when physical interfaces are present.
|
|
for pass := 0; pass < 2; pass++ {
|
|
for _, iface := range interfaces {
|
|
if len(iface.HardwareAddr) == 0 {
|
|
continue
|
|
}
|
|
if iface.Flags&net.FlagLoopback != 0 {
|
|
continue
|
|
}
|
|
if pass == 0 && isLikelyVirtualInterfaceName(iface.Name) {
|
|
continue
|
|
}
|
|
|
|
mac := strings.ToLower(iface.HardwareAddr.String())
|
|
normalized := strings.NewReplacer(":", "", "-", "", ".", "").Replace(mac)
|
|
if normalized == "" {
|
|
continue
|
|
}
|
|
return "mac-" + normalized
|
|
}
|
|
}
|
|
|
|
return ""
|
|
}
|
|
|
|
func isLikelyVirtualInterfaceName(name string) bool {
|
|
name = strings.ToLower(strings.TrimSpace(name))
|
|
switch {
|
|
case name == "":
|
|
return true
|
|
case name == "lo":
|
|
return true
|
|
case strings.HasPrefix(name, "docker"):
|
|
return true
|
|
case strings.HasPrefix(name, "veth"):
|
|
return true
|
|
case strings.HasPrefix(name, "br-"):
|
|
return true
|
|
case strings.HasPrefix(name, "cni"):
|
|
return true
|
|
case strings.HasPrefix(name, "flannel"):
|
|
return true
|
|
case strings.HasPrefix(name, "virbr"):
|
|
return true
|
|
case strings.HasPrefix(name, "zt"):
|
|
return true
|
|
default:
|
|
return false
|
|
}
|
|
}
|