From d699dc66a05e338de5e30df778b33598afc069e5 Mon Sep 17 00:00:00 2001 From: rcourtman Date: Fri, 7 Aug 2026 11:40:46 +0100 Subject: [PATCH] feat(agent): converge agent self-update within one report cycle MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The server now echoes its version on unified-agent report acks, and the agent nudges its auto-updater the moment an ack carries a newer version. After a server upgrade, agents converge within one report interval instead of waiting out the hourly update check, so the "older Pulse agent" notice self-resolves in seconds once upgraded agents report in. The hourly loop stays as the retry and backstop path. Nudges dedupe per server version, refuse downgrades, skip disabled and development-mode updaters, and never fire from observer destination acks — only the authoritative server may steer an agent's updater, and a nudged check re-validates against the server and runs the existing checksum and self-test pipeline before swapping binaries. Agents deployed before this change still converge on their old hourly cadence once; every upgrade after that lands within a report cycle. Contract deltas recorded in agent-lifecycle and api-contracts, with boundary notes in security-privacy (no update authority in the echo), performance-and-scalability (no steady-state work), and storage-recovery (nothing persisted). --- cmd/pulse-agent/main.go | 11 +- cmd/pulse-agent/main_test.go | 54 +++++++ .../v6/internal/subsystems/agent-lifecycle.md | 23 +++ .../v6/internal/subsystems/api-contracts.md | 12 ++ .../subsystems/performance-and-scalability.md | 8 ++ .../internal/subsystems/security-privacy.md | 11 ++ .../internal/subsystems/storage-recovery.md | 8 ++ internal/agentupdate/update.go | 56 ++++++++ internal/agentupdate/update_http_test.go | 135 ++++++++++++++++++ internal/api/agent_ingest.go | 15 ++ internal/api/contract_test.go | 53 +++++++ ...gent_removal_lifecycle_integration_test.go | 11 ++ internal/api/router.go | 1 + internal/hostagent/agent.go | 21 ++- internal/hostagent/send_report_test.go | 94 ++++++++++++ 15 files changed, 509 insertions(+), 4 deletions(-) diff --git a/cmd/pulse-agent/main.go b/cmd/pulse-agent/main.go index 8cf1ff219..f3661bf23 100644 --- a/cmd/pulse-agent/main.go +++ b/cmd/pulse-agent/main.go @@ -108,6 +108,15 @@ var ( remoteConfigRefreshInterval = 1 * time.Minute ) +// wireUpdaterHooks connects the self-updater to the host module's report loop: +// update status snapshots flow out on reports, and server versions carried on +// report acks nudge the updater so a server upgrade converges within one +// report cycle instead of the next hourly check. +func wireUpdaterHooks(hostCfg *hostagent.Config, updater *agentupdate.Updater) { + hostCfg.UpdateStatus = updater.Snapshot + hostCfg.OnServerVersion = updater.NudgeVersion +} + type multiValue []string func (m *multiValue) String() string { @@ -413,12 +422,12 @@ func run(ctx context.Context, args []string, getenv func(string) string) error { DisableCeph: cfg.DisableCeph, AvailabilityTargets: cfg.AvailabilityTargets, AppliedConfig: cfg.AppliedConfig, - UpdateStatus: updater.Snapshot, ModuleStatus: runtimeStatus.moduleStatuses, Observers: hostObserverTargets(cfg.Observers), DockerContainerUpdater: dockerUpdaterBridge, } + wireUpdaterHooks(&hostCfg, updater) agent, err := newHostAgent(hostCfg) if err != nil { diff --git a/cmd/pulse-agent/main_test.go b/cmd/pulse-agent/main_test.go index dd9d14423..b41c3cea3 100644 --- a/cmd/pulse-agent/main_test.go +++ b/cmd/pulse-agent/main_test.go @@ -2393,3 +2393,57 @@ func TestApplyRemoteSettingsCarriesAvailabilityAssignmentsToStartup(t *testing.T t.Fatalf("availability targets = %+v, want an unreadable payload ignored", cfg.AvailabilityTargets) } } + +func TestWireUpdaterHooksNudgesUpdaterOnNewerAckVersion(t *testing.T) { + t.Parallel() + + hits := make(chan string, 8) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + hits <- r.URL.Path + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"version":"9.9.9"}`)) + })) + defer srv.Close() + + updater := newUpdater(agentupdate.Config{ + CurrentVersion: "1.0.0", + PulseURL: srv.URL, + // Keep the built-in initial check out of the way so the only thing + // that can reach the server inside the assertion window is the nudge. + InitialCheckDelay: time.Hour, + CheckInterval: time.Hour, + }) + + var hostCfg hostagent.Config + wireUpdaterHooks(&hostCfg, updater) + if hostCfg.UpdateStatus == nil { + t.Fatal("UpdateStatus hook not wired") + } + if hostCfg.OnServerVersion == nil { + t.Fatal("OnServerVersion hook not wired") + } + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := make(chan struct{}) + go func() { + defer close(done) + updater.RunLoop(ctx) + }() + + // Simulate what the host module does when a report ack carries a newer + // server version. The wired updater must check for updates immediately. + hostCfg.OnServerVersion("9.9.9") + + select { + case path := <-hits: + if !strings.Contains(path, "/api/agent/version") { + t.Fatalf("first updater request hit %q, want the version check endpoint", path) + } + case <-time.After(10 * time.Second): + t.Fatal("ack-carried server version did not trigger an immediate update check") + } + + cancel() + <-done +} diff --git a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md index 90804f06e..b99fa7eec 100644 --- a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md +++ b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md @@ -5905,3 +5905,26 @@ a fresh handle rather than using a closed one (`TestResourceHandlers_CloseTenantStoreReleasesTheHandle`), and both entry points are idempotent and nil-safe (`TestResourceHandlers_CloseIsIdempotentAndNilSafe`). +### Server upgrades nudge agent self-update within one report cycle + +The unified-agent report ack now carries the server's version +(`serverVersion`), and the host module surfaces it through +`hostagent.Config.OnServerVersion`, which `cmd/pulse-agent` wires to +`agentupdate.Updater.NudgeVersion` via `wireUpdaterHooks` +(`TestWireUpdaterHooksNudgesUpdaterOnNewerAckVersion`). A nudge wakes the +update loop immediately, including during the initial-check delay, so after a +server upgrade agents converge within one report interval instead of waiting +out the hourly check (`TestRunLoopRunsCheckOnNudge`). Nudges queue only for +versions strictly newer than the running agent, at most once per distinct +server version, and never on disabled or development-mode updaters +(`TestNudgeVersionQueuesOnlyForNewerVersions`, +`TestNudgeVersionComparesPrereleaseIdentifiers`, +`TestNudgeVersionNudgesEachDistinctVersionOnce`, +`TestNudgeVersionRespectsDisabledAndDevelopmentGates`); the hourly loop stays +the retry path when a nudged check fails. Only the authoritative destination's +ack invokes the hook — observer acks never steer the updater +(`TestAgentSendReport_SurfacesAckServerVersion`, +`TestAgentSendReport_SkipsCallbackWithoutAckServerVersion`, +`TestAgentSendReport_ObserverAckNeverInvokesCallback`). +`agentupdate.Config.InitialCheckDelay` overrides the five-second initial check +delay; zero keeps the default. diff --git a/docs/release-control/v6/internal/subsystems/api-contracts.md b/docs/release-control/v6/internal/subsystems/api-contracts.md index 0dc8b6546..41f20c04c 100644 --- a/docs/release-control/v6/internal/subsystems/api-contracts.md +++ b/docs/release-control/v6/internal/subsystems/api-contracts.md @@ -8935,3 +8935,15 @@ a fresh handle rather than using a closed one (`TestResourceHandlers_CloseTenantStoreReleasesTheHandle`), and both entry points are idempotent and nil-safe (`TestResourceHandlers_CloseIsIdempotentAndNilSafe`). +### Unified-agent report ack optionally carries serverVersion + +The `POST /api/agents/agent/report` acknowledgement now includes a +`serverVersion` field carrying the running server's trimmed version string, +supplied at wiring time via `UnifiedAgentHandlers.SetServerVersion` and +omitted entirely when unset +(`TestUnifiedAgentHandlers_HandleReportAckCarriesServerVersion`, +`TestUnifiedAgentHandlers_HandleReportAckOmitsEmptyServerVersion`); the live +router echoes it end-to-end on enrollment acks +(`TestHostAgentRemovalLifecycleThroughAuthenticatedRouterAndRestart`). Agents +use it to trigger an immediate self-update check after a server upgrade. The +field is additive: agents that predate it ignore it. diff --git a/docs/release-control/v6/internal/subsystems/performance-and-scalability.md b/docs/release-control/v6/internal/subsystems/performance-and-scalability.md index d07d828e5..0e2b611e0 100644 --- a/docs/release-control/v6/internal/subsystems/performance-and-scalability.md +++ b/docs/release-control/v6/internal/subsystems/performance-and-scalability.md @@ -2268,3 +2268,11 @@ a fresh handle rather than using a closed one (`TestResourceHandlers_CloseTenantStoreReleasesTheHandle`), and both entry points are idempotent and nil-safe (`TestResourceHandlers_CloseIsIdempotentAndNilSafe`). +### Report-ack version echo adds no steady-state work + +The `serverVersion` field on unified-agent report acks is a constant string +echo per report. Agents queue an update check only when the ack version is +strictly newer than the running agent, and at most once per distinct version +(`TestNudgeVersionNudgesEachDistinctVersionOnce`), so steady-state report +traffic — equal versions on every cycle — never wakes the update loop and adds +no recurring work on either side. diff --git a/docs/release-control/v6/internal/subsystems/security-privacy.md b/docs/release-control/v6/internal/subsystems/security-privacy.md index 73b387241..c66410c8c 100644 --- a/docs/release-control/v6/internal/subsystems/security-privacy.md +++ b/docs/release-control/v6/internal/subsystems/security-privacy.md @@ -1951,3 +1951,14 @@ a fresh handle rather than using a closed one (`TestResourceHandlers_CloseTenantStoreReleasesTheHandle`), and both entry points are idempotent and nil-safe (`TestResourceHandlers_CloseIsIdempotentAndNilSafe`). +### Report-ack server versions cannot steer agents beyond one validated check + +The unified-agent report ack's `serverVersion` echo gives acks a version +channel, but it carries no update authority. Only the authoritative Pulse +destination's ack reaches the updater hook — observer destination acks are +discarded before config parsing +(`TestAgentSendReport_ObserverAckNeverInvokesCallback`) — and a nudged updater +re-fetches the server version itself and runs the existing download, checksum, +and self-test pipeline before swapping binaries, so a stale or spoofed ack +version can at most trigger one extra validated check +(`TestRunLoopRunsCheckOnNudge`). diff --git a/docs/release-control/v6/internal/subsystems/storage-recovery.md b/docs/release-control/v6/internal/subsystems/storage-recovery.md index ffab53eb5..61a499d7f 100644 --- a/docs/release-control/v6/internal/subsystems/storage-recovery.md +++ b/docs/release-control/v6/internal/subsystems/storage-recovery.md @@ -5017,3 +5017,11 @@ a fresh handle rather than using a closed one (`TestResourceHandlers_CloseTenantStoreReleasesTheHandle`), and both entry points are idempotent and nil-safe (`TestResourceHandlers_CloseIsIdempotentAndNilSafe`). +### Report-ack version echo persists nothing + +The unified-agent report ack's `serverVersion` field is supplied from the +running server's version at router wiring time +(`UnifiedAgentHandlers.SetServerVersion`) and is never persisted; the report +ingest path's persistence surface — agent-id continuity and host state — is +unchanged by the echo +(`TestHostAgentRemovalLifecycleThroughAuthenticatedRouterAndRestart`). diff --git a/internal/agentupdate/update.go b/internal/agentupdate/update.go index 3bd08ac64..f4498f3af 100644 --- a/internal/agentupdate/update.go +++ b/internal/agentupdate/update.go @@ -136,6 +136,10 @@ type Config struct { // CheckInterval is how often to check for updates (default: 1 hour) CheckInterval time.Duration + // InitialCheckDelay overrides the short delay before the first update + // check after RunLoop starts (default: 5 seconds). Zero keeps the default. + InitialCheckDelay time.Duration + // InsecureSkipVerify skips TLS certificate verification InsecureSkipVerify bool @@ -168,6 +172,10 @@ type Updater struct { selfTestFn func(context.Context, string) error initialDelay time.Duration newTicker func(time.Duration) *time.Ticker + + nudgeCh chan struct{} + nudgeMu sync.Mutex + lastNudgedVersion string } // New creates a new Updater with the given configuration. @@ -233,7 +241,11 @@ func New(cfg Config) *Updater { u.performUpdateFn = u.performUpdateForVersion u.selfTestFn = u.runDownloadedBinarySelfTest u.initialDelay = 5 * time.Second + if cfg.InitialCheckDelay > 0 { + u.initialDelay = cfg.InitialCheckDelay + } u.newTicker = time.NewTicker + u.nudgeCh = make(chan struct{}, 1) return u } @@ -309,6 +321,8 @@ func (u *Updater) RunLoop(ctx context.Context) { return case <-initialDelayTimer.C: u.CheckAndUpdate(ctx) + case <-u.nudgeCh: + u.CheckAndUpdate(ctx) } ticker := u.newTicker(u.cfg.CheckInterval) @@ -320,10 +334,52 @@ func (u *Updater) RunLoop(ctx context.Context) { return case <-ticker.C: u.CheckAndUpdate(ctx) + case <-u.nudgeCh: + u.CheckAndUpdate(ctx) } } } +// NudgeVersion asks the update loop to run a check now, outside the hourly +// cadence. Callers pass the version the server just reported (for example on a +// report acknowledgement); the nudge is dropped unless that version is +// strictly newer than the running agent, so steady-state report traffic never +// wakes the loop. Each distinct server version nudges at most once — if the +// resulting check fails, the hourly loop remains the retry path — and the +// nudged check re-validates against the server before updating, so a stale or +// spoofed version string can trigger nothing beyond one extra check. +func (u *Updater) NudgeVersion(serverVersion string) { + serverVersion = strings.TrimSpace(serverVersion) + if serverVersion == "" || u.cfg.Disabled || u.configErr != nil { + return + } + if u.cfg.CurrentVersion == developmentVersion || serverVersion == developmentVersion { + return + } + if utils.CompareVersions(utils.NormalizeVersion(serverVersion), utils.NormalizeVersion(u.cfg.CurrentVersion)) <= 0 { + return + } + + u.nudgeMu.Lock() + alreadyNudged := u.lastNudgedVersion == serverVersion + if !alreadyNudged { + u.lastNudgedVersion = serverVersion + } + u.nudgeMu.Unlock() + if alreadyNudged { + return + } + + select { + case u.nudgeCh <- struct{}{}: + u.logger.Info(). + Str("currentVersion", u.cfg.CurrentVersion). + Str("serverVersion", serverVersion). + Msg("server reported a newer version; scheduling immediate update check") + default: + } +} + // CheckAndUpdate checks for a new version and performs the update if available. func (u *Updater) CheckAndUpdate(ctx context.Context) { if !u.startCheck() { diff --git a/internal/agentupdate/update_http_test.go b/internal/agentupdate/update_http_test.go index a260fd08d..e8674cd68 100644 --- a/internal/agentupdate/update_http_test.go +++ b/internal/agentupdate/update_http_test.go @@ -531,3 +531,138 @@ func TestUpdater_performUpdateWithExecPath_RejectsRedirects(t *testing.T) { t.Fatalf("expected no redirected download request to be sent") } } + +func drainNudge(u *Updater) bool { + select { + case <-u.nudgeCh: + return true + default: + return false + } +} + +func TestNudgeVersionQueuesOnlyForNewerVersions(t *testing.T) { + t.Parallel() + + u := New(Config{CurrentVersion: "6.2.0", PulseURL: "http://127.0.0.1:7655"}) + + for _, version := range []string{"", "6.2.0", "6.1.9", "v6.2.0", "not-a-version"} { + u.NudgeVersion(version) + if drainNudge(u) { + t.Fatalf("NudgeVersion(%q) queued a nudge, want none", version) + } + } + + u.NudgeVersion("6.2.1") + if !drainNudge(u) { + t.Fatal("NudgeVersion(6.2.1) queued no nudge, want one") + } +} + +func TestNudgeVersionComparesPrereleaseIdentifiers(t *testing.T) { + t.Parallel() + + u := New(Config{CurrentVersion: "6.2.0-rc.8", PulseURL: "http://127.0.0.1:7655"}) + + u.NudgeVersion("6.2.0-rc.8") + if drainNudge(u) { + t.Fatal("equal prerelease queued a nudge, want none") + } + + u.NudgeVersion("v6.2.0-rc.9") + if !drainNudge(u) { + t.Fatal("newer prerelease queued no nudge, want one") + } + + // A stable release outranks its own release candidates. + u.NudgeVersion("6.2.0") + if !drainNudge(u) { + t.Fatal("stable release above rc queued no nudge, want one") + } +} + +func TestNudgeVersionNudgesEachDistinctVersionOnce(t *testing.T) { + t.Parallel() + + u := New(Config{CurrentVersion: "6.2.0", PulseURL: "http://127.0.0.1:7655"}) + + u.NudgeVersion("6.2.1") + if !drainNudge(u) { + t.Fatal("first nudge for 6.2.1 not queued") + } + + // Repeats of the same server version — one per report cycle in production — + // must not re-wake the loop even after the first nudge was consumed. + u.NudgeVersion("6.2.1") + if drainNudge(u) { + t.Fatal("repeated nudge for 6.2.1 queued, want dedupe") + } + + u.NudgeVersion("6.2.2") + if !drainNudge(u) { + t.Fatal("nudge for distinct newer version 6.2.2 not queued") + } +} + +func TestNudgeVersionRespectsDisabledAndDevelopmentGates(t *testing.T) { + t.Parallel() + + disabled := New(Config{CurrentVersion: "6.2.0", PulseURL: "http://127.0.0.1:7655", Disabled: true}) + disabled.NudgeVersion("6.2.1") + if drainNudge(disabled) { + t.Fatal("disabled updater queued a nudge, want none") + } + + dev := New(Config{CurrentVersion: developmentVersion, PulseURL: "http://127.0.0.1:7655"}) + dev.NudgeVersion("6.2.1") + if drainNudge(dev) { + t.Fatal("development-mode updater queued a nudge, want none") + } + + current := New(Config{CurrentVersion: "6.2.0", PulseURL: "http://127.0.0.1:7655"}) + current.NudgeVersion(developmentVersion) + if drainNudge(current) { + t.Fatal("development server version queued a nudge, want none") + } +} + +func TestRunLoopRunsCheckOnNudge(t *testing.T) { + t.Parallel() + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _ = json.NewEncoder(w).Encode(serverVersionResponse{Version: "6.2.1"}) + })) + defer srv.Close() + + u := New(Config{CurrentVersion: "6.2.0", PulseURL: srv.URL}) + // Keep the initial check and the hourly ticker out of the way so the only + // thing that can trigger a check is the nudge. + u.initialDelay = time.Hour + updated := make(chan string, 1) + u.performUpdateFn = func(_ context.Context, targetVersion string) error { + updated <- targetVersion + return nil + } + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := make(chan struct{}) + go func() { + defer close(done) + u.RunLoop(ctx) + }() + + u.NudgeVersion("6.2.1") + + select { + case targetVersion := <-updated: + if targetVersion != "6.2.1" { + t.Fatalf("performUpdate target = %q, want %q", targetVersion, "6.2.1") + } + case <-time.After(10 * time.Second): + t.Fatal("nudge did not trigger an update check") + } + + cancel() + <-done +} diff --git a/internal/api/agent_ingest.go b/internal/api/agent_ingest.go index 86a196541..ab17a305d 100644 --- a/internal/api/agent_ingest.go +++ b/internal/api/agent_ingest.go @@ -36,6 +36,18 @@ var configSigningState struct { // UnifiedAgentHandlers manages ingest from the runtime-side Unified Agent module of pulse-agent. type UnifiedAgentHandlers struct { baseAgentHandlers + + // serverVersion is echoed on report acknowledgements so agents can spot a + // server upgrade on their next report instead of their next hourly update + // check. Empty (the default) omits it from acks. + serverVersion string +} + +// SetServerVersion supplies the running server version to include on report +// acknowledgements. Call once at wiring time; an empty version keeps acks +// version-free. +func (h *UnifiedAgentHandlers) SetServerVersion(version string) { + h.serverVersion = strings.TrimSpace(version) } func trimUnifiedAgentRoutePath(path string) string { @@ -109,6 +121,9 @@ func (h *UnifiedAgentHandlers) HandleReport(w http.ResponseWriter, r *http.Reque "osName": host.OSName, "osVersion": host.OSVersion, } + if h.serverVersion != "" { + resp["serverVersion"] = h.serverVersion + } // Only include config if there are actual overrides if serverConfig.CommandsEnabled != nil { diff --git a/internal/api/contract_test.go b/internal/api/contract_test.go index e276aab93..0068073f7 100644 --- a/internal/api/contract_test.go +++ b/internal/api/contract_test.go @@ -22413,3 +22413,56 @@ func TestCleanupTenantReleasesPerTenantResourceStore(t *testing.T) { t.Error("Router must expose ShutdownResourceStores for shutdown-time release") } } + +func postUnifiedAgentReport(t *testing.T, handler *UnifiedAgentHandlers) map[string]any { + t.Helper() + + report := agentshost.Report{ + Agent: agentshost.AgentInfo{ + ID: "agent-ack", + Version: "1.0.0", + }, + Host: agentshost.HostInfo{ + ID: "machine-ack", + Hostname: "host-ack.local", + Platform: "linux", + }, + Timestamp: time.Now().UTC(), + } + body, err := json.Marshal(report) + if err != nil { + t.Fatalf("marshal report: %v", err) + } + + req := httptest.NewRequest(http.MethodPost, "/api/agents/agent/report", bytes.NewReader(body)) + rec := httptest.NewRecorder() + handler.HandleReport(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("status = %d, want 200: %s", rec.Code, rec.Body.String()) + } + + var ack map[string]any + if err := json.Unmarshal(rec.Body.Bytes(), &ack); err != nil { + t.Fatalf("unmarshal ack: %v", err) + } + return ack +} + +func TestUnifiedAgentHandlers_HandleReportAckCarriesServerVersion(t *testing.T) { + handler, _ := newUnifiedAgentHandlers(t, nil) + handler.SetServerVersion(" 6.2.1 ") + + ack := postUnifiedAgentReport(t, handler) + if got, ok := ack["serverVersion"].(string); !ok || got != "6.2.1" { + t.Fatalf("ack serverVersion = %v, want trimmed %q", ack["serverVersion"], "6.2.1") + } +} + +func TestUnifiedAgentHandlers_HandleReportAckOmitsEmptyServerVersion(t *testing.T) { + handler, _ := newUnifiedAgentHandlers(t, nil) + + ack := postUnifiedAgentReport(t, handler) + if _, present := ack["serverVersion"]; present { + t.Fatalf("ack unexpectedly carries serverVersion: %v", ack["serverVersion"]) + } +} diff --git a/internal/api/host_agent_removal_lifecycle_integration_test.go b/internal/api/host_agent_removal_lifecycle_integration_test.go index 0e02b0ad2..f0b07d764 100644 --- a/internal/api/host_agent_removal_lifecycle_integration_test.go +++ b/internal/api/host_agent_removal_lifecycle_integration_test.go @@ -178,6 +178,17 @@ func TestHostAgentRemovalLifecycleThroughAuthenticatedRouterAndRestart(t *testin if targetID == "" || keeperID == "" || targetID == keeperID { t.Fatalf("active IDs target=%q keeper=%q, want distinct", targetID, keeperID) } + // Report acks from the live router echo the server version it was wired + // with, so an enrolled agent can spot a server upgrade on its next report. + var targetAck struct { + ServerVersion string `json:"serverVersion"` + } + if err := json.Unmarshal([]byte(targetBody), &targetAck); err != nil { + t.Fatalf("decode target ack: %v", err) + } + if targetAck.ServerVersion != "6.1.1" { + t.Fatalf("target ack serverVersion = %q, want %q", targetAck.ServerVersion, "6.1.1") + } deleteRec := serveHostRemovalLifecycleRequest( t, diff --git a/internal/api/router.go b/internal/api/router.go index 841982b4e..9d3c65d6b 100644 --- a/internal/api/router.go +++ b/internal/api/router.go @@ -536,6 +536,7 @@ func (r *Router) setupRoutes() { r.dockerAgentHandlers = NewDockerAgentHandlers(r.mtMonitor, r.monitor, r.wsHub, r.config) r.kubernetesAgentHandlers = NewKubernetesAgentHandlers(r.mtMonitor, r.monitor, r.wsHub) r.unifiedAgentHandlers = NewUnifiedAgentHandlers(r.mtMonitor, r.monitor, r.wsHub) + r.unifiedAgentHandlers.SetServerVersion(r.serverVersion) r.kubernetesAgentHandlers.SetRecoveryIngestor(r.recoveryHandlers) r.resourceHandlers = NewResourceHandlers(r.config) actionOrgChecker := NewAuthorizationChecker(NewMultiTenantOrganizationLoader(r.multiTenant)) diff --git a/internal/hostagent/agent.go b/internal/hostagent/agent.go index 6128cb53b..6d50fb60c 100644 --- a/internal/hostagent/agent.go +++ b/internal/hostagent/agent.go @@ -92,6 +92,12 @@ type Config struct { UpdateStatus func() agentupdate.Status ModuleStatus func() []agentshost.ModuleStatus + // 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) + Collector SystemCollector // Optional: override default system information collector (for testing) // DockerContainerUpdater bridges typed container update operations to the @@ -1333,9 +1339,10 @@ func (a *Agent) sendReportToDestination(ctx context.Context, report agentshost.R // Parse response to check for server-side config overrides var reportResp struct { - Success bool `json:"success"` - AgentID string `json:"agentId"` - Config *struct { + Success bool `json:"success"` + AgentID string `json:"agentId"` + ServerVersion string `json:"serverVersion"` + Config *struct { CommandsEnabled *bool `json:"commandsEnabled"` } `json:"config,omitempty"` } @@ -1345,6 +1352,14 @@ func (a *Agent) sendReportToDestination(ctx context.Context, report agentshost.R 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 != "" { diff --git a/internal/hostagent/send_report_test.go b/internal/hostagent/send_report_test.go index 97c77ad64..9c0f783d5 100644 --- a/internal/hostagent/send_report_test.go +++ b/internal/hostagent/send_report_test.go @@ -366,3 +366,97 @@ func TestAgentProcess_SuccessLogsUnifiedAgentReport(t *testing.T) { t.Fatalf("expected success log to mention Unified Agent report, got %q", logBuf.String()) } } + +func newServerVersionAckServer(body string) *httptest.Server { + return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(body)) + })) +} + +func TestAgentSendReport_SurfacesAckServerVersion(t *testing.T) { + t.Parallel() + + server := newServerVersionAckServer(`{"success":true,"agentId":"agent-1","serverVersion":"6.2.1"}`) + defer server.Close() + + var seen []string + agent := &Agent{ + cfg: Config{ + APIToken: "test-token", + OnServerVersion: func(version string) { seen = append(seen, version) }, + }, + httpClient: server.Client(), + trimmedPulseURL: server.URL, + } + + if err := agent.sendReport(context.Background(), agentshost.Report{ + Agent: agentshost.AgentInfo{ID: "agent-1"}, + Host: agentshost.HostInfo{Hostname: "test-host"}, + }); err != nil { + t.Fatalf("sendReport: %v", err) + } + + if len(seen) != 1 || seen[0] != "6.2.1" { + t.Fatalf("OnServerVersion calls = %v, want exactly [6.2.1]", seen) + } +} + +func TestAgentSendReport_SkipsCallbackWithoutAckServerVersion(t *testing.T) { + t.Parallel() + + server := newServerVersionAckServer(`{"success":true,"agentId":"agent-1"}`) + defer server.Close() + + called := false + agent := &Agent{ + cfg: Config{ + APIToken: "test-token", + OnServerVersion: func(string) { called = true }, + }, + httpClient: server.Client(), + trimmedPulseURL: server.URL, + } + + if err := agent.sendReport(context.Background(), agentshost.Report{ + Agent: agentshost.AgentInfo{ID: "agent-1"}, + Host: agentshost.HostInfo{Hostname: "test-host"}, + }); err != nil { + t.Fatalf("sendReport: %v", err) + } + + if called { + t.Fatal("OnServerVersion called for an ack without serverVersion") + } +} + +func TestAgentSendReport_ObserverAckNeverInvokesCallback(t *testing.T) { + t.Parallel() + + // Observer destinations may run any Pulse version; only the authoritative + // server may steer this agent's updater. + server := newServerVersionAckServer(`{"success":true,"agentId":"agent-1","serverVersion":"9.9.9"}`) + defer server.Close() + + called := false + agent := &Agent{ + cfg: Config{ + APIToken: "test-token", + OnServerVersion: func(string) { called = true }, + }, + httpClient: server.Client(), + trimmedPulseURL: server.URL, + } + + report := agentshost.Report{ + Agent: agentshost.AgentInfo{ID: "agent-1"}, + Host: agentshost.HostInfo{Hostname: "test-host"}, + } + if err := agent.sendReportToDestination(context.Background(), report, server.URL, "observer-token", server.Client(), false); err != nil { + t.Fatalf("sendReportToDestination: %v", err) + } + + if called { + t.Fatal("OnServerVersion called from a non-authoritative destination ack") + } +}