Keep incomplete Proxmox guest inventories out of publication

Require whole bounded, unambiguous enumerations on the efficient and node fallback routes. Preserve scoped empty inventories, LXC integer compatibility and ordinary counter/receipt semantics. Validate retained readings, History, breaches and template policy through actual fallback and recovery.

Change-source: pulse-maintainer
This commit is contained in:
pulse-triage[bot]
2026-10-14 11:35:55 +01:00
parent a0e352eaf9
commit 9dcd5697cc
9 changed files with 514 additions and 33 deletions
@@ -7661,3 +7661,35 @@ The canonical Docker-ingest source guard follows both guest enumeration routes
through `publishPVEGuests` to the canonical state update. Publication reconciliation
updates its former direct-call assertions without changing Docker ingestion or
the runtime ownership gate; the behavioral withdrawal controls remain required.
### Complete guest enumeration before publication (14 October 2026)
PVE `cluster/resources`, node `qemu` and node `lxc` enumeration consume and
close the whole existing 8 MiB-bounded HTTP response before admission. Only
HTTP 200 containing a complete UTF-8 JSON object with an explicit canonical
`data` array is authoritative inventory. Missing/null data, duplicate or
case-variant supported fields, conflicting errors, trailing values, incomplete
bodies, invalid rows and duplicate guest identities fail the whole enumeration;
no partial inventory or provider body is returned as a successful reading.
Cluster guest identities require a node and positive cluster-wide unique VMID.
Node lists require positive unique VMIDs; existing LXC string integers remain.
Scoped empty arrays, optional names/status/metrics, null counter presence,
templates, unique provider metadata and unfiltered non-guest resources remain
supported. Observation timestamps are assigned only to an admitted inventory.
A refused efficient enumeration can use the existing ordinary node fallback.
Failed node enumerations retain their earlier guest observations, History and
backup-template eligibility; a complete fallback can still remove deleted
guests and restore ordinary sampling. Complete cluster recovery returns to
its efficient route. This changes no TLS/pin, credentials, permissions,
polling cadence, guest-command/backup-lock checks, completion uncertainty,
publication ownership or grace-period rules.
`guest_inventory_response_test.go` covers complete-body and all-or-nothing
admission on all three real HTTP readers; `io_counters_test.go` covers
scoped-empty, extended fields, counter presence and receipt-time controls.
`guest_inventory_admission_test.go` covers actual efficient/fallback collection,
owned publication, retained observation times, breach/History/template
continuity, independent-connection preservation and normal recovery/removal.
These controlled source proofs do not establish native thaw/restart recovery,
installed trust, report attribution or release availability.
@@ -0,0 +1,208 @@
package monitoring
import (
"context"
"encoding/json"
"fmt"
"net/http"
"net/http/httptest"
"reflect"
"strings"
"sync/atomic"
"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"
"github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources"
"github.com/rcourtman/pulse-go-rewrite/pkg/proxmox"
)
// Walk the actual efficient enumeration -> fallback -> accepted publication ->
// canonical projection. Malformed successes cannot remove guests, refresh their
// last observation, clear a breach or invalidate template backup eligibility.
// All endpoints are synthetic; no backup, guest command or live send occurs.
func TestGuestInventoryFailurePreservesOwnedReadingsAndRecovers(t *testing.T) {
for _, failure := range []string{"missing-data", "trailing-value", "truncated-prefix", "duplicate-id"} {
t.Run(failure, func(t *testing.T) {
t.Setenv("PULSE_DATA_DIR", t.TempDir())
var phase, commands, clusterReads, listReads atomic.Int32
corrupt := func(w http.ResponseWriter, prefix string) {
switch failure {
case "missing-data":
fmt.Fprint(w, `{}`)
case "trailing-value":
fmt.Fprint(w, prefix+` {"errors":{"message":"fixture-private-detail"}}`)
case "truncated-prefix":
w.Header().Set("Content-Length", fmt.Sprint(len(prefix)+100))
fmt.Fprint(w, prefix)
case "duplicate-id":
var payload struct {
Data []json.RawMessage `json:"data"`
}
if err := json.Unmarshal([]byte(prefix), &payload); err != nil {
t.Error(err)
return
}
payload.Data = append(payload.Data, payload.Data[0])
_ = json.NewEncoder(w).Encode(payload)
}
}
const vm = `{"vmid":105,"type":"qemu","node":"member","name":"kept-vm","status":"running","lock":"backup","cpu":0.95,"maxmem":1073741824,"maxdisk":2147483648}`
const extra = `{"vmid":106,"type":"qemu","node":"member","name":"later-deleted-vm","status":"running","lock":"backup","cpu":0.1,"maxmem":1073741824}`
const ct = `{"vmid":205,"type":"lxc","node":"member","name":"kept-container","status":"running","lock":"backup","cpu":0.15,"mem":104857600,"maxmem":1073741824,"disk":268435456,"maxdisk":2147483648}`
const template = `{"vmid":999,"type":"qemu","node":"member","template":1}`
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
p := phase.Load()
switch {
case strings.HasSuffix(r.URL.Path, "/cluster/resources"):
clusterReads.Add(1)
if p == 1 || p == 2 {
corrupt(w, `{"data":[`+vm+`]}`)
return
}
if p == 4 {
http.NotFound(w, r)
return
}
rows := vm + `,` + ct
if p == 0 {
rows += `,` + extra + `,` + template
}
if p == 3 {
rows += `,{"vmid":108,"type":"qemu","node":"member","name":"new-vm","status":"running","lock":"backup","cpu":0,"maxmem":1073741824}`
}
fmt.Fprint(w, `{"data":[`+rows+`]}`)
case strings.HasSuffix(r.URL.Path, "/member/qemu"):
listReads.Add(1)
if p == 1 {
corrupt(w, `{"data":[`+vm+`]}`)
return
}
if p == 4 {
fmt.Fprint(w, `{"data":[]}`)
return
}
fmt.Fprint(w, `{"data":[`+vm+`]}`)
case strings.HasSuffix(r.URL.Path, "/member/lxc"):
listReads.Add(1)
if p == 1 {
corrupt(w, `{"data":[`+ct+`]}`)
return
}
if p == 4 {
fmt.Fprint(w, `{"data":[]}`)
return
}
// Keep the existing older-PVE string identifier compatibility on fallback.
fmt.Fprint(w, `{"data":[`+strings.Replace(ct, `"vmid":205`, `"vmid":"205"`, 1)+`]}`)
case strings.Contains(r.URL.Path, "/agent/"):
commands.Add(1)
t.Error("inventory proof dispatched an in-guest command")
http.Error(w, "unexpected command", 500)
case strings.HasSuffix(r.URL.Path, "/qemu/105/status/current"), strings.HasSuffix(r.URL.Path, "/qemu/106/status/current"), strings.HasSuffix(r.URL.Path, "/qemu/108/status/current"):
fmt.Fprint(w, `{"data":{"status":"running","lock":"backup","maxmem":1073741824,"mem":104857600}}`)
case strings.HasSuffix(r.URL.Path, "/lxc/205/status/current"):
fmt.Fprint(w, `{"data":{"vmid":205,"status":"running","lock":"backup","maxmem":1073741824,"mem":104857600,"maxdisk":2147483648,"disk":268435456}}`)
case strings.HasSuffix(r.URL.Path, "/config"):
fmt.Fprint(w, `{"data":{"lock":"backup"}}`)
default:
http.NotFound(w, r)
}
}))
defer server.Close()
client, err := proxmox.NewClient(proxmox.ClientConfig{Host: server.URL, TokenName: "fixture@pve!pulse", TokenValue: "fixture", Timeout: 10 * time.Second})
if err != nil {
t.Fatal(err)
}
m := newTestPVEMonitor("inventory")
t.Cleanup(m.alertManager.Stop)
t.Cleanup(m.notificationMgr.Stop)
m.replacePVEClient("inventory", client)
alertCfg := m.alertManager.GetConfig()
alertCfg.Enabled = true
alertCfg.GuestDefaults.CPU = &alerts.HysteresisThreshold{Trigger: 80, Clear: 75}
alertCfg.MetricTimeThresholds = map[string]map[string]int{"guest": {"cpu": 0}}
m.alertManager.UpdateConfig(alertCfg)
run := func() models.StateSnapshot {
t.Helper()
if err := m.pollGuestsWithFallback(context.Background(), "inventory", &config.PVEInstance{MonitorVMs: true, MonitorContainers: true}, client, []proxmox.Node{{Node: "member", Status: "online"}}, map[string]string{"member": "online"}); err != nil {
t.Fatal(err)
}
return m.state.GetSnapshot()
}
// A second registered connection has its own observation and must survive
// both failed enumerations and actual removals on this one.
peer := models.VM{ID: "peer:other:305", Instance: "peer", Node: "other", VMID: 305, Name: "independent-peer", Type: "qemu", Status: "running", LastSeen: time.Now()}
m.state.UpdateVMsForInstance("peer", []models.VM{peer})
original := run()
if len(original.VMs) != 3 || len(original.Containers) != 1 {
t.Fatalf("initial complete inventory not established: %+v", original)
}
templateKey := pveBackupTemplateSubjectKey("inventory", "qemu", "member", 999)
if _, ok := m.pveBackupTemplateSubjects["inventory"][templateKey]; !ok {
t.Fatal("initial template not established")
}
originalHistory := m.metricsHistory.GetGuestMetrics("inventory:member:105", "cpu", time.Hour)
originalAlerts := m.alertManager.GetActiveAlerts()
if len(originalHistory) != 1 || len(originalAlerts) != 1 {
t.Fatalf("initial history/breach not established: %d/%d", len(originalHistory), len(originalAlerts))
}
emit := func(label string, snapshot models.StateSnapshot) {
registry := unifiedresources.NewRegistry(nil)
registry.IngestSnapshot(snapshot)
projection, err := json.Marshal(map[string]any{"case": failure, "phase": label, "snapshot": snapshot, "resources": registry.List()})
if err != nil {
t.Fatal(err)
}
t.Logf("GUEST_INVENTORY_PROJECTION %s", projection)
}
emit("initial", original)
phase.Store(1)
failed := run()
if !reflect.DeepEqual(failed.VMs, original.VMs) || !reflect.DeepEqual(failed.Containers, original.Containers) {
t.Error("incomplete inventory changed retained guests or their observation time")
}
if !reflect.DeepEqual(m.metricsHistory.GetGuestMetrics("inventory:member:105", "cpu", time.Hour), originalHistory) {
t.Error("failed inventory fabricated a new History sample")
}
if !reflect.DeepEqual(m.alertManager.GetActiveAlerts(), originalAlerts) {
t.Error("failed inventory changed an existing breach")
}
if _, ok := m.pveBackupTemplateSubjects["inventory"][templateKey]; !ok {
t.Error("failed inventory erased backup-template eligibility")
}
emit("both-inventories-refused", failed)
phase.Store(2)
fallback := run()
if len(fallback.VMs) != 2 || len(fallback.Containers) != 1 || fallback.VMs[0].VMID == 106 {
t.Error("complete fallback failed to publish a real deletion")
}
if _, ok := m.pveBackupTemplateSubjects["inventory"][templateKey]; ok {
t.Error("complete fallback retained a deleted template")
}
if len(m.metricsHistory.GetGuestMetrics("inventory:member:105", "cpu", time.Hour)) != 2 {
t.Error("complete fallback failed to resume History")
}
emit("complete-node-fallback", fallback)
before := listReads.Load()
phase.Store(3)
recovered := run()
if len(recovered.VMs) != 3 || len(recovered.Containers) != 1 || listReads.Load() != before {
t.Error("complete cluster recovery did not return to the efficient route")
}
emit("complete-cluster-recovery", recovered)
phase.Store(4)
empty := run()
if len(empty.VMs) != 1 || empty.VMs[0].ID != peer.ID || len(empty.Containers) != 0 {
t.Error("explicit empty node inventories failed to remove guests within their scope")
}
if commands.Load() != 0 || clusterReads.Load() != 5 || listReads.Load() != 6 || len(m.pveGuestPolls) != 0 {
t.Errorf("unexpected command/retry/owner leak: %d/%d/%d/%d", commands.Load(), clusterReads.Load(), listReads.Load(), len(m.pveGuestPolls))
}
emit("complete-scoped-empty", empty)
})
}
}
+17 -27
View File
@@ -1272,18 +1272,13 @@ func (c *Client) GetVMs(ctx context.Context, node string) ([]VM, error) {
if err != nil {
return nil, err
}
defer resp.Body.Close()
var result struct {
Data []VM `json:"data"`
}
if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
values, err := readGuestInventory(resp, func(vm VM) (string, bool) { return guestInventoryID(vm.VMID) })
if err != nil {
return nil, err
}
stampVMObservation(result.Data, time.Now().UTC())
return result.Data, nil
stampVMObservation(values, time.Now().UTC())
return values, nil
}
// GetContainers returns all LXC containers on a specific node
@@ -1292,18 +1287,13 @@ func (c *Client) GetContainers(ctx context.Context, node string) ([]Container, e
if err != nil {
return nil, err
}
defer resp.Body.Close()
var result struct {
Data []Container `json:"data"`
}
if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
values, err := readGuestInventory(resp, func(container Container) (string, bool) { return guestInventoryID(int(container.VMID)) })
if err != nil {
return nil, err
}
stampContainerObservation(result.Data, time.Now().UTC())
return result.Data, nil
stampContainerObservation(values, time.Now().UTC())
return values, nil
}
// GetStorage returns storage information for a specific node
@@ -2580,18 +2570,18 @@ func (c *Client) GetClusterResources(ctx context.Context, resourceType string) (
if err != nil {
return nil, err
}
defer resp.Body.Close()
var result struct {
Data []ClusterResource `json:"data"`
}
if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
values, err := readGuestInventory(resp, func(resource ClusterResource) (string, bool) {
if resourceType == "vm" && resource.Type != "qemu" && resource.Type != "lxc" {
return "", false
}
return clusterInventoryID(resource)
})
if err != nil {
return nil, err
}
stampClusterResourceObservation(result.Data, time.Now().UTC())
return result.Data, nil
stampClusterResourceObservation(values, time.Now().UTC())
return values, nil
}
// ClusterOptions holds selected Proxmox datacenter configuration options.
+1 -1
View File
@@ -297,7 +297,7 @@ func TestClientStatusAndResources(t *testing.T) {
})
case "/api2/json/cluster/resources":
writeJSON(t, w, map[string]interface{}{
"data": []ClusterResource{{ID: "qemu/100", Type: "vm", VMID: 100, Pool: "prod-vms"}},
"data": []ClusterResource{{ID: "qemu/100", Type: "qemu", Node: "node1", VMID: 100, Pool: "prod-vms"}},
})
case "/api2/json/nodes/node1/lxc/101/interfaces":
writeJSON(t, w, map[string]interface{}{
@@ -275,7 +275,7 @@ func TestClusterClient_GetClusterResources(t *testing.T) {
return
}
if strings.HasPrefix(r.URL.Path, "/api2/json/cluster/resources") {
fmt.Fprint(w, `{"data":[{"id":"qemu/100","type":"qemu","node":"node1"}]}`)
fmt.Fprint(w, `{"data":[{"id":"qemu/100","type":"qemu","node":"node1","vmid":100}]}`)
return
}
w.WriteHeader(http.StatusNotFound)
@@ -302,7 +302,7 @@ func TestClusterClient_GetClusterResources_AllTypes(t *testing.T) {
return
}
if r.URL.Path == "/api2/json/cluster/resources" {
fmt.Fprint(w, `{"data":[{"id":"lxc/200","type":"lxc","node":"node1"}]}`)
fmt.Fprint(w, `{"data":[{"id":"lxc/200","type":"lxc","node":"node1","vmid":200}]}`)
return
}
w.WriteHeader(http.StatusNotFound)
+85
View File
@@ -0,0 +1,85 @@
package proxmox
import (
"bytes"
"encoding/json"
"errors"
"net/http"
"strconv"
"strings"
"unicode/utf8"
)
// Inventories own removal. A malformed success is neither a complete empty
// enumeration nor an authentication refusal; keep its body out of diagnostics.
var errInvalidGuestInventoryResponse = errors.New("invalid Proxmox guest inventory response")
// The three enumerations share identity, telemetry and metadata fields. Reject
// case-insensitive struct overwrites without excluding unique provider metadata
// or changing the existing endpoint-specific decoders (notably LXC FlexInt).
var guestInventoryFields = []string{
"id", "type", "node", "pool", "status", "name", "vmid", "cpu", "cpus", "maxcpu", "mem", "maxmem", "swap", "maxswap", "disk", "maxdisk", "netin", "netout", "diskread", "diskwrite", "uptime", "template", "tags", "lock", "agent", "hostname", "ip", "ip6", "ipv4", "ipv6", "network", "diskinfo", "rootfs",
}
func readGuestInventory[T any](resp *http.Response, identity func(T) (string, bool)) ([]T, error) {
body, readErr := readResponseBodyLimited(resp.Body)
closeErr := resp.Body.Close()
if readErr != nil || closeErr != nil {
return nil, errors.Join(readErr, closeErr)
}
if resp.StatusCode != http.StatusOK || !utf8.Valid(body) {
return nil, errInvalidGuestInventoryResponse
}
envelope, valid := inventoryResponseObject(body, "data", "errors")
if !valid {
return nil, errInvalidGuestInventoryResponse
}
if failure, exists := envelope["errors"]; exists && !bytes.Equal(bytes.TrimSpace(failure), []byte("null")) {
return nil, errInvalidGuestInventoryResponse
}
data := bytes.TrimSpace(envelope["data"])
// Preserve a scoped empty array, never a missing/null inventory. Read and
// close the whole bounded HTTP body before publishing even a valid prefix.
if len(data) == 0 || data[0] != '[' {
return nil, errInvalidGuestInventoryResponse
}
var rows []json.RawMessage
if json.Unmarshal(data, &rows) != nil {
return nil, errInvalidGuestInventoryResponse
}
values := make([]T, 0, len(rows))
seen := make(map[string]bool, len(rows))
for _, row := range rows {
if _, valid := inventoryResponseObject(row, guestInventoryFields...); !valid {
return nil, errInvalidGuestInventoryResponse
}
var value T
if json.Unmarshal(row, &value) != nil {
return nil, errInvalidGuestInventoryResponse
}
key, valid := identity(value)
if !valid || seen[key] {
return nil, errInvalidGuestInventoryResponse
}
seen[key] = true
values = append(values, value)
}
return values, nil
}
func guestInventoryID(vmid int) (string, bool) {
return strconv.Itoa(vmid), vmid > 0
}
func clusterInventoryID(resource ClusterResource) (string, bool) {
switch resource.Type {
case "qemu", "lxc":
key, valid := guestInventoryID(resource.VMID)
// VMIDs are cluster-wide, including across a migration or guest kind.
return "guest/" + key, valid && strings.TrimSpace(resource.Node) != ""
default:
// Unfiltered calls may include nodes, storage, pools and future provider
// resource kinds. They retain their provider identity and optional fields.
return resource.ID, strings.TrimSpace(resource.Type) != "" && strings.TrimSpace(resource.ID) != ""
}
}
@@ -0,0 +1,102 @@
package proxmox
import (
"context"
"fmt"
"net/http"
"net/http/httptest"
"reflect"
"strings"
"sync/atomic"
"testing"
"time"
)
// Enumerations drive removal, backup-template policy and History. Exercise the
// real HTTP readers on both production routes, with no QGA commands or retries.
func TestGuestInventoryResponseAdmission(t *testing.T) {
for _, kind := range []string{"cluster", "qemu", "lxc"} {
t.Run(kind, func(t *testing.T) {
row := `{"vmid":105,"type":"qemu","node":"member","name":"fixture","status":"running"}`
inventory := `{"data":[` + row + `]}`
cases := []struct {
name, body string
incomplete bool
}{
{"missing-data", `{}`, false},
{"null-data", `{"data":null}`, false},
{"null-envelope", `null`, false},
{"conflicting-error", `{"data":[],"errors":{"message":"fixture-private-detail"}}`, false},
{"case-data", `{"Data":[]}`, false},
{"duplicate-data", `{"data":[` + row + `],"data":[]}`, false},
{"escaped-duplicate-data", `{"data":[],"\u0064ata":[` + row + `]}`, false},
{"case-overwrite", `{"data":[` + row + `],"Data":[]}`, false},
{"trailing-value", inventory + ` {"data":[]}`, false},
{"trailing-garbage", inventory + `fixture-private-detail`, false},
{"truncated-after-prefix", inventory, true},
{"oversized-after-prefix", inventory + strings.Repeat(" ", int(maxResponseBodyBytes)), false},
{"invalid-utf8", `{"data":[` + row + `],"metadata":"` + string([]byte{0xff}) + `"}`, false},
{"null-row", `{"data":[` + row + `,null]}`, false},
{"missing-guest-id", `{"data":[{"type":"qemu","node":"member"}]}`, false},
{"zero-guest-id", `{"data":[{"vmid":0,"type":"qemu","node":"member"}]}`, false},
{"negative-guest-id", `{"data":[{"vmid":-105,"type":"qemu","node":"member"}]}`, false},
{"duplicate-field", `{"data":[{"vmid":106,"vmid":105,"type":"qemu","node":"member"}]}`, false},
{"case-id", `{"data":[{"vmid":105,"VMID":106,"type":"qemu","node":"member"}]}`, false},
{"duplicate-row", `{"data":[` + row + `,` + row + `]}`, false},
{"bad-row-after-good", `{"data":[` + row + `,{"vmid":106,"type":"qemu","node":"member","mem":"not-a-number"}]}`, false},
}
if kind == "cluster" {
cases = append(cases, struct {
name, body string
incomplete bool
}{"missing-node", `{"data":[{"vmid":105,"type":"qemu"}]}`, false},
struct {
name, body string
incomplete bool
}{"missing-type", `{"data":[{"vmid":105,"node":"member"}]}`, false},
struct {
name, body string
incomplete bool
}{"duplicate-migrated-id", `{"data":[` + row + `,{"vmid":105,"type":"qemu","node":"other"}]}`, false})
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
var requests atomic.Int32
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
requests.Add(1)
if tc.incomplete {
w.Header().Set("Content-Length", fmt.Sprint(len(tc.body)+100))
} else {
w.(http.Flusher).Flush()
}
fmt.Fprint(w, tc.body)
}))
defer server.Close()
c, err := NewClient(ClientConfig{Host: server.URL, TokenName: "fixture@pve!pulse", TokenValue: "fixture", Timeout: 15 * time.Second})
if err != nil {
t.Fatal(err)
}
defer c.httpClient.CloseIdleConnections()
var values any
switch kind {
case "cluster":
values, err = c.GetClusterResources(context.Background(), "vm")
case "qemu":
values, err = c.GetVMs(context.Background(), "member")
case "lxc":
values, err = c.GetContainers(context.Background(), "member")
}
if err == nil || !reflect.ValueOf(values).IsNil() {
t.Errorf("unverified %s inventory admitted: rows=%+v err=%v", kind, values, err)
}
if err != nil && strings.Contains(err.Error(), "fixture-private-detail") {
t.Error("private response body became an inventory diagnostic")
}
if requests.Load() != 1 {
t.Errorf("enumeration was replayed: %d requests", requests.Load())
}
})
}
})
}
}
+64
View File
@@ -253,3 +253,67 @@ func TestVMStatusMemoryAvailabilityPresence(t *testing.T) {
t.Run("invalid-numbers", testVMMemInfoAvailabilityRejectsInvalidNumbers)
t.Run("reused-decode", testVMMemInfoAvailabilityPresenceDoesNotLeakAcrossDecode)
}
// Complete enumeration still owns the original counter-presence and receipt-
// time semantics; scoped-empty and older LXC encodings remain compatible.
func TestGuestInventoryAcceptedScopedAndExtendedResponses(t *testing.T) {
for _, kind := range []string{"cluster", "qemu", "lxc"} {
for _, body := range []string{
`{"data":[]}`, `{"data":[],"errors":null,"total":0}`,
`{"data":[{"vmid":105,"type":"qemu","node":"member","name":"fixture","status":"stopped","pool":"department","cpu":0,"maxcpu":4,"cpus":4,"mem":0,"maxmem":1024,"disk":0,"maxdisk":4096,"diskread":0,"netin":null,"uptime":0,"template":1,"tags":"ops","lock":"backup","agent":1,"vendor-metadata":{"version":1}}],"digest":"fixture-metadata"}`,
} {
t.Run(kind+"/"+body, func(t *testing.T) {
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { fmt.Fprint(w, body) }))
defer server.Close()
c := backupTestClient(t, server.URL)
defer c.httpClient.CloseIdleConnections()
switch kind {
case "cluster":
got, err := c.GetClusterResources(context.Background(), "vm")
if err != nil {
t.Fatal(err)
}
if strings.Contains(body, `"vmid"`) && (len(got) != 1 || got[0].VMID != 105 || got[0].Node != "member" || got[0].MaxCPU != 4 || got[0].Pool != "department" || got[0].Lock != "backup" || got[0].Template != 1 || got[0].IOCounters.DiskRead != true || got[0].IOCounters.NetworkIn || got[0].ObservedAt.IsZero()) {
t.Fatalf("supported cluster fields changed: %+v", got)
}
case "qemu":
got, err := c.GetVMs(context.Background(), "member")
if err != nil {
t.Fatal(err)
}
if strings.Contains(body, `"vmid"`) && (len(got) != 1 || got[0].VMID != 105 || got[0].CPUs != 4 || got[0].Agent != 1 || got[0].Lock != "backup" || got[0].Pool != "department" || got[0].IOCounters.NetworkIn || got[0].ObservedAt.IsZero()) {
t.Fatalf("supported VM fields changed: %+v", got)
}
case "lxc":
got, err := c.GetContainers(context.Background(), "member")
if err != nil {
t.Fatal(err)
}
if strings.Contains(body, `"vmid"`) && (len(got) != 1 || int(got[0].VMID) != 105 || int(got[0].CPUs) != 4 || got[0].Pool != "department" || got[0].IOCounters.NetworkIn || got[0].ObservedAt.IsZero()) {
t.Fatalf("supported container fields changed: %+v", got)
}
}
})
}
}
t.Run("lxc-string-identifiers", func(t *testing.T) {
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
fmt.Fprint(w, `{"data":[{"vmid":"205","cpus":"2","name":"container"}]}`)
}))
defer server.Close()
got, err := backupTestClient(t, server.URL).GetContainers(context.Background(), "member")
if err != nil || len(got) != 1 || int(got[0].VMID) != 205 || int(got[0].CPUs) != 2 {
t.Fatalf("string LXC identifiers lost: %+v %v", got, err)
}
})
t.Run("all-resource-types", func(t *testing.T) {
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
fmt.Fprint(w, `{"data":[{"id":"node/member","type":"node","node":"member"},{"id":"storage/member/local","type":"storage","node":"member"},{"id":"pool/department","type":"pool"},{"id":"future/provider","type":"future-type"}]}`)
}))
defer server.Close()
got, err := backupTestClient(t, server.URL).GetClusterResources(context.Background(), "")
if err != nil || len(got) != 4 {
t.Fatalf("non-guest resource compatibility lost: %+v %v", got, err)
}
})
}
+3 -3
View File
@@ -16,7 +16,7 @@ func decodeNodeInventory(body []byte) ([]Node, error) {
if !utf8.Valid(body) {
return nil, errInvalidNodeInventoryResponse
}
envelope, valid := nodeInventoryObject(body, "data", "errors")
envelope, valid := inventoryResponseObject(body, "data", "errors")
if !valid {
return nil, errInvalidNodeInventoryResponse
}
@@ -36,7 +36,7 @@ func decodeNodeInventory(body []byte) ([]Node, error) {
nodes := make([]Node, 0, len(rows))
seen := make(map[string]bool, len(rows))
for _, row := range rows {
fields, valid := nodeInventoryObject(row, "node", "status", "cpu", "maxcpu", "mem", "maxmem", "disk", "maxdisk", "uptime", "level")
fields, valid := inventoryResponseObject(row, "node", "status", "cpu", "maxcpu", "mem", "maxmem", "disk", "maxdisk", "uptime", "level")
if !valid {
return nil, errInvalidNodeInventoryResponse
}
@@ -57,7 +57,7 @@ func decodeNodeInventory(body []byte) ([]Node, error) {
// The existing object reader rejects duplicates and extra JSON values. Keep
// unique provider metadata, but never let encoding/json's case-insensitive
// struct matching replace canonical inventory fields with another spelling.
func nodeInventoryObject(body []byte, canonical ...string) (map[string]json.RawMessage, bool) {
func inventoryResponseObject(body []byte, canonical ...string) (map[string]json.RawMessage, bool) {
fields, valid := guestAgentResponseFields(body)
if !valid {
return nil, false