mirror of
https://github.com/rcourtman/Pulse.git
synced 2026-10-23 20:25:56 +00:00
fix(proxmox): defer guest reads after uncertain HTTP failures
Preserve the actual HTTP response status rather than trusting quoted API errors. Apply the VM-wide cooldown after request timeouts and unexplained server or gateway failures, while retaining exact command-bound terminal refusals as ordinary errors. Reject ambiguous error envelopes and preserve cached guest observations alongside fresh live counters. Change-source: pulse-maintainer
This commit is contained in:
@@ -112,6 +112,26 @@ replay through another cluster endpoint. Coordination retains at most 4,096
|
||||
active/cooling endpoint-guest entries and expires cooldowns. Explicit API
|
||||
permission refusals and unsupported commands remain errors, not successes.
|
||||
|
||||
The completed-error boundary uses the actual HTTP status retained by the request
|
||||
layer, not an `API error` phrase in provider or proxy text. HTTP 408 and
|
||||
unexplained server/proxy failures (including PVE 595) retain the same per-VM
|
||||
uncertainty cooldown, even when their bodies finish: command completion is still
|
||||
unverified. Another method, diagnostic client or configured cluster alias cannot
|
||||
immediately send a command. Only complete, exact command-bound terminal QGA
|
||||
rejections under HTTP 500 (unsupported/missing command, explicit stopped agent,
|
||||
and OS-info's known missing os-release) retain ordinary error handling. Ambiguous
|
||||
or conflicting envelopes, duplicate keys, other VM/command identities, extra
|
||||
values and gateway responses quoting a rejection do not gain that exemption.
|
||||
Existing authentication hints, error text, ordinary API session/retry policy and
|
||||
endpoint health remain unchanged. Unknown server failures now defer enrichment
|
||||
for a minute rather than inviting a sequence of diagnostic reads; this is a
|
||||
safety trade-off, not evidence of a freeze cause or native recovery.
|
||||
`TestGuestAgentHTTPFailureDefersEveryRead` exercises all six guest readers,
|
||||
independent status counters and cross-client cooldown; cluster and semantic
|
||||
controls pin no failover, definitive refusals and conservative error provenance.
|
||||
The connected transport/History lifecycle and counter-receipt controls also
|
||||
cover complete server/gateway failures without refreshing old guest evidence.
|
||||
|
||||
Guest commands also prevent transport-level replay: each uses a fresh single-use
|
||||
HTTP/1 connection, with no pooled-connection or HTTP/2-stream retries and no
|
||||
redirect following. The client preserves its configured TLS verification or
|
||||
|
||||
@@ -2976,7 +2976,7 @@ func TestGuestAgentBackupMonitoringContract(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestGuestAgentTransportMonitoringContract(t *testing.T) {
|
||||
for _, reason := range []string{"agent-redirect", "agent-transport-unverified"} {
|
||||
for _, reason := range []string{"agent-redirect", "agent-transport-unverified", "agent-completion-unverified"} {
|
||||
for _, prefix := range []string{"", "prev-"} {
|
||||
if !guestAgentDiskDeferred(prefix + reason) {
|
||||
t.Errorf("transport uncertainty %q is not labelled deferred", prefix+reason)
|
||||
|
||||
@@ -18,7 +18,7 @@ import (
|
||||
)
|
||||
|
||||
func testGuestAgentTransportDeferralKeepsLastKnownHistory(t *testing.T) {
|
||||
for _, kind := range []string{"lost reply", "redirect"} {
|
||||
for _, kind := range []string{"lost reply", "redirect", "server error", "gateway error"} {
|
||||
t.Run(kind, func(t *testing.T) {
|
||||
const mib = uint64(1024 * 1024)
|
||||
var phase, calls, redirected atomic.Int32
|
||||
@@ -35,6 +35,15 @@ func testGuestAgentTransportDeferralKeepsLastKnownHistory(t *testing.T) {
|
||||
redirected.Add(1)
|
||||
}
|
||||
if phase.Load() == 1 && lost.CompareAndSwap(false, true) {
|
||||
if kind == "server error" || kind == "gateway error" {
|
||||
status := http.StatusInternalServerError
|
||||
if kind == "gateway error" {
|
||||
status = http.StatusBadGateway
|
||||
}
|
||||
w.WriteHeader(status)
|
||||
fmt.Fprint(w, "upstream unavailable")
|
||||
return
|
||||
}
|
||||
if kind == "redirect" {
|
||||
http.Redirect(w, r, r.URL.Path+"?redirect=1", http.StatusTemporaryRedirect)
|
||||
return
|
||||
|
||||
@@ -891,7 +891,7 @@ func (m *Monitor) updateVMDisksFromGuestAgentFSInfo(
|
||||
|
||||
func guestAgentDiskDeferred(reason string) bool {
|
||||
switch strings.TrimPrefix(reason, "prev-") {
|
||||
case "vm-locked", "lock-unverified", "agent-busy", "agent-cooldown", "agent-capacity", "agent-timeout", "agent-response-incomplete", "agent-redirect", "agent-transport-unverified":
|
||||
case "vm-locked", "lock-unverified", "agent-busy", "agent-cooldown", "agent-capacity", "agent-timeout", "agent-response-incomplete", "agent-redirect", "agent-transport-unverified", "agent-completion-unverified":
|
||||
return true
|
||||
}
|
||||
return false
|
||||
|
||||
@@ -636,11 +636,14 @@ func (c *Client) requestWithRetryUsingClient(ctx context.Context, method, path s
|
||||
// Preserve the established returned-error classification for callers;
|
||||
// this distinction changes only the emitted diagnostic severity.
|
||||
if resp.StatusCode == 401 || resp.StatusCode == 403 || resp.StatusCode == 595 {
|
||||
// Import errors package at top of file
|
||||
return nil, fmt.Errorf("authentication error: %w", apiErr)
|
||||
apiErr = fmt.Errorf("authentication error: %w", apiErr)
|
||||
}
|
||||
|
||||
return nil, apiErr
|
||||
return nil, &apiResponseError{
|
||||
statusCode: resp.StatusCode,
|
||||
guestCommandRejected: resp.StatusCode == http.StatusInternalServerError && guestAgentTerminalFailure(path, body),
|
||||
cause: apiErr,
|
||||
}
|
||||
}
|
||||
|
||||
if err := securityutil.LimitResponseBody(resp, maxResponseBodyBytes); err != nil {
|
||||
|
||||
@@ -186,13 +186,24 @@ func (c *Client) getGuestAgent(ctx context.Context, path, node string, vmid int)
|
||||
uncertain = true
|
||||
return nil, &guestAgentDeferredError{reason: "agent-response-incomplete", cause: err}
|
||||
}
|
||||
// Explicit refusals/unsupported commands remain ordinary errors. An
|
||||
// uncertain completion blocks every command, not just this method.
|
||||
// A complete gateway/server error can follow a consumed command. Use
|
||||
// the actual wire status, never an "API error" quoted in body text.
|
||||
// Specific terminal rejections remain errors, not successful telemetry.
|
||||
lower := strings.ToLower(err.Error())
|
||||
uncertain = !strings.Contains(lower, "api error") || ctx.Err() != nil || strings.Contains(lower, "timeout") || strings.Contains(lower, "timed out") || strings.Contains(lower, "wrong command id")
|
||||
uncertain = ctx.Err() != nil || strings.Contains(lower, "timeout") || strings.Contains(lower, "timed out") || strings.Contains(lower, "wrong command id")
|
||||
if uncertain {
|
||||
return nil, &guestAgentDeferredError{reason: "agent-timeout", cause: err}
|
||||
}
|
||||
var response *apiResponseError
|
||||
if !errors.As(err, &response) {
|
||||
uncertain = true
|
||||
return nil, &guestAgentDeferredError{reason: "agent-timeout", cause: err}
|
||||
}
|
||||
if response.statusCode == http.StatusRequestTimeout ||
|
||||
(response.statusCode >= 500 && !response.guestCommandRejected) {
|
||||
uncertain = true
|
||||
return nil, &guestAgentDeferredError{reason: "agent-completion-unverified", cause: err}
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
// Hold admission until the entire bounded response has arrived. Returning
|
||||
|
||||
@@ -0,0 +1,315 @@
|
||||
package proxmox
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// A complete HTTP error is not necessarily a completed QGA command: an
|
||||
// intermediary may fail after the command was accepted. Error text cannot
|
||||
// turn a gateway status into a definitive refusal.
|
||||
func TestGuestAgentHTTPFailureDefersEveryRead(t *testing.T) {
|
||||
cases := []struct {
|
||||
name string
|
||||
status int
|
||||
body string
|
||||
}{
|
||||
{"request-timeout", 408, "request did not finish"},
|
||||
{"plain-server-error", 500, `{"data":null}`},
|
||||
{"gateway", 502, "upstream unavailable"},
|
||||
{"service-unavailable", 503, "temporary failure"},
|
||||
{"gateway-deadline", 504, "<html>request did not finish</html>"},
|
||||
{"node-proxy", 595, "no ticket"},
|
||||
{"unexplained-not-implemented", 501, "upstream unavailable"},
|
||||
{"gateway-quoting-refusal", 502, "upstream API error 403: permission denied"},
|
||||
{"gateway-quoting-unsupported", 502, "unsupported command: guest-get-osinfo"},
|
||||
{"server-quoting-refusal", 500, "upstream API error 403: permission denied"},
|
||||
}
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
for name, first := range backupAgentReads() {
|
||||
t.Run(name, func(t *testing.T) {
|
||||
var calls atomic.Int32
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
switch {
|
||||
case strings.HasSuffix(r.URL.Path, "/config"):
|
||||
fmt.Fprint(w, `{"data":{}}`)
|
||||
case strings.HasSuffix(r.URL.Path, "/status/current"):
|
||||
fmt.Fprint(w, `{"data":{"status":"running","cpu":0.25,"diskread":4096}}`)
|
||||
default:
|
||||
if calls.Add(1) == 1 {
|
||||
w.WriteHeader(tc.status)
|
||||
fmt.Fprint(w, tc.body)
|
||||
return
|
||||
}
|
||||
backupAgentPayload(w, r)
|
||||
}
|
||||
}))
|
||||
defer server.Close()
|
||||
c := backupTestClient(t, server.URL)
|
||||
err := first(context.Background(), c, 105)
|
||||
if !errors.Is(err, ErrGuestAgentDeferred) || GuestAgentDeferredReason(err) != "agent-completion-unverified" {
|
||||
t.Errorf("unknown completion = %v, want completion-unverified deferral", err)
|
||||
}
|
||||
// A separately constructed diagnostic cannot follow with any command.
|
||||
diagnostic := backupTestClient(t, server.URL)
|
||||
for later, read := range backupAgentReads() {
|
||||
if err := read(context.Background(), diagnostic, 105); GuestAgentDeferredReason(err) != "agent-cooldown" {
|
||||
t.Errorf("next %s = %v, want VM-wide cooldown", later, err)
|
||||
}
|
||||
}
|
||||
status, err := c.GetVMStatus(context.Background(), "node", 105)
|
||||
if err != nil || status.CPU != 0.25 || status.DiskRead != 4096 {
|
||||
t.Errorf("live non-agent counters lost: %+v %v", status, err)
|
||||
}
|
||||
if calls.Load() != 1 {
|
||||
t.Errorf("wire guest commands = %d, want one", calls.Load())
|
||||
}
|
||||
})
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestGuestAgentHTTPFailureDoesNotFailOver(t *testing.T) {
|
||||
var calls atomic.Int32
|
||||
handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
switch {
|
||||
case strings.HasSuffix(r.URL.Path, "/nodes"):
|
||||
fmt.Fprint(w, `{"data":[]}`)
|
||||
case strings.HasSuffix(r.URL.Path, "/config"):
|
||||
fmt.Fprint(w, `{"data":{}}`)
|
||||
default:
|
||||
calls.Add(1)
|
||||
w.WriteHeader(502)
|
||||
fmt.Fprint(w, "upstream unavailable")
|
||||
}
|
||||
})
|
||||
a, b := httptest.NewServer(handler), httptest.NewServer(handler)
|
||||
defer a.Close()
|
||||
defer b.Close()
|
||||
cc := NewClusterClient("http-uncertainty", ClientConfig{Host: a.URL, TokenName: "fixture@pve!pulse", TokenValue: "fixture"}, []string{a.URL, b.URL}, nil)
|
||||
if _, err := cc.GetVMFSInfo(context.Background(), "node", 105); GuestAgentDeferredReason(err) != "agent-completion-unverified" {
|
||||
t.Errorf("first command not deferred: %v", err)
|
||||
}
|
||||
for _, endpoint := range []string{a.URL, b.URL} {
|
||||
if _, err := backupTestClient(t, endpoint).GetVMAgentInfo(context.Background(), "migrated", 105); GuestAgentDeferredReason(err) != "agent-cooldown" {
|
||||
t.Errorf("alias escaped cooldown: %v", err)
|
||||
}
|
||||
}
|
||||
if calls.Load() != 1 {
|
||||
t.Errorf("cluster wire commands = %d, want one", calls.Load())
|
||||
}
|
||||
for endpoint, healthy := range cc.GetHealthStatus() {
|
||||
if !healthy {
|
||||
t.Errorf("VM error poisoned endpoint %s", endpoint)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestGuestAgentHTTPExplicitRefusalsRemainErrors(t *testing.T) {
|
||||
cases := []struct {
|
||||
name string
|
||||
status int
|
||||
body string
|
||||
}{
|
||||
{"invalid-request", 400, "invalid argument"},
|
||||
{"unauthorized", 401, "unauthorized"},
|
||||
{"forbidden", 403, "permission denied"},
|
||||
{"missing-command", 404, "command not found"},
|
||||
{"rate-limited", 429, "too many requests"},
|
||||
{"unsupported-plain", 500, "unsupported command: guest-get-osinfo"},
|
||||
{"unsupported-message", 500, `{"data":null,"message":"unsupported command: guest-get-osinfo"}`},
|
||||
{"unsupported-errors", 500, `{"data":null,"errors":{"message":"unsupported command: guest-get-osinfo"}}`},
|
||||
{"missing-os-release", 500, `{"errors":{"message":"guest agent command failed: Failed to open file '/etc/os-release': No such file or directory"}}`},
|
||||
{"missing-usr-os-release", 500, "guest agent command failed: Failed to open file '/usr/lib/os-release': No such file or directory"},
|
||||
{"missing-os-release-qmp-wrapper", 500, "VM 105 qmp command 'guest-get-osinfo' failed - Failed to open file '/etc/os-release': No such file or directory"},
|
||||
}
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
var calls 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":{}}`)
|
||||
return
|
||||
}
|
||||
if calls.Add(1) == 1 {
|
||||
w.WriteHeader(tc.status)
|
||||
fmt.Fprint(w, tc.body)
|
||||
return
|
||||
}
|
||||
backupAgentPayload(w, r)
|
||||
}))
|
||||
defer server.Close()
|
||||
c := backupTestClient(t, server.URL)
|
||||
if _, err := c.GetVMAgentInfo(context.Background(), "node", 105); err == nil || errors.Is(err, ErrGuestAgentDeferred) || !strings.Contains(err.Error(), fmt.Sprintf("API error %d", tc.status)) {
|
||||
t.Fatalf("explicit refusal lost: %v", err)
|
||||
}
|
||||
if _, err := c.GetVMFSInfo(context.Background(), "node", 105); err != nil {
|
||||
t.Fatalf("known refusal poisoned healthy command: %v", err)
|
||||
}
|
||||
if calls.Load() != 2 {
|
||||
t.Errorf("refused/healthy commands = %d", calls.Load())
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestGuestAgentHTTPAmbiguousFailuresDoNotClearAdmission(t *testing.T) {
|
||||
cases := []struct {
|
||||
name string
|
||||
body string
|
||||
}{
|
||||
{"wrong-command", "unsupported command: guest-get-osinfo"},
|
||||
{"proxy-prefix", "upstream rejected: unsupported command: guest-get-fsinfo"},
|
||||
{"wrong-vm", "VM 106 qmp command 'guest-get-fsinfo' failed - unsupported command: guest-get-fsinfo"},
|
||||
{"conflicting-messages", `{"message":"unsupported command: guest-get-fsinfo","errors":{"message":"upstream unavailable"}}`},
|
||||
{"duplicate-message", `{"message":"upstream unavailable","message":"unsupported command: guest-get-fsinfo"}`},
|
||||
{"escaped-duplicate-message", `{"message":"upstream unavailable","\u006dessage":"unsupported command: guest-get-fsinfo"}`},
|
||||
{"duplicate-errors", `{"errors":{"message":"upstream unavailable"},"errors":{"message":"unsupported command: guest-get-fsinfo"}}`},
|
||||
{"duplicate-nested-message", `{"errors":{"message":"upstream unavailable","message":"unsupported command: guest-get-fsinfo"}}`},
|
||||
{"success-and-error", `{"data":{"result":[]},"message":"unsupported command: guest-get-fsinfo"}`},
|
||||
{"unrecognised-detail", `{"errors":{"message":"unsupported command: guest-get-fsinfo","other":"unavailable"}}`},
|
||||
{"trailing-value", `{"message":"unsupported command: guest-get-fsinfo"} {}`},
|
||||
{"truncated-json", `{"message":"unsupported command: guest-get-fsinfo"`},
|
||||
}
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
var calls 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":{}}`)
|
||||
return
|
||||
}
|
||||
calls.Add(1)
|
||||
w.WriteHeader(500)
|
||||
fmt.Fprint(w, tc.body)
|
||||
}))
|
||||
defer server.Close()
|
||||
c := backupTestClient(t, server.URL)
|
||||
if _, err := c.GetVMFSInfo(context.Background(), "node", 105); GuestAgentDeferredReason(err) != "agent-completion-unverified" {
|
||||
t.Errorf("ambiguous error released admission: %v", err)
|
||||
}
|
||||
if _, err := c.GetVMAgentInfo(context.Background(), "node", 105); GuestAgentDeferredReason(err) != "agent-cooldown" {
|
||||
t.Errorf("ambiguous error allowed next command: %v", err)
|
||||
}
|
||||
if calls.Load() != 1 {
|
||||
t.Errorf("wire commands = %d, want one", calls.Load())
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestGuestAgentHTTPTerminalFailuresRemainCommandBound(t *testing.T) {
|
||||
cases := []struct{ name, command, body string }{
|
||||
{"fs-not-found", "get-fsinfo", "The command guest-get-fsinfo has not been found"},
|
||||
{"network-not-supported", "network-get-interfaces", "unsupported command: guest-network-get-interfaces"},
|
||||
{"version-not-supported", "info", "unsupported command: guest-info"},
|
||||
{"file-read-not-supported", "file-read", "unsupported command: guest-file-read"},
|
||||
{"pve-wrapper", "get-fsinfo", "VM 105 qmp command 'guest-get-fsinfo' failed - unsupported command: guest-get-fsinfo"},
|
||||
{"not-running", "get-fsinfo", `{"data":null,"message":"QEMU guest agent is not running"}`},
|
||||
}
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
var calls 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":{}}`)
|
||||
return
|
||||
}
|
||||
if calls.Add(1) == 1 {
|
||||
w.WriteHeader(500)
|
||||
fmt.Fprint(w, tc.body)
|
||||
return
|
||||
}
|
||||
backupAgentPayload(w, r)
|
||||
}))
|
||||
defer server.Close()
|
||||
c := backupTestClient(t, server.URL)
|
||||
if _, err := c.get(context.Background(), "/nodes/node/qemu/105/agent/"+tc.command); err == nil || errors.Is(err, ErrGuestAgentDeferred) {
|
||||
t.Fatalf("known terminal failure lost: %v", err)
|
||||
}
|
||||
if _, err := c.GetVMFSInfo(context.Background(), "node", 105); err != nil {
|
||||
t.Fatalf("known completed command blocked healthy command: %v", err)
|
||||
}
|
||||
if calls.Load() != 2 {
|
||||
t.Errorf("refused/healthy commands = %d", calls.Load())
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestGuestAgentHTTPOrdinaryRequestPreservesStatusAndError(t *testing.T) {
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.WriteHeader(502)
|
||||
fmt.Fprint(w, "API error 403: permission denied")
|
||||
}))
|
||||
defer server.Close()
|
||||
c := backupTestClient(t, server.URL)
|
||||
_, err := c.get(context.Background(), "/nodes")
|
||||
var response *apiResponseError
|
||||
if !errors.As(err, &response) || response.statusCode != 502 || response.guestCommandRejected || err.Error() != "API error 502: API error 403: permission denied" {
|
||||
t.Fatalf("wire status/ordinary diagnostic lost: %v", err)
|
||||
}
|
||||
if errors.Is(err, ErrGuestAgentDeferred) {
|
||||
t.Fatal("ordinary API acquired guest uncertainty")
|
||||
}
|
||||
}
|
||||
|
||||
func TestGuestAgentHTTPUncertaintyExpiresAndKeepsOtherVMsIndependent(t *testing.T) {
|
||||
var calls 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":{}}`)
|
||||
return
|
||||
}
|
||||
if calls.Add(1) == 1 {
|
||||
w.WriteHeader(502)
|
||||
fmt.Fprint(w, "upstream unavailable")
|
||||
return
|
||||
}
|
||||
backupAgentPayload(w, r)
|
||||
}))
|
||||
defer server.Close()
|
||||
c := backupTestClient(t, server.URL)
|
||||
if _, err := c.GetVMFSInfo(context.Background(), "node", 105); GuestAgentDeferredReason(err) != "agent-completion-unverified" {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := c.GetVMFSInfo(context.Background(), "node", 106); err != nil {
|
||||
t.Fatalf("other VM blocked: %v", err)
|
||||
}
|
||||
if _, err := c.GetVMAgentInfo(context.Background(), "node", 105); GuestAgentDeferredReason(err) != "agent-cooldown" {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// Advance only this synthetic cooldown, not wall time or a native target.
|
||||
guestAgentGuards.Lock()
|
||||
key := guestAgentGuardKey{endpoint: c.baseURL, vmid: 105}
|
||||
entry, ok := guestAgentGuards.entries[key]
|
||||
if ok {
|
||||
entry.until = time.Now().Add(-time.Second)
|
||||
guestAgentGuards.entries[key] = entry
|
||||
}
|
||||
guestAgentGuards.Unlock()
|
||||
if !ok {
|
||||
t.Fatal("uncertain VM had no retained admission")
|
||||
}
|
||||
if _, err := c.GetVMFSInfo(context.Background(), "node", 105); err != nil {
|
||||
t.Fatalf("unlocked completed response did not resume: %v", err)
|
||||
}
|
||||
guestAgentGuards.Lock()
|
||||
_, retained := guestAgentGuards.entries[key]
|
||||
guestAgentGuards.Unlock()
|
||||
if retained {
|
||||
t.Fatal("healthy resumption retained cooldown")
|
||||
}
|
||||
if calls.Load() != 3 {
|
||||
t.Errorf("wire commands = %d, want uncertain/other/resumed", calls.Load())
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,128 @@
|
||||
package proxmox
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"strings"
|
||||
)
|
||||
|
||||
// Keep the wire status separate from the existing diagnostic text. In
|
||||
// particular, a proxy's error body can quote a different HTTP status.
|
||||
type apiResponseError struct {
|
||||
statusCode int
|
||||
guestCommandRejected bool
|
||||
cause error
|
||||
}
|
||||
|
||||
func (e *apiResponseError) Error() string { return e.cause.Error() }
|
||||
func (e *apiResponseError) Unwrap() error { return e.cause }
|
||||
|
||||
// Only specific, complete terminal failures from the command endpoint allow
|
||||
// another command after an HTTP 500. An unexplained 5xx, even with a complete
|
||||
// body, does not establish that a serial QGA command has finished.
|
||||
func guestAgentTerminalFailure(path string, body []byte) bool {
|
||||
_, vmid, ok := guestAgentPath(path)
|
||||
if !ok {
|
||||
return false
|
||||
}
|
||||
message, ok := guestAgentErrorMessage(body)
|
||||
if !ok {
|
||||
return false
|
||||
}
|
||||
commandPath := strings.SplitN(path, "?", 2)[0]
|
||||
command := "guest-" + commandPath[strings.LastIndex(commandPath, "/")+1:]
|
||||
message = strings.TrimPrefix(message, fmt.Sprintf("VM %d qmp command '%s' failed - ", vmid, command))
|
||||
switch message {
|
||||
case "QEMU guest agent is not running",
|
||||
"unsupported command: " + command,
|
||||
"The command " + command + " has not been found":
|
||||
return true
|
||||
}
|
||||
// Preserve the existing OpenBSD OS-info skip rather than repeatedly probing
|
||||
// its missing os-release file. Do not apply it to another guest command.
|
||||
if command == "guest-get-osinfo" {
|
||||
message = strings.TrimPrefix(message, "guest agent command failed: ")
|
||||
for _, filename := range []string{"/etc/os-release", "/usr/lib/os-release"} {
|
||||
if message == "Failed to open file '"+filename+"': No such file or directory" {
|
||||
return true
|
||||
}
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func guestAgentErrorMessage(body []byte) (string, bool) {
|
||||
body = bytes.TrimSpace(body)
|
||||
if len(body) == 0 {
|
||||
return "", false
|
||||
}
|
||||
if body[0] != '{' {
|
||||
return string(body), true // Still requires an exact, command-bound match.
|
||||
}
|
||||
fields, ok := guestAgentErrorFields(body, "data", "message", "errors")
|
||||
if !ok {
|
||||
return "", false
|
||||
}
|
||||
if data, exists := fields["data"]; exists && string(bytes.TrimSpace(data)) != "null" {
|
||||
return "", false
|
||||
}
|
||||
message, hasMessage := fields["message"]
|
||||
if nested, hasErrors := fields["errors"]; hasErrors {
|
||||
if hasMessage {
|
||||
return "", false // Conflicting or duplicate failure evidence.
|
||||
}
|
||||
errors, ok := guestAgentErrorFields(nested, "message")
|
||||
if !ok {
|
||||
return "", false
|
||||
}
|
||||
message = errors["message"]
|
||||
}
|
||||
var text string
|
||||
if json.Unmarshal(message, &text) != nil {
|
||||
return "", false
|
||||
}
|
||||
return strings.TrimSpace(text), true
|
||||
}
|
||||
|
||||
// Reject ambiguous envelopes (including duplicate keys and trailing values)
|
||||
// instead of letting a favourable final field override other error evidence.
|
||||
func guestAgentErrorFields(body []byte, allowed ...string) (map[string]json.RawMessage, bool) {
|
||||
decoder := json.NewDecoder(bytes.NewReader(body))
|
||||
start, err := decoder.Token()
|
||||
if err != nil || start != json.Delim('{') {
|
||||
return nil, false
|
||||
}
|
||||
fields := make(map[string]json.RawMessage)
|
||||
for decoder.More() {
|
||||
key, err := decoder.Token()
|
||||
if err != nil {
|
||||
return nil, false
|
||||
}
|
||||
name, ok := key.(string)
|
||||
if !ok {
|
||||
return nil, false
|
||||
}
|
||||
permitted := false
|
||||
for _, candidate := range allowed {
|
||||
permitted = permitted || name == candidate
|
||||
}
|
||||
if _, duplicate := fields[name]; !permitted || duplicate {
|
||||
return nil, false
|
||||
}
|
||||
var value json.RawMessage
|
||||
if decoder.Decode(&value) != nil {
|
||||
return nil, false
|
||||
}
|
||||
fields[name] = value
|
||||
}
|
||||
end, err := decoder.Token()
|
||||
if err != nil || end != json.Delim('}') {
|
||||
return nil, false
|
||||
}
|
||||
if _, err := decoder.Token(); err != io.EOF {
|
||||
return nil, false
|
||||
}
|
||||
return fields, true
|
||||
}
|
||||
@@ -178,7 +178,7 @@ func TestUnverifiedVMConfigPreservesStatusCounterObservations(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestGuestAgentTransportDeferralPreservesLiveCounterReceipts(t *testing.T) {
|
||||
for _, kind := range []string{"lost reply", "redirect"} {
|
||||
for _, kind := range []string{"lost reply", "redirect", "server error", "gateway error"} {
|
||||
t.Run(kind, func(t *testing.T) {
|
||||
var commands, statusCalls atomic.Int32
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
@@ -190,6 +190,15 @@ func TestGuestAgentTransportDeferralPreservesLiveCounterReceipts(t *testing.T) {
|
||||
fmt.Fprint(w, `{"data":{"status":"running","cpu":0.25,"diskread":0,"diskwrite":null,"netin":42}}`)
|
||||
case strings.Contains(r.URL.Path, "/agent/"):
|
||||
commands.Add(1)
|
||||
if kind == "server error" || kind == "gateway error" {
|
||||
status := http.StatusInternalServerError
|
||||
if kind == "gateway error" {
|
||||
status = http.StatusBadGateway
|
||||
}
|
||||
w.WriteHeader(status)
|
||||
fmt.Fprint(w, "upstream unavailable")
|
||||
return
|
||||
}
|
||||
if kind == "redirect" {
|
||||
http.Redirect(w, r, "/unverified/agent", http.StatusTemporaryRedirect)
|
||||
return
|
||||
@@ -210,6 +219,9 @@ func TestGuestAgentTransportDeferralPreservesLiveCounterReceipts(t *testing.T) {
|
||||
if kind == "redirect" {
|
||||
wantReason = "agent-redirect"
|
||||
}
|
||||
if kind == "server error" || kind == "gateway error" {
|
||||
wantReason = "agent-completion-unverified"
|
||||
}
|
||||
if _, err := client.GetVMFSInfo(context.Background(), "node", 105); GuestAgentDeferredReason(err) != wantReason {
|
||||
t.Fatalf("uncertain command not deferred: %v", err)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user