From 49217d284d2aa875cc69a44746fca87a27cc6984 Mon Sep 17 00:00:00 2001 From: rcourtman Date: Fri, 24 Jul 2026 10:17:21 +0100 Subject: [PATCH] fix: correct Proxmox guest rate sampling --- .../v6/internal/subsystems/agent-lifecycle.md | 5 + .../v6/internal/subsystems/alerts.md | 7 + .../v6/internal/subsystems/monitoring.md | 41 +- .../v6/internal/subsystems/registry.json | 27 + .../internal/subsystems/storage-recovery.md | 5 + .../internal/subsystems/unified-resources.md | 8 + .../Workloads/__tests__/GuestRow.test.tsx | 17 + internal/alerts/filter_evaluation.go | 41 +- internal/alerts/filter_evaluation_test.go | 22 + internal/alerts/guest.go | 35 +- internal/alerts/guest_snapshot.go | 26 +- internal/alerts/guest_snapshot_test.go | 24 + internal/api/router_state_test.go | 78 +++ internal/models/metrics_types.go | 85 ++- internal/models/metrics_types_test.go | 58 ++ internal/models/models.go | 2 + .../monitoring/canonical_guardrails_test.go | 8 +- .../monitoring/issue1613_contract_test.go | 180 +++++++ .../monitoring/memory_source_catalog_test.go | 3 +- internal/monitoring/monitor_helpers.go | 57 ++ .../monitoring/monitor_pbs_fallback_test.go | 1 + .../monitoring/monitor_polling_containers.go | 71 ++- .../monitor_polling_node_helpers.go | 5 +- internal/monitoring/monitor_polling_vm.go | 22 +- internal/monitoring/monitor_pve.go | 38 +- .../monitoring/monitor_pve_guest_builders.go | 164 ++++-- .../monitoring/monitor_pve_guest_helpers.go | 23 +- internal/monitoring/monitor_pve_guest_lxc.go | 47 +- .../monitoring/monitor_pve_guest_lxc_test.go | 60 ++- .../monitoring/monitor_pve_node_vm_builder.go | 42 +- internal/monitoring/ratetracker.go | 226 ++++---- internal/monitoring/ratetracker_test.go | 509 ++++++------------ .../monitoring/resource_stale_thresholds.go | 4 + .../unifiedresources/code_standards_test.go | 17 + internal/unifiedresources/metrics.go | 16 +- internal/unifiedresources/metrics_test.go | 25 + internal/unifiedresources/types.go | 2 +- pkg/proxmox/client.go | 187 ++++--- pkg/proxmox/io_counters.go | 124 +++++ pkg/proxmox/io_counters_test.go | 117 ++++ .../canonical_completion_guard_test.py | 6 + .../release_control/subsystem_lookup_test.py | 1 + 42 files changed, 1717 insertions(+), 719 deletions(-) create mode 100644 internal/monitoring/issue1613_contract_test.go create mode 100644 pkg/proxmox/io_counters.go create mode 100644 pkg/proxmox/io_counters_test.go diff --git a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md index 96e744c5f..0edcb7a54 100644 --- a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md +++ b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md @@ -36,6 +36,11 @@ one compressed unified-agent report without a collector-side suffix cap. websocket payloads. Carrying plural availability facets through that serializer is an adjacent monitoring/API projection and does not change agent enrollment, report admission, removal, update, profile, or command authority. +The JSON-excluded Proxmox VM/LXC I/O-rate validity fields carried by +`internal/models/models.go` are likewise monitoring-owned sidecar evidence. +They distinguish a valid idle interval from an unknown rate for history, +alerts, and unified-resource projection; they do not describe agent report +admission, heartbeat, enrollment, command reachability, or fleet liveness. Docker / Podman report sizing is a two-dimensional transport contract: `pkg/agents/docker/report_limits.go` owns the inclusive encoded HTTP-body and decoded JSON byte ceilings, derives the 80% early-warning boundaries, and diff --git a/docs/release-control/v6/internal/subsystems/alerts.md b/docs/release-control/v6/internal/subsystems/alerts.md index b5597d4ae..dd241253a 100644 --- a/docs/release-control/v6/internal/subsystems/alerts.md +++ b/docs/release-control/v6/internal/subsystems/alerts.md @@ -43,6 +43,13 @@ used/free/total evidence must not open or clear a memory alert. If such a sample follows an active alert, the alert remains active with its last trusted value until a subsequent trusted sample proves recovery; missing evidence is not evidence that pressure disappeared. +Proxmox guest disk-read, disk-write, network-in, and network-out thresholds +consume only monitoring-owned valid rate observations. A valid idle interval +is explicit zero and may prove recovery; a first sample, missing/null counter, +partial response, or rejected out-of-order sample is unknown and must not +start, clear, or match a custom metric filter. Alert units remain MiB/s at the +threshold boundary (`bytes/s / 1024 / 1024`); that display/threshold conversion +must not be applied to the upstream cumulative-counter divisor. ## Canonical Files diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md index ff1ab4719..27afd14f5 100644 --- a/docs/release-control/v6/internal/subsystems/monitoring.md +++ b/docs/release-control/v6/internal/subsystems/monitoring.md @@ -144,6 +144,31 @@ no core-count division and no in-guest host-agent substitution. A linked host agent may supplement metrics the platform does not provide, but guest CPU and its `vm` / `system-container` history target remain platform-owned so dashboard, details, API state, alerts, and history cannot select different authorities. +Proxmox guest disk and network throughput has one cumulative-counter sampling +contract. `diskread`, `diskwrite`, `netin`, and `netout` are cumulative bytes; +the canonical rate is `(current counter - previous counter) / elapsed +observation seconds`, in bytes per second, with no 1024 divisor. Elapsed time +comes from the receipt time stamped immediately after the relevant Proxmox API +response is decoded, not from later guest-agent, filesystem, or metadata +enrichment. Each counter keeps an independent adjacent-sample baseline: +explicitly unchanged counters produce a valid zero, missing/null fields produce +unknown, out-of-order samples produce unknown without moving the baseline, and +a counter decrease caused by restart, reconnect, migration epoch change, or +wrap rebases that counter and produces a valid zero for the reset interval. A +source-uptime rollback rebases the complete counter epoch and leaves the first +post-restart rate unknown, including when a busy guest already surpassed its +pre-restart counter value before the next poll. +First-sample and missing-field unknowns remain internal validity state; the +legacy API/websocket guest number fields stay numeric, while history, unified +metrics, and alerts omit the unknown observation instead of manufacturing +zero. The rate-tracker identity is `(configured PVE instance, guest kind, +VMID)`: it survives node migration, separates QEMU from LXC, and prevents +duplicate configured cluster identities from sharing a concurrent baseline. +Idle and partial samples still refresh tracker liveness. +Proxmox row liveness uses the same cadence-derived threshold as source +freshness (`max(2 * configured poll interval, 60s)`). Node offline grace and +guest preservation must not expire between healthy 60- or 90-second polls, and +must not use a separate fixed 60-second timer. Tenant monitor enumeration is monitoring-owned runtime topology, not a reporting source of truth. `MultiTenantMonitor.ListOrganizationIDs` may expose persisted organization IDs to API-owned background workers, but it must not @@ -202,6 +227,8 @@ node-local Agent evidence. 20. `internal/dockeragent/collect.go` 21. `pkg/proxmox/ceph.go` 21a. `pkg/proxmox/cluster_client.go` +21b. `pkg/proxmox/client.go` +21c. `pkg/proxmox/io_counters.go` 22. `pkg/proxmox/zfs.go` 23. `internal/monitoring/guest_memory_sources.go` 24. `internal/monitoring/guest_memory_stability.go` @@ -222,6 +249,7 @@ node-local Agent evidence. 37. `pkg/agents/docker/report.go` 38. `internal/models/models.go` 38a. `internal/models/proxmox_guest_state.go` +38b. `internal/models/metrics_types.go` 39. `internal/models/models_frontend.go` 40. `internal/models/converters.go` 41. `internal/models/deepcopy.go` @@ -1076,14 +1104,17 @@ legacy PBS and PMG summaries on current `LastSeen` and health state each tick, so long-lived infrastructure, workloads, storage, and recovery demos do not decay into synthetic stale-state warnings while mock mode remains enabled. That same Proxmox container monitoring boundary now also owns runtime counter -recovery when the lower-fidelity container list or cluster-resources payload -reports stale or zero I/O totals. `internal/monitoring/monitor_pve.go`, +authority when the lower-fidelity container list or cluster-resources payload +and the current-status payload differ. `internal/monitoring/monitor_pve.go`, `internal/monitoring/monitor_pve_guest_lxc.go`, and `internal/monitoring/monitor_polling_containers.go` must merge the current `GetContainerStatus` counters through one canonical `mergeContainerRuntimeCounters` -path before LXC rate calculation and must reuse the same prefetched status -snapshot for metadata enrichment instead of paying disconnected metric and -metadata status reads that can diverge. +path before LXC rate calculation. A present status field is newer authority +even when it is zero or lower after a restart; an absent/null status field +retains the listing field and its presence state. The merge must retain the +status response receipt time and reuse the same prefetched status snapshot for +metadata enrichment instead of paying disconnected metric and metadata status +reads that can diverge. That same Proxmox backup/snapshot boundary owns bounded concurrent guest snapshot enumeration. `internal/monitoring/monitor_backups.go` must query VM and LXC snapshot endpoints through one capped worker pool and preserve diff --git a/docs/release-control/v6/internal/subsystems/registry.json b/docs/release-control/v6/internal/subsystems/registry.json index 51401af61..6b2752320 100644 --- a/docs/release-control/v6/internal/subsystems/registry.json +++ b/docs/release-control/v6/internal/subsystems/registry.json @@ -1349,6 +1349,7 @@ "exact_files": [ "internal/api/host_agent_removal_lifecycle_integration_test.go", "internal/config/host_continuity_test.go", + "internal/models/metrics_types_test.go", "internal/monitoring/monitor_host_agent_removal_lifecycle_test.go", "internal/monitoring/monitor_host_agents_test.go", "scripts/installtests/agent_state_dir_lifecycle_test.go", @@ -5184,6 +5185,7 @@ "internal/mock/platform_fixtures.go", "internal/models/converters.go", "internal/models/deepcopy.go", + "internal/models/metrics_types.go", "internal/models/models.go", "internal/models/models_frontend.go", "internal/models/proxmox_guest_state.go", @@ -5192,7 +5194,9 @@ "pkg/agents/host/report.go", "pkg/agents/kubernetes/report.go", "pkg/proxmox/ceph.go", + "pkg/proxmox/client.go", "pkg/proxmox/cluster_client.go", + "pkg/proxmox/io_counters.go", "pkg/proxmox/zfs.go" ], "verification": { @@ -5397,6 +5401,7 @@ "match_files": [ "internal/models/converters.go", "internal/models/deepcopy.go", + "internal/models/metrics_types.go", "internal/models/models.go", "internal/models/models_frontend.go", "internal/models/proxmox_guest_state.go", @@ -5408,6 +5413,7 @@ "internal/dockeragent/swarm_coverage_test.go", "internal/models/deepcopy_test.go", "internal/models/host_transient_expiry_test.go", + "internal/models/metrics_types_test.go", "internal/models/state_host_test.go", "internal/monitoring/issue1595_collection_trust_test.go", "internal/monitoring/monitor_full_coverage_test.go", @@ -5418,6 +5424,24 @@ "internal/unifiedresources/registry_test.go" ] }, + { + "id": "proxmox-guest-counter-runtime", + "label": "Proxmox guest cumulative-counter presence and receipt-time proof", + "match_prefixes": [], + "match_files": [ + "pkg/proxmox/client.go", + "pkg/proxmox/io_counters.go" + ], + "allow_same_subsystem_tests": false, + "test_prefixes": [], + "exact_files": [ + "internal/api/router_state_test.go", + "internal/monitoring/issue1613_contract_test.go", + "internal/monitoring/monitor_pve_guest_lxc_test.go", + "internal/monitoring/ratetracker_test.go", + "pkg/proxmox/io_counters_test.go" + ] + }, { "id": "proxmox-zfs-runtime", "label": "proxmox zfs compatibility proof", @@ -5587,6 +5611,7 @@ "internal/monitoring/canonical_guardrails_test.go", "internal/monitoring/issue1485_unraid_lifecycle_test.go", "internal/monitoring/issue1595_collection_trust_test.go", + "internal/monitoring/issue1613_contract_test.go", "internal/monitoring/monitor_additional_test.go", "internal/monitoring/monitor_alert_intent_test.go", "internal/monitoring/monitor_alert_override_migration_test.go", @@ -5595,6 +5620,8 @@ "internal/monitoring/monitor_host_agent_removal_lifecycle_test.go", "internal/monitoring/monitor_host_agents_test.go", "internal/monitoring/monitor_pve_cluster_refresh_test.go", + "internal/monitoring/monitor_pve_guest_lxc_test.go", + "internal/monitoring/ratetracker_test.go", "internal/unifiedresources/code_standards_test.go" ] } diff --git a/docs/release-control/v6/internal/subsystems/storage-recovery.md b/docs/release-control/v6/internal/subsystems/storage-recovery.md index 2b0c01fed..78ae70df7 100644 --- a/docs/release-control/v6/internal/subsystems/storage-recovery.md +++ b/docs/release-control/v6/internal/subsystems/storage-recovery.md @@ -1568,6 +1568,11 @@ recovery scope, or a storage/recovery-owned secret source. remain absent optional timestamps on that shared facet; storage and recovery must not reinterpret absence or a year-one zero-time serialization as recovery age, missed backup cadence, or restore freshness. + The same shared type file keeps `MetricValue.value` numeric whenever a + metric object is present. Storage/recovery consumers must preserve explicit + zero as an observed idle value and use absence of the metric object for + unknown Proxmox guest I/O; they must not reinterpret zero as missing or + accept a JSON `null` value as recovery, backup, or storage-health evidence. ## Forbidden Paths diff --git a/docs/release-control/v6/internal/subsystems/unified-resources.md b/docs/release-control/v6/internal/subsystems/unified-resources.md index 8146e0315..5c1d1fd39 100644 --- a/docs/release-control/v6/internal/subsystems/unified-resources.md +++ b/docs/release-control/v6/internal/subsystems/unified-resources.md @@ -55,6 +55,14 @@ at the adapter boundary; a mutation that arrives during a rebuild must apply to the newly published registry rather than a superseded pointer. Authoritative snapshot omissions still remove resources and emit the normal canonical history change. +Proxmox guest I/O projection preserves the monitoring-owned rate validity +contract. A valid idle interval publishes an explicit zero-valued `bytes/s` +metric; an unknown first, missing, partial, or out-of-order observation leaves +that metric absent. `MetricValue.value` is always a numeric JSON field when a +metric object exists, so `/api/resources`, `/api/state`, websocket resources, +history, alerts, and UI adapters never need to interpret `null` as either idle +or unknown. Unified resources must not infer validity from the numeric guest +compatibility field alone. Physical-disk resources own cross-source disk identity. When Proxmox inventory and host-agent SMART telemetry describe the same device, the merged resource must retain Proxmox node/instance source payloads while carrying SMART diff --git a/frontend-modern/src/components/Workloads/__tests__/GuestRow.test.tsx b/frontend-modern/src/components/Workloads/__tests__/GuestRow.test.tsx index 90f399670..7219ce21a 100644 --- a/frontend-modern/src/components/Workloads/__tests__/GuestRow.test.tsx +++ b/frontend-modern/src/components/Workloads/__tests__/GuestRow.test.tsx @@ -1199,6 +1199,23 @@ describe('getOutlierEmphasis (via I/O column rendering)', () => { expect(screen.getAllByText('—').length).toBeGreaterThan(0); }); + it('renders valid zero I/O rates for a running guest on an online parent', () => { + renderGuestRow({ + guest: makeGuest({ + status: 'running', + networkIn: 0, + networkOut: 0, + diskRead: 0, + diskWrite: 0, + }), + parentNodeOnline: true, + visibleColumnIds: ['name', 'netIo', 'diskIo'], + }); + + expect(screen.getAllByText('0 B/s')).toHaveLength(4); + expect(screen.queryByText('—')).toBeNull(); + }); + it('shows dash for disk I/O when guest is stopped', () => { renderGuestRow({ guest: makeGuest({ status: 'stopped', diskRead: 0, diskWrite: 0 }), diff --git a/internal/alerts/filter_evaluation.go b/internal/alerts/filter_evaluation.go index 5467db432..4db28e799 100644 --- a/internal/alerts/filter_evaluation.go +++ b/internal/alerts/filter_evaluation.go @@ -20,17 +20,18 @@ func (m *Manager) evaluateFilterCondition(guest any, condition FilterCondition) // guestMetrics holds common metrics for filter evaluation type guestMetrics struct { - CPU float64 // CPU usage as percentage (0-100) - MemUsage float64 // Memory usage percentage - DiskUsage float64 // Disk usage percentage - DiskRead int64 // Bytes/s - DiskWrite int64 // Bytes/s - NetworkIn int64 // Bytes/s - NetworkOut int64 // Bytes/s - Name string - Node string - ID string - Status string + CPU float64 // CPU usage as percentage (0-100) + MemUsage float64 // Memory usage percentage + DiskUsage float64 // Disk usage percentage + DiskRead int64 // Bytes/s + DiskWrite int64 // Bytes/s + NetworkIn int64 // Bytes/s + NetworkOut int64 // Bytes/s + IORateValidity models.IORateValidity + Name string + Node string + ID string + Status string } // extractGuestMetrics extracts common metrics from a VM or Container @@ -47,6 +48,12 @@ func evaluateGuestCondition(metrics guestMetrics, condition FilterCondition) boo switch condition.Type { case "metric": value := 0.0 + validity := metrics.IORateValidity.EffectiveForRates( + metrics.DiskRead, + metrics.DiskWrite, + metrics.NetworkIn, + metrics.NetworkOut, + ) switch strings.ToLower(condition.Field) { case "cpu": value = metrics.CPU @@ -55,12 +62,24 @@ func evaluateGuestCondition(metrics guestMetrics, condition FilterCondition) boo case "disk": value = metrics.DiskUsage case "diskread": + if !validity.DiskRead { + return false + } value = float64(metrics.DiskRead) / 1024 / 1024 // Convert to MB/s case "diskwrite": + if !validity.DiskWrite { + return false + } value = float64(metrics.DiskWrite) / 1024 / 1024 case "networkin": + if !validity.NetworkIn { + return false + } value = float64(metrics.NetworkIn) / 1024 / 1024 case "networkout": + if !validity.NetworkOut { + return false + } value = float64(metrics.NetworkOut) / 1024 / 1024 default: return false diff --git a/internal/alerts/filter_evaluation_test.go b/internal/alerts/filter_evaluation_test.go index 9183f37ec..a3c3d3497 100644 --- a/internal/alerts/filter_evaluation_test.go +++ b/internal/alerts/filter_evaluation_test.go @@ -374,6 +374,28 @@ func TestEvaluateVMCondition(t *testing.T) { } } +func TestEvaluateVMConditionTreatsUnknownIORateAsNonMatch(t *testing.T) { + manager := NewManager() + vm := models.VM{ + DiskRead: 0, + IORateValidity: models.IORateValidity{Explicit: true}, + } + condition := FilterCondition{ + Type: "metric", + Field: "diskread", + Operator: "<=", + Value: 0, + } + if manager.evaluateVMCondition(vm, condition) { + t.Fatal("unknown disk read rate matched a zero-valued filter") + } + + vm.IORateValidity.DiskRead = true + if !manager.evaluateVMCondition(vm, condition) { + t.Fatal("valid zero disk read rate did not match a zero-valued filter") + } +} + func TestEvaluateContainerCondition(t *testing.T) { // t.Parallel() m := NewManager() diff --git a/internal/alerts/guest.go b/internal/alerts/guest.go index d8a1bcef2..c0c62b3a1 100644 --- a/internal/alerts/guest.go +++ b/internal/alerts/guest.go @@ -124,10 +124,6 @@ func (m *Manager) CheckGuest(guest any, instanceName string) { cpu := snapshot.CPUPercent memUsage := snapshot.MemUsage diskUsage := snapshot.DiskUsage - diskRead := snapshot.DiskRead - diskWrite := snapshot.DiskWrite - netIn := snapshot.NetworkIn - netOut := snapshot.NetworkOut disks := snapshot.Disks // Debug logging for high memory VMs @@ -265,6 +261,7 @@ func (m *Manager) CheckGuest(guest any, instanceName string) { if !snapshot.MemoryUnavailable { memoryMetric = &UnifiedResourceMetric{Percent: memUsage} } + diskReadMetric, diskWriteMetric, networkInMetric, networkOutMetric := guestIORateMetrics(snapshot) m.evaluateUnifiedMetrics(&UnifiedResourceInput{ ID: guestID, Type: snapshot.resourceType(), @@ -274,10 +271,10 @@ func (m *Manager) CheckGuest(guest any, instanceName string) { CPU: &UnifiedResourceMetric{Percent: cpu}, Memory: memoryMetric, Disk: &UnifiedResourceMetric{Percent: diskUsage}, - DiskRead: &UnifiedResourceMetric{Value: float64(diskRead) / 1024 / 1024}, - DiskWrite: &UnifiedResourceMetric{Value: float64(diskWrite) / 1024 / 1024}, - NetworkIn: &UnifiedResourceMetric{Value: float64(netIn) / 1024 / 1024}, - NetworkOut: &UnifiedResourceMetric{Value: float64(netOut) / 1024 / 1024}, + DiskRead: diskReadMetric, + DiskWrite: diskWriteMetric, + NetworkIn: networkInMetric, + NetworkOut: networkOutMetric, }, thresholds, evalOpts) if thresholds.Disk != nil && thresholds.Disk.Trigger > 0 && len(disks) > 0 { @@ -369,6 +366,28 @@ func (m *Manager) CheckGuest(guest any, instanceName string) { } } +func guestIORateMetrics(snapshot guestSnapshot) (diskRead, diskWrite, networkIn, networkOut *UnifiedResourceMetric) { + validity := snapshot.IORateValidity.EffectiveForRates( + snapshot.DiskRead, + snapshot.DiskWrite, + snapshot.NetworkIn, + snapshot.NetworkOut, + ) + if validity.DiskRead { + diskRead = &UnifiedResourceMetric{Value: float64(snapshot.DiskRead) / 1024 / 1024} + } + if validity.DiskWrite { + diskWrite = &UnifiedResourceMetric{Value: float64(snapshot.DiskWrite) / 1024 / 1024} + } + if validity.NetworkIn { + networkIn = &UnifiedResourceMetric{Value: float64(snapshot.NetworkIn) / 1024 / 1024} + } + if validity.NetworkOut { + networkOut = &UnifiedResourceMetric{Value: float64(snapshot.NetworkOut) / 1024 / 1024} + } + return +} + // checkGuestPoweredOff creates an alert for powered-off guests. func (m *Manager) checkGuestPoweredOff(guestID, name, node, instanceName, guestType string, monitorOnly bool) { m.mu.RLock() diff --git a/internal/alerts/guest_snapshot.go b/internal/alerts/guest_snapshot.go index 5113c5f34..f9741f964 100644 --- a/internal/alerts/guest_snapshot.go +++ b/internal/alerts/guest_snapshot.go @@ -36,6 +36,7 @@ type guestSnapshot struct { DiskWrite int64 NetworkIn int64 NetworkOut int64 + IORateValidity models.IORateValidity Disks []models.Disk Tags []string @@ -80,17 +81,18 @@ func (g guestSnapshot) resourceType() string { func (g guestSnapshot) metrics() guestMetrics { return guestMetrics{ - CPU: g.CPUPercent, - MemUsage: g.MemUsage, - DiskUsage: g.DiskUsage, - DiskRead: g.DiskRead, - DiskWrite: g.DiskWrite, - NetworkIn: g.NetworkIn, - NetworkOut: g.NetworkOut, - Name: g.Name, - Node: g.Node, - ID: g.ID, - Status: g.Status, + CPU: g.CPUPercent, + MemUsage: g.MemUsage, + DiskUsage: g.DiskUsage, + DiskRead: g.DiskRead, + DiskWrite: g.DiskWrite, + NetworkIn: g.NetworkIn, + NetworkOut: g.NetworkOut, + IORateValidity: g.IORateValidity, + Name: g.Name, + Node: g.Node, + ID: g.ID, + Status: g.Status, } } @@ -112,6 +114,7 @@ func guestSnapshotFromVM(vm models.VM) guestSnapshot { DiskWrite: vm.DiskWrite, NetworkIn: vm.NetworkIn, NetworkOut: vm.NetworkOut, + IORateValidity: vm.IORateValidity, Disks: append([]models.Disk(nil), vm.Disks...), Tags: append([]string(nil), vm.Tags...), OnBoot: vm.OnBoot, @@ -136,6 +139,7 @@ func guestSnapshotFromContainer(container models.Container) guestSnapshot { DiskWrite: container.DiskWrite, NetworkIn: container.NetworkIn, NetworkOut: container.NetworkOut, + IORateValidity: container.IORateValidity, Disks: append([]models.Disk(nil), container.Disks...), Tags: append([]string(nil), container.Tags...), OnBoot: container.OnBoot, diff --git a/internal/alerts/guest_snapshot_test.go b/internal/alerts/guest_snapshot_test.go index d9a9679af..28c7e4e65 100644 --- a/internal/alerts/guest_snapshot_test.go +++ b/internal/alerts/guest_snapshot_test.go @@ -94,3 +94,27 @@ func TestExtractGuestSnapshot_UnknownTypeReturnsCanonicalEmptySnapshot(t *testin t.Fatalf("expected canonical empty guest snapshot on unknown type, got %#v", snapshot) } } + +func TestGuestIORateMetricsDistinguishValidZeroFromUnknown(t *testing.T) { + validZero := guestSnapshotFromVM(models.VM{ + IORateValidity: models.IORateValidity{ + Explicit: true, + DiskRead: true, + DiskWrite: true, + NetworkIn: true, + NetworkOut: true, + }, + }) + diskRead, diskWrite, networkIn, networkOut := guestIORateMetrics(validZero) + if diskRead == nil || diskRead.Value != 0 || diskWrite == nil || networkIn == nil || networkOut == nil { + t.Fatalf("valid zero rates were not alert candidates: %+v %+v %+v %+v", diskRead, diskWrite, networkIn, networkOut) + } + + unknown := guestSnapshotFromVM(models.VM{ + IORateValidity: models.IORateValidity{Explicit: true}, + }) + diskRead, diskWrite, networkIn, networkOut = guestIORateMetrics(unknown) + if diskRead != nil || diskWrite != nil || networkIn != nil || networkOut != nil { + t.Fatalf("unknown rates became alert candidates: %+v %+v %+v %+v", diskRead, diskWrite, networkIn, networkOut) + } +} diff --git a/internal/api/router_state_test.go b/internal/api/router_state_test.go index 464ff6e32..9526bd697 100644 --- a/internal/api/router_state_test.go +++ b/internal/api/router_state_test.go @@ -4,6 +4,7 @@ import ( "encoding/json" "net/http" "net/http/httptest" + "strings" "testing" "time" @@ -107,6 +108,83 @@ func TestRouter_HandleState_MockIsolation(t *testing.T) { }) } +func TestRouterHandleStatePreservesNumericIdleRatesAndOmitsUnknownRates(t *testing.T) { + dataPath := t.TempDir() + hp, err := auth.HashPassword("password") + if err != nil { + t.Fatalf("hash password: %v", err) + } + cfg := &config.Config{ + DataPath: dataPath, + AuthUser: "admin", + AuthPass: hp, + } + InitSessionStore(dataPath) + InitCSRFStore(dataPath) + + monitor, state, _ := newTestMonitor(t) + state.VMs = []models.VM{ + { + ID: "site-a:pve-a:100", + VMID: 100, + Name: "idle-vm", + Node: "pve-a", + Instance: "site-a", + Status: "running", + IORateValidity: models.IORateValidity{ + Explicit: true, + DiskRead: true, + DiskWrite: true, + NetworkIn: true, + NetworkOut: true, + }, + }, + { + ID: "site-a:pve-a:101", + VMID: 101, + Name: "unknown-vm", + Node: "pve-a", + Instance: "site-a", + Status: "running", + IORateValidity: models.IORateValidity{ + Explicit: true, + }, + }, + } + syncTestResourceStore(t, monitor, state) + + router := &Router{config: cfg, monitor: monitor} + rec := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodGet, "/api/state", nil) + req.SetBasicAuth("admin", "password") + router.handleState(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("/api/state status = %d, body=%s", rec.Code, rec.Body.String()) + } + + var payload struct { + Resources []models.ResourceFrontend `json:"resources"` + } + if err := json.NewDecoder(rec.Body).Decode(&payload); err != nil { + t.Fatalf("decode /api/state: %v", err) + } + + byName := make(map[string]models.ResourceFrontend, len(payload.Resources)) + for _, resource := range payload.Resources { + byName[resource.Name] = resource + } + idle := byName["idle-vm"] + if idle.DiskIO == nil || idle.DiskIO.ReadRate != 0 || idle.DiskIO.WriteRate != 0 { + t.Fatalf("valid idle rates were not emitted as numeric zero: %+v", idle.DiskIO) + } + if unknown := byName["unknown-vm"]; unknown.DiskIO != nil { + t.Fatalf("unknown rates projected a disk I/O object: %+v", unknown.DiskIO) + } + if strings.Contains(rec.Body.String(), `"readRate":null`) || strings.Contains(rec.Body.String(), `"writeRate":null`) { + t.Fatalf("/api/state emitted unstable null rates: %s", rec.Body.String()) + } +} + func TestRouter_HandleStateSummary(t *testing.T) { dataPath := t.TempDir() hp, _ := auth.HashPassword("password") diff --git a/internal/models/metrics_types.go b/internal/models/metrics_types.go index d18d8ef9e..b64e8e631 100644 --- a/internal/models/metrics_types.go +++ b/internal/models/metrics_types.go @@ -10,10 +10,83 @@ type MetricPoint struct { // IOMetrics represents I/O metrics at a point in time type IOMetrics struct { - DiskRead int64 `json:"diskRead"` - DiskWrite int64 `json:"diskWrite"` - DiskBusy int64 `json:"diskBusy,omitempty"` - NetworkIn int64 `json:"networkIn"` - NetworkOut int64 `json:"networkOut"` - Timestamp time.Time `json:"timestamp"` + DiskRead int64 `json:"diskRead"` + DiskWrite int64 `json:"diskWrite"` + DiskBusy int64 `json:"diskBusy,omitempty"` + NetworkIn int64 `json:"networkIn"` + NetworkOut int64 `json:"networkOut"` + Timestamp time.Time `json:"timestamp"` + Presence IOCounterPresence `json:"-"` + ObservedAt IOCounterObservationTimes `json:"-"` + // SourceUptime is an optional counter-epoch hint. A decrease proves the + // source restarted even when a busy guest has already surpassed its old + // counter value before the next poll. + SourceUptime uint64 `json:"-"` +} + +// IOCounterObservationTimes keeps the receipt time of each independently +// sampled counter. A zero field falls back to IOMetrics.Timestamp for legacy +// producers that obtain every counter in one response. +type IOCounterObservationTimes struct { + DiskRead time.Time + DiskWrite time.Time + DiskBusy time.Time + NetworkIn time.Time + NetworkOut time.Time +} + +// IOCounterPresence distinguishes a counter that was explicitly observed at +// zero from one that was absent in the upstream sample. Explicit=false keeps +// legacy producers compatible by treating every counter as present. +type IOCounterPresence struct { + Explicit bool + DiskRead bool + DiskWrite bool + DiskBusy bool + NetworkIn bool + NetworkOut bool +} + +// Effective returns the presence contract used by rate calculation. +func (p IOCounterPresence) Effective() IOCounterPresence { + if p.Explicit { + return p + } + return IOCounterPresence{ + Explicit: true, + DiskRead: true, + DiskWrite: true, + DiskBusy: true, + NetworkIn: true, + NetworkOut: true, + } +} + +// IORateValidity records which numeric guest rate fields represent an +// observed rate. It is intentionally excluded from JSON: legacy API and +// websocket contracts continue to expose stable numbers, while internal +// history, alerting, and unified-resource consumers can preserve unknown. +type IORateValidity struct { + Explicit bool + DiskRead bool + DiskWrite bool + NetworkIn bool + NetworkOut bool +} + +// EffectiveForRates preserves compatibility with legacy producers that did +// not carry explicit validity. Their non-zero values are usable evidence, but +// a legacy zero remains ambiguous and must not override a supplemental source +// or prove alert recovery. +func (v IORateValidity) EffectiveForRates(diskRead, diskWrite, networkIn, networkOut int64) IORateValidity { + if v.Explicit { + return v + } + return IORateValidity{ + Explicit: true, + DiskRead: diskRead != 0, + DiskWrite: diskWrite != 0, + NetworkIn: networkIn != 0, + NetworkOut: networkOut != 0, + } } diff --git a/internal/models/metrics_types_test.go b/internal/models/metrics_types_test.go index 4090c1c06..a8d97887c 100644 --- a/internal/models/metrics_types_test.go +++ b/internal/models/metrics_types_test.go @@ -2,6 +2,7 @@ package models import ( "encoding/json" + "strings" "testing" "time" ) @@ -26,6 +27,7 @@ func TestIOMetrics_Fields(t *testing.T) { NetworkIn: 3000, NetworkOut: 4000, Timestamp: now, + ObservedAt: IOCounterObservationTimes{DiskRead: now}, } if metrics.DiskRead != 1000 { @@ -116,4 +118,60 @@ func TestIOMetrics_JSONSerializationUsesCamelCaseFields(t *testing.T) { if _, ok := decoded["Timestamp"]; ok { t.Error("expected Timestamp key to be absent") } + if _, ok := decoded["ObservedAt"]; ok { + t.Error("expected per-counter observation times to be absent") + } +} + +func TestIORateValidityLegacyFallbackTreatsOnlyNonZeroRatesAsKnown(t *testing.T) { + validity := (IORateValidity{}).EffectiveForRates(1024, 0, 2048, 0) + if !validity.DiskRead || validity.DiskWrite || !validity.NetworkIn || validity.NetworkOut { + t.Fatalf("legacy inferred validity = %+v", validity) + } + + explicitIdle := IORateValidity{ + Explicit: true, + DiskRead: true, + DiskWrite: true, + NetworkIn: true, + NetworkOut: true, + }.EffectiveForRates(0, 0, 0, 0) + if !explicitIdle.DiskRead || !explicitIdle.DiskWrite || !explicitIdle.NetworkIn || !explicitIdle.NetworkOut { + t.Fatalf("explicit idle validity was not preserved: %+v", explicitIdle) + } +} + +func TestUnknownGuestRatesRemainNumericOnAPIAndWebsocketShapes(t *testing.T) { + vm := VM{ + ID: "site:node:100", + DiskRead: 0, + DiskWrite: 0, + IORateValidity: IORateValidity{ + Explicit: true, + }, + } + modelPayload, err := json.Marshal(vm) + if err != nil { + t.Fatal(err) + } + if strings.Contains(string(modelPayload), "IORateValidity") || strings.Contains(string(modelPayload), "ioRateValidity") { + t.Fatalf("internal validity leaked into guest model JSON: %s", modelPayload) + } + + payload, err := json.Marshal(vm.ToFrontend()) + if err != nil { + t.Fatal(err) + } + wire := string(payload) + for _, field := range []string{`"diskRead":0`, `"diskWrite":0`, `"networkIn":0`, `"networkOut":0`} { + if !strings.Contains(wire, field) { + t.Fatalf("wire payload %s does not contain numeric field %s", wire, field) + } + } + if strings.Contains(wire, "null") { + t.Fatalf("guest wire payload contains unstable null: %s", wire) + } + if strings.Contains(wire, "IORateValidity") { + t.Fatalf("internal validity leaked into wire payload: %s", wire) + } } diff --git a/internal/models/models.go b/internal/models/models.go index c14bcd4d9..d092acb87 100644 --- a/internal/models/models.go +++ b/internal/models/models.go @@ -162,6 +162,7 @@ type VM struct { NetworkOut int64 `json:"networkOut"` DiskRead int64 `json:"diskRead"` DiskWrite int64 `json:"diskWrite"` + IORateValidity IORateValidity `json:"-"` Uptime int64 `json:"uptime"` Template bool `json:"template"` OnBoot *bool `json:"onBoot,omitempty"` @@ -209,6 +210,7 @@ type Container struct { NetworkOut int64 `json:"networkOut"` DiskRead int64 `json:"diskRead"` DiskWrite int64 `json:"diskWrite"` + IORateValidity IORateValidity `json:"-"` Uptime int64 `json:"uptime"` Template bool `json:"template"` OnBoot *bool `json:"onBoot,omitempty"` diff --git a/internal/monitoring/canonical_guardrails_test.go b/internal/monitoring/canonical_guardrails_test.go index 86e0e44d7..6e8e823e2 100644 --- a/internal/monitoring/canonical_guardrails_test.go +++ b/internal/monitoring/canonical_guardrails_test.go @@ -1819,7 +1819,7 @@ func TestProxmoxGuestPollersCarryPoolIntoCanonicalModels(t *testing.T) { requiredSnippets := map[string][]string{ "monitor_pve_guest_builders.go": {"Pool: strings.TrimSpace(res.Pool)"}, "monitor_pve_guest_lxc.go": {"Pool: strings.TrimSpace(res.Pool)"}, - "monitor_pve_node_vm_builder.go": {"Pool: vm.Pool"}, + "monitor_pve_node_vm_builder.go": {"Pool: vm.Pool"}, "monitor_polling_containers.go": {"Pool: strings.TrimSpace(container.Pool)"}, } @@ -2021,14 +2021,16 @@ func TestMockNativePollersDeferToCanonicalMockSampler(t *testing.T) { file: "monitor_polling_vm.go", snippets: []string{ "if !shouldSkipNativeMockStateMetricWrites() {", - `m.recordGuestMetric("vm", vm.ID, unifiedresources.ProxmoxGuestCPUPercent(vm.CPU), historyMemoryUsage(vm.Memory), vm.Disk.Usage, -1, -1, -1, -1, now)`, + "guestHistoryRates(", + "m.recordGuestMetric(", }, }, { file: "monitor_polling_containers.go", snippets: []string{ "if !shouldSkipNativeMockStateMetricWrites() {", - `m.recordGuestMetric("container", ct.ID, unifiedresources.ProxmoxGuestCPUPercent(ct.CPU), historyMemoryUsage(ct.Memory), ct.Disk.Usage, -1, -1, -1, -1, now)`, + "guestHistoryRates(", + "m.recordGuestMetric(", }, }, { diff --git a/internal/monitoring/issue1613_contract_test.go b/internal/monitoring/issue1613_contract_test.go new file mode 100644 index 000000000..c3ebd73e9 --- /dev/null +++ b/internal/monitoring/issue1613_contract_test.go @@ -0,0 +1,180 @@ +package monitoring + +import ( + "encoding/json" + "strings" + "testing" + "time" + + "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" +) + +func TestIssue1613GuestRateIdentitySurvivesMigrationAndIsolatesDuplicates(t *testing.T) { + beforeMigration := makeGuestRateKey("site-a", "qemu", 100) + afterMigration := makeGuestRateKey("site-a", "qemu", 100) + if beforeMigration != afterMigration { + t.Fatalf("migration changed rate key: %q != %q", beforeMigration, afterMigration) + } + if beforeMigration == makeGuestRateKey("site-b", "qemu", 100) { + t.Fatal("duplicate cluster registrations shared a rate key") + } + if beforeMigration == makeGuestRateKey("site-a", "lxc", 100) { + t.Fatal("QEMU and LXC with the same VMID shared a rate key") + } +} + +func TestIssue1613QEMUStatusUsesPresentLowerCounterAndKeepsMissingListingCounter(t *testing.T) { + listingAt := time.Unix(1_700_000_000, 0) + statusAt := listingAt.Add(3 * time.Second) + state := vmBuildState{ + diskReadBytes: 9_000, + diskWriteBytes: 8_000, + networkInBytes: 7_000, + networkOutBytes: 6_000, + counterPresence: models.IOCounterPresence{Explicit: true, DiskRead: true, DiskWrite: true, NetworkIn: true, NetworkOut: true}, + counterObservedAt: listingAt, + counterObservationTimes: models.IOCounterObservationTimes{ + DiskRead: listingAt, + DiskWrite: listingAt, + NetworkIn: listingAt, + NetworkOut: listingAt, + }, + counterUptime: 10_000, + } + mergeVMRuntimeCounters(&state, &proxmox.VMStatus{ + DiskRead: 0, + NetOut: 500, + Uptime: 20, + IOCounters: proxmox.IOCounterPresence{ + Explicit: true, + DiskRead: true, + NetworkOut: true, + }, + ObservedAt: statusAt, + }) + + if state.diskReadBytes != 0 || state.networkOutBytes != 500 { + t.Fatalf("present status counters did not replace listing values: %+v", state) + } + if state.diskWriteBytes != 8_000 || state.networkInBytes != 7_000 { + t.Fatalf("missing status counters replaced listing values: %+v", state) + } + if !state.counterObservationTimes.DiskRead.Equal(statusAt) || + !state.counterObservationTimes.NetworkOut.Equal(statusAt) || + !state.counterObservationTimes.DiskWrite.Equal(listingAt) || + !state.counterObservationTimes.NetworkIn.Equal(listingAt) { + t.Fatalf("per-counter receipt authority was not preserved: %+v", state.counterObservationTimes) + } + if state.counterUptime != 20 { + t.Fatalf("status uptime = %d, want 20", state.counterUptime) + } +} + +func TestIssue1613PVEGraceTracksConfiguredPollingCadence(t *testing.T) { + for _, test := range []struct { + interval time.Duration + want time.Duration + }{ + {interval: 30 * time.Second, want: 60 * time.Second}, + {interval: 60 * time.Second, want: 120 * time.Second}, + {interval: 90 * time.Second, want: 180 * time.Second}, + } { + monitor := &Monitor{config: &config.Config{PVEPollingInterval: test.interval}} + if got := monitor.pveNodeOfflineGracePeriod(); got != test.want { + t.Fatalf("interval %s grace = %s, want %s", test.interval, got, test.want) + } + } +} + +func TestIssue1613HistoryWritesValidZeroAndSkipsUnknownRates(t *testing.T) { + monitor := &Monitor{metricsHistory: NewMetricsHistory(32, time.Hour)} + now := time.Now() + + diskRead, diskWrite, networkIn, networkOut := guestHistoryRates(0, 0, 0, 0, models.IORateValidity{Explicit: true}) + monitor.recordGuestMetric("vm", "unknown", 0, 0, 0, diskRead, diskWrite, networkIn, networkOut, now) + if points := monitor.metricsHistory.GetGuestMetrics("unknown", "diskread", time.Hour); len(points) != 0 { + t.Fatalf("unknown disk read rate was written to history: %+v", points) + } + + diskRead, diskWrite, networkIn, networkOut = guestHistoryRates(0, 0, 0, 0, models.IORateValidity{ + Explicit: true, + DiskRead: true, + DiskWrite: true, + NetworkIn: true, + NetworkOut: true, + }) + monitor.recordGuestMetric("vm", "idle", 0, 0, 0, diskRead, diskWrite, networkIn, networkOut, now) + points := monitor.metricsHistory.GetGuestMetrics("idle", "diskread", time.Hour) + if len(points) != 1 || points[0].Value != 0 { + t.Fatalf("valid idle disk read rate was not written as zero: %+v", points) + } +} + +func TestIssue1613NodeDoesNotGreyBetweenNinetySecondPolls(t *testing.T) { + instance := &config.PVEInstance{Name: "site-a"} + monitor := &Monitor{ + config: &config.Config{PVEPollingInterval: 90 * time.Second}, + nodeLastOnline: map[string]time.Time{"site-a-pve-a": time.Now().Add(-100 * time.Second)}, + } + + _, status := monitor.determineNodeIDAndStatus("site-a", instance, proxmox.Node{ + Node: "pve-a", + Status: "offline", + }) + if status != "online" { + t.Fatalf("node status = %q inside cadence grace, want online", status) + } + + monitor.nodeLastOnline["site-a-pve-a"] = time.Now().Add(-181 * time.Second) + _, status = monitor.determineNodeIDAndStatus("site-a", instance, proxmox.Node{ + Node: "pve-a", + Status: "offline", + }) + if status != "offline" { + t.Fatalf("node status = %q after cadence grace, want offline", status) + } +} + +func TestIssue1613WebsocketStateKeepsUnknownRatesNumeric(t *testing.T) { + monitor := &Monitor{ + state: models.NewState(), + resourceStore: &resourceOnlyStore{resources: []unifiedresources.Resource{ + { + ID: "site-a:pve-a:100", + Type: unifiedresources.ResourceTypeVM, + Name: "idle-vm", + Status: unifiedresources.StatusOnline, + Sources: []unifiedresources.DataSource{unifiedresources.SourceProxmox}, + Metrics: &unifiedresources.ResourceMetrics{ + DiskRead: &unifiedresources.MetricValue{Value: 0, Unit: "bytes/s"}, + }, + }, + { + ID: "site-a:pve-a:101", + Type: unifiedresources.ResourceTypeVM, + Name: "unknown-vm", + Status: unifiedresources.StatusOnline, + Sources: []unifiedresources.DataSource{unifiedresources.SourceProxmox}, + Metrics: &unifiedresources.ResourceMetrics{}, + }, + }}, + } + frontend := monitor.BuildBroadcastFrontendState() + payload, err := json.Marshal(frontend) + if err != nil { + t.Fatal(err) + } + wire := string(payload) + if !strings.Contains(wire, `"diskIO":{"readRate":0,"writeRate":0}`) { + t.Fatalf("websocket payload does not contain numeric valid zero disk rate: %s", wire) + } + if strings.Count(wire, `"diskIO"`) != 1 { + t.Fatalf("unknown rate projected a disk I/O object: %s", wire) + } + if strings.Contains(wire, `"diskWrite":null`) || strings.Contains(wire, `"netIn":null`) { + t.Fatalf("websocket payload contains unstable null I/O values: %s", wire) + } +} diff --git a/internal/monitoring/memory_source_catalog_test.go b/internal/monitoring/memory_source_catalog_test.go index 8b2874589..d705e9ab8 100644 --- a/internal/monitoring/memory_source_catalog_test.go +++ b/internal/monitoring/memory_source_catalog_test.go @@ -146,7 +146,8 @@ func TestMockVMPollingDefersMemoryHistoryToCanonicalSampler(t *testing.T) { requiredSnippets := []string{ "if !shouldSkipNativeMockStateMetricWrites() {", - `m.recordGuestMetric("vm", vm.ID, unifiedresources.ProxmoxGuestCPUPercent(vm.CPU), historyMemoryUsage(vm.Memory), vm.Disk.Usage, -1, -1, -1, -1, now)`, + "guestHistoryRates(", + "m.recordGuestMetric(", } for _, snippet := range requiredSnippets { if !strings.Contains(source, snippet) { diff --git a/internal/monitoring/monitor_helpers.go b/internal/monitoring/monitor_helpers.go index 3216e8b98..dc622708f 100644 --- a/internal/monitoring/monitor_helpers.go +++ b/internal/monitoring/monitor_helpers.go @@ -12,6 +12,7 @@ import ( agentsdocker "github.com/rcourtman/pulse-go-rewrite/pkg/agents/docker" agentshost "github.com/rcourtman/pulse-go-rewrite/pkg/agents/host" "github.com/rcourtman/pulse-go-rewrite/pkg/diskinventory" + "github.com/rcourtman/pulse-go-rewrite/pkg/proxmox" "github.com/rs/zerolog/log" ) @@ -62,6 +63,62 @@ func makeGuestID(instanceName string, node string, vmid int) string { return fmt.Sprintf("%s:%s:%d", instanceName, node, vmid) } +// makeGuestRateKey identifies one counter stream independently of placement. +// A live migration changes the canonical row ID's node coordinate but not the +// cumulative counter epoch. The configured instance remains in the key so +// duplicate cluster registrations never race on one baseline. +func makeGuestRateKey(instanceName, guestType string, vmid int) string { + return fmt.Sprintf("pve:%s:%s:%d", instanceName, strings.ToLower(strings.TrimSpace(guestType)), vmid) +} + +func pveCounterPresence(p proxmox.IOCounterPresence) models.IOCounterPresence { + effective := p.Effective() + return models.IOCounterPresence{ + Explicit: true, + DiskRead: effective.DiskRead, + DiskWrite: effective.DiskWrite, + NetworkIn: effective.NetworkIn, + NetworkOut: effective.NetworkOut, + } +} + +func observedAtOr(value, fallback time.Time) time.Time { + if !value.IsZero() { + return value + } + return fallback +} + +func counterObservationTimes(observedAt time.Time) models.IOCounterObservationTimes { + return models.IOCounterObservationTimes{ + DiskRead: observedAt, + DiskWrite: observedAt, + NetworkIn: observedAt, + NetworkOut: observedAt, + } +} + +func numericGuestRate(rate float64) (int64, bool) { + if rate < 0 || math.IsNaN(rate) || math.IsInf(rate, 0) { + return 0, false + } + return max(0, int64(rate)), true +} + +func guestRateValues(diskRead, diskWrite, networkIn, networkOut float64) (int64, int64, int64, int64, models.IORateValidity) { + diskReadValue, diskReadKnown := numericGuestRate(diskRead) + diskWriteValue, diskWriteKnown := numericGuestRate(diskWrite) + networkInValue, networkInKnown := numericGuestRate(networkIn) + networkOutValue, networkOutKnown := numericGuestRate(networkOut) + return diskReadValue, diskWriteValue, networkInValue, networkOutValue, models.IORateValidity{ + Explicit: true, + DiskRead: diskReadKnown, + DiskWrite: diskWriteKnown, + NetworkIn: networkInKnown, + NetworkOut: networkOutKnown, + } +} + // parseBoolEnv parses a boolean from an environment variable, returning defaultVal if not set or invalid func parseBoolEnv(key string, defaultVal bool) bool { val := strings.TrimSpace(os.Getenv(key)) diff --git a/internal/monitoring/monitor_pbs_fallback_test.go b/internal/monitoring/monitor_pbs_fallback_test.go index 80e8abbdf..40cfa9de6 100644 --- a/internal/monitoring/monitor_pbs_fallback_test.go +++ b/internal/monitoring/monitor_pbs_fallback_test.go @@ -65,6 +65,7 @@ func TestMonitor_PollPBSInstance_Fallback_Extra(t *testing.T) { }, state: models.NewState(), stalenessTracker: NewStalenessTracker(nil), + pollStatusMap: make(map[string]*pollStatus), } ctx := context.Background() diff --git a/internal/monitoring/monitor_polling_containers.go b/internal/monitoring/monitor_polling_containers.go index 6df34aaa7..19cb2047b 100644 --- a/internal/monitoring/monitor_polling_containers.go +++ b/internal/monitoring/monitor_polling_containers.go @@ -105,12 +105,16 @@ func (m *Monitor) collectContainersWithNodes(ctx context.Context, instanceName s guestID := makeGuestID(instanceName, n.Node, int(container.VMID)) sampleTime := time.Now() + counterObservedAt := observedAtOr(container.ObservedAt, sampleTime) currentMetrics := IOMetrics{ - DiskRead: int64(container.DiskRead), - DiskWrite: int64(container.DiskWrite), - NetworkIn: int64(container.NetIn), - NetworkOut: int64(container.NetOut), - Timestamp: sampleTime, + DiskRead: int64(container.DiskRead), + DiskWrite: int64(container.DiskWrite), + NetworkIn: int64(container.NetIn), + NetworkOut: int64(container.NetOut), + Timestamp: counterObservedAt, + Presence: pveCounterPresence(container.IOCounters), + ObservedAt: counterObservationTimes(counterObservedAt), + SourceUptime: uint64(container.Uptime), } statusSnapshot := (*proxmox.Container)(nil) if container.Status == "running" { @@ -124,7 +128,16 @@ func (m *Monitor) collectContainersWithNodes(ctx context.Context, instanceName s ) currentMetrics = mergeContainerRuntimeCounters(currentMetrics, statusSnapshot) } - diskReadRate, diskWriteRate, netInRate, netOutRate := m.rateTracker.CalculateRates(guestID, currentMetrics) + diskReadRate, diskWriteRate, netInRate, netOutRate := m.rateTracker.CalculateRates( + makeGuestRateKey(instanceName, "lxc", int(container.VMID)), + currentMetrics, + ) + diskReadValue, diskWriteValue, networkInValue, networkOutValue, rateValidity := guestRateValues( + diskReadRate, + diskWriteRate, + netInRate, + netOutRate, + ) // Set CPU to 0 for non-running containers cpuUsage := safeFloat(container.CPU) @@ -201,14 +214,15 @@ func (m *Monitor) collectContainersWithNodes(ctx context.Context, instanceName s Free: diskFreeBytes, Usage: diskUsagePercent, }, - NetworkIn: max(0, int64(netInRate)), - NetworkOut: max(0, int64(netOutRate)), - DiskRead: max(0, int64(diskReadRate)), - DiskWrite: max(0, int64(diskWriteRate)), - Uptime: int64(container.Uptime), - Template: container.Template == 1, - LastSeen: sampleTime, - Tags: tags, + NetworkIn: networkInValue, + NetworkOut: networkOutValue, + DiskRead: diskReadValue, + DiskWrite: diskWriteValue, + IORateValidity: rateValidity, + Uptime: int64(container.Uptime), + Template: container.Template == 1, + LastSeen: sampleTime, + Tags: tags, } if prevContainerIsOCI[modelContainer.VMID] { @@ -258,6 +272,13 @@ func (m *Monitor) collectContainersWithNodes(ctx context.Context, instanceName s modelContainer.NetworkOut = 0 modelContainer.DiskRead = 0 modelContainer.DiskWrite = 0 + modelContainer.IORateValidity = models.IORateValidity{ + Explicit: true, + DiskRead: true, + DiskWrite: true, + NetworkIn: true, + NetworkOut: true, + } } // Trigger guest metadata migration if old format exists @@ -349,9 +370,25 @@ func (m *Monitor) collectContainersWithNodes(ctx context.Context, instanceName s if ct.Status != "running" { continue } - // IO/network series are not recorded on the traditional polling - // path (parity with the historical inline writes). - m.recordGuestMetric("container", ct.ID, unifiedresources.ProxmoxGuestCPUPercent(ct.CPU), historyMemoryUsage(ct.Memory), ct.Disk.Usage, -1, -1, -1, -1, now) + diskRead, diskWrite, networkIn, networkOut := guestHistoryRates( + ct.DiskRead, + ct.DiskWrite, + ct.NetworkIn, + ct.NetworkOut, + ct.IORateValidity, + ) + m.recordGuestMetric( + "container", + ct.ID, + unifiedresources.ProxmoxGuestCPUPercent(ct.CPU), + historyMemoryUsage(ct.Memory), + ct.Disk.Usage, + diskRead, + diskWrite, + networkIn, + networkOut, + now, + ) } } diff --git a/internal/monitoring/monitor_polling_node_helpers.go b/internal/monitoring/monitor_polling_node_helpers.go index 2b01497ee..b275f5e65 100644 --- a/internal/monitoring/monitor_polling_node_helpers.go +++ b/internal/monitoring/monitor_polling_node_helpers.go @@ -47,6 +47,7 @@ func (m *Monitor) determineNodeIDAndStatus(instanceName string, instanceCfg *con } effectiveStatus := node.Status now := time.Now() + gracePeriod := m.pveNodeOfflineGracePeriod() m.mu.Lock() if strings.ToLower(node.Status) == "online" { @@ -55,14 +56,14 @@ func (m *Monitor) determineNodeIDAndStatus(instanceName string, instanceCfg *con } else { // Node is reported as offline - check grace period lastOnline, exists := m.nodeLastOnline[nodeID] - if exists && now.Sub(lastOnline) < nodeOfflineGracePeriod { + if exists && now.Sub(lastOnline) < gracePeriod { // Still within grace period - preserve online status effectiveStatus = "online" log.Debug(). Str("instance", instanceName). Str("node", node.Node). Dur("timeSinceOnline", now.Sub(lastOnline)). - Dur("gracePeriod", nodeOfflineGracePeriod). + Dur("gracePeriod", gracePeriod). Msg("Node offline but within grace period - preserving online status") } else { // Grace period expired or never seen online - mark as offline diff --git a/internal/monitoring/monitor_polling_vm.go b/internal/monitoring/monitor_polling_vm.go index d4a6c0f87..12f2ba9b6 100644 --- a/internal/monitoring/monitor_polling_vm.go +++ b/internal/monitoring/monitor_polling_vm.go @@ -126,9 +126,25 @@ func (m *Monitor) collectVMsWithNodes(ctx context.Context, instanceName string, if vm.Status != "running" { continue } - // IO/network series are not recorded on the traditional polling - // path (parity with the historical inline writes). - m.recordGuestMetric("vm", vm.ID, unifiedresources.ProxmoxGuestCPUPercent(vm.CPU), historyMemoryUsage(vm.Memory), vm.Disk.Usage, -1, -1, -1, -1, now) + diskRead, diskWrite, networkIn, networkOut := guestHistoryRates( + vm.DiskRead, + vm.DiskWrite, + vm.NetworkIn, + vm.NetworkOut, + vm.IORateValidity, + ) + m.recordGuestMetric( + "vm", + vm.ID, + unifiedresources.ProxmoxGuestCPUPercent(vm.CPU), + historyMemoryUsage(vm.Memory), + vm.Disk.Usage, + diskRead, + diskWrite, + networkIn, + networkOut, + now, + ) } } diff --git a/internal/monitoring/monitor_pve.go b/internal/monitoring/monitor_pve.go index 4fa2984c4..9c2bbdcea 100644 --- a/internal/monitoring/monitor_pve.go +++ b/internal/monitoring/monitor_pve.go @@ -63,10 +63,35 @@ func mergeContainerRuntimeCounters(current IOMetrics, status *proxmox.Container) return current } - current.DiskRead = max(current.DiskRead, int64(status.DiskRead)) - current.DiskWrite = max(current.DiskWrite, int64(status.DiskWrite)) - current.NetworkIn = max(current.NetworkIn, int64(status.NetIn)) - current.NetworkOut = max(current.NetworkOut, int64(status.NetOut)) + currentPresence := current.Presence.Effective() + statusPresence := status.IOCounters.Effective() + if statusPresence.DiskRead { + current.DiskRead = int64(status.DiskRead) + currentPresence.DiskRead = true + current.ObservedAt.DiskRead = status.ObservedAt + } + if statusPresence.DiskWrite { + current.DiskWrite = int64(status.DiskWrite) + currentPresence.DiskWrite = true + current.ObservedAt.DiskWrite = status.ObservedAt + } + if statusPresence.NetworkIn { + current.NetworkIn = int64(status.NetIn) + currentPresence.NetworkIn = true + current.ObservedAt.NetworkIn = status.ObservedAt + } + if statusPresence.NetworkOut { + current.NetworkOut = int64(status.NetOut) + currentPresence.NetworkOut = true + current.ObservedAt.NetworkOut = status.ObservedAt + } + current.Presence = currentPresence + if !status.ObservedAt.IsZero() { + current.Timestamp = status.ObservedAt + } + if status.Uptime > 0 { + current.SourceUptime = status.Uptime + } return current } @@ -712,6 +737,7 @@ func (m *Monitor) placeholderNodesForInstance(instanceName string) []models.Node func (m *Monitor) preserveOrExpireNodes(prevInstanceNodes []models.Node) []models.Node { preserved := make([]models.Node, 0, len(prevInstanceNodes)) now := time.Now() + gracePeriod := m.pveNodeOfflineGracePeriod() for _, prevNode := range prevInstanceNodes { nodeCopy := prevNode @@ -729,9 +755,9 @@ func (m *Monitor) preserveOrExpireNodes(prevInstanceNodes []models.Node) []model lastOnline, sawOnline := m.nodeLastOnline[prevNode.ID] m.mu.Unlock() - withinGrace := sawOnline && now.Sub(lastOnline) < nodeOfflineGracePeriod + withinGrace := sawOnline && now.Sub(lastOnline) < gracePeriod if !withinGrace && strings.EqualFold(strings.TrimSpace(prevNode.Status), "online") { - withinGrace = !prevNode.LastSeen.IsZero() && now.Sub(prevNode.LastSeen) < nodeOfflineGracePeriod + withinGrace = !prevNode.LastSeen.IsZero() && now.Sub(prevNode.LastSeen) < gracePeriod } if withinGrace { if strings.TrimSpace(nodeCopy.Status) == "" || strings.EqualFold(nodeCopy.Status, "offline") { diff --git a/internal/monitoring/monitor_pve_guest_builders.go b/internal/monitoring/monitor_pve_guest_builders.go index b84f9e2d2..af16aa7db 100644 --- a/internal/monitoring/monitor_pve_guest_builders.go +++ b/internal/monitoring/monitor_pve_guest_builders.go @@ -13,30 +13,34 @@ import ( ) type vmBuildState struct { - memTotal uint64 - memUsed uint64 - memorySource string - guestRaw VMMemoryRaw - diskReadBytes int64 - diskWriteBytes int64 - networkInBytes int64 - networkOutBytes int64 - diskTotal uint64 - diskUsed uint64 - diskFree uint64 - diskUsage float64 - diskFromAgent bool - diskStatusReason string - guestAgentStatus string - guestAgentExpected bool - individualDisks []models.Disk - ipAddresses []string - networkInterfaces []models.GuestNetworkInterface - osName string - osVersion string - agentVersion string - detailedStatus *proxmox.VMStatus - onBoot *bool + memTotal uint64 + memUsed uint64 + memorySource string + guestRaw VMMemoryRaw + diskReadBytes int64 + diskWriteBytes int64 + networkInBytes int64 + networkOutBytes int64 + counterPresence models.IOCounterPresence + counterObservedAt time.Time + counterObservationTimes models.IOCounterObservationTimes + counterUptime uint64 + diskTotal uint64 + diskUsed uint64 + diskFree uint64 + diskUsage float64 + diskFromAgent bool + diskStatusReason string + guestAgentStatus string + guestAgentExpected bool + individualDisks []models.Disk + ipAddresses []string + networkInterfaces []models.GuestNetworkInterface + osName string + osVersion string + agentVersion string + detailedStatus *proxmox.VMStatus + onBoot *bool } func (m *Monitor) applyVMStatusDetails( @@ -71,11 +75,7 @@ func (m *Monitor) applyVMStatusDetails( &state.guestRaw, ) - // Use actual disk I/O values from detailed status - state.diskReadBytes = int64(status.DiskRead) - state.diskWriteBytes = int64(status.DiskWrite) - state.networkInBytes = int64(status.NetIn) - state.networkOutBytes = int64(status.NetOut) + mergeVMRuntimeCounters(state, status) // Gather guest metadata from the agent when available guestIPs, guestIfaces, guestOSName, guestOSVersion, guestAgentVersion := m.fetchGuestAgentMetadata(ctx, client, instanceName, res.Node, res.Name, res.VMID, status, false) @@ -132,6 +132,42 @@ func (m *Monitor) applyVMStatusDetails( } +func mergeVMRuntimeCounters(state *vmBuildState, status *proxmox.VMStatus) { + if state == nil || status == nil { + return + } + + // A detailed status response is newer and more authoritative than the + // cluster listing, but only for fields it actually contains. + statusPresence := status.IOCounters.Effective() + if statusPresence.DiskRead { + state.diskReadBytes = int64(status.DiskRead) + state.counterPresence.DiskRead = true + state.counterObservationTimes.DiskRead = status.ObservedAt + } + if statusPresence.DiskWrite { + state.diskWriteBytes = int64(status.DiskWrite) + state.counterPresence.DiskWrite = true + state.counterObservationTimes.DiskWrite = status.ObservedAt + } + if statusPresence.NetworkIn { + state.networkInBytes = int64(status.NetIn) + state.counterPresence.NetworkIn = true + state.counterObservationTimes.NetworkIn = status.ObservedAt + } + if statusPresence.NetworkOut { + state.networkOutBytes = int64(status.NetOut) + state.counterPresence.NetworkOut = true + state.counterObservationTimes.NetworkOut = status.ObservedAt + } + if !status.ObservedAt.IsZero() { + state.counterObservedAt = status.ObservedAt + } + if status.Uptime > 0 { + state.counterUptime = status.Uptime + } +} + func vmGuestAgentRuntimeState(status *proxmox.VMStatus, recentGuestAgentEvidence bool) (string, bool) { if status == nil { return "", false @@ -171,16 +207,20 @@ func (m *Monitor) buildVMFromClusterResource( m.hasRecentGuestMetadataEvidence(instanceName, res.Node, res.VMID, prePollTime) state := vmBuildState{ - memTotal: res.MaxMem, - memUsed: res.Mem, - memorySource: "cluster-resources", - guestRaw: VMMemoryRaw{ListingMem: res.Mem, ListingMaxMem: res.MaxMem}, - diskReadBytes: int64(res.DiskRead), - diskWriteBytes: int64(res.DiskWrite), - networkInBytes: int64(res.NetIn), - networkOutBytes: int64(res.NetOut), - diskTotal: res.MaxDisk, - diskUsed: res.Disk, + memTotal: res.MaxMem, + memUsed: res.Mem, + memorySource: "cluster-resources", + guestRaw: VMMemoryRaw{ListingMem: res.Mem, ListingMaxMem: res.MaxMem}, + diskReadBytes: int64(res.DiskRead), + diskWriteBytes: int64(res.DiskWrite), + networkInBytes: int64(res.NetIn), + networkOutBytes: int64(res.NetOut), + counterPresence: pveCounterPresence(res.IOCounters), + counterObservedAt: observedAtOr(res.ObservedAt, prePollTime), + counterObservationTimes: counterObservationTimes(observedAtOr(res.ObservedAt, prePollTime)), + counterUptime: res.Uptime, + diskTotal: res.MaxDisk, + diskUsed: res.Disk, } state.diskFree = state.diskTotal - state.diskUsed state.diskUsage = safePercentage(float64(state.diskUsed), float64(state.diskTotal)) @@ -339,13 +379,25 @@ func (m *Monitor) buildVMFromClusterResource( memFree = state.memTotal - state.memUsed } currentMetrics := IOMetrics{ - DiskRead: state.diskReadBytes, - DiskWrite: state.diskWriteBytes, - NetworkIn: state.networkInBytes, - NetworkOut: state.networkOutBytes, - Timestamp: sampleTime, + DiskRead: state.diskReadBytes, + DiskWrite: state.diskWriteBytes, + NetworkIn: state.networkInBytes, + NetworkOut: state.networkOutBytes, + Timestamp: state.counterObservedAt, + Presence: state.counterPresence, + ObservedAt: state.counterObservationTimes, + SourceUptime: state.counterUptime, } - diskReadRate, diskWriteRate, netInRate, netOutRate := m.rateTracker.CalculateRates(guestID, currentMetrics) + diskReadRate, diskWriteRate, netInRate, netOutRate := m.rateTracker.CalculateRates( + makeGuestRateKey(instanceName, "qemu", res.VMID), + currentMetrics, + ) + diskReadValue, diskWriteValue, networkInValue, networkOutValue, rateValidity := guestRateValues( + diskReadRate, + diskWriteRate, + netInRate, + netOutRate, + ) memory := models.UnavailableMemory(clampToInt64(state.memTotal)) if CanonicalMemorySource(state.memorySource) != "unavailable" { @@ -403,10 +455,11 @@ func (m *Monitor) buildVMFromClusterResource( OSVersion: state.osVersion, AgentVersion: state.agentVersion, NetworkInterfaces: state.networkInterfaces, - NetworkIn: max(0, int64(netInRate)), - NetworkOut: max(0, int64(netOutRate)), - DiskRead: max(0, int64(diskReadRate)), - DiskWrite: max(0, int64(diskWriteRate)), + NetworkIn: networkInValue, + NetworkOut: networkOutValue, + DiskRead: diskReadValue, + DiskWrite: diskWriteValue, + IORateValidity: rateValidity, Uptime: int64(res.Uptime), Template: res.Template == 1, OnBoot: state.onBoot, @@ -429,6 +482,19 @@ func (m *Monitor) buildVMFromClusterResource( } } } + if res.Status != "running" { + vm.NetworkIn = 0 + vm.NetworkOut = 0 + vm.DiskRead = 0 + vm.DiskWrite = 0 + vm.IORateValidity = models.IORateValidity{ + Explicit: true, + DiskRead: true, + DiskWrite: true, + NetworkIn: true, + NetworkOut: true, + } + } return vm, state.guestRaw, state.memorySource, snapshotNotes, sampleTime, true } diff --git a/internal/monitoring/monitor_pve_guest_helpers.go b/internal/monitoring/monitor_pve_guest_helpers.go index f156349d1..47fdc9ec2 100644 --- a/internal/monitoring/monitor_pve_guest_helpers.go +++ b/internal/monitoring/monitor_pve_guest_helpers.go @@ -110,16 +110,35 @@ func (m *Monitor) recordGuestMetrics(allVMs []models.VM, allContainers []models. now := time.Now() for _, vm := range allVMs { if vm.Status == "running" { - m.recordGuestMetric("vm", vm.ID, unifiedresources.ProxmoxGuestCPUPercent(vm.CPU), historyMemoryUsage(vm.Memory), vm.Disk.Usage, vm.DiskRead, vm.DiskWrite, vm.NetworkIn, vm.NetworkOut, now) + diskRead, diskWrite, networkIn, networkOut := guestHistoryRates(vm.DiskRead, vm.DiskWrite, vm.NetworkIn, vm.NetworkOut, vm.IORateValidity) + m.recordGuestMetric("vm", vm.ID, unifiedresources.ProxmoxGuestCPUPercent(vm.CPU), historyMemoryUsage(vm.Memory), vm.Disk.Usage, diskRead, diskWrite, networkIn, networkOut, now) } } for _, ct := range allContainers { if ct.Status == "running" { - m.recordGuestMetric("container", ct.ID, unifiedresources.ProxmoxGuestCPUPercent(ct.CPU), historyMemoryUsage(ct.Memory), ct.Disk.Usage, ct.DiskRead, ct.DiskWrite, ct.NetworkIn, ct.NetworkOut, now) + diskRead, diskWrite, networkIn, networkOut := guestHistoryRates(ct.DiskRead, ct.DiskWrite, ct.NetworkIn, ct.NetworkOut, ct.IORateValidity) + m.recordGuestMetric("container", ct.ID, unifiedresources.ProxmoxGuestCPUPercent(ct.CPU), historyMemoryUsage(ct.Memory), ct.Disk.Usage, diskRead, diskWrite, networkIn, networkOut, now) } } } +func guestHistoryRates(diskRead, diskWrite, networkIn, networkOut int64, validity models.IORateValidity) (int64, int64, int64, int64) { + effective := validity.EffectiveForRates(diskRead, diskWrite, networkIn, networkOut) + if !effective.DiskRead { + diskRead = -1 + } + if !effective.DiskWrite { + diskWrite = -1 + } + if !effective.NetworkIn { + networkIn = -1 + } + if !effective.NetworkOut { + networkOut = -1 + } + return diskRead, diskWrite, networkIn, networkOut +} + func historyMemoryUsage(memory models.Memory) float64 { if !memory.HasKnownUsage() { return -1 diff --git a/internal/monitoring/monitor_pve_guest_lxc.go b/internal/monitoring/monitor_pve_guest_lxc.go index f2e96029a..d9c531d22 100644 --- a/internal/monitoring/monitor_pve_guest_lxc.go +++ b/internal/monitoring/monitor_pve_guest_lxc.go @@ -101,12 +101,16 @@ func (m *Monitor) buildContainerFromClusterResource( guestID := makeGuestID(instanceName, res.Node, res.VMID) sampleTime := time.Now() + counterObservedAt := observedAtOr(res.ObservedAt, sampleTime) currentMetrics := IOMetrics{ - DiskRead: int64(res.DiskRead), - DiskWrite: int64(res.DiskWrite), - NetworkIn: int64(res.NetIn), - NetworkOut: int64(res.NetOut), - Timestamp: sampleTime, + DiskRead: int64(res.DiskRead), + DiskWrite: int64(res.DiskWrite), + NetworkIn: int64(res.NetIn), + NetworkOut: int64(res.NetOut), + Timestamp: counterObservedAt, + Presence: pveCounterPresence(res.IOCounters), + ObservedAt: counterObservationTimes(counterObservedAt), + SourceUptime: res.Uptime, } statusSnapshot := (*proxmox.Container)(nil) if res.Status == "running" { @@ -120,7 +124,16 @@ func (m *Monitor) buildContainerFromClusterResource( ) currentMetrics = mergeContainerRuntimeCounters(currentMetrics, statusSnapshot) } - diskReadRate, diskWriteRate, netInRate, netOutRate := m.rateTracker.CalculateRates(guestID, currentMetrics) + diskReadRate, diskWriteRate, netInRate, netOutRate := m.rateTracker.CalculateRates( + makeGuestRateKey(instanceName, "lxc", res.VMID), + currentMetrics, + ) + diskReadValue, diskWriteValue, networkInValue, networkOutValue, rateValidity := guestRateValues( + diskReadRate, + diskWriteRate, + netInRate, + netOutRate, + ) memTotal, memUsed, memorySource, guestRaw := m.calculateLXCMemory(ctx, instanceName, res, client) memUsed, memorySource, _ = stabilizeGuestLowTrustMemory( @@ -181,13 +194,14 @@ func (m *Monitor) buildContainerFromClusterResource( Free: diskFree, Usage: safePercentage(float64(diskUsed), float64(res.MaxDisk)), }, - NetworkIn: max(0, int64(netInRate)), - NetworkOut: max(0, int64(netOutRate)), - DiskRead: max(0, int64(diskReadRate)), - DiskWrite: max(0, int64(diskWriteRate)), - Uptime: int64(res.Uptime), - Template: res.Template == 1, - LastSeen: lastSeen, + NetworkIn: networkInValue, + NetworkOut: networkOutValue, + DiskRead: diskReadValue, + DiskWrite: diskWriteValue, + IORateValidity: rateValidity, + Uptime: int64(res.Uptime), + Template: res.Template == 1, + LastSeen: lastSeen, } if prevContainerIsOCI[container.VMID] { @@ -231,6 +245,13 @@ func (m *Monitor) buildContainerFromClusterResource( container.NetworkOut = 0 container.DiskRead = 0 container.DiskWrite = 0 + container.IORateValidity = models.IORateValidity{ + Explicit: true, + DiskRead: true, + DiskWrite: true, + NetworkIn: true, + NetworkOut: true, + } } return container, guestRaw, memorySource, sampleTime, true diff --git a/internal/monitoring/monitor_pve_guest_lxc_test.go b/internal/monitoring/monitor_pve_guest_lxc_test.go index ed6d6823b..764002ace 100644 --- a/internal/monitoring/monitor_pve_guest_lxc_test.go +++ b/internal/monitoring/monitor_pve_guest_lxc_test.go @@ -5,6 +5,7 @@ import ( "testing" "time" + "github.com/rcourtman/pulse-go-rewrite/internal/models" "github.com/rcourtman/pulse-go-rewrite/pkg/proxmox" ) @@ -20,7 +21,7 @@ func (s *stubPVEClientLXCStatus) GetContainerStatus(ctx context.Context, node st return s.containerStatus, nil } -func TestMergeContainerRuntimeCounters_PrefersHigherStatusCounters(t *testing.T) { +func TestMergeContainerRuntimeCounters_PrefersNewerStatusCounters(t *testing.T) { t.Parallel() current := IOMetrics{ @@ -41,17 +42,66 @@ func TestMergeContainerRuntimeCounters_PrefersHigherStatusCounters(t *testing.T) if merged.DiskRead != 128 { t.Fatalf("expected DiskRead to upgrade from status snapshot, got %d", merged.DiskRead) } - if merged.DiskWrite != 8 { - t.Fatalf("expected DiskWrite to preserve the higher baseline counter, got %d", merged.DiskWrite) + if merged.DiskWrite != 4 { + t.Fatalf("expected DiskWrite to follow the newer reset counter, got %d", merged.DiskWrite) } - if merged.NetworkIn != 12 { - t.Fatalf("expected NetworkIn to preserve the higher baseline counter, got %d", merged.NetworkIn) + if merged.NetworkIn != 10 { + t.Fatalf("expected NetworkIn to follow the newer reset counter, got %d", merged.NetworkIn) } if merged.NetworkOut != 256 { t.Fatalf("expected NetworkOut to upgrade from status snapshot, got %d", merged.NetworkOut) } } +func TestMergeContainerRuntimeCounters_OverridesOnlyPresentStatusFields(t *testing.T) { + t.Parallel() + + listingObservedAt := time.Unix(10, 0) + current := IOMetrics{ + DiskRead: 8, + DiskWrite: 16, + NetworkIn: 32, + NetworkOut: 64, + Timestamp: listingObservedAt, + ObservedAt: counterObservationTimes(listingObservedAt), + Presence: models.IOCounterPresence{ + Explicit: true, + DiskRead: true, + DiskWrite: true, + NetworkIn: true, + NetworkOut: true, + }, + } + status := &proxmox.Container{ + DiskRead: 0, + DiskWrite: 999, + IOCounters: proxmox.IOCounterPresence{ + Explicit: true, + DiskRead: true, + }, + ObservedAt: time.Unix(20, 0), + } + + merged := mergeContainerRuntimeCounters(current, status) + if merged.DiskRead != 0 { + t.Fatalf("explicit status zero was not authoritative: %d", merged.DiskRead) + } + if merged.DiskWrite != 16 || merged.NetworkIn != 32 || merged.NetworkOut != 64 { + t.Fatalf("missing status fields overwrote listing counters: %+v", merged) + } + if !merged.Timestamp.Equal(status.ObservedAt) { + t.Fatalf("timestamp = %v, want status receipt time %v", merged.Timestamp, status.ObservedAt) + } + if !merged.ObservedAt.DiskRead.Equal(status.ObservedAt) { + t.Fatalf("disk-read receipt = %v, want status receipt %v", merged.ObservedAt.DiskRead, status.ObservedAt) + } + if !merged.ObservedAt.DiskWrite.Equal(listingObservedAt) || + !merged.ObservedAt.NetworkIn.Equal(listingObservedAt) || + !merged.ObservedAt.NetworkOut.Equal(listingObservedAt) { + t.Fatalf("missing status fields lost listing receipt times: %+v", merged.ObservedAt) + } +} + func TestBuildContainerFromClusterResource_UsesContainerStatusCountersForRates(t *testing.T) { t.Parallel() diff --git a/internal/monitoring/monitor_pve_node_vm_builder.go b/internal/monitoring/monitor_pve_node_vm_builder.go index 28442f92e..755c249e4 100644 --- a/internal/monitoring/monitor_pve_node_vm_builder.go +++ b/internal/monitoring/monitor_pve_node_vm_builder.go @@ -31,26 +31,28 @@ func (m *Monitor) pollNodeVMsWithClusterResourceBuilder( resources = append(resources, indexedClusterResource{ order: len(resources), resource: proxmox.ClusterResource{ - ID: guestID, - Type: "qemu", - Node: node, - Pool: vm.Pool, - Status: vm.Status, - Name: vm.Name, - VMID: vm.VMID, - CPU: vm.CPU, - MaxCPU: vm.CPUs, - Mem: vm.Mem, - MaxMem: vm.MaxMem, - Disk: vm.Disk, - MaxDisk: vm.MaxDisk, - NetIn: vm.NetIn, - NetOut: vm.NetOut, - DiskRead: vm.DiskRead, - DiskWrite: vm.DiskWrite, - Uptime: vm.Uptime, - Template: vm.Template, - Tags: vm.Tags, + ID: guestID, + Type: "qemu", + Node: node, + Pool: vm.Pool, + Status: vm.Status, + Name: vm.Name, + VMID: vm.VMID, + CPU: vm.CPU, + MaxCPU: vm.CPUs, + Mem: vm.Mem, + MaxMem: vm.MaxMem, + Disk: vm.Disk, + MaxDisk: vm.MaxDisk, + NetIn: vm.NetIn, + NetOut: vm.NetOut, + DiskRead: vm.DiskRead, + DiskWrite: vm.DiskWrite, + Uptime: vm.Uptime, + Template: vm.Template, + Tags: vm.Tags, + IOCounters: vm.IOCounters, + ObservedAt: vm.ObservedAt, }, guestID: guestID, }) diff --git a/internal/monitoring/ratetracker.go b/internal/monitoring/ratetracker.go index 61d1163bf..13a59cd8c 100644 --- a/internal/monitoring/ratetracker.go +++ b/internal/monitoring/ratetracker.go @@ -7,81 +7,49 @@ import ( "github.com/rcourtman/pulse-go-rewrite/internal/models" ) -// IOMetrics is an alias for models.IOMetrics +// IOMetrics is an alias for models.IOMetrics. type IOMetrics = models.IOMetrics -// rateWindowSize is the number of counter samples retained per guest. -// Rate is computed from the oldest to the newest sample, giving an average -// over (rateWindowSize-1) polling intervals. With a 10s poll interval and -// window size 4, this produces a 30-second sliding window — the same approach -// Prometheus rate() uses to smooth out per-interval counter jitter. -const rateWindowSize = 4 - -// counterRing is a fixed-size ring buffer of IOMetrics samples. -type counterRing struct { - entries [rateWindowSize]IOMetrics - count int // number of entries stored (up to rateWindowSize) - head int // next write position +type counterBaseline struct { + value int64 + observedAt time.Time + initialized bool } -func (r *counterRing) add(m IOMetrics) { - r.entries[r.head] = m - r.head = (r.head + 1) % rateWindowSize - if r.count < rateWindowSize { - r.count++ - } +type counterHistory struct { + diskRead counterBaseline + diskWrite counterBaseline + diskBusy counterBaseline + networkIn counterBaseline + networkOut counterBaseline + lastObservedAt time.Time + sourceUptime uint64 } -func (r *counterRing) oldest() IOMetrics { - if r.count < rateWindowSize { - return r.entries[0] - } - return r.entries[r.head] // head points to the oldest when full -} - -func (r *counterRing) newest() IOMetrics { - return r.entries[(r.head-1+rateWindowSize)%rateWindowSize] -} - -// newestTimestamp returns the timestamp of the most recent entry. -func (r *counterRing) newestTimestamp() time.Time { - return r.newest().Timestamp -} - -// RateTracker tracks I/O metrics to calculate rates +// RateTracker converts cumulative byte counters into adjacent-sample rates. +// Each counter has an independent baseline because Proxmox may omit only part +// of an otherwise valid status payload. type RateTracker struct { - mu sync.RWMutex - history map[string]*counterRing - lastRates map[string]RateCache + mu sync.RWMutex + history map[string]*counterHistory } -// RateCache stores the last calculated rates for a guest -type RateCache struct { - DiskReadRate float64 - DiskWriteRate float64 - DiskBusyPct float64 - NetInRate float64 - NetOutRate float64 -} - -// NewRateTracker creates a new rate tracker +// NewRateTracker creates a new rate tracker. func NewRateTracker() *RateTracker { - return &RateTracker{ - history: make(map[string]*counterRing), - lastRates: make(map[string]RateCache), - } + return &RateTracker{history: make(map[string]*counterHistory)} } -// CalculateRates calculates I/O rates for a guest -// Returns -1 for rates that don't have enough data yet (will be converted to null in JSON) +// CalculateRates calculates disk and network rates in bytes per second. +// A negative result means the upstream counter was absent, the sample was +// out-of-order, or no earlier observation exists. A returned zero is a valid +// observed idle/reset interval. func (rt *RateTracker) CalculateRates(guestID string, current IOMetrics) (diskReadRate, diskWriteRate, netInRate, netOutRate float64) { diskReadRate, diskWriteRate, _, netInRate, netOutRate = rt.calculateRates(guestID, current) return } -// CalculateRatesWithBusy calculates disk/network rates plus disk busy percent -// from cumulative counters. Returns -1 for metrics that do not yet have enough -// data to produce a rate. +// CalculateRatesWithBusy also calculates disk busy percent from a cumulative +// millisecond counter. func (rt *RateTracker) CalculateRatesWithBusy(guestID string, current IOMetrics) (diskReadRate, diskWriteRate, diskBusyPct, netInRate, netOutRate float64) { return rt.calculateRates(guestID, current) } @@ -90,102 +58,104 @@ func (rt *RateTracker) calculateRates(guestID string, current IOMetrics) (diskRe rt.mu.Lock() defer rt.mu.Unlock() - ring, exists := rt.history[guestID] - - if !exists { - // No previous data, store it and return -1 to indicate no data available - ring = &counterRing{} - ring.add(current) - rt.history[guestID] = ring + if current.Timestamp.IsZero() { return -1, -1, -1, -1, -1 } - prev := ring.newest() - - // Check if the values have actually changed (detect stale data) - // If all cumulative values are the same, we're getting cached data from Proxmox - if current.DiskRead == prev.DiskRead && - current.DiskWrite == prev.DiskWrite && - current.DiskBusy == prev.DiskBusy && - current.NetworkIn == prev.NetworkIn && - current.NetworkOut == prev.NetworkOut { - // Data hasn't changed - return last known good rates - if lastRate, hasRate := rt.lastRates[guestID]; hasRate { - return lastRate.DiskReadRate, lastRate.DiskWriteRate, lastRate.DiskBusyPct, lastRate.NetInRate, lastRate.NetOutRate + history := rt.history[guestID] + if history == nil { + history = &counterHistory{} + rt.history[guestID] = history + } + if current.SourceUptime > 0 { + if history.sourceUptime > 0 && current.SourceUptime < history.sourceUptime { + history.resetCounterEpoch() } - // No last rates available, return 0 - return 0, 0, 0, 0, 0 + history.sourceUptime = current.SourceUptime } - // Data has changed, add to ring buffer - ring.add(current) + presence := current.Presence.Effective() + diskReadRate = calculateCounterRate(&history.diskRead, current.DiskRead, observationTime(current.ObservedAt.DiskRead, current.Timestamp), presence.DiskRead) + diskWriteRate = calculateCounterRate(&history.diskWrite, current.DiskWrite, observationTime(current.ObservedAt.DiskWrite, current.Timestamp), presence.DiskWrite) + diskBusyRate := calculateCounterRate(&history.diskBusy, current.DiskBusy, observationTime(current.ObservedAt.DiskBusy, current.Timestamp), presence.DiskBusy) + netInRate = calculateCounterRate(&history.networkIn, current.NetworkIn, observationTime(current.ObservedAt.NetworkIn, current.Timestamp), presence.NetworkIn) + netOutRate = calculateCounterRate(&history.networkOut, current.NetworkOut, observationTime(current.ObservedAt.NetworkOut, current.Timestamp), presence.NetworkOut) - // Calculate rate over the full window (oldest to current), like Prometheus rate(). - // This naturally smooths out per-interval jitter from Proxmox's lumpy counter - // reporting by averaging over a wider time span. - oldest := ring.oldest() - timeDiff := current.Timestamp.Sub(oldest.Timestamp).Seconds() - if timeDiff <= 0 { - // Return last known rates if time hasn't advanced - if lastRate, hasRate := rt.lastRates[guestID]; hasRate { - return lastRate.DiskReadRate, lastRate.DiskWriteRate, lastRate.DiskBusyPct, lastRate.NetInRate, lastRate.NetOutRate - } - return 0, 0, 0, 0, 0 - } - - // Calculate rates (bytes per second) over the window - if current.DiskRead >= oldest.DiskRead { - diskReadRate = float64(current.DiskRead-oldest.DiskRead) / timeDiff - } - if current.DiskWrite >= oldest.DiskWrite { - diskWriteRate = float64(current.DiskWrite-oldest.DiskWrite) / timeDiff - } - if current.DiskBusy >= oldest.DiskBusy { - diskBusyPct = (float64(current.DiskBusy-oldest.DiskBusy) / (timeDiff * 1000)) * 100 - if diskBusyPct < 0 { - diskBusyPct = 0 - } + if diskBusyRate < 0 { + diskBusyPct = -1 + } else { + // DiskBusy is cumulative busy milliseconds, so ms/s divided by ten is + // the percentage of wall time spent busy. + diskBusyPct = diskBusyRate / 10 if diskBusyPct > 100 { diskBusyPct = 100 } } - if current.NetworkIn >= oldest.NetworkIn { - netInRate = float64(current.NetworkIn-oldest.NetworkIn) / timeDiff - } - if current.NetworkOut >= oldest.NetworkOut { - netOutRate = float64(current.NetworkOut-oldest.NetworkOut) / timeDiff - } - // Cache the calculated rates - rt.lastRates[guestID] = RateCache{ - DiskReadRate: diskReadRate, - DiskWriteRate: diskWriteRate, - DiskBusyPct: diskBusyPct, - NetInRate: netInRate, - NetOutRate: netOutRate, + if current.Timestamp.After(history.lastObservedAt) { + history.lastObservedAt = current.Timestamp } - return } -// Clear removes all tracked data +func (history *counterHistory) resetCounterEpoch() { + history.diskRead = counterBaseline{} + history.diskWrite = counterBaseline{} + history.diskBusy = counterBaseline{} + history.networkIn = counterBaseline{} + history.networkOut = counterBaseline{} +} + +func observationTime(counterTime, fallback time.Time) time.Time { + if counterTime.IsZero() { + return fallback + } + return counterTime +} + +func calculateCounterRate(baseline *counterBaseline, value int64, observedAt time.Time, present bool) float64 { + if !present { + return -1 + } + if !baseline.initialized { + baseline.value = value + baseline.observedAt = observedAt + baseline.initialized = true + return -1 + } + if !observedAt.After(baseline.observedAt) { + return -1 + } + + elapsed := observedAt.Sub(baseline.observedAt).Seconds() + previous := baseline.value + baseline.value = value + baseline.observedAt = observedAt + + if value < previous { + // A guest restart, migration reconnect, or counter wrap starts a new + // cumulative epoch. Rebase without inventing a negative or huge rate. + return 0 + } + return float64(value-previous) / elapsed +} + +// Clear removes all tracked data. func (rt *RateTracker) Clear() { rt.mu.Lock() defer rt.mu.Unlock() - rt.history = make(map[string]*counterRing) - rt.lastRates = make(map[string]RateCache) + rt.history = make(map[string]*counterHistory) } -// Cleanup removes entries for resources that haven't reported data since the cutoff time. -// This prevents unbounded memory growth when containers/VMs are deleted. +// Cleanup removes resources that have not supplied any newer sample since the +// cutoff. Idle and partial samples still refresh resource observation time. func (rt *RateTracker) Cleanup(cutoff time.Time) (removed int) { rt.mu.Lock() defer rt.mu.Unlock() - for guestID, ring := range rt.history { - if ring.newestTimestamp().Before(cutoff) { + for guestID, history := range rt.history { + if history.lastObservedAt.Before(cutoff) { delete(rt.history, guestID) - delete(rt.lastRates, guestID) removed++ } } diff --git a/internal/monitoring/ratetracker_test.go b/internal/monitoring/ratetracker_test.go index 46e9aca99..3421a8ab4 100644 --- a/internal/monitoring/ratetracker_test.go +++ b/internal/monitoring/ratetracker_test.go @@ -1,397 +1,236 @@ package monitoring import ( + "math" "testing" "time" "github.com/rcourtman/pulse-go-rewrite/internal/models" ) -func TestCalculateRates_FirstCallReturnsNegativeOnes(t *testing.T) { - rt := NewRateTracker() - d, w, ni, no := rt.CalculateRates("vm-100", models.IOMetrics{ - DiskRead: 1000, DiskWrite: 2000, NetworkIn: 3000, NetworkOut: 4000, - Timestamp: time.Now(), - }) - if d != -1 || w != -1 || ni != -1 || no != -1 { - t.Errorf("first call: got (%v, %v, %v, %v), want (-1, -1, -1, -1)", d, w, ni, no) +func fullCounterSample(at time.Time, diskRead, diskWrite, networkIn, networkOut int64) models.IOMetrics { + return models.IOMetrics{ + DiskRead: diskRead, + DiskWrite: diskWrite, + NetworkIn: networkIn, + NetworkOut: networkOut, + Timestamp: at, } } -func TestCalculateRates_StaleDataReturnsCachedRates(t *testing.T) { - rt := NewRateTracker() - base := time.Unix(1000, 0) +func TestRateTrackerConstantRatesUseActualElapsedTime(t *testing.T) { + for _, interval := range []time.Duration{30 * time.Second, 60 * time.Second, 90 * time.Second} { + t.Run(interval.String(), func(t *testing.T) { + tracker := NewRateTracker() + start := time.Unix(1_700_000_000, 0) + tracker.CalculateRates("guest", fullCounterSample(start, 1_000, 2_000, 3_000, 4_000)) - // Seed - rt.CalculateRates("vm-100", models.IOMetrics{ - DiskRead: 1000, DiskWrite: 2000, NetworkIn: 3000, NetworkOut: 4000, - Timestamp: base, - }) - // Establish rates - rt.CalculateRates("vm-100", models.IOMetrics{ - DiskRead: 6000, DiskWrite: 12000, NetworkIn: 18000, NetworkOut: 24000, - Timestamp: base.Add(10 * time.Second), - }) - - // Send stale data (same counter values, different timestamp) - d, w, ni, no := rt.CalculateRates("vm-100", models.IOMetrics{ - DiskRead: 6000, DiskWrite: 12000, NetworkIn: 18000, NetworkOut: 24000, - Timestamp: base.Add(20 * time.Second), - }) - if d != 500 || w != 1000 || ni != 1500 || no != 2000 { - t.Errorf("stale data: got (%v, %v, %v, %v), want (500, 1000, 1500, 2000)", d, w, ni, no) + seconds := int64(interval / time.Second) + read, write, in, out := tracker.CalculateRates( + "guest", + fullCounterSample( + start.Add(interval), + 1_000+seconds*1_024, + 2_000+seconds*2_048, + 3_000+seconds*4_096, + 4_000+seconds*8_192, + ), + ) + if read != 1_024 || write != 2_048 || in != 4_096 || out != 8_192 { + t.Fatalf("rates = (%v, %v, %v, %v), want (1024, 2048, 4096, 8192)", read, write, in, out) + } + }) } } -func TestCalculateRates_StaleDataWithoutCachedRatesReturnsZeros(t *testing.T) { - rt := NewRateTracker() - base := time.Unix(1000, 0) +func TestRateTrackerBurstUsesAdjacentObservationInterval(t *testing.T) { + tracker := NewRateTracker() + start := time.Unix(1_700_000_000, 0) + tracker.CalculateRates("guest", fullCounterSample(start, 0, 0, 0, 0)) - // Seed with values - rt.CalculateRates("vm-100", models.IOMetrics{ - DiskRead: 1000, DiskWrite: 2000, NetworkIn: 3000, NetworkOut: 4000, - Timestamp: base, - }) + read, _, _, _ := tracker.CalculateRates("guest", fullCounterSample(start.Add(90*time.Second), 90_000, 0, 0, 0)) + if read != 1_000 { + t.Fatalf("burst rate = %v, want 1000 B/s", read) + } - // Send identical values (stale) — no cached rates yet - d, w, ni, no := rt.CalculateRates("vm-100", models.IOMetrics{ - DiskRead: 1000, DiskWrite: 2000, NetworkIn: 3000, NetworkOut: 4000, - Timestamp: base.Add(10 * time.Second), - }) - if d != 0 || w != 0 || ni != 0 || no != 0 { - t.Errorf("stale without cache: got (%v, %v, %v, %v), want (0, 0, 0, 0)", d, w, ni, no) + read, _, _, _ = tracker.CalculateRates("guest", fullCounterSample(start.Add(180*time.Second), 90_000, 0, 0, 0)) + if read != 0 { + t.Fatalf("idle interval rate = %v, want valid zero", read) } } -func TestCalculateRates_NormalRateCalculation(t *testing.T) { - rt := NewRateTracker() - base := time.Unix(1000, 0) - - // Seed - rt.CalculateRates("vm-100", models.IOMetrics{ - DiskRead: 1000, DiskWrite: 2000, NetworkIn: 3000, NetworkOut: 4000, - Timestamp: base, +func TestRateTrackerMissingAndExplicitZeroAreDistinct(t *testing.T) { + tracker := NewRateTracker() + start := time.Unix(1_700_000_000, 0) + diskOnly := models.IOCounterPresence{Explicit: true, DiskRead: true} + tracker.CalculateRates("guest", models.IOMetrics{ + DiskRead: 0, + Timestamp: start, + Presence: diskOnly, }) - // Second call — rate over 1 interval (ring only has 2 entries) - d, w, ni, no := rt.CalculateRates("vm-100", models.IOMetrics{ - DiskRead: 6000, DiskWrite: 12000, NetworkIn: 18000, NetworkOut: 24000, - Timestamp: base.Add(10 * time.Second), + read, write, in, out := tracker.CalculateRates("guest", models.IOMetrics{ + DiskRead: 0, + Timestamp: start.Add(60 * time.Second), + Presence: diskOnly, }) - if d != 500 || w != 1000 || ni != 1500 || no != 2000 { - t.Errorf("normal rate: got (%v, %v, %v, %v), want (500, 1000, 1500, 2000)", d, w, ni, no) + if read != 0 { + t.Fatalf("explicit zero rate = %v, want 0", read) + } + if write != -1 || in != -1 || out != -1 { + t.Fatalf("missing rates = (%v, %v, %v), want unknown", write, in, out) } } -func TestCalculateRates_CounterRolloverReturnsZero(t *testing.T) { - rt := NewRateTracker() - base := time.Unix(1000, 0) +func TestRateTrackerPartialSampleKeepsIndependentBaseline(t *testing.T) { + tracker := NewRateTracker() + start := time.Unix(1_700_000_000, 0) + tracker.CalculateRates("guest", fullCounterSample(start, 0, 0, 0, 0)) - rt.CalculateRates("vm-100", models.IOMetrics{ - DiskRead: 5000, DiskWrite: 2000, NetworkIn: 3000, NetworkOut: 4000, - Timestamp: base, + read, write, in, out := tracker.CalculateRates("guest", models.IOMetrics{ + DiskRead: 60_000, + Timestamp: start.Add(60 * time.Second), + Presence: models.IOCounterPresence{ + Explicit: true, + DiskRead: true, + }, }) - - // DiskRead decreased (counter rollover) - d, w, ni, no := rt.CalculateRates("vm-100", models.IOMetrics{ - DiskRead: 1000, DiskWrite: 12000, NetworkIn: 18000, NetworkOut: 24000, - Timestamp: base.Add(10 * time.Second), - }) - if d != 0 { - t.Errorf("DiskRead rollover: got %v, want 0", d) + if read != 1_000 || write != -1 || in != -1 || out != -1 { + t.Fatalf("partial rates = (%v, %v, %v, %v)", read, write, in, out) } - if w != 1000 || ni != 1500 || no != 2000 { - t.Errorf("other rates: got (%v, %v, %v), want (1000, 1500, 2000)", w, ni, no) + + _, write, _, _ = tracker.CalculateRates("guest", models.IOMetrics{ + DiskWrite: 180_000, + Timestamp: start.Add(90 * time.Second), + Presence: models.IOCounterPresence{ + Explicit: true, + DiskWrite: true, + }, + }) + if write != 2_000 { + t.Fatalf("disk write rate = %v, want 2000 over its 90s observation gap", write) } } -func TestCalculateRates_AllCountersRollover(t *testing.T) { - rt := NewRateTracker() - base := time.Unix(1000, 0) +func TestRateTrackerUsesEachCounterReceiptTimeForPartialSources(t *testing.T) { + tracker := NewRateTracker() + start := time.Unix(1_700_000_000, 0) + tracker.CalculateRates("guest", fullCounterSample(start, 0, 0, 0, 0)) - rt.CalculateRates("vm-100", models.IOMetrics{ - DiskRead: 6000, DiskWrite: 12000, NetworkIn: 18000, NetworkOut: 24000, - Timestamp: base, + read, write, _, _ := tracker.CalculateRates("guest", models.IOMetrics{ + DiskRead: 30_000, + DiskWrite: 180_000, + Timestamp: start.Add(90 * time.Second), + Presence: models.IOCounterPresence{ + Explicit: true, + DiskRead: true, + DiskWrite: true, + }, + ObservedAt: models.IOCounterObservationTimes{ + DiskRead: start.Add(30 * time.Second), + DiskWrite: start.Add(90 * time.Second), + }, }) - - d, w, ni, no := rt.CalculateRates("vm-100", models.IOMetrics{ - DiskRead: 1000, DiskWrite: 2000, NetworkIn: 3000, NetworkOut: 4000, - Timestamp: base.Add(10 * time.Second), - }) - if d != 0 || w != 0 || ni != 0 || no != 0 { - t.Errorf("all rollover: got (%v, %v, %v, %v), want (0, 0, 0, 0)", d, w, ni, no) + if read != 1_000 || write != 2_000 { + t.Fatalf("rates = (%v, %v), want per-counter receipt rates (1000, 2000)", read, write) } } -func TestCalculateRates_FractionalTimeDifference(t *testing.T) { - rt := NewRateTracker() - base := time.Unix(1000, 0) +func TestRateTrackerResetOrWrapRebasesCounterEpoch(t *testing.T) { + tracker := NewRateTracker() + start := time.Unix(1_700_000_000, 0) + tracker.CalculateRates("guest", fullCounterSample(start, math.MaxInt64-1_000, 50_000, 10_000, 20_000)) - rt.CalculateRates("vm-100", models.IOMetrics{ - DiskRead: 1000, DiskWrite: 2000, NetworkIn: 3000, NetworkOut: 4000, - Timestamp: base, - }) + read, write, in, out := tracker.CalculateRates( + "guest", + fullCounterSample(start.Add(30*time.Second), 500, 1_000, 100, 200), + ) + if read != 0 || write != 0 || in != 0 || out != 0 { + t.Fatalf("reset rates = (%v, %v, %v, %v), want zeros", read, write, in, out) + } - d, w, ni, no := rt.CalculateRates("vm-100", models.IOMetrics{ - DiskRead: 1500, DiskWrite: 2500, NetworkIn: 3500, NetworkOut: 4500, - Timestamp: base.Add(500 * time.Millisecond), - }) - // 500 / 0.5 = 1000 - if d != 1000 || w != 1000 || ni != 1000 || no != 1000 { - t.Errorf("fractional time: got (%v, %v, %v, %v), want (1000, 1000, 1000, 1000)", d, w, ni, no) + read, _, _, _ = tracker.CalculateRates("guest", fullCounterSample(start.Add(60*time.Second), 30_500, 1_000, 100, 200)) + if read != 1_000 { + t.Fatalf("post-reset rate = %v, want 1000", read) } } -func TestCalculateRates_LargeValues(t *testing.T) { - rt := NewRateTracker() - base := time.Unix(1000, 0) +func TestRateTrackerUptimeRollbackRebasesEvenWhenCounterSurpassesOldValue(t *testing.T) { + tracker := NewRateTracker() + start := time.Unix(1_700_000_000, 0) + beforeRestart := fullCounterSample(start, 1_000, 2_000, 3_000, 4_000) + beforeRestart.SourceUptime = 10_000 + tracker.CalculateRates("guest", beforeRestart) - rt.CalculateRates("vm-100", models.IOMetrics{ - DiskRead: 1000000000, DiskWrite: 2000000000, NetworkIn: 3000000000, NetworkOut: 4000000000, - Timestamp: base, - }) + afterRestart := fullCounterSample(start.Add(90*time.Second), 91_000, 182_000, 273_000, 364_000) + afterRestart.SourceUptime = 30 + read, write, in, out := tracker.CalculateRates("guest", afterRestart) + if read != -1 || write != -1 || in != -1 || out != -1 { + t.Fatalf("first sample in restarted epoch = (%v, %v, %v, %v), want unknown", read, write, in, out) + } - d, w, ni, no := rt.CalculateRates("vm-100", models.IOMetrics{ - DiskRead: 1100000000, DiskWrite: 2200000000, NetworkIn: 3300000000, NetworkOut: 4400000000, - Timestamp: base.Add(100 * time.Second), - }) - if d != 1000000 || w != 2000000 || ni != 3000000 || no != 4000000 { - t.Errorf("large values: got (%v, %v, %v, %v), want (1000000, 2000000, 3000000, 4000000)", d, w, ni, no) + next := fullCounterSample(start.Add(120*time.Second), 121_000, 242_000, 363_000, 484_000) + next.SourceUptime = 60 + read, write, in, out = tracker.CalculateRates("guest", next) + if read != 1_000 || write != 2_000 || in != 3_000 || out != 4_000 { + t.Fatalf("post-restart rates = (%v, %v, %v, %v), want (1000, 2000, 3000, 4000)", read, write, in, out) } } -func TestCalculateRates_WindowSmooths(t *testing.T) { - rt := NewRateTracker() - base := time.Unix(1000, 0) +func TestRateTrackerRejectsOutOfOrderSamplesWithoutChangingBaseline(t *testing.T) { + tracker := NewRateTracker() + start := time.Unix(1_700_000_000, 0) + tracker.CalculateRates("guest", fullCounterSample(start, 0, 0, 0, 0)) + tracker.CalculateRates("guest", fullCounterSample(start.Add(60*time.Second), 60_000, 0, 0, 0)) - // Simulate a steady 1000 bytes/sec download with Proxmox's lumpy counter updates. - // Over 4 intervals (40 seconds), 40000 bytes should arrive. - // But Proxmox distributes them unevenly across intervals. - - // T=0: seed - rt.CalculateRates("vm-100", models.IOMetrics{ - NetworkIn: 0, Timestamp: base, - }) - - // T=10: normal interval (10000 bytes in 10s = 1000 B/s) - rt.CalculateRates("vm-100", models.IOMetrics{ - NetworkIn: 10000, Timestamp: base.Add(10 * time.Second), - }) - - // T=20: short-changed interval (only 5000 bytes reported) - rt.CalculateRates("vm-100", models.IOMetrics{ - NetworkIn: 15000, Timestamp: base.Add(20 * time.Second), - }) - - // T=30: lumpy interval (15000 bytes — makes up for the deficit + normal) - // Without windowing, raw rate would be 15000/10 = 1500 B/s (50% spike). - // With windowing (oldest=T=0, current=T=30), rate = 30000/30 = 1000 B/s. - _, _, ni, _ := rt.CalculateRates("vm-100", models.IOMetrics{ - NetworkIn: 30000, Timestamp: base.Add(30 * time.Second), - }) - - if ni != 1000 { - t.Errorf("windowed rate during lumpy interval: got %v, want 1000", ni) + read, _, _, _ := tracker.CalculateRates("guest", fullCounterSample(start.Add(30*time.Second), 90_000, 0, 0, 0)) + if read != -1 { + t.Fatalf("out-of-order rate = %v, want unknown", read) } - - // T=40: ring is now full (4 entries), oldest is T=10. - // Rate = (40000-10000)/(40-10) = 30000/30 = 1000 B/s - _, _, ni, _ = rt.CalculateRates("vm-100", models.IOMetrics{ - NetworkIn: 40000, Timestamp: base.Add(40 * time.Second), - }) - - if ni != 1000 { - t.Errorf("windowed rate after ring full: got %v, want 1000", ni) + read, _, _, _ = tracker.CalculateRates("guest", fullCounterSample(start.Add(90*time.Second), 90_000, 0, 0, 0)) + if read != 1_000 { + t.Fatalf("rate after rejected sample = %v, want 1000", read) } } -func TestCalculateRates_MultipleGuestsTrackedIndependently(t *testing.T) { - rt := NewRateTracker() - baseTime := time.Unix(1000, 0) - - // First call for guest A - should return -1 for all - diskReadA1, diskWriteA1, netInA1, netOutA1 := rt.CalculateRates("vm-100", models.IOMetrics{ - DiskRead: 1000, DiskWrite: 2000, NetworkIn: 3000, NetworkOut: 4000, - Timestamp: baseTime, - }) - if diskReadA1 != -1 || diskWriteA1 != -1 || netInA1 != -1 || netOutA1 != -1 { - t.Errorf("first call for vm-100: got (%v, %v, %v, %v), want (-1, -1, -1, -1)", - diskReadA1, diskWriteA1, netInA1, netOutA1) - } - - // First call for guest B - should also return -1 for all - diskReadB1, diskWriteB1, netInB1, netOutB1 := rt.CalculateRates("vm-200", models.IOMetrics{ - DiskRead: 5000, DiskWrite: 6000, NetworkIn: 7000, NetworkOut: 8000, - Timestamp: baseTime, - }) - if diskReadB1 != -1 || diskWriteB1 != -1 || netInB1 != -1 || netOutB1 != -1 { - t.Errorf("first call for vm-200: got (%v, %v, %v, %v), want (-1, -1, -1, -1)", - diskReadB1, diskWriteB1, netInB1, netOutB1) - } - - // Second call for guest A - diskReadA2, diskWriteA2, netInA2, netOutA2 := rt.CalculateRates("vm-100", models.IOMetrics{ - DiskRead: 11000, DiskWrite: 22000, NetworkIn: 33000, NetworkOut: 44000, - Timestamp: baseTime.Add(10 * time.Second), - }) - if diskReadA2 != 1000 || diskWriteA2 != 2000 || netInA2 != 3000 || netOutA2 != 4000 { - t.Errorf("second call for vm-100: got (%v, %v, %v, %v), want (1000, 2000, 3000, 4000)", - diskReadA2, diskWriteA2, netInA2, netOutA2) - } - - // Second call for guest B - different rates - diskReadB2, diskWriteB2, netInB2, netOutB2 := rt.CalculateRates("vm-200", models.IOMetrics{ - DiskRead: 10000, DiskWrite: 16000, NetworkIn: 22000, NetworkOut: 28000, - Timestamp: baseTime.Add(5 * time.Second), - }) - if diskReadB2 != 1000 || diskWriteB2 != 2000 || netInB2 != 3000 || netOutB2 != 4000 { - t.Errorf("second call for vm-200: got (%v, %v, %v, %v), want (1000, 2000, 3000, 4000)", - diskReadB2, diskWriteB2, netInB2, netOutB2) - } -} - -func TestCalculateRates_CachesRates(t *testing.T) { - rt := NewRateTracker() - baseTime := time.Unix(1000, 0) - - // Seed - rt.CalculateRates("vm-100", models.IOMetrics{ - DiskRead: 1000, DiskWrite: 2000, NetworkIn: 3000, NetworkOut: 4000, - Timestamp: baseTime, - }) - - // Calculate rates - diskRead2, diskWrite2, netIn2, netOut2 := rt.CalculateRates("vm-100", models.IOMetrics{ - DiskRead: 11000, DiskWrite: 22000, NetworkIn: 33000, NetworkOut: 44000, - Timestamp: baseTime.Add(10 * time.Second), - }) - - // Verify rates are cached - cachedRates, exists := rt.lastRates["vm-100"] - if !exists { - t.Fatal("expected rates to be cached for vm-100") - } - if cachedRates.DiskReadRate != diskRead2 { - t.Errorf("cached DiskReadRate = %v, want %v", cachedRates.DiskReadRate, diskRead2) - } - if cachedRates.DiskWriteRate != diskWrite2 { - t.Errorf("cached DiskWriteRate = %v, want %v", cachedRates.DiskWriteRate, diskWrite2) - } - if cachedRates.NetInRate != netIn2 { - t.Errorf("cached NetInRate = %v, want %v", cachedRates.NetInRate, netIn2) - } - if cachedRates.NetOutRate != netOut2 { - t.Errorf("cached NetOutRate = %v, want %v", cachedRates.NetOutRate, netOut2) - } - - // Stale data returns cached rates - diskRead3, diskWrite3, netIn3, netOut3 := rt.CalculateRates("vm-100", models.IOMetrics{ - DiskRead: 11000, DiskWrite: 22000, NetworkIn: 33000, NetworkOut: 44000, - Timestamp: baseTime.Add(15 * time.Second), - }) - if diskRead3 != diskRead2 || diskWrite3 != diskWrite2 || netIn3 != netIn2 || netOut3 != netOut2 { - t.Errorf("stale data call: got (%v, %v, %v, %v), want cached (%v, %v, %v, %v)", - diskRead3, diskWrite3, netIn3, netOut3, diskRead2, diskWrite2, netIn2, netOut2) - } -} - -func TestCalculateRates_DoesNotAddStaleDataToRing(t *testing.T) { - rt := NewRateTracker() - base := time.Unix(1000, 0) - - // Seed - rt.CalculateRates("vm-100", models.IOMetrics{ - DiskRead: 1000, NetworkIn: 3000, Timestamp: base, - }) - - // Real data - rt.CalculateRates("vm-100", models.IOMetrics{ - DiskRead: 6000, NetworkIn: 18000, Timestamp: base.Add(10 * time.Second), - }) - - // Stale data — should not be added to ring - rt.CalculateRates("vm-100", models.IOMetrics{ - DiskRead: 6000, NetworkIn: 18000, Timestamp: base.Add(20 * time.Second), - }) - - // Ring should still have 2 entries (seed + one real update) - ring := rt.history["vm-100"] - if ring.count != 2 { - t.Errorf("ring count after stale data: got %d, want 2", ring.count) - } -} - -func TestClear(t *testing.T) { - rt := NewRateTracker() - base := time.Unix(1000, 0) - - rt.CalculateRates("vm-100", models.IOMetrics{ - DiskRead: 1000, Timestamp: base, - }) - rt.CalculateRates("vm-200", models.IOMetrics{ - DiskRead: 1000, Timestamp: base, - }) - - if len(rt.history) != 2 { - t.Fatalf("expected 2 entries in history, got %d", len(rt.history)) - } - - rt.Clear() - - if len(rt.history) != 0 { - t.Errorf("expected history to be empty after Clear, got %d entries", len(rt.history)) - } - if len(rt.lastRates) != 0 { - t.Errorf("expected lastRates to be empty after Clear, got %d entries", len(rt.lastRates)) - } - - // After clear, first call returns -1 - d, w, ni, no := rt.CalculateRates("vm-100", models.IOMetrics{ - DiskRead: 1000, Timestamp: base, - }) - if d != -1 || w != -1 || ni != -1 || no != -1 { - t.Errorf("after Clear: got (%v, %v, %v, %v), want (-1, -1, -1, -1)", d, w, ni, no) - } -} - -func TestRateTrackerCleanup(t *testing.T) { - rt := NewRateTracker() +func TestRateTrackerCleanupUsesLatestSampleEvenWhenIdleOrPartial(t *testing.T) { + tracker := NewRateTracker() now := time.Now() - - rt.CalculateRates("active-guest", models.IOMetrics{ - DiskRead: 1000, Timestamp: now.Add(-1 * time.Hour), - }) - rt.CalculateRates("stale-guest", models.IOMetrics{ - DiskRead: 1000, Timestamp: now.Add(-48 * time.Hour), + tracker.CalculateRates("idle", fullCounterSample(now.Add(-2*time.Hour), 100, 100, 100, 100)) + tracker.CalculateRates("idle", models.IOMetrics{ + Timestamp: now, + Presence: models.IOCounterPresence{Explicit: true}, }) + tracker.CalculateRates("stale", fullCounterSample(now.Add(-2*time.Hour), 100, 100, 100, 100)) - if len(rt.history) != 2 { - t.Fatalf("expected 2 entries, got %d", len(rt.history)) + if removed := tracker.Cleanup(now.Add(-time.Hour)); removed != 1 { + t.Fatalf("removed = %d, want 1", removed) } - - cutoff := now.Add(-24 * time.Hour) - removed := rt.Cleanup(cutoff) - - if removed != 1 { - t.Errorf("expected 1 entry removed, got %d", removed) - } - if len(rt.history) != 1 { - t.Errorf("expected 1 entry remaining, got %d", len(rt.history)) - } - if _, exists := rt.history["active-guest"]; !exists { - t.Error("active-guest should still exist after cleanup") - } - if _, exists := rt.history["stale-guest"]; exists { - t.Error("stale-guest should be removed after cleanup") + if _, ok := tracker.history["idle"]; !ok { + t.Fatal("idle resource was removed despite a recent sample") } } -func TestRateTrackerCleanupEmpty(t *testing.T) { - rt := NewRateTracker() - cutoff := time.Now().Add(-24 * time.Hour) - - removed := rt.Cleanup(cutoff) - if removed != 0 { - t.Errorf("expected 0 entries removed from empty tracker, got %d", removed) +func TestRateTrackerDiskBusyUsesElapsedMilliseconds(t *testing.T) { + tracker := NewRateTracker() + start := time.Unix(1_700_000_000, 0) + tracker.CalculateRatesWithBusy("disk", models.IOMetrics{DiskBusy: 100, Timestamp: start}) + _, _, busy, _, _ := tracker.CalculateRatesWithBusy("disk", models.IOMetrics{ + DiskBusy: 15_100, + Timestamp: start.Add(30 * time.Second), + }) + if busy != 50 { + t.Fatalf("busy = %v, want 50%%", busy) + } +} + +func TestRateTrackerClearRestoresFirstSampleUnknown(t *testing.T) { + tracker := NewRateTracker() + now := time.Now() + tracker.CalculateRates("guest", fullCounterSample(now, 1, 2, 3, 4)) + tracker.Clear() + read, write, in, out := tracker.CalculateRates("guest", fullCounterSample(now.Add(time.Second), 2, 3, 4, 5)) + if read != -1 || write != -1 || in != -1 || out != -1 { + t.Fatalf("first rates after clear = (%v, %v, %v, %v)", read, write, in, out) } } diff --git a/internal/monitoring/resource_stale_thresholds.go b/internal/monitoring/resource_stale_thresholds.go index b98c65369..cc5bccf3b 100644 --- a/internal/monitoring/resource_stale_thresholds.go +++ b/internal/monitoring/resource_stale_thresholds.go @@ -39,6 +39,10 @@ func (m *Monitor) resourceStaleThresholds() map[unifiedresources.DataSource]time return ResourceStaleThresholdsForConfig(m.config) } +func (m *Monitor) pveNodeOfflineGracePeriod() time.Duration { + return m.resourceStaleThresholds()[unifiedresources.SourceProxmox] +} + func effectivePVEPollingIntervalForConfig(cfg *config.Config) time.Duration { const minInterval = 10 * time.Second const maxInterval = time.Hour diff --git a/internal/unifiedresources/code_standards_test.go b/internal/unifiedresources/code_standards_test.go index 102ea7912..ec8e73cb8 100644 --- a/internal/unifiedresources/code_standards_test.go +++ b/internal/unifiedresources/code_standards_test.go @@ -59,6 +59,7 @@ package unifiedresources // banned direct-state access patterns. import ( + "encoding/json" "os" "path/filepath" "reflect" @@ -70,6 +71,22 @@ import ( "github.com/rcourtman/pulse-go-rewrite/internal/models" ) +func TestMetricValueAlwaysCarriesNumericValueWhenObjectExists(t *testing.T) { + payload, err := json.Marshal(MetricValue{Unit: "bytes/s", Source: SourceProxmox}) + if err != nil { + t.Fatal(err) + } + + var decoded map[string]any + if err := json.Unmarshal(payload, &decoded); err != nil { + t.Fatal(err) + } + value, ok := decoded["value"].(float64) + if !ok || value != 0 { + t.Fatalf("MetricValue JSON must contain numeric zero, got %s", payload) + } +} + func TestActionTruthTypesStayUnifiedResourceOwned(t *testing.T) { repoRoot := filepath.Join("..", "..") roots := []string{"internal/api", "internal/ai", "internal/agentexec", "internal/hostagent", "internal/dockeragent", "internal/relay", "internal/workflow", "internal/workflows"} diff --git a/internal/unifiedresources/metrics.go b/internal/unifiedresources/metrics.go index f3c401c79..6c50f57b1 100644 --- a/internal/unifiedresources/metrics.go +++ b/internal/unifiedresources/metrics.go @@ -159,6 +159,7 @@ func metricsFromVM(vm models.VM) *ResourceMetrics { vm.NetworkOut, vm.DiskRead, vm.DiskWrite, + vm.IORateValidity, SourceProxmox, ) } @@ -172,6 +173,7 @@ func metricsFromContainer(ct models.Container) *ResourceMetrics { ct.NetworkOut, ct.DiskRead, ct.DiskWrite, + ct.IORateValidity, SourceProxmox, ) } @@ -181,6 +183,7 @@ func buildVMMetricPayload( memory models.Memory, disk models.Disk, netIn, netOut, diskRead, diskWrite int64, + rateValidity models.IORateValidity, source DataSource, ) *ResourceMetrics { metrics := &ResourceMetrics{} @@ -194,7 +197,7 @@ func buildVMMetricPayload( percent := percentFromUsage(disk.Usage) metrics.Disk = &MetricValue{Used: &disk.Used, Total: &disk.Total, Percent: percent, Unit: "bytes", Source: source} } - setNetworkAndDiskIOMetricsVM(metrics, netIn, netOut, diskRead, diskWrite, source) + setNetworkAndDiskIOMetricsVM(metrics, netIn, netOut, diskRead, diskWrite, rateValidity, source) return metrics } @@ -822,17 +825,18 @@ func setNetworkAndDiskIOMetricsHost(metrics *ResourceMetrics, netIn, netOut, dis } } -func setNetworkAndDiskIOMetricsVM(metrics *ResourceMetrics, netIn, netOut, diskRead, diskWrite int64, source DataSource) { - if netIn != 0 { +func setNetworkAndDiskIOMetricsVM(metrics *ResourceMetrics, netIn, netOut, diskRead, diskWrite int64, validity models.IORateValidity, source DataSource) { + effective := validity.EffectiveForRates(diskRead, diskWrite, netIn, netOut) + if effective.NetworkIn { metrics.NetIn = &MetricValue{Value: float64(netIn), Unit: "bytes/s", Source: source} } - if netOut != 0 { + if effective.NetworkOut { metrics.NetOut = &MetricValue{Value: float64(netOut), Unit: "bytes/s", Source: source} } - if diskRead != 0 { + if effective.DiskRead { metrics.DiskRead = &MetricValue{Value: float64(diskRead), Unit: "bytes/s", Source: source} } - if diskWrite != 0 { + if effective.DiskWrite { metrics.DiskWrite = &MetricValue{Value: float64(diskWrite), Unit: "bytes/s", Source: source} } } diff --git a/internal/unifiedresources/metrics_test.go b/internal/unifiedresources/metrics_test.go index b42486031..d99e54af5 100644 --- a/internal/unifiedresources/metrics_test.go +++ b/internal/unifiedresources/metrics_test.go @@ -33,6 +33,31 @@ func TestMetricsFromDockerHostIncludesIORates(t *testing.T) { } } +func TestProxmoxGuestMetricsDistinguishValidZeroFromUnknown(t *testing.T) { + validZero := metricsFromVM(models.VM{ + Status: "running", + IORateValidity: models.IORateValidity{ + Explicit: true, + DiskRead: true, + DiskWrite: true, + NetworkIn: true, + NetworkOut: true, + }, + }) + if validZero.DiskRead == nil || validZero.DiskRead.Value != 0 || + validZero.DiskWrite == nil || validZero.NetIn == nil || validZero.NetOut == nil { + t.Fatalf("valid zero rates were not projected: %+v", validZero) + } + + unknown := metricsFromVM(models.VM{ + Status: "running", + IORateValidity: models.IORateValidity{Explicit: true}, + }) + if unknown.DiskRead != nil || unknown.DiskWrite != nil || unknown.NetIn != nil || unknown.NetOut != nil { + t.Fatalf("unknown rates were projected as measurements: %+v", unknown) + } +} + func TestUnavailableMemoryDoesNotProjectOrOverwriteTrustedCrossSourceMetric(t *testing.T) { unavailable := models.UnavailableMemory(8 * 1024 * 1024 * 1024) diff --git a/internal/unifiedresources/types.go b/internal/unifiedresources/types.go index a0cd9fe33..e7306e638 100644 --- a/internal/unifiedresources/types.go +++ b/internal/unifiedresources/types.go @@ -321,7 +321,7 @@ type ResourceMetrics struct { // MetricValue represents a metric value, optionally with totals. type MetricValue struct { - Value float64 `json:"value,omitempty"` + Value float64 `json:"value"` Used *int64 `json:"used,omitempty"` Total *int64 `json:"total,omitempty"` Percent float64 `json:"percent,omitempty"` diff --git a/pkg/proxmox/client.go b/pkg/proxmox/client.go index 57a709a27..876fac5f5 100644 --- a/pkg/proxmox/client.go +++ b/pkg/proxmox/client.go @@ -943,59 +943,63 @@ func (c *Client) GetVMRRDData(ctx context.Context, node string, vmid int, timefr // VM represents a Proxmox VE virtual machine type VM struct { - VMID int `json:"vmid"` - Name string `json:"name"` - Node string `json:"node"` - Pool string `json:"pool,omitempty"` - Status string `json:"status"` - CPU float64 `json:"cpu"` - CPUs int `json:"cpus"` - Mem uint64 `json:"mem"` - MaxMem uint64 `json:"maxmem"` - Disk uint64 `json:"disk"` - MaxDisk uint64 `json:"maxdisk"` - NetIn uint64 `json:"netin"` - NetOut uint64 `json:"netout"` - DiskRead uint64 `json:"diskread"` - DiskWrite uint64 `json:"diskwrite"` - Uptime uint64 `json:"uptime"` - Template int `json:"template"` - Tags string `json:"tags"` - Lock string `json:"lock"` - Agent int `json:"agent"` + VMID int `json:"vmid"` + Name string `json:"name"` + Node string `json:"node"` + Pool string `json:"pool,omitempty"` + Status string `json:"status"` + CPU float64 `json:"cpu"` + CPUs int `json:"cpus"` + Mem uint64 `json:"mem"` + MaxMem uint64 `json:"maxmem"` + Disk uint64 `json:"disk"` + MaxDisk uint64 `json:"maxdisk"` + NetIn uint64 `json:"netin"` + NetOut uint64 `json:"netout"` + DiskRead uint64 `json:"diskread"` + DiskWrite uint64 `json:"diskwrite"` + Uptime uint64 `json:"uptime"` + Template int `json:"template"` + Tags string `json:"tags"` + Lock string `json:"lock"` + Agent int `json:"agent"` + IOCounters IOCounterPresence `json:"-"` + ObservedAt time.Time `json:"-"` } // Container represents a Proxmox VE LXC container type Container struct { - VMID FlexInt `json:"vmid"` // Changed to FlexInt to handle string VMIDs from some Proxmox versions - Name string `json:"name"` - Node string `json:"node"` - Pool string `json:"pool,omitempty"` - Status string `json:"status"` - CPU float64 `json:"cpu"` - CPUs FlexInt `json:"cpus"` - Mem uint64 `json:"mem"` - MaxMem uint64 `json:"maxmem"` - Swap uint64 `json:"swap"` - MaxSwap uint64 `json:"maxswap"` - Disk uint64 `json:"disk"` - MaxDisk uint64 `json:"maxdisk"` - NetIn uint64 `json:"netin"` - NetOut uint64 `json:"netout"` - DiskRead uint64 `json:"diskread"` - DiskWrite uint64 `json:"diskwrite"` - Uptime uint64 `json:"uptime"` - Template int `json:"template"` - Tags string `json:"tags"` - Lock string `json:"lock"` - Hostname string `json:"hostname,omitempty"` - IP string `json:"ip,omitempty"` - IP6 string `json:"ip6,omitempty"` - IPv4 json.RawMessage `json:"ipv4,omitempty"` - IPv6 json.RawMessage `json:"ipv6,omitempty"` - Network map[string]ContainerNetworkConfig `json:"network,omitempty"` - DiskInfo map[string]ContainerDiskUsage `json:"diskinfo,omitempty"` - RootFS string `json:"rootfs,omitempty"` + VMID FlexInt `json:"vmid"` // Changed to FlexInt to handle string VMIDs from some Proxmox versions + Name string `json:"name"` + Node string `json:"node"` + Pool string `json:"pool,omitempty"` + Status string `json:"status"` + CPU float64 `json:"cpu"` + CPUs FlexInt `json:"cpus"` + Mem uint64 `json:"mem"` + MaxMem uint64 `json:"maxmem"` + Swap uint64 `json:"swap"` + MaxSwap uint64 `json:"maxswap"` + Disk uint64 `json:"disk"` + MaxDisk uint64 `json:"maxdisk"` + NetIn uint64 `json:"netin"` + NetOut uint64 `json:"netout"` + DiskRead uint64 `json:"diskread"` + DiskWrite uint64 `json:"diskwrite"` + Uptime uint64 `json:"uptime"` + Template int `json:"template"` + Tags string `json:"tags"` + Lock string `json:"lock"` + Hostname string `json:"hostname,omitempty"` + IP string `json:"ip,omitempty"` + IP6 string `json:"ip6,omitempty"` + IPv4 json.RawMessage `json:"ipv4,omitempty"` + IPv6 json.RawMessage `json:"ipv6,omitempty"` + Network map[string]ContainerNetworkConfig `json:"network,omitempty"` + DiskInfo map[string]ContainerDiskUsage `json:"diskinfo,omitempty"` + RootFS string `json:"rootfs,omitempty"` + IOCounters IOCounterPresence `json:"-"` + ObservedAt time.Time `json:"-"` } // ContainerNetworkConfig captures basic container network status information. @@ -1153,6 +1157,7 @@ func (c *Client) GetVMs(ctx context.Context, node string) ([]VM, error) { return nil, err } + stampVMObservation(result.Data, time.Now().UTC()) return result.Data, nil } @@ -1172,6 +1177,7 @@ func (c *Client) GetContainers(ctx context.Context, node string) ([]Container, e return nil, err } + stampContainerObservation(result.Data, time.Now().UTC()) return result.Data, nil } @@ -2290,6 +2296,7 @@ func (c *Client) GetVMStatus(ctx context.Context, node string, vmid int) (*VMSta return nil, err } + result.Data.ObservedAt = time.Now().UTC() return &result.Data, nil } @@ -2309,31 +2316,34 @@ func (c *Client) GetContainerStatus(ctx context.Context, node string, vmid int) return nil, err } + result.Data.ObservedAt = time.Now().UTC() return &result.Data, nil } // ClusterResource represents a resource from /cluster/resources type ClusterResource struct { - ID string `json:"id"` - Type string `json:"type"` - Node string `json:"node"` - Pool string `json:"pool,omitempty"` - Status string `json:"status"` - Name string `json:"name,omitempty"` - VMID int `json:"vmid,omitempty"` - CPU float64 `json:"cpu,omitempty"` - MaxCPU int `json:"maxcpu,omitempty"` - Mem uint64 `json:"mem,omitempty"` - MaxMem uint64 `json:"maxmem,omitempty"` - Disk uint64 `json:"disk,omitempty"` - MaxDisk uint64 `json:"maxdisk,omitempty"` - NetIn uint64 `json:"netin,omitempty"` - NetOut uint64 `json:"netout,omitempty"` - DiskRead uint64 `json:"diskread,omitempty"` - DiskWrite uint64 `json:"diskwrite,omitempty"` - Uptime uint64 `json:"uptime,omitempty"` - Template int `json:"template,omitempty"` - Tags string `json:"tags,omitempty"` + ID string `json:"id"` + Type string `json:"type"` + Node string `json:"node"` + Pool string `json:"pool,omitempty"` + Status string `json:"status"` + Name string `json:"name,omitempty"` + VMID int `json:"vmid,omitempty"` + CPU float64 `json:"cpu,omitempty"` + MaxCPU int `json:"maxcpu,omitempty"` + Mem uint64 `json:"mem,omitempty"` + MaxMem uint64 `json:"maxmem,omitempty"` + Disk uint64 `json:"disk,omitempty"` + MaxDisk uint64 `json:"maxdisk,omitempty"` + NetIn uint64 `json:"netin,omitempty"` + NetOut uint64 `json:"netout,omitempty"` + DiskRead uint64 `json:"diskread,omitempty"` + DiskWrite uint64 `json:"diskwrite,omitempty"` + Uptime uint64 `json:"uptime,omitempty"` + Template int `json:"template,omitempty"` + Tags string `json:"tags,omitempty"` + IOCounters IOCounterPresence `json:"-"` + ObservedAt time.Time `json:"-"` } // GetClusterResources returns all resources (VMs, containers) across the cluster @@ -2357,6 +2367,7 @@ func (c *Client) GetClusterResources(ctx context.Context, resourceType string) ( return nil, err } + stampClusterResourceObservation(result.Data, time.Now().UTC()) return result.Data, nil } @@ -2596,24 +2607,26 @@ func (a *VMAgentField) UnmarshalJSON(data []byte) error { // VMStatus represents detailed VM status returned by Proxmox. type VMStatus struct { - Status string `json:"status"` - CPU float64 `json:"cpu"` - CPUs int `json:"cpus"` - Mem uint64 `json:"mem"` - MaxMem uint64 `json:"maxmem"` - Balloon uint64 `json:"balloon"` - BalloonMin uint64 `json:"balloon_min"` - BalloonInfo *VMBalloonInfo `json:"ballooninfo,omitempty"` - FreeMem uint64 `json:"freemem"` - MemInfo *VMMemInfo `json:"meminfo,omitempty"` - Disk uint64 `json:"disk"` - MaxDisk uint64 `json:"maxdisk"` - DiskRead uint64 `json:"diskread"` - DiskWrite uint64 `json:"diskwrite"` - NetIn uint64 `json:"netin"` - NetOut uint64 `json:"netout"` - Uptime uint64 `json:"uptime"` - Agent VMAgentField `json:"agent"` + Status string `json:"status"` + CPU float64 `json:"cpu"` + CPUs int `json:"cpus"` + Mem uint64 `json:"mem"` + MaxMem uint64 `json:"maxmem"` + Balloon uint64 `json:"balloon"` + BalloonMin uint64 `json:"balloon_min"` + BalloonInfo *VMBalloonInfo `json:"ballooninfo,omitempty"` + FreeMem uint64 `json:"freemem"` + MemInfo *VMMemInfo `json:"meminfo,omitempty"` + Disk uint64 `json:"disk"` + MaxDisk uint64 `json:"maxdisk"` + DiskRead uint64 `json:"diskread"` + DiskWrite uint64 `json:"diskwrite"` + NetIn uint64 `json:"netin"` + NetOut uint64 `json:"netout"` + Uptime uint64 `json:"uptime"` + Agent VMAgentField `json:"agent"` + IOCounters IOCounterPresence `json:"-"` + ObservedAt time.Time `json:"-"` } // GetZFSPoolStatus gets the status of ZFS pools on a node diff --git a/pkg/proxmox/io_counters.go b/pkg/proxmox/io_counters.go new file mode 100644 index 000000000..717ddecd3 --- /dev/null +++ b/pkg/proxmox/io_counters.go @@ -0,0 +1,124 @@ +package proxmox + +import ( + "bytes" + "encoding/json" + "time" +) + +// IOCounterPresence distinguishes explicit zero counters from fields omitted +// by a Proxmox endpoint or permission-limited response. +type IOCounterPresence struct { + Explicit bool + DiskRead bool + DiskWrite bool + NetworkIn bool + NetworkOut bool +} + +// Effective keeps manually constructed fixtures and older producers +// compatible while decoded API responses retain exact field presence. +func (p IOCounterPresence) Effective() IOCounterPresence { + if p.Explicit { + return p + } + return IOCounterPresence{ + Explicit: true, + DiskRead: true, + DiskWrite: true, + NetworkIn: true, + NetworkOut: true, + } +} + +func counterPresence(raw map[string]json.RawMessage) IOCounterPresence { + return IOCounterPresence{ + Explicit: true, + DiskRead: jsonFieldPresent(raw, "diskread"), + DiskWrite: jsonFieldPresent(raw, "diskwrite"), + NetworkIn: jsonFieldPresent(raw, "netin"), + NetworkOut: jsonFieldPresent(raw, "netout"), + } +} + +func jsonFieldPresent(raw map[string]json.RawMessage, key string) bool { + value, ok := raw[key] + return ok && !bytes.Equal(bytes.TrimSpace(value), []byte("null")) +} + +func decodeWithCounterPresence(data []byte, target any) (IOCounterPresence, error) { + var raw map[string]json.RawMessage + if err := json.Unmarshal(data, &raw); err != nil { + return IOCounterPresence{}, err + } + if err := json.Unmarshal(data, target); err != nil { + return IOCounterPresence{}, err + } + return counterPresence(raw), nil +} + +func (v *VM) UnmarshalJSON(data []byte) error { + type alias VM + var decoded alias + presence, err := decodeWithCounterPresence(data, &decoded) + if err != nil { + return err + } + *v = VM(decoded) + v.IOCounters = presence + return nil +} + +func (c *Container) UnmarshalJSON(data []byte) error { + type alias Container + var decoded alias + presence, err := decodeWithCounterPresence(data, &decoded) + if err != nil { + return err + } + *c = Container(decoded) + c.IOCounters = presence + return nil +} + +func (r *ClusterResource) UnmarshalJSON(data []byte) error { + type alias ClusterResource + var decoded alias + presence, err := decodeWithCounterPresence(data, &decoded) + if err != nil { + return err + } + *r = ClusterResource(decoded) + r.IOCounters = presence + return nil +} + +func (s *VMStatus) UnmarshalJSON(data []byte) error { + type alias VMStatus + var decoded alias + presence, err := decodeWithCounterPresence(data, &decoded) + if err != nil { + return err + } + *s = VMStatus(decoded) + s.IOCounters = presence + return nil +} + +func stampVMObservation(values []VM, observedAt time.Time) { + for i := range values { + values[i].ObservedAt = observedAt + } +} + +func stampContainerObservation(values []Container, observedAt time.Time) { + for i := range values { + values[i].ObservedAt = observedAt + } +} + +func stampClusterResourceObservation(values []ClusterResource, observedAt time.Time) { + for i := range values { + values[i].ObservedAt = observedAt + } +} diff --git a/pkg/proxmox/io_counters_test.go b/pkg/proxmox/io_counters_test.go new file mode 100644 index 000000000..8259242b3 --- /dev/null +++ b/pkg/proxmox/io_counters_test.go @@ -0,0 +1,117 @@ +package proxmox + +import ( + "encoding/json" + "testing" + "time" +) + +func TestClusterResourceCounterPresenceDistinguishesZeroNullAndMissing(t *testing.T) { + var resource ClusterResource + if err := json.Unmarshal([]byte(`{ + "type":"qemu", + "diskread":0, + "diskwrite":null, + "netin":42 + }`), &resource); err != nil { + t.Fatal(err) + } + + presence := resource.IOCounters.Effective() + if !presence.DiskRead || !presence.NetworkIn { + t.Fatalf("explicit zero/value fields were not present: %+v", presence) + } + if presence.DiskWrite || presence.NetworkOut { + t.Fatalf("null/missing fields were incorrectly present: %+v", presence) + } +} + +func TestGuestStatusTypesRetainCounterPresence(t *testing.T) { + tests := []struct { + name string + read func() IOCounterPresence + }{ + { + name: "vm listing", + read: func() IOCounterPresence { + var value VM + _ = json.Unmarshal([]byte(`{"diskread":0}`), &value) + return value.IOCounters + }, + }, + { + name: "lxc status", + read: func() IOCounterPresence { + var value Container + _ = json.Unmarshal([]byte(`{"diskread":0}`), &value) + return value.IOCounters + }, + }, + { + name: "qemu status", + read: func() IOCounterPresence { + var value VMStatus + _ = json.Unmarshal([]byte(`{"diskread":0}`), &value) + return value.IOCounters + }, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + presence := test.read().Effective() + if !presence.DiskRead || presence.DiskWrite || presence.NetworkIn || presence.NetworkOut { + t.Fatalf("presence = %+v, want only diskread", presence) + } + }) + } +} + +func TestObservationStampUsesOneReceiptTimeForResponse(t *testing.T) { + observedAt := time.Date(2026, time.July, 24, 8, 30, 0, 0, time.UTC) + + vms := []VM{{VMID: 100}, {VMID: 101}} + stampVMObservation(vms, observedAt) + for _, vm := range vms { + if !vm.ObservedAt.Equal(observedAt) { + t.Fatalf("VM %d observedAt = %v", vm.VMID, vm.ObservedAt) + } + } + + containers := []Container{{VMID: 200}, {VMID: 201}} + stampContainerObservation(containers, observedAt) + for _, container := range containers { + if !container.ObservedAt.Equal(observedAt) { + t.Fatalf("container %d observedAt = %v", container.VMID, container.ObservedAt) + } + } + + resources := []ClusterResource{{VMID: 300}, {VMID: 301}} + stampClusterResourceObservation(resources, observedAt) + for _, resource := range resources { + if !resource.ObservedAt.Equal(observedAt) { + t.Fatalf("resource %d observedAt = %v", resource.VMID, resource.ObservedAt) + } + } +} + +func TestInternalCounterMetadataNeverChangesProxmoxWireShape(t *testing.T) { + payload, err := json.Marshal(ClusterResource{ + VMID: 100, + DiskRead: 0, + IOCounters: IOCounterPresence{Explicit: true, DiskRead: true}, + ObservedAt: time.Now(), + }) + if err != nil { + t.Fatal(err) + } + var raw map[string]json.RawMessage + if err := json.Unmarshal(payload, &raw); err != nil { + t.Fatal(err) + } + for _, key := range []string{"ioCounters", "IOCounters", "observedAt", "ObservedAt"} { + if _, ok := raw[key]; ok { + t.Fatalf("internal counter metadata %q leaked into JSON", key) + } + } +} diff --git a/scripts/release_control/canonical_completion_guard_test.py b/scripts/release_control/canonical_completion_guard_test.py index c0bb033f3..a42c11ec1 100644 --- a/scripts/release_control/canonical_completion_guard_test.py +++ b/scripts/release_control/canonical_completion_guard_test.py @@ -216,6 +216,7 @@ class CanonicalCompletionGuardTest(unittest.TestCase): "exact_files": [ "internal/api/host_agent_removal_lifecycle_integration_test.go", "internal/config/host_continuity_test.go", + "internal/models/metrics_types_test.go", "internal/monitoring/monitor_host_agent_removal_lifecycle_test.go", "internal/monitoring/monitor_host_agents_test.go", "scripts/installtests/agent_state_dir_lifecycle_test.go", @@ -258,6 +259,7 @@ class CanonicalCompletionGuardTest(unittest.TestCase): "docker-swarm-runtime", "kubernetes-native-agent-runtime", "runtime-report-model", + "proxmox-guest-counter-runtime", "proxmox-zfs-runtime", "proxmox-cluster-client-runtime", "proxmox-ceph-runtime", @@ -287,6 +289,7 @@ class CanonicalCompletionGuardTest(unittest.TestCase): "internal/monitoring/canonical_guardrails_test.go", "internal/monitoring/issue1485_unraid_lifecycle_test.go", "internal/monitoring/issue1595_collection_trust_test.go", + "internal/monitoring/issue1613_contract_test.go", "internal/monitoring/monitor_additional_test.go", "internal/monitoring/monitor_alert_intent_test.go", "internal/monitoring/monitor_alert_override_migration_test.go", @@ -295,6 +298,8 @@ class CanonicalCompletionGuardTest(unittest.TestCase): "internal/monitoring/monitor_host_agent_removal_lifecycle_test.go", "internal/monitoring/monitor_host_agents_test.go", "internal/monitoring/monitor_pve_cluster_refresh_test.go", + "internal/monitoring/monitor_pve_guest_lxc_test.go", + "internal/monitoring/ratetracker_test.go", "internal/unifiedresources/code_standards_test.go", ], } @@ -392,6 +397,7 @@ class CanonicalCompletionGuardTest(unittest.TestCase): "exact_files": [ "internal/api/host_agent_removal_lifecycle_integration_test.go", "internal/config/host_continuity_test.go", + "internal/models/metrics_types_test.go", "internal/monitoring/monitor_host_agent_removal_lifecycle_test.go", "internal/monitoring/monitor_host_agents_test.go", "scripts/installtests/agent_state_dir_lifecycle_test.go", diff --git a/scripts/release_control/subsystem_lookup_test.py b/scripts/release_control/subsystem_lookup_test.py index db4737470..5d416da9b 100644 --- a/scripts/release_control/subsystem_lookup_test.py +++ b/scripts/release_control/subsystem_lookup_test.py @@ -4159,6 +4159,7 @@ class SubsystemLookupTest(unittest.TestCase): [ "internal/api/host_agent_removal_lifecycle_integration_test.go", "internal/config/host_continuity_test.go", + "internal/models/metrics_types_test.go", "internal/monitoring/monitor_host_agent_removal_lifecycle_test.go", "internal/monitoring/monitor_host_agents_test.go", "scripts/installtests/agent_state_dir_lifecycle_test.go",