diff --git a/cmd/pulse-agent/main.go b/cmd/pulse-agent/main.go index 5abde3d2a..7376d2224 100644 --- a/cmd/pulse-agent/main.go +++ b/cmd/pulse-agent/main.go @@ -28,6 +28,7 @@ import ( "github.com/rcourtman/pulse-go-rewrite/internal/agentexec" "github.com/rcourtman/pulse-go-rewrite/internal/agenttarget" "github.com/rcourtman/pulse-go-rewrite/internal/agentupdate" + pulseconfig "github.com/rcourtman/pulse-go-rewrite/internal/config" "github.com/rcourtman/pulse-go-rewrite/internal/dockeragent" "github.com/rcourtman/pulse-go-rewrite/internal/hostagent" "github.com/rcourtman/pulse-go-rewrite/internal/kubernetesagent" @@ -387,32 +388,33 @@ func run(ctx context.Context, args []string, getenv func(string) string) error { // 8. Start Host Agent (if enabled) if cfg.EnableHost { hostCfg := hostagent.Config{ - PulseURL: cfg.PulseURL, - APIToken: cfg.APIToken, - Interval: cfg.Interval, - HostnameOverride: cfg.HostnameOverride, - AgentID: cfg.AgentID, - AgentType: "unified", - AgentVersion: Version, - Tags: cfg.Tags, - InsecureSkipVerify: cfg.InsecureSkipVerify, - CACertPath: cfg.CACertPath, - ServerFingerprint: cfg.ServerFingerprint, - DeploySSHUser: cfg.DeploySSHUser, - LogLevel: cfg.LogLevel, - Logger: &logger, - EnableProxmox: cfg.EnableProxmox, - ProxmoxType: cfg.ProxmoxType, - EnableCommands: cfg.EnableCommands, - Enroll: cfg.Enroll, - DiskExclude: cfg.DiskExclude, - StateDir: cfg.StateDir, - ReportIP: cfg.ReportIP, - DisableCeph: cfg.DisableCeph, - AppliedConfig: cfg.AppliedConfig, - UpdateStatus: updater.Snapshot, - ModuleStatus: runtimeStatus.moduleStatuses, - Observers: hostObserverTargets(cfg.Observers), + PulseURL: cfg.PulseURL, + APIToken: cfg.APIToken, + Interval: cfg.Interval, + HostnameOverride: cfg.HostnameOverride, + AgentID: cfg.AgentID, + AgentType: "unified", + AgentVersion: Version, + Tags: cfg.Tags, + InsecureSkipVerify: cfg.InsecureSkipVerify, + CACertPath: cfg.CACertPath, + ServerFingerprint: cfg.ServerFingerprint, + DeploySSHUser: cfg.DeploySSHUser, + LogLevel: cfg.LogLevel, + Logger: &logger, + EnableProxmox: cfg.EnableProxmox, + ProxmoxType: cfg.ProxmoxType, + EnableCommands: cfg.EnableCommands, + Enroll: cfg.Enroll, + DiskExclude: cfg.DiskExclude, + StateDir: cfg.StateDir, + ReportIP: cfg.ReportIP, + DisableCeph: cfg.DisableCeph, + AvailabilityTargets: cfg.AvailabilityTargets, + AppliedConfig: cfg.AppliedConfig, + UpdateStatus: updater.Snapshot, + ModuleStatus: runtimeStatus.moduleStatuses, + Observers: hostObserverTargets(cfg.Observers), DockerContainerUpdater: dockerUpdaterBridge, } @@ -848,6 +850,10 @@ type Config struct { DisableCeph bool // Disable local Ceph status polling SelfTest bool // Perform self-test and exit + // AvailabilityTargets are externally probed availability checks assigned to + // this agent by the server. Remote config is the only source. + AvailabilityTargets []pulseconfig.AvailabilityTarget + // Health/metrics server HealthAddr string AppliedConfig *agentshost.ConfigFingerprint @@ -1530,6 +1536,7 @@ func retryLogEvent(logger *zerolog.Logger, attempt int) *zerolog.Event { // - interval (string/duration) // - report_ip (string) // - disable_ceph (bool) +// - availabilityTargets (array of assigned availability checks) func applyRemoteSettings(cfg *Config, settings map[string]interface{}, logger *zerolog.Logger) { for k, v := range settings { switch k { @@ -1633,6 +1640,14 @@ func applyRemoteSettings(cfg *Config, settings map[string]interface{}, logger *z cfg.DisableCeph = b logger.Info().Bool("val", b).Msg("Remote config: disable_ceph") } + case "availabilityTargets": + targets, err := hostagent.AvailabilityTargetsFromSetting(v) + if err != nil { + logger.Warn().Err(err).Msg("Remote config: ignoring unreadable availabilityTargets value") + continue + } + cfg.AvailabilityTargets = targets + logger.Info().Int("count", len(targets)).Msg("Remote config: availabilityTargets") } } } diff --git a/cmd/pulse-agent/main_test.go b/cmd/pulse-agent/main_test.go index c1204c431..e8a575c89 100644 --- a/cmd/pulse-agent/main_test.go +++ b/cmd/pulse-agent/main_test.go @@ -2354,3 +2354,35 @@ func TestAllowPlaintextHTTPFlagParsesAndDefaultsClosed(t *testing.T) { t.Fatal("PULSE_AGENT_ALLOW_PLAINTEXT_HTTP env was not applied") } } + +func TestApplyRemoteSettingsCarriesAvailabilityAssignmentsToStartup(t *testing.T) { + logger := zerolog.New(io.Discard) + cfg := &Config{} + + applyRemoteSettings(cfg, map[string]interface{}{ + "availabilityTargets": []interface{}{ + map[string]interface{}{ + "id": "remote-a", + "address": "a.local", + "protocol": "icmp", + "enabled": true, + "pollIntervalSeconds": float64(30), + }, + }, + }, &logger) + + if len(cfg.AvailabilityTargets) != 1 { + t.Fatalf("availability targets = %+v, want the assignment applied at boot", cfg.AvailabilityTargets) + } + if cfg.AvailabilityTargets[0].ID != "remote-a" { + t.Fatalf("availability target = %+v", cfg.AvailabilityTargets[0]) + } + + applyRemoteSettings(cfg, map[string]interface{}{ + "availabilityTargets": "not-a-list", + }, &logger) + + if len(cfg.AvailabilityTargets) != 1 { + t.Fatalf("availability targets = %+v, want an unreadable payload ignored", cfg.AvailabilityTargets) + } +} diff --git a/cmd/pulse-agent/service_windows.go b/cmd/pulse-agent/service_windows.go index 8a6f89024..d4e2499e1 100644 --- a/cmd/pulse-agent/service_windows.go +++ b/cmd/pulse-agent/service_windows.go @@ -85,24 +85,25 @@ func (ws *windowsService) Execute(args []string, r <-chan svc.ChangeRequest, cha // Start Host Agent (if enabled) if ws.cfg.EnableHost { hostCfg := hostagent.Config{ - PulseURL: ws.cfg.PulseURL, - APIToken: ws.cfg.APIToken, - Interval: ws.cfg.Interval, - HostnameOverride: ws.cfg.HostnameOverride, - AgentID: ws.cfg.AgentID, - AgentType: "unified", - AgentVersion: Version, - Tags: ws.cfg.Tags, - InsecureSkipVerify: ws.cfg.InsecureSkipVerify, - CACertPath: ws.cfg.CACertPath, - ServerFingerprint: ws.cfg.ServerFingerprint, - DeploySSHUser: ws.cfg.DeploySSHUser, - LogLevel: ws.cfg.LogLevel, - Logger: &ws.logger, - AppliedConfig: ws.cfg.AppliedConfig, - UpdateStatus: updater.Snapshot, - ModuleStatus: runtimeStatus.moduleStatuses, - Observers: hostObserverTargets(ws.cfg.Observers), + PulseURL: ws.cfg.PulseURL, + APIToken: ws.cfg.APIToken, + Interval: ws.cfg.Interval, + HostnameOverride: ws.cfg.HostnameOverride, + AgentID: ws.cfg.AgentID, + AgentType: "unified", + AgentVersion: Version, + Tags: ws.cfg.Tags, + InsecureSkipVerify: ws.cfg.InsecureSkipVerify, + CACertPath: ws.cfg.CACertPath, + ServerFingerprint: ws.cfg.ServerFingerprint, + DeploySSHUser: ws.cfg.DeploySSHUser, + LogLevel: ws.cfg.LogLevel, + Logger: &ws.logger, + AppliedConfig: ws.cfg.AppliedConfig, + AvailabilityTargets: ws.cfg.AvailabilityTargets, + UpdateStatus: updater.Snapshot, + ModuleStatus: runtimeStatus.moduleStatuses, + Observers: hostObserverTargets(ws.cfg.Observers), } agent, err := hostagent.New(hostCfg) if err != nil { diff --git a/internal/hostagent/agent.go b/internal/hostagent/agent.go index f011da9cf..47f79537b 100644 --- a/internal/hostagent/agent.go +++ b/internal/hostagent/agent.go @@ -26,6 +26,7 @@ import ( "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" @@ -78,6 +79,11 @@ type Config struct { 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. @@ -156,6 +162,7 @@ type Agent struct { runCommandClient func(*CommandClient, context.Context) error packageUpdates *packageUpdateManager storageCleanup *storageCleanupManager + availability *availabilityProbeModule reportStreamID string reportSequence atomic.Uint64 @@ -387,6 +394,7 @@ func New(cfg Config) (*Agent, error) { runCommandClient: runCommandClientFn, packageUpdates: packageUpdates, storageCleanup: storageCleanup, + availability: newAvailabilityProbeModule(logger, cfg.AvailabilityTargets), reportStreamID: reportStreamID, } @@ -541,6 +549,10 @@ func (a *Agent) Run(ctx context.Context) error { 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() @@ -643,10 +655,16 @@ func (a *Agent) currentUpdateStatus() *agentshost.UpdateStatus { } func (a *Agent) currentModuleStatus() []agentshost.ModuleStatus { - if a.cfg.ModuleStatus == nil { - return nil + var statuses []agentshost.ModuleStatus + if a.cfg.ModuleStatus != nil { + statuses = append(statuses, a.cfg.ModuleStatus()...) } - return append([]agentshost.ModuleStatus(nil), a.cfg.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() { @@ -736,13 +754,14 @@ func (a *Agent) deliverPrimaryReport(ctx context.Context, report agentshost.Repo if !a.reportBuffer.IsEmpty() { a.flushBuffer(ctx) if !a.reportBuffer.IsEmpty() { - a.reportBuffer.Push(report) + 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(). @@ -763,7 +782,7 @@ func (a *Agent) deliverPrimaryReport(ctx context.Context, report agentshost.Repo return nil } - a.reportBuffer.Push(report) + a.bufferPrimaryReport(report) event := a.logger.Warn(). Err(err). Str("endpoint", agentReportEndpoint). @@ -777,6 +796,10 @@ func (a *Agent) deliverPrimaryReport(ctx context.Context, report agentshost.Repo } agenttarget.MarkDelivery("host", "primary", "primary", true) + // 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{} @@ -791,6 +814,17 @@ func (a *Agent) deliverPrimaryReport(ctx context.Context, report agentshost.Repo 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) @@ -1135,9 +1169,12 @@ func (a *Agent) buildReport(ctx context.Context) (agentshost.Report, error) { Unraid: unraidData, Ceph: cephData, ClusterSensors: clusterSensors, - Tags: append([]string(nil), runtimeConfig.tags...), - Timestamp: a.collector.Now(), - SequenceID: a.nextReportSequenceID(), + // 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 @@ -1322,6 +1359,7 @@ func (a *Agent) ApplyRemoteConfig(settings map[string]interface{}, commandsEnabl a.configMu.Unlock() a.logger.Info().Bool("disable_ceph", disableCeph).Msg("Applied remote Ceph collection setting") } + a.applyRemoteAvailabilityTargets(settings) if remoteConfigAppliedWithoutRestart(settings) && remoteconfig.HasAppliedDesiredConfig(commandsEnabled, settings) { metadata, err := remoteconfig.BuildDesiredConfigMetadata(commandsEnabled, settings) if err != nil { @@ -1334,10 +1372,27 @@ func (a *Agent) ApplyRemoteConfig(settings map[string]interface{}, commandsEnabl } } +// 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": + case "interval", "report_ip", "disable_ceph", availabilitySettingsKey: continue default: return false diff --git a/internal/hostagent/availability.go b/internal/hostagent/availability.go new file mode 100644 index 000000000..4b12e0dab --- /dev/null +++ b/internal/hostagent/availability.go @@ -0,0 +1,392 @@ +package hostagent + +import ( + "context" + "encoding/json" + "fmt" + "sort" + "strings" + "sync" + "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/availabilityprobe" + "github.com/rcourtman/pulse-go-rewrite/internal/config" + "github.com/rcourtman/pulse-go-rewrite/internal/utils" + agentshost "github.com/rcourtman/pulse-go-rewrite/pkg/agents/host" + "github.com/rs/zerolog" +) + +// availabilitySettingsKey is the remote-config key carrying the availability +// targets the server assigned to this agent. +const availabilitySettingsKey = "availabilityTargets" + +// availabilityModuleName is how an active probe assignment is surfaced in +// AgentInfo.Modules, alongside host/docker/kubernetes. +const availabilityModuleName = "availability" + +const ( + availabilityModuleStateStarting = "starting" + availabilityModuleStateRunning = "running" +) + +// availabilityPendingCapacity bounds the results waiting for a report. It +// mirrors the report buffer's spirit: an agent that cannot reach Pulse keeps +// the newest observations and drops the oldest rather than growing without +// limit. +const availabilityPendingCapacity = 200 + +// availabilityMinInterval and availabilityMaxInterval mirror the server-side +// clamp so an assignment cannot make the agent probe faster (or rarer) than +// the poller would have locally. +const ( + availabilityMinInterval = 10 * time.Second + availabilityMaxInterval = time.Hour +) + +// availabilityErrorLimit bounds the failure text a single check contributes to +// a report. The server only shows the latest error, so an unbounded message +// buys nothing and inflates every report. +const availabilityErrorLimit = 240 + +// pendingAvailabilityResult tags each queued result with a monotonic sequence. +// Delivery is confirmed by sequence rather than by count so a queue overflow +// while a report is in flight cannot drop an observation that was never sent. +type pendingAvailabilityResult struct { + sequence uint64 + result agentshost.AvailabilityProbeResult +} + +// availabilityProbeModule runs the availability checks the server assigned to +// this agent and queues their results for the next host report. It owns its +// scheduling entirely: the server sends assignments, never a schedule. +type availabilityProbeModule struct { + logger zerolog.Logger + now func() time.Time + probe func(context.Context, config.AvailabilityTarget) (availabilityprobe.Outcome, error) + + mu sync.Mutex + targets []config.AvailabilityTarget + updatedAt time.Time + running bool + sequence uint64 + inFlight uint64 + + pending *utils.Queue[pendingAvailabilityResult] + reload chan struct{} +} + +func newAvailabilityProbeModule(logger zerolog.Logger, targets []config.AvailabilityTarget) *availabilityProbeModule { + module := &availabilityProbeModule{ + logger: logger.With().Str("module", availabilityModuleName).Logger(), + now: time.Now, + probe: availabilityprobe.Result, + pending: utils.New[pendingAvailabilityResult](availabilityPendingCapacity), + reload: make(chan struct{}, 1), + } + module.applyTargets(targets) + return module +} + +// applyTargets replaces the assignment set. It reports whether the schedule +// actually changed: the remote config is re-fetched on a fixed interval and an +// unchanged assignment must not restart (and therefore re-run) every check. +func (m *availabilityProbeModule) applyTargets(targets []config.AvailabilityTarget) bool { + if m == nil { + return false + } + normalized := normalizeAvailabilityAssignments(targets) + + m.mu.Lock() + if availabilityAssignmentsEqual(m.targets, normalized) { + m.mu.Unlock() + return false + } + m.targets = normalized + m.updatedAt = m.now().UTC() + m.mu.Unlock() + + select { + case m.reload <- struct{}{}: + default: + } + return true +} + +func normalizeAvailabilityAssignments(targets []config.AvailabilityTarget) []config.AvailabilityTarget { + normalized := make([]config.AvailabilityTarget, 0, len(targets)) + for _, target := range targets { + // The ID is checked before normalization: defaults would mint a fresh + // one, and a target the server cannot recognise is unreportable. + if strings.TrimSpace(target.ID) == "" { + continue + } + normalized = append(normalized, config.NormalizeAvailabilityTarget(target)) + } + sort.Slice(normalized, func(i, j int) bool { return normalized[i].ID < normalized[j].ID }) + return normalized +} + +func availabilityAssignmentsEqual(left, right []config.AvailabilityTarget) bool { + if len(left) != len(right) { + return false + } + for i := range left { + if left[i] != right[i] { + return false + } + } + return true +} + +func (m *availabilityProbeModule) assignments() []config.AvailabilityTarget { + m.mu.Lock() + defer m.mu.Unlock() + return append([]config.AvailabilityTarget(nil), m.targets...) +} + +// Run supervises the probe schedule until the context is cancelled. Each +// assignment change stops the current workers and rebuilds them, which keeps a +// removed target from ever running one more time. +func (m *availabilityProbeModule) Run(ctx context.Context) { + if m == nil { + return + } + m.mu.Lock() + m.running = true + m.mu.Unlock() + defer func() { + m.mu.Lock() + m.running = false + m.mu.Unlock() + }() + + for { + // Consume a reload queued before this pass: the assignment about to be + // read already contains it, and acting on it again would restart every + // worker and re-run each check immediately. + select { + case <-m.reload: + default: + } + + runCtx, cancel := context.WithCancel(ctx) + var wg sync.WaitGroup + for _, target := range m.assignments() { + if !target.Enabled { + continue + } + wg.Add(1) + go func(target config.AvailabilityTarget) { + defer wg.Done() + m.runTarget(runCtx, target) + }(target) + } + + select { + case <-ctx.Done(): + cancel() + wg.Wait() + return + case <-m.reload: + cancel() + wg.Wait() + } + } +} + +func (m *availabilityProbeModule) runTarget(ctx context.Context, target config.AvailabilityTarget) { + interval := availabilityProbeInterval(target) + ticker := time.NewTicker(interval) + defer ticker.Stop() + + // Probe immediately so a fresh assignment reports before the server's + // staleness window opens instead of after a full interval. + m.check(ctx, target) + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + m.check(ctx, target) + } + } +} + +func availabilityProbeInterval(target config.AvailabilityTarget) time.Duration { + interval := time.Duration(target.EffectivePollIntervalSecs()) * time.Second + if interval < availabilityMinInterval { + return availabilityMinInterval + } + if interval > availabilityMaxInterval { + return availabilityMaxInterval + } + return interval +} + +func (m *availabilityProbeModule) check(ctx context.Context, target config.AvailabilityTarget) { + timeout := time.Duration(target.EffectiveTimeoutMillis()) * time.Millisecond + probeCtx, cancel := context.WithTimeout(ctx, timeout) + start := m.now() + outcome, err := m.probe(probeCtx, target) + latency := m.now().Sub(start) + cancel() + + if ctx.Err() != nil { + // A shutdown or reassignment cancelled the check mid-flight. The + // failure it produced describes the agent, not the target. + return + } + if latency < 0 { + latency = 0 + } + + result := agentshost.AvailabilityProbeResult{ + TargetID: target.ID, + Outcome: string(outcome), + LatencyMillis: latency.Milliseconds(), + CheckedAt: m.now().UTC(), + } + if err != nil { + message := strings.TrimSpace(err.Error()) + if len(message) > availabilityErrorLimit { + message = message[:availabilityErrorLimit] + } + result.Error = message + } + + m.enqueue(result) + + m.logger.Debug(). + Str("targetID", target.ID). + Str("outcome", result.Outcome). + Int64("latencyMillis", result.LatencyMillis). + Msg("Completed assigned availability check") +} + +// enqueue queues one completed observation for the next report. The queue is +// bounded and drops the oldest entry when full, so a long outage costs the +// earliest observations rather than unbounded memory. +func (m *availabilityProbeModule) enqueue(result agentshost.AvailabilityProbeResult) { + m.mu.Lock() + m.sequence++ + pending := pendingAvailabilityResult{sequence: m.sequence, result: result} + m.mu.Unlock() + m.pending.Push(pending) +} + +// snapshotForReport returns every queued result without removing it and marks +// the batch as in flight. Nothing leaves the queue until the primary +// destination has accepted it. +func (m *availabilityProbeModule) snapshotForReport() []agentshost.AvailabilityProbeResult { + if m == nil { + return nil + } + items := m.pending.Items() + if len(items) == 0 { + m.mu.Lock() + m.inFlight = 0 + m.mu.Unlock() + return nil + } + results := make([]agentshost.AvailabilityProbeResult, 0, len(items)) + for _, item := range items { + results = append(results, item.result) + } + + m.mu.Lock() + m.inFlight = items[len(items)-1].sequence + m.mu.Unlock() + return results +} + +// commitDelivered drops the results the primary destination accepted. Results +// queued while the report was in flight carry a higher sequence and survive +// for the next report, so no observation is delivered twice or lost. +func (m *availabilityProbeModule) commitDelivered() { + if m == nil { + return + } + m.mu.Lock() + delivered := m.inFlight + m.inFlight = 0 + m.mu.Unlock() + if delivered == 0 { + return + } + for { + item, ok := m.pending.Peek() + if !ok || item.sequence > delivered { + return + } + m.pending.Pop() + } +} + +// discardInFlight forgets the in-flight marker when a report is buffered +// instead of sent, so a later success cannot retire results the server never +// received. +func (m *availabilityProbeModule) discardInFlight() { + if m == nil { + return + } + m.mu.Lock() + m.inFlight = 0 + m.mu.Unlock() +} + +// moduleStatus surfaces the probe module in AgentInfo.Modules only while the +// server has something assigned to this agent. An agent with no assignment has +// no module to report on. +func (m *availabilityProbeModule) moduleStatus() (agentshost.ModuleStatus, bool) { + if m == nil { + return agentshost.ModuleStatus{}, false + } + m.mu.Lock() + defer m.mu.Unlock() + if len(m.targets) == 0 { + return agentshost.ModuleStatus{}, false + } + state := availabilityModuleStateStarting + if m.running { + state = availabilityModuleStateRunning + } + return agentshost.ModuleStatus{ + Name: availabilityModuleName, + Enabled: true, + State: state, + UpdatedAt: m.updatedAt, + }, true +} + +// AvailabilityTargetsFromSetting decodes the availability assignments carried +// by a remote-config setting value. The payload is re-marshalled through JSON +// so the agent shares the server's field names and ignores keys it does not +// know. +func AvailabilityTargetsFromSetting(value interface{}) ([]config.AvailabilityTarget, error) { + if value == nil { + return nil, nil + } + raw, err := json.Marshal(value) + if err != nil { + return nil, fmt.Errorf("encode availability targets: %w", err) + } + var targets []config.AvailabilityTarget + if err := json.Unmarshal(raw, &targets); err != nil { + return nil, fmt.Errorf("decode availability targets: %w", err) + } + return normalizeAvailabilityAssignments(targets), nil +} + +// availabilityTargetsFromSettings reports the assignments in a remote-config +// payload. The server omits the key entirely when it has nothing assigned to +// this agent, so a missing key means "no assignments" rather than "unchanged". +func availabilityTargetsFromSettings(settings map[string]interface{}) ([]config.AvailabilityTarget, error) { + if settings == nil { + return nil, nil + } + value, ok := settings[availabilitySettingsKey] + if !ok { + return nil, nil + } + return AvailabilityTargetsFromSetting(value) +} diff --git a/internal/hostagent/availability_test.go b/internal/hostagent/availability_test.go new file mode 100644 index 000000000..da201c08e --- /dev/null +++ b/internal/hostagent/availability_test.go @@ -0,0 +1,556 @@ +package hostagent + +import ( + "compress/gzip" + "context" + "encoding/json" + "errors" + "io" + "net/http" + "net/http/httptest" + "sync" + "testing" + "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/availabilityprobe" + "github.com/rcourtman/pulse-go-rewrite/internal/config" + agentshost "github.com/rcourtman/pulse-go-rewrite/pkg/agents/host" + "github.com/rs/zerolog" +) + +func testProbeModule(t *testing.T, probe func(context.Context, config.AvailabilityTarget) (availabilityprobe.Outcome, error)) *availabilityProbeModule { + t.Helper() + logger := zerolog.New(io.Discard) + module := newAvailabilityProbeModule(logger, nil) + if probe != nil { + module.probe = probe + } + return module +} + +func availabilitySetting(entries ...map[string]interface{}) map[string]interface{} { + values := make([]interface{}, 0, len(entries)) + for _, entry := range entries { + values = append(values, entry) + } + return map[string]interface{}{availabilitySettingsKey: values} +} + +func TestAvailabilityTargetsFromSettingsDecodesServerPayload(t *testing.T) { + // Mirrors the payload shape built by the server for an assigned agent, + // including a key the agent does not know about. + settings := availabilitySetting(map[string]interface{}{ + "id": "udp-check", + "name": "UDP check", + "targetKind": "device", + "address": "sensor.local", + "protocol": "udp", + "port": float64(5353), + "udpMode": "response-required", + "udpRequest": "ping", + "udpExpectedResponse": "pong", + "enabled": true, + "pollIntervalSeconds": float64(45), + "timeoutMillis": float64(1500), + "unknownFutureField": "ignored", + }) + + targets, err := availabilityTargetsFromSettings(settings) + if err != nil { + t.Fatalf("availabilityTargetsFromSettings() error = %v", err) + } + if len(targets) != 1 { + t.Fatalf("targets = %+v, want one assignment", targets) + } + target := targets[0] + if target.ID != "udp-check" || target.Address != "sensor.local" { + t.Fatalf("target = %+v", target) + } + if target.Protocol != config.AvailabilityProbeUDP || target.Port != 5353 { + t.Fatalf("target protocol/port = %v/%d", target.Protocol, target.Port) + } + if target.UDPRequest != "ping" || target.UDPExpected != "pong" { + t.Fatalf("target udp payloads = %q/%q", target.UDPRequest, target.UDPExpected) + } + if target.EffectivePollIntervalSecs() != 45 || target.EffectiveTimeoutMillis() != 1500 { + t.Fatalf("target schedule = %ds/%dms", target.EffectivePollIntervalSecs(), target.EffectiveTimeoutMillis()) + } + if !target.Enabled { + t.Fatal("target should be enabled") + } + // Server-only concerns are never carried in an agent assignment. + if target.ProbeAgentID != "" || target.LinkedResourceID != "" { + t.Fatalf("target carried server-side fields: %+v", target) + } +} + +func TestAvailabilityTargetsFromSettingsHandlesMissingAndUnusableValues(t *testing.T) { + targets, err := availabilityTargetsFromSettings(nil) + if err != nil || targets != nil { + t.Fatalf("nil settings = (%+v, %v), want no assignments", targets, err) + } + + targets, err = availabilityTargetsFromSettings(map[string]interface{}{"interval": "30s"}) + if err != nil || targets != nil { + t.Fatalf("settings without the key = (%+v, %v), want no assignments", targets, err) + } + + targets, err = availabilityTargetsFromSettings(map[string]interface{}{availabilitySettingsKey: nil}) + if err != nil || targets != nil { + t.Fatalf("null value = (%+v, %v), want no assignments", targets, err) + } + + if _, err := availabilityTargetsFromSettings(map[string]interface{}{availabilitySettingsKey: "not-a-list"}); err == nil { + t.Fatal("garbage value error = nil, want a decode failure") + } + + // An entry without an ID cannot be reported against and is dropped. + targets, err = availabilityTargetsFromSettings(availabilitySetting( + map[string]interface{}{"address": "orphan.local", "protocol": "icmp", "enabled": true}, + map[string]interface{}{"id": "keep", "address": "keep.local", "protocol": "icmp", "enabled": true}, + )) + if err != nil { + t.Fatalf("availabilityTargetsFromSettings() error = %v", err) + } + if len(targets) != 1 || targets[0].ID != "keep" { + t.Fatalf("targets = %+v, want only the identified assignment", targets) + } +} + +func TestApplyRemoteAvailabilityTargetsReconcilesAssignments(t *testing.T) { + agent := &Agent{ + logger: zerolog.New(io.Discard), + availability: testProbeModule(t, nil), + } + + assigned := availabilitySetting(map[string]interface{}{ + "id": "remote-a", "address": "a.local", "protocol": "icmp", "enabled": true, + }) + agent.ApplyRemoteConfig(assigned, nil) + if got := agent.availability.assignments(); len(got) != 1 || got[0].ID != "remote-a" { + t.Fatalf("assignments = %+v, want remote-a", got) + } + + // A repeated fetch of the same assignment must not restart the schedule. + if agent.availability.applyTargets(agent.availability.assignments()) { + t.Fatal("unchanged assignments reported a schedule change") + } + + // An unreadable payload leaves the current schedule alone. + agent.ApplyRemoteConfig(map[string]interface{}{availabilitySettingsKey: "not-a-list"}, nil) + if got := agent.availability.assignments(); len(got) != 1 { + t.Fatalf("assignments after a garbage payload = %+v, want the previous schedule", got) + } + + // The server omits the key once nothing is assigned: that unassigns. + agent.ApplyRemoteConfig(map[string]interface{}{"interval": "30s"}, nil) + if got := agent.availability.assignments(); len(got) != 0 { + t.Fatalf("assignments after unassignment = %+v, want none", got) + } +} + +func TestNewSeedsAvailabilityAssignmentsFromStartupConfig(t *testing.T) { + recorder := &availabilityReportRecorder{} + agent := newAvailabilityDeliveryAgent(t, recorder, config.AvailabilityTarget{ + ID: "boot-target", + Address: "boot.local", + Protocol: config.AvailabilityProbeICMP, + Enabled: true, + }) + + // The assignment fetched before startup must be live without waiting for + // the first remote-config refresh. + if got := agent.availability.assignments(); len(got) != 1 || got[0].ID != "boot-target" { + t.Fatalf("assignments = %+v, want the startup assignment", got) + } + status, ok := agent.availability.moduleStatus() + if !ok || status.Name != availabilityModuleName { + t.Fatalf("module status = (%+v, %v), want the availability module", status, ok) + } +} + +func TestAvailabilityProbeIntervalClampsToServerBounds(t *testing.T) { + tests := []struct { + seconds int + want time.Duration + }{ + {seconds: 0, want: time.Duration(config.DefaultAvailabilityPollIntervalSecs) * time.Second}, + {seconds: 1, want: availabilityMinInterval}, + {seconds: 45, want: 45 * time.Second}, + {seconds: 86400, want: availabilityMaxInterval}, + } + for _, test := range tests { + target := config.AvailabilityTarget{PollIntervalSecs: test.seconds} + if got := availabilityProbeInterval(target); got != test.want { + t.Fatalf("availabilityProbeInterval(%ds) = %v, want %v", test.seconds, got, test.want) + } + } +} + +func TestAvailabilityModuleSchedulerRunsAssignedTargetsAndReschedules(t *testing.T) { + var ( + mu sync.Mutex + checked []string + ) + module := testProbeModule(t, func(_ context.Context, target config.AvailabilityTarget) (availabilityprobe.Outcome, error) { + mu.Lock() + checked = append(checked, target.ID) + mu.Unlock() + if target.ID == "down" { + return availabilityprobe.OutcomeUnreachable, errors.New("icmp probe timed out") + } + return availabilityprobe.OutcomeReachable, nil + }) + module.applyTargets([]config.AvailabilityTarget{ + {ID: "up", Address: "up.local", Protocol: config.AvailabilityProbeICMP, Enabled: true}, + {ID: "down", Address: "down.local", Protocol: config.AvailabilityProbeICMP, Enabled: true}, + {ID: "paused", Address: "paused.local", Protocol: config.AvailabilityProbeICMP, Enabled: false}, + }) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := make(chan struct{}) + go func() { + defer close(done) + module.Run(ctx) + }() + + waitForPending := func(want int) { + t.Helper() + deadline := time.After(5 * time.Second) + for { + if module.pending.Len() >= want { + return + } + select { + case <-time.After(5 * time.Millisecond): + case <-deadline: + t.Fatalf("timed out waiting for %d queued results, have %d", want, module.pending.Len()) + } + } + } + + // Enabled targets probe immediately rather than after a full interval. + waitForPending(2) + results := module.snapshotForReport() + if len(results) != 2 { + t.Fatalf("results = %+v, want one per enabled target", results) + } + byTarget := make(map[string]agentshost.AvailabilityProbeResult, len(results)) + for _, result := range results { + byTarget[result.TargetID] = result + } + if got := byTarget["up"].Outcome; got != string(availabilityprobe.OutcomeReachable) { + t.Fatalf("up outcome = %q", got) + } + if got := byTarget["down"]; got.Outcome != string(availabilityprobe.OutcomeUnreachable) || got.Error == "" { + t.Fatalf("down result = %+v, want a failure with an explanation", got) + } + if byTarget["up"].CheckedAt.IsZero() { + t.Fatal("result is missing its observation time") + } + if _, ok := byTarget["paused"]; ok { + t.Fatalf("disabled target was probed: %+v", results) + } + + module.commitDelivered() + if module.pending.Len() != 0 { + t.Fatalf("pending = %d after delivery, want 0", module.pending.Len()) + } + + // Reassignment stops the old workers and starts the new set at once. + mu.Lock() + checked = nil + mu.Unlock() + module.applyTargets([]config.AvailabilityTarget{ + {ID: "replacement", Address: "new.local", Protocol: config.AvailabilityProbeICMP, Enabled: true}, + }) + waitForPending(1) + reassigned := module.snapshotForReport() + if len(reassigned) != 1 || reassigned[0].TargetID != "replacement" { + t.Fatalf("results after reassignment = %+v, want only the replacement", reassigned) + } + mu.Lock() + for _, id := range checked { + if id != "replacement" { + t.Fatalf("target %q ran after it was unassigned", id) + } + } + mu.Unlock() + + cancel() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("scheduler did not stop on context cancellation") + } +} + +func TestAvailabilityModuleQueueDropsOldestBeyondCapacity(t *testing.T) { + module := testProbeModule(t, nil) + for i := 0; i < availabilityPendingCapacity+50; i++ { + module.enqueue(agentshost.AvailabilityProbeResult{ + TargetID: "target", + Outcome: string(availabilityprobe.OutcomeReachable), + CheckedAt: time.Unix(int64(i), 0).UTC(), + }) + } + + if got := module.pending.Len(); got != availabilityPendingCapacity { + t.Fatalf("pending = %d, want the capacity %d", got, availabilityPendingCapacity) + } + results := module.snapshotForReport() + if len(results) != availabilityPendingCapacity { + t.Fatalf("results = %d, want %d", len(results), availabilityPendingCapacity) + } + // The oldest 50 observations were dropped, not the newest. + if got := results[0].CheckedAt; !got.Equal(time.Unix(50, 0).UTC()) { + t.Fatalf("oldest retained result = %v, want the 51st observation", got) + } + if got := results[len(results)-1].CheckedAt; !got.Equal(time.Unix(availabilityPendingCapacity+49, 0).UTC()) { + t.Fatalf("newest retained result = %v", got) + } +} + +func TestAvailabilityModuleStatusOnlyWhileAssigned(t *testing.T) { + module := testProbeModule(t, nil) + if _, ok := module.moduleStatus(); ok { + t.Fatal("module status reported without an assignment") + } + + module.applyTargets([]config.AvailabilityTarget{ + {ID: "remote", Address: "remote.local", Protocol: config.AvailabilityProbeICMP, Enabled: true}, + }) + status, ok := module.moduleStatus() + if !ok { + t.Fatal("module status missing for an assigned agent") + } + if status.Name != availabilityModuleName || !status.Enabled { + t.Fatalf("status = %+v", status) + } + if status.State != availabilityModuleStateStarting { + t.Fatalf("state = %q before the scheduler runs, want %q", status.State, availabilityModuleStateStarting) + } + if status.UpdatedAt.IsZero() { + t.Fatal("status is missing its assignment time") + } + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + defer close(done) + module.Run(ctx) + }() + deadline := time.After(5 * time.Second) + for { + status, _ = module.moduleStatus() + if status.State == availabilityModuleStateRunning { + break + } + select { + case <-time.After(5 * time.Millisecond): + case <-deadline: + t.Fatalf("state = %q, want %q once the scheduler is up", status.State, availabilityModuleStateRunning) + } + } + cancel() + <-done + + module.applyTargets(nil) + if _, ok := module.moduleStatus(); ok { + t.Fatal("module status survived unassignment") + } +} + +func TestAgentModuleStatusAppendsAvailabilityToRuntimeModules(t *testing.T) { + agent := &Agent{ + logger: zerolog.New(io.Discard), + availability: testProbeModule(t, nil), + cfg: Config{ + ModuleStatus: func() []agentshost.ModuleStatus { + return []agentshost.ModuleStatus{{Name: "host", Enabled: true, State: "running"}} + }, + }, + } + + statuses := agent.currentModuleStatus() + if len(statuses) != 1 || statuses[0].Name != "host" { + t.Fatalf("modules = %+v, want only the runtime modules", statuses) + } + + agent.availability.applyTargets([]config.AvailabilityTarget{ + {ID: "remote", Address: "remote.local", Protocol: config.AvailabilityProbeICMP, Enabled: true}, + }) + statuses = agent.currentModuleStatus() + if len(statuses) != 2 || statuses[1].Name != availabilityModuleName { + t.Fatalf("modules = %+v, want the availability module appended", statuses) + } +} + +// availabilityReportRecorder is a Pulse report endpoint that can be switched +// into failure so retention across a delivery outage is observable. +type availabilityReportRecorder struct { + mu sync.Mutex + reports []agentshost.Report + fail bool +} + +func (r *availabilityReportRecorder) setFail(fail bool) { + r.mu.Lock() + defer r.mu.Unlock() + r.fail = fail +} + +func (r *availabilityReportRecorder) received() []agentshost.Report { + r.mu.Lock() + defer r.mu.Unlock() + return append([]agentshost.Report(nil), r.reports...) +} + +func (r *availabilityReportRecorder) ServeHTTP(w http.ResponseWriter, req *http.Request) { + r.mu.Lock() + fail := r.fail + r.mu.Unlock() + if fail { + w.WriteHeader(http.StatusInternalServerError) + return + } + + var body io.Reader = req.Body + if req.Header.Get("Content-Encoding") == "gzip" { + gz, err := gzip.NewReader(req.Body) + if err != nil { + w.WriteHeader(http.StatusBadRequest) + return + } + defer gz.Close() + body = gz + } + var report agentshost.Report + if err := json.NewDecoder(body).Decode(&report); err != nil { + w.WriteHeader(http.StatusBadRequest) + return + } + + r.mu.Lock() + r.reports = append(r.reports, report) + r.mu.Unlock() + _ = json.NewEncoder(w).Encode(map[string]any{"success": true}) +} + +func newAvailabilityDeliveryAgent(t *testing.T, recorder *availabilityReportRecorder, targets ...config.AvailabilityTarget) *Agent { + t.Helper() + server := httptest.NewServer(recorder) + t.Cleanup(server.Close) + + logger := zerolog.New(io.Discard) + agent, err := New(Config{ + PulseURL: server.URL, + APIToken: "test-token", + Interval: time.Minute, + HostnameOverride: "probe-agent", + StateDir: t.TempDir(), + Logger: &logger, + Collector: &mockCollector{}, + AvailabilityTargets: targets, + packageUpdates: newPackageUpdateManager("windows", nil), + storageCleanup: newStorageCleanupManager("windows", nil), + }) + if err != nil { + t.Fatalf("New() error = %v", err) + } + return agent +} + +func TestAvailabilityResultsSurviveFailedDeliveryAndAreSentOnce(t *testing.T) { + recorder := &availabilityReportRecorder{} + agent := newAvailabilityDeliveryAgent(t, recorder) + ctx := context.Background() + + for _, id := range []string{"target-a", "target-b"} { + agent.availability.enqueue(agentshost.AvailabilityProbeResult{ + TargetID: id, + Outcome: string(availabilityprobe.OutcomeReachable), + CheckedAt: time.Now().UTC(), + }) + } + + recorder.setFail(true) + if err := agent.process(ctx); err != nil { + t.Fatalf("process() during outage error = %v", err) + } + if got := agent.availability.pending.Len(); got != 2 { + t.Fatalf("pending = %d after a failed delivery, want the results retained", got) + } + buffered, ok := agent.reportBuffer.Peek() + if !ok { + t.Fatal("failed report was not buffered") + } + if len(buffered.AvailabilityResults) != 0 { + t.Fatalf("buffered report kept %d results; a retry would double count them", len(buffered.AvailabilityResults)) + } + + // A result observed during the outage joins the same batch. + agent.availability.enqueue(agentshost.AvailabilityProbeResult{ + TargetID: "target-c", + Outcome: string(availabilityprobe.OutcomeUnreachable), + CheckedAt: time.Now().UTC(), + Error: "icmp probe timed out", + }) + + recorder.setFail(false) + if err := agent.process(ctx); err != nil { + t.Fatalf("process() after recovery error = %v", err) + } + if got := agent.availability.pending.Len(); got != 0 { + t.Fatalf("pending = %d after a successful delivery, want 0", got) + } + + // A later report must not repeat what the server already counted. + if err := agent.process(ctx); err != nil { + t.Fatalf("process() error = %v", err) + } + + delivered := map[string]int{} + for _, report := range recorder.received() { + for _, result := range report.AvailabilityResults { + delivered[result.TargetID]++ + } + } + if len(delivered) != 3 { + t.Fatalf("delivered targets = %+v, want all three observations", delivered) + } + for id, count := range delivered { + if count != 1 { + t.Fatalf("target %q was delivered %d times, want exactly once", id, count) + } + } +} + +func TestAvailabilityResultsAreNotRetiredWhenTheTokenIsRejected(t *testing.T) { + recorder := &availabilityReportRecorder{} + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusUnauthorized) + })) + defer server.Close() + + agent := newAvailabilityDeliveryAgent(t, recorder) + agent.trimmedPulseURL = server.URL + agent.availability.enqueue(agentshost.AvailabilityProbeResult{ + TargetID: "target-a", + Outcome: string(availabilityprobe.OutcomeReachable), + CheckedAt: time.Now().UTC(), + }) + + // A rejected token drops the report instead of buffering it, so the + // results have to stay queued for whenever the token is replaced. + if err := agent.process(context.Background()); err != nil { + t.Fatalf("process() error = %v", err) + } + if got := agent.availability.pending.Len(); got != 1 { + t.Fatalf("pending = %d after a rejected report, want the result retained", got) + } + if !agent.reportBuffer.IsEmpty() { + t.Fatal("a rejected report should not be buffered") + } +} diff --git a/internal/monitoring/availability_poller.go b/internal/monitoring/availability_poller.go index 1817fdbed..0953bbc99 100644 --- a/internal/monitoring/availability_poller.go +++ b/internal/monitoring/availability_poller.go @@ -412,6 +412,7 @@ func availabilityResourceFromTarget(target config.AvailabilityTarget, status Ava Address: target.Address, Protocol: string(target.Protocol), ProbeOutcome: status.Outcome, + ProbeAgentID: status.ProbeAgentID, UDPMode: string(target.UDPMode), Port: target.Port, Path: target.Path, diff --git a/internal/monitoring/availability_probe_agent.go b/internal/monitoring/availability_probe_agent.go index 1bffe60ac..2c65d6a39 100644 --- a/internal/monitoring/availability_probe_agent.go +++ b/internal/monitoring/availability_probe_agent.go @@ -7,6 +7,7 @@ import ( "github.com/rcourtman/pulse-go-rewrite/internal/availabilityprobe" "github.com/rcourtman/pulse-go-rewrite/internal/config" + agentshost "github.com/rcourtman/pulse-go-rewrite/pkg/agents/host" pkglicensing "github.com/rcourtman/pulse-go-rewrite/pkg/licensing" "github.com/rs/zerolog/log" ) @@ -29,6 +30,33 @@ type ProbeAvailabilityResult struct { Error string } +// probeAvailabilityResultsFromReport converts the wire results carried by a +// host agent report. Outcome vocabulary is normalized here so the ingestion +// path never has to trust an agent's spelling; probeResultOutcome then decides +// what an unknown outcome means for failure accounting. +func probeAvailabilityResultsFromReport(reported []agentshost.AvailabilityProbeResult) []ProbeAvailabilityResult { + if len(reported) == 0 { + return nil + } + results := make([]ProbeAvailabilityResult, 0, len(reported)) + for _, entry := range reported { + outcome := availabilityprobe.Outcome(strings.ToLower(strings.TrimSpace(entry.Outcome))) + switch outcome { + case availabilityprobe.OutcomeReachable, availabilityprobe.OutcomeUnreachable, availabilityprobe.OutcomeIndeterminate: + default: + outcome = availabilityprobe.OutcomeIndeterminate + } + results = append(results, ProbeAvailabilityResult{ + TargetID: strings.TrimSpace(entry.TargetID), + Outcome: outcome, + LatencyMillis: entry.LatencyMillis, + CheckedAt: entry.CheckedAt, + Error: strings.TrimSpace(entry.Error), + }) + } + return results +} + // effectiveProbeAgentID returns the host agent that currently owns execution of // the target. It collapses to local execution ("") whenever the external probe // entitlement is absent, so a license lapse resumes local polling instead of diff --git a/internal/monitoring/availability_probe_agent_test.go b/internal/monitoring/availability_probe_agent_test.go index bcfaa536c..6ce9ef9cd 100644 --- a/internal/monitoring/availability_probe_agent_test.go +++ b/internal/monitoring/availability_probe_agent_test.go @@ -4,8 +4,10 @@ import ( "testing" "time" + "github.com/rcourtman/pulse-go-rewrite/internal/alerts" "github.com/rcourtman/pulse-go-rewrite/internal/config" "github.com/rcourtman/pulse-go-rewrite/internal/models" + agentshost "github.com/rcourtman/pulse-go-rewrite/pkg/agents/host" pkglicensing "github.com/rcourtman/pulse-go-rewrite/pkg/licensing" ) @@ -393,9 +395,94 @@ func TestAvailabilitySupplementalRecordsPresentStaleProbeAsIndeterminate(t *test if data.LastError != availabilityProbeStaleError { t.Fatalf("availability last error = %q, want %q", data.LastError, availabilityProbeStaleError) } + if data.ProbeAgentID != "agent-1" { + t.Fatalf("availability probe agent = %q, want agent-1 so the UI can attribute the source", data.ProbeAgentID) + } statuses := availabilityPollProvider{}.ConnectionStatuses(monitor) if statuses["availability-remote"] { t.Fatalf("connection statuses = %+v, want a stale probe to read as not connected", statuses) } } + +func TestApplyHostReportIngestsAssignedAvailabilityResults(t *testing.T) { + monitor := newProbeAgentTestMonitor(t, + probeAgentTarget("remote", "probe-host"), + probeAgentTarget("foreign", "other-agent"), + ) + monitor.alertManager = alerts.NewManager() + t.Cleanup(func() { monitor.alertManager.Stop() }) + monitor.config = &config.Config{} + monitor.rateTracker = NewRateTracker() + monitor.hostTokenBindings = make(map[string]string) + monitor.SetLicenseChecker(licenseWithExternalProbe(true)) + + checkedAt := time.Now().UTC() + host, err := monitor.ApplyHostReport(agentshost.Report{ + Agent: agentshost.AgentInfo{ID: "probe-host", Version: "6.0.0", IntervalSeconds: 30}, + Host: agentshost.HostInfo{ + ID: "probe-host", + Hostname: "probe-host.local", + Platform: "linux", + }, + AvailabilityResults: []agentshost.AvailabilityProbeResult{ + {TargetID: "remote", Outcome: "reachable", LatencyMillis: 17, CheckedAt: checkedAt}, + {TargetID: "foreign", Outcome: "reachable", LatencyMillis: 3, CheckedAt: checkedAt}, + }, + Timestamp: checkedAt, + }, nil) + if err != nil { + t.Fatalf("ApplyHostReport() error = %v", err) + } + // The ownership check compares the host ID that GetHostAgentConfig is + // keyed by, so the report must resolve to exactly that identity. + if host.ID != "probe-host" { + t.Fatalf("host ID = %q, want the assigned probe agent ID", host.ID) + } + + statuses := monitor.AvailabilityStatusSnapshot() + status, ok := statuses["remote"] + if !ok { + t.Fatalf("statuses = %+v, want the assigned target applied", statuses) + } + if !status.Available || status.LatencyMillis != 17 { + t.Fatalf("status = %+v, want the reported observation", status) + } + if status.ProbeAgentID != "probe-host" { + t.Fatalf("probe agent attribution = %q, want probe-host", status.ProbeAgentID) + } + if !status.LastChecked.Equal(checkedAt) { + t.Fatalf("last checked = %v, want %v", status.LastChecked, checkedAt) + } + if _, ok := statuses["foreign"]; ok { + t.Fatal("a report claimed a target assigned to another agent") + } +} + +func TestProbeAvailabilityResultsFromReportNormalizesOutcomes(t *testing.T) { + if got := probeAvailabilityResultsFromReport(nil); got != nil { + t.Fatalf("results = %+v, want nil for a report without probe work", got) + } + + checkedAt := time.Now().UTC() + results := probeAvailabilityResultsFromReport([]agentshost.AvailabilityProbeResult{ + {TargetID: " padded ", Outcome: " REACHABLE ", LatencyMillis: 4, CheckedAt: checkedAt}, + {TargetID: "b", Outcome: "unreachable", Error: " icmp probe timed out "}, + {TargetID: "c", Outcome: "who-knows"}, + {TargetID: "d", Outcome: ""}, + }) + if len(results) != 4 { + t.Fatalf("results = %+v, want one per reported observation", results) + } + if results[0].TargetID != "padded" || results[0].Outcome != AvailabilityProbeReachable { + t.Fatalf("first result = %+v", results[0]) + } + if results[1].Error != "icmp probe timed out" || results[1].Outcome != AvailabilityProbeUnreachable { + t.Fatalf("second result = %+v", results[1]) + } + for _, result := range results[2:] { + if result.Outcome != AvailabilityProbeIndeterminate { + t.Fatalf("result %+v: unknown outcomes must read as indeterminate", result) + } + } +} diff --git a/internal/monitoring/monitor_agents.go b/internal/monitoring/monitor_agents.go index 052066b1f..83e10f145 100644 --- a/internal/monitoring/monitor_agents.go +++ b/internal/monitoring/monitor_agents.go @@ -2851,6 +2851,9 @@ func (m *Monitor) ApplyHostReport(report agentshost.Report, tokenRecord *config. // Store cluster peer sensor data if present and evict stale entries m.applyClusterSensors(report.ClusterSensors, observedAt) + // Availability results are ingested only once the host identity is + // committed, because ownership is checked against that host ID. + m.ApplyProbeAvailabilityResults(host.ID, probeAvailabilityResultsFromReport(report.AvailabilityResults)) m.persistHostContinuity(host, report, reportOrder) m.refreshUnifiedResourceStoreAfterAgentReport() diff --git a/internal/unifiedresources/types.go b/internal/unifiedresources/types.go index d6efb2769..1aaca576e 100644 --- a/internal/unifiedresources/types.go +++ b/internal/unifiedresources/types.go @@ -1686,6 +1686,7 @@ type AvailabilityData struct { Address string `json:"address,omitempty"` Protocol string `json:"protocol,omitempty"` ProbeOutcome string `json:"probeOutcome,omitempty"` + ProbeAgentID string `json:"probeAgentId,omitempty"` UDPMode string `json:"udpMode,omitempty"` Port int `json:"port,omitempty"` Path string `json:"path,omitempty"` diff --git a/pkg/agents/host/report.go b/pkg/agents/host/report.go index c2b3ef98e..04a4b1b00 100644 --- a/pkg/agents/host/report.go +++ b/pkg/agents/host/report.go @@ -54,9 +54,25 @@ type Report struct { Unraid *UnraidStorage `json:"unraid,omitempty"` Ceph *CephCluster `json:"ceph,omitempty"` ClusterSensors []ClusterNodeSensors `json:"clusterSensors,omitempty"` - Tags []string `json:"tags,omitempty"` - Timestamp time.Time `json:"timestamp"` - SequenceID string `json:"sequenceId,omitempty"` + // AvailabilityResults carries availability checks the agent executed on + // behalf of the server for targets assigned to it. The server owns failure + // accounting; the agent only reports what each check observed. + AvailabilityResults []AvailabilityProbeResult `json:"availabilityResults,omitempty"` + Tags []string `json:"tags,omitempty"` + Timestamp time.Time `json:"timestamp"` + SequenceID string `json:"sequenceId,omitempty"` +} + +// AvailabilityProbeResult is one completed availability check reported by a +// remote probe agent. Outcome mirrors the shared probe vocabulary +// ("reachable", "unreachable", "indeterminate"); anything else is treated as +// indeterminate by the server. +type AvailabilityProbeResult struct { + TargetID string `json:"targetId"` + Outcome string `json:"outcome"` + LatencyMillis int64 `json:"latencyMillis"` + CheckedAt time.Time `json:"checkedAt"` + Error string `json:"error,omitempty"` } // ClusterNodeSensors contains temperature sensor data collected from a Proxmox diff --git a/pkg/agents/host/report_test.go b/pkg/agents/host/report_test.go index 139f665e8..5db97bec1 100644 --- a/pkg/agents/host/report_test.go +++ b/pkg/agents/host/report_test.go @@ -510,3 +510,54 @@ func contains(s, substr string) bool { } return false } + +func TestReportAvailabilityResultsJSONRoundTrip(t *testing.T) { + checkedAt := time.Date(2026, 7, 27, 10, 30, 0, 0, time.UTC) + report := Report{ + Agent: AgentInfo{ID: "probe-agent"}, + Host: HostInfo{Hostname: "probe-agent.local"}, + AvailabilityResults: []AvailabilityProbeResult{ + {TargetID: "target-1", Outcome: "reachable", LatencyMillis: 12, CheckedAt: checkedAt}, + {TargetID: "target-2", Outcome: "unreachable", CheckedAt: checkedAt, Error: "icmp probe timed out"}, + }, + Timestamp: checkedAt, + } + + data, err := json.Marshal(report) + if err != nil { + t.Fatalf("Failed to marshal: %v", err) + } + if !contains(string(data), `"availabilityResults":[`) { + t.Fatalf("availability results missing from payload: %s", data) + } + if contains(string(data), `"error":""`) { + t.Fatalf("empty probe error should be omitted: %s", data) + } + + var decoded Report + if err := json.Unmarshal(data, &decoded); err != nil { + t.Fatalf("Failed to unmarshal: %v", err) + } + if len(decoded.AvailabilityResults) != 2 { + t.Fatalf("availability results = %+v, want 2", decoded.AvailabilityResults) + } + first := decoded.AvailabilityResults[0] + if first.TargetID != "target-1" || first.Outcome != "reachable" || first.LatencyMillis != 12 { + t.Fatalf("first result = %+v", first) + } + if !first.CheckedAt.Equal(checkedAt) { + t.Fatalf("checkedAt = %v, want %v", first.CheckedAt, checkedAt) + } + if decoded.AvailabilityResults[1].Error != "icmp probe timed out" { + t.Fatalf("second result = %+v", decoded.AvailabilityResults[1]) + } + + // A report without probe assignments must not carry the section at all. + bare, err := json.Marshal(Report{Host: HostInfo{Hostname: "plain"}}) + if err != nil { + t.Fatalf("Failed to marshal bare report: %v", err) + } + if contains(string(bare), "availabilityResults") { + t.Fatalf("empty availability results should be omitted: %s", bare) + } +}