diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md index 06862e071c..cf047d13ba 100644 --- a/docs/release-control/v6/internal/subsystems/monitoring.md +++ b/docs/release-control/v6/internal/subsystems/monitoring.md @@ -44,6 +44,15 @@ backup), missing/malformed config or an unverifiable lock defers the command. Known locks in node inventory, cluster resources or detailed status suppress polling enrichment even when status fails and recent agent evidence exists. +The VM config body must finish within the existing response-size and request +bounds and contain exactly one complete JSON value. An unlocked JSON prefix +does not establish lock clearance: a short Content-Length body, stalled suffix, +extra JSON/garbage or chunked body exceeding 8 MiB yields `lock-unverified` both +before dispatch and after a completed guest command. Post-command verification +holds per-guest admission until body completion; its failure discards the +payload. Complete responses at the limit, including trailing whitespace, remain +valid. Ordinary PVE CPU/I/O observations remain independent of this deferral. + Admission is held through the complete response, not just headers. A transport failure, timeout, wrong-command-ID response or incomplete body defers every command for that guest for a minute; callers neither immediately retry nor @@ -63,6 +72,12 @@ resource lock with failed status → node fallback → unlocked resumption throu the real client/builders and History. The client controls cover every read, unknown lock state, diagnostic overlap, aliases/source/VM isolation, in-flight lock changes, incomplete bodies, cooldown and absence of failover replay. +`TestGuestAgentConfigResponseMustBeComplete` covers every guest read before and +after dispatch and subsequent healthy resumption; the stalled-prefix and +post-config-body controls pin completion and cross-client admission. +`TestUnverifiedVMConfigPreservesStatusCounterObservations` and the monitoring +lifecycle pin live counter presence and rejection of cached disk/memory History +refresh during an incomplete lock read, not just a decoder error. These reproduce missing Pulse safety controls, **not** the reported native PVE/QGA command-ID collision or successful backup thaw. HTTP lock checks cannot atomically reserve PVE's serial channel against a backup beginning afterwards, diff --git a/internal/monitoring/guest_agent_backup_integration_test.go b/internal/monitoring/guest_agent_backup_integration_test.go index 3afe1c8a27..349103ec48 100644 --- a/internal/monitoring/guest_agent_backup_integration_test.go +++ b/internal/monitoring/guest_agent_backup_integration_test.go @@ -26,7 +26,10 @@ func testGuestAgentBackupMonitoringLifecycle(t *testing.T) { w.Header().Set("Content-Type", "application/json") switch { case strings.HasSuffix(r.URL.Path, "/config"): - if p == 1 || p == 2 { + if p == 4 { + w.Header().Set("Content-Length", "1000") + fmt.Fprint(w, `{"data":{}}`) + } else if p == 1 || p == 2 { fmt.Fprint(w, `{"data":{"lock":"backup"}}`) } else { fmt.Fprint(w, `{"data":{}}`) @@ -98,6 +101,22 @@ func testGuestAgentBackupMonitoringLifecycle(t *testing.T) { memKey := guestMemoryCacheKey("fixture", "node", 105) originalMetadata := m.guestMetadataCache[metadataKey] originalMemory := m.vmAgentMemCache[memKey] + // A syntactically unlocked prefix is not a completed config observation. + // Warm guest caches must stay labelled last-known rather than becoming new + // disk/memory History samples when the lock response is incomplete. + phase.Store(4) + unverified, _ := build() + if unverified.ID != initial.ID || unverified.GuestAgentStatus != "deferred" || unverified.DiskStatusReason != "prev-lock-unverified" || unverified.Disk.Used != initial.Disk.Used || unverified.Memory.Used != initial.Memory.Used { + t.Fatalf("incomplete lock response lost truthful continuity: %#v", unverified) + } + if calls.Load() != before || !reflect.DeepEqual(m.guestMetadataCache[metadataKey], originalMetadata) || !reflect.DeepEqual(m.vmAgentMemCache[memKey], originalMemory) { + t.Fatal("incomplete lock response sent a guest command or renewed caches") + } + for metric, want := range map[string]int{"cpu": 2, "memory": 1, "memoryused": 1, "disk": 1} { + if got := len(m.metricsHistory.GetGuestMetrics(identity, metric, time.Hour)); got != want { + t.Errorf("incomplete lock %s history points = %d, want %d", metric, got, want) + } + } phase.Store(1) res.CPU = 0.2 locked, source := build() @@ -113,7 +132,7 @@ func testGuestAgentBackupMonitoringLifecycle(t *testing.T) { if !reflect.DeepEqual(m.guestMetadataCache[metadataKey], originalMetadata) || !reflect.DeepEqual(m.vmAgentMemCache[memKey], originalMemory) { t.Fatal("lock renewed/replaced cached observations") } - for metric, want := range map[string]int{"cpu": 2, "memory": 1, "memoryused": 1, "disk": 1} { + for metric, want := range map[string]int{"cpu": 3, "memory": 1, "memoryused": 1, "disk": 1} { if got := len(m.metricsHistory.GetGuestMetrics(identity, metric, time.Hour)); got != want { t.Errorf("%s history points = %d, want %d", metric, got, want) } diff --git a/pkg/proxmox/client.go b/pkg/proxmox/client.go index dec68ba461..12750b34e0 100644 --- a/pkg/proxmox/client.go +++ b/pkg/proxmox/client.go @@ -1596,11 +1596,18 @@ func (c *Client) GetVMConfig(ctx context.Context, node string, vmid int) (map[st } defer resp.Body.Close() + // Guest-agent admission uses this response as authoritative lock evidence. + // Decoding only the first JSON value can accept an unlocked prefix before a + // truncated, stalled, oversized or malformed response has been received. + body, err := readResponseBodyLimited(resp.Body) + if err != nil { + return nil, fmt.Errorf("read VM configuration: %w", err) + } var result struct { Data map[string]interface{} `json:"data"` } - if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + if err := json.Unmarshal(body, &result); err != nil { return nil, err } diff --git a/pkg/proxmox/guest_agent_lock_response_test.go b/pkg/proxmox/guest_agent_lock_response_test.go new file mode 100644 index 0000000000..58f5cd0083 --- /dev/null +++ b/pkg/proxmox/guest_agent_lock_response_test.go @@ -0,0 +1,164 @@ +package proxmox + +import ( + "context" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + "time" +) + +func TestGuestAgentConfigResponseMustBeComplete(t *testing.T) { + for _, test := range []struct { + name string + write func(http.ResponseWriter) + }{ + {"truncated", func(w http.ResponseWriter) { + w.Header().Set("Content-Length", "1000") + fmt.Fprint(w, `{"data":{}}`) + }}, + {"trailing-json", func(w http.ResponseWriter) { + fmt.Fprint(w, `{"data":{}}{"data":{"lock":"backup"}}`) + }}, + {"trailing-garbage", func(w http.ResponseWriter) { + fmt.Fprint(w, `{"data":{}}incomplete-response`) + }}, + {"chunked-over-limit", func(w http.ResponseWriter) { + fmt.Fprint(w, `{"data":{}}`) + w.(http.Flusher).Flush() // No declared length: enforce the read boundary. + fmt.Fprint(w, strings.Repeat(" ", int(maxResponseBodyBytes))) + }}, + } { + t.Run(test.name, func(t *testing.T) { + for name, read := range backupAgentReads() { + for _, postCommand := range []bool{false, true} { + t.Run(fmt.Sprintf("%s/post=%t", name, postCommand), func(t *testing.T) { + var configs, commands atomic.Int32 + var healthy atomic.Bool + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if strings.HasSuffix(r.URL.Path, "/config") { + if configs.Add(1) == 1 && postCommand || healthy.Load() { + fmt.Fprint(w, `{"data":{}}`) + } else { + test.write(w) + } + return + } + commands.Add(1) + backupAgentPayload(w, r) + })) + defer server.Close() + c := backupTestClient(t, server.URL) + if err := read(context.Background(), c, 105); GuestAgentDeferredReason(err) != "lock-unverified" { + t.Errorf("incomplete config was accepted: %v", err) + } + wantCommands := int32(0) + if postCommand { + wantCommands = 1 + } + if got := commands.Load(); got != wantCommands { + t.Errorf("guest commands = %d, want %d", got, wantCommands) + } + // This config failure says nothing about the completed guest + // command. A later ordinary read must recheck the lock and resume. + healthy.Store(true) + if err := read(context.Background(), c, 105); err != nil { + t.Fatalf("complete unlocked config did not resume: %v", err) + } + if got := commands.Load(); got != wantCommands+1 { + t.Errorf("resumed guest commands = %d, want %d", got, wantCommands+1) + } + }) + } + } + }) + } +} + +func TestGuestAgentConfigPrefixDoesNotAdmitCommand(t *testing.T) { + var commands atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if strings.HasSuffix(r.URL.Path, "/config") { + fmt.Fprint(w, `{"data":{}}`) + w.(http.Flusher).Flush() + <-r.Context().Done() // The body never completes, despite a valid prefix. + return + } + commands.Add(1) + backupAgentPayload(w, r) + })) + defer server.Close() + ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond) + defer cancel() + if _, err := backupTestClient(t, server.URL).GetVMAgentInfo(ctx, "node", 105); GuestAgentDeferredReason(err) != "lock-unverified" { + t.Errorf("stalled config admitted a command: %v", err) + } + if got := commands.Load(); got != 0 { + t.Errorf("commands before config completion = %d, want zero", got) + } +} + +func TestGuestAgentAdmissionIncludesPostConfigBody(t *testing.T) { + var configs, commands atomic.Int32 + entered := make(chan struct{}) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if strings.HasSuffix(r.URL.Path, "/config") { + n := configs.Add(1) + fmt.Fprint(w, `{"data":{}}`) + if n == 2 { + w.(http.Flusher).Flush() + close(entered) + <-r.Context().Done() + } + return + } + commands.Add(1) + backupAgentPayload(w, r) + })) + defer server.Close() + c1, c2 := backupTestClient(t, server.URL), backupTestClient(t, server.URL) + ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond) + defer cancel() + done := make(chan error, 1) + go func() { + value, err := c1.GetVMFSInfo(ctx, "node", 105) + if value != nil { + t.Error("filesystem payload escaped the incomplete post-command lock check") + } + done <- err + }() + select { + case <-entered: + case err := <-done: + t.Fatalf("post-command config check never started: %v", err) + } + if _, err := c2.GetVMNetworkInterfaces(context.Background(), "node", 105); GuestAgentDeferredReason(err) != "agent-busy" { + t.Errorf("second client bypassed incomplete config admission: %v", err) + } + if err := <-done; GuestAgentDeferredReason(err) != "lock-unverified" { + t.Errorf("post-command config prefix accepted: %v", err) + } + if got := commands.Load(); got != 1 { + t.Errorf("commands during incomplete post-check = %d, want one", got) + } + if _, err := c2.GetVMFSInfo(context.Background(), "node", 105); err != nil { + t.Fatalf("admission not released after failed lock check: %v", err) + } +} + +func TestVMConfigCompleteWhitespaceAtResponseLimit(t *testing.T) { + const config = `{"data":{"name":"fixture","agent":1}}` + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + fmt.Fprint(w, config) + w.(http.Flusher).Flush() + fmt.Fprint(w, strings.Repeat(" ", int(maxResponseBodyBytes)-len(config))) + })) + defer server.Close() + got, err := backupTestClient(t, server.URL).GetVMConfig(context.Background(), "node", 105) + if err != nil || got["name"] != "fixture" || got["agent"] != float64(1) { + t.Fatalf("complete bounded config rejected or changed: %#v %v", got, err) + } +} diff --git a/pkg/proxmox/io_counters_test.go b/pkg/proxmox/io_counters_test.go index fc210c76ab..1354c757e6 100644 --- a/pkg/proxmox/io_counters_test.go +++ b/pkg/proxmox/io_counters_test.go @@ -1,7 +1,13 @@ package proxmox import ( + "context" "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" "testing" "time" ) @@ -138,3 +144,35 @@ func TestBackupLocksRetainCumulativeCounterPresence(t *testing.T) { }) } } + +func TestUnverifiedVMConfigPreservesStatusCounterObservations(t *testing.T) { + var commands atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch { + case strings.HasSuffix(r.URL.Path, "/config"): + w.Header().Set("Content-Length", "1000") + fmt.Fprint(w, `{"data":{}}`) + case strings.HasSuffix(r.URL.Path, "/status/current"): + fmt.Fprint(w, `{"data":{"status":"running","cpu":0.25,"diskread":0,"diskwrite":null,"netin":42}}`) + default: + commands.Add(1) + backupAgentPayload(w, r) + } + })) + defer server.Close() + c := backupTestClient(t, server.URL) + if _, err := c.GetVMFSInfo(context.Background(), "node", 105); GuestAgentDeferredReason(err) != "lock-unverified" { + t.Errorf("incomplete config did not defer the guest command: %v", err) + } + status, err := c.GetVMStatus(context.Background(), "node", 105) + if err != nil { + t.Fatal(err) + } + p := status.IOCounters.Effective() + if status.CPU != 0.25 || status.NetIn != 42 || status.DiskRead != 0 || status.ObservedAt.IsZero() || !p.DiskRead || !p.NetworkIn || p.DiskWrite || p.NetworkOut { + t.Fatalf("unverified guest lock suppressed/changed current PVE counters: %+v", status) + } + if got := commands.Load(); got != 0 { + t.Errorf("guest commands with incomplete config = %d, want zero", got) + } +}