mirror of
https://github.com/rcourtman/Pulse.git
synced 2026-09-11 14:00:29 +00:00
Fix Proxmox workload refresh coherence
This commit is contained in:
@@ -27,6 +27,14 @@ Monitoring owns source freshness cadence for Proxmox, PBS, and PMG resources:
|
|||||||
the stale threshold is derived from the configured polling interval with a
|
the stale threshold is derived from the configured polling interval with a
|
||||||
minimum floor, so API-facing resource status must not degrade merely because a
|
minimum floor, so API-facing resource status must not degrade merely because a
|
||||||
healthy source is between normal poll cycles.
|
healthy source is between normal poll cycles.
|
||||||
|
Proxmox guest enumeration is a generation boundary. VM and LXC collection and
|
||||||
|
enrichment must finish before one `State.UpdateGuestsForInstance` publication,
|
||||||
|
so readers never observe a VM-only or LXC-only intermediate snapshot. A failed
|
||||||
|
online cluster member retains only that member's last coherent guests and their
|
||||||
|
source-native `{instance}:{node}:{vmid}` IDs; a successful empty member
|
||||||
|
enumeration is authoritative and removes genuinely deleted guests. Collection
|
||||||
|
failure remains visible through source freshness/error state and must not be
|
||||||
|
converted into an authoritative empty inventory.
|
||||||
Host-agent report liveness is server-observed, not agent-clock-observed:
|
Host-agent report liveness is server-observed, not agent-clock-observed:
|
||||||
`ApplyHostReport` must stamp `Host.LastSeen`, agent-sourced Ceph cluster
|
`ApplyHostReport` must stamp `Host.LastSeen`, agent-sourced Ceph cluster
|
||||||
freshness, and host-agent cluster sensor freshness from Pulse receipt time, so
|
freshness, and host-agent cluster sensor freshness from Pulse receipt time, so
|
||||||
@@ -213,6 +221,7 @@ node-local Agent evidence.
|
|||||||
36. `internal/dockeragent/docker_client.go`
|
36. `internal/dockeragent/docker_client.go`
|
||||||
37. `pkg/agents/docker/report.go`
|
37. `pkg/agents/docker/report.go`
|
||||||
38. `internal/models/models.go`
|
38. `internal/models/models.go`
|
||||||
|
38a. `internal/models/proxmox_guest_state.go`
|
||||||
39. `internal/models/models_frontend.go`
|
39. `internal/models/models_frontend.go`
|
||||||
40. `internal/models/converters.go`
|
40. `internal/models/converters.go`
|
||||||
41. `internal/models/deepcopy.go`
|
41. `internal/models/deepcopy.go`
|
||||||
|
|||||||
@@ -854,6 +854,19 @@ shell clickable behind another overlay.
|
|||||||
|
|
||||||
## Current State
|
## Current State
|
||||||
|
|
||||||
|
### Workload refreshes retain one coherent paged generation
|
||||||
|
|
||||||
|
The workload polling cache accepts a REST refresh only when every advertised
|
||||||
|
page succeeds. A failed later page leaves the prior array and its stable row
|
||||||
|
identities in place, ends loading, and exposes the refresh error; it must not
|
||||||
|
publish a partial list, clear rows as a loading transition, or silently hide
|
||||||
|
the error. Proxmox workload filtering consumes the source-authored
|
||||||
|
`proxmox.runtimeStatus` before aggregate health so an availability-only
|
||||||
|
freshness update cannot collapse a Running view. A later complete response may
|
||||||
|
still remove an authoritatively deleted guest. These rules preserve sort,
|
||||||
|
selection, drawer, and virtualized viewport state without adding another
|
||||||
|
resource scan, websocket subscription, or browser-local source of truth.
|
||||||
|
|
||||||
### Canonical mutation-plane dependency
|
### Canonical mutation-plane dependency
|
||||||
|
|
||||||
Router wiring now exposes only typed action planning for model-originated
|
Router wiring now exposes only typed action planning for model-originated
|
||||||
|
|||||||
@@ -5186,6 +5186,7 @@
|
|||||||
"internal/models/deepcopy.go",
|
"internal/models/deepcopy.go",
|
||||||
"internal/models/models.go",
|
"internal/models/models.go",
|
||||||
"internal/models/models_frontend.go",
|
"internal/models/models_frontend.go",
|
||||||
|
"internal/models/proxmox_guest_state.go",
|
||||||
"internal/proxmoxidentity/backup_identity.go",
|
"internal/proxmoxidentity/backup_identity.go",
|
||||||
"pkg/agents/docker/report.go",
|
"pkg/agents/docker/report.go",
|
||||||
"pkg/agents/host/report.go",
|
"pkg/agents/host/report.go",
|
||||||
@@ -5398,6 +5399,7 @@
|
|||||||
"internal/models/deepcopy.go",
|
"internal/models/deepcopy.go",
|
||||||
"internal/models/models.go",
|
"internal/models/models.go",
|
||||||
"internal/models/models_frontend.go",
|
"internal/models/models_frontend.go",
|
||||||
|
"internal/models/proxmox_guest_state.go",
|
||||||
"pkg/agents/docker/report.go"
|
"pkg/agents/docker/report.go"
|
||||||
],
|
],
|
||||||
"allow_same_subsystem_tests": false,
|
"allow_same_subsystem_tests": false,
|
||||||
|
|||||||
@@ -1917,6 +1917,17 @@ that Safe auto-fix or Autopilot remediation is verified.
|
|||||||
|
|
||||||
## Current State
|
## Current State
|
||||||
|
|
||||||
|
### Proxmox runtime continuity is not protection evidence
|
||||||
|
|
||||||
|
The additive `ProxmoxData.RuntimeStatus` field preserves VM/LXC power-state
|
||||||
|
presentation while a platform snapshot is stale or refreshing. Storage and
|
||||||
|
recovery consumers must not interpret `running`, `stopped`, a retained guest
|
||||||
|
row, or an availability facet as backup freshness, protection coverage,
|
||||||
|
restore-chain verification, or recoverability. Authoritative guest deletion
|
||||||
|
may remove the inventory row and its canonical history identity, while
|
||||||
|
Recovery Assurance remains governed solely by its own backup and verification
|
||||||
|
evidence.
|
||||||
|
|
||||||
### Storage history remount state is bounded
|
### Storage history remount state is bounded
|
||||||
|
|
||||||
Storage summary history no longer treats every node/range combination visited
|
Storage summary history no longer treats every node/range combination visited
|
||||||
|
|||||||
@@ -43,6 +43,18 @@ comparable fallback because LXC observes the shared kernel and QEMU accounting
|
|||||||
is independently scoped. Agent CPU may fill the field only when the platform
|
is independently scoped. Agent CPU may fill the field only when the platform
|
||||||
has no CPU observation. Agent-only fields that the platform does not provide
|
has no CPU observation. Agent-only fields that the platform does not provide
|
||||||
remain eligible per metric.
|
remain eligible per metric.
|
||||||
|
Proxmox guest power state is likewise source-authored semantic state, separate
|
||||||
|
from aggregate resource health. `ProxmoxData.RuntimeStatus` retains the last
|
||||||
|
coherent `running`/`stopped` observation while `Resource.Status` and
|
||||||
|
`SourceStatus` continue to report stale, warning, offline, availability, and
|
||||||
|
error evidence honestly. Availability checks may add facets and influence
|
||||||
|
aggregate health after Proxmox becomes stale, but they must not overwrite or
|
||||||
|
erase the Proxmox runtime state used by workload power filters. Full monitor
|
||||||
|
registry generations and incremental agent/availability mutations serialize
|
||||||
|
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.
|
||||||
Physical-disk resources own cross-source disk identity. When Proxmox inventory
|
Physical-disk resources own cross-source disk identity. When Proxmox inventory
|
||||||
and host-agent SMART telemetry describe the same device, the merged resource
|
and host-agent SMART telemetry describe the same device, the merged resource
|
||||||
must retain Proxmox node/instance source payloads while carrying SMART
|
must retain Proxmox node/instance source payloads while carrying SMART
|
||||||
|
|||||||
@@ -228,6 +228,126 @@ describe('useWorkloads', () => {
|
|||||||
dispose();
|
dispose();
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it('keeps Proxmox power state stable while aggregate freshness changes, then removes an authoritative deletion', async () => {
|
||||||
|
const guest = (vmid: number, status: string, runtimeStatus: string) => ({
|
||||||
|
...sampleResource,
|
||||||
|
id: `cluster-a-pve1-${vmid}`,
|
||||||
|
type: 'system-container',
|
||||||
|
name: `lxc-${vmid}`,
|
||||||
|
status,
|
||||||
|
vmid,
|
||||||
|
proxmox: {
|
||||||
|
vmid,
|
||||||
|
nodeName: 'pve1',
|
||||||
|
instance: 'cluster-a',
|
||||||
|
runtimeStatus,
|
||||||
|
},
|
||||||
|
});
|
||||||
|
|
||||||
|
apiFetchJSONMock.mockResolvedValueOnce({
|
||||||
|
data: [
|
||||||
|
guest(101, 'online', 'running'),
|
||||||
|
guest(102, 'online', 'running'),
|
||||||
|
guest(103, 'online', 'running'),
|
||||||
|
],
|
||||||
|
meta: { totalPages: 1 },
|
||||||
|
});
|
||||||
|
|
||||||
|
let dispose = () => {};
|
||||||
|
let result: ReturnType<UseWorkloadsModule['useWorkloads']> | undefined;
|
||||||
|
createRoot((d) => {
|
||||||
|
dispose = d;
|
||||||
|
const [enabled] = createSignal(true);
|
||||||
|
result = useWorkloads(enabled);
|
||||||
|
});
|
||||||
|
|
||||||
|
await waitForWorkloadCount(() => result!.workloads().length, 3);
|
||||||
|
const initialIds = result!.workloads().map((workload) => workload.id);
|
||||||
|
|
||||||
|
apiFetchJSONMock.mockResolvedValueOnce({
|
||||||
|
data: [
|
||||||
|
guest(101, 'warning', 'running'),
|
||||||
|
{
|
||||||
|
...guest(102, 'online', 'running'),
|
||||||
|
availability: { targetId: 'probe-102', protocol: 'icmp', enabled: true, available: true },
|
||||||
|
},
|
||||||
|
guest(103, 'warning', 'running'),
|
||||||
|
],
|
||||||
|
meta: { totalPages: 1 },
|
||||||
|
});
|
||||||
|
await result!.refetch();
|
||||||
|
|
||||||
|
expect(result!.workloads().map((workload) => workload.id)).toEqual(initialIds);
|
||||||
|
expect(result!.workloads().map((workload) => workload.status)).toEqual([
|
||||||
|
'running',
|
||||||
|
'running',
|
||||||
|
'running',
|
||||||
|
]);
|
||||||
|
|
||||||
|
apiFetchJSONMock.mockResolvedValueOnce({
|
||||||
|
data: [guest(101, 'online', 'running'), guest(102, 'online', 'running')],
|
||||||
|
meta: { totalPages: 1 },
|
||||||
|
});
|
||||||
|
await result!.refetch();
|
||||||
|
|
||||||
|
expect(result!.workloads().map((workload) => workload.id)).toEqual(initialIds.slice(0, 2));
|
||||||
|
expect(result!.error()).toBeUndefined();
|
||||||
|
|
||||||
|
dispose();
|
||||||
|
});
|
||||||
|
|
||||||
|
it('rejects a partial paged refresh and retains the last coherent snapshot', async () => {
|
||||||
|
const secondResource = {
|
||||||
|
...sampleResource,
|
||||||
|
id: 'cluster-a-pve1-102',
|
||||||
|
name: 'vm-102',
|
||||||
|
vmid: 102,
|
||||||
|
};
|
||||||
|
apiFetchJSONMock.mockResolvedValueOnce({
|
||||||
|
data: [sampleResource, secondResource],
|
||||||
|
meta: { totalPages: 1 },
|
||||||
|
});
|
||||||
|
|
||||||
|
let dispose = () => {};
|
||||||
|
let result: ReturnType<UseWorkloadsModule['useWorkloads']> | undefined;
|
||||||
|
createRoot((d) => {
|
||||||
|
dispose = d;
|
||||||
|
const [enabled] = createSignal(true);
|
||||||
|
result = useWorkloads(enabled);
|
||||||
|
});
|
||||||
|
|
||||||
|
await waitForWorkloadCount(() => result!.workloads().length, 2);
|
||||||
|
const coherentSnapshot = result!.workloads();
|
||||||
|
|
||||||
|
apiFetchJSONMock
|
||||||
|
.mockResolvedValueOnce({
|
||||||
|
data: [sampleResource],
|
||||||
|
meta: { totalPages: 2 },
|
||||||
|
})
|
||||||
|
.mockRejectedValueOnce(new Error('page 2 unavailable'));
|
||||||
|
|
||||||
|
await expect(result!.refetch()).rejects.toThrow('page 2 unavailable');
|
||||||
|
expect(result!.workloads()).toBe(coherentSnapshot);
|
||||||
|
expect(result!.workloads()).toHaveLength(2);
|
||||||
|
expect(result!.error()).toBeInstanceOf(Error);
|
||||||
|
|
||||||
|
apiFetchJSONMock
|
||||||
|
.mockResolvedValueOnce({
|
||||||
|
data: [sampleResource],
|
||||||
|
meta: { totalPages: 2 },
|
||||||
|
})
|
||||||
|
.mockResolvedValueOnce({
|
||||||
|
data: [],
|
||||||
|
meta: { totalPages: 2 },
|
||||||
|
});
|
||||||
|
await result!.refetch();
|
||||||
|
|
||||||
|
expect(result!.workloads()).toHaveLength(1);
|
||||||
|
expect(result!.error()).toBeUndefined();
|
||||||
|
|
||||||
|
dispose();
|
||||||
|
});
|
||||||
|
|
||||||
it('does not apply in-flight workload results after the hook is disabled', async () => {
|
it('does not apply in-flight workload results after the hook is disabled', async () => {
|
||||||
const pendingFetch = deferred<unknown>();
|
const pendingFetch = deferred<unknown>();
|
||||||
apiFetchJSONMock.mockImplementationOnce(() => pendingFetch.promise as Promise<any>);
|
apiFetchJSONMock.mockImplementationOnce(() => pendingFetch.promise as Promise<any>);
|
||||||
|
|||||||
@@ -87,6 +87,7 @@ type APIResource = {
|
|||||||
node?: string;
|
node?: string;
|
||||||
instance?: string;
|
instance?: string;
|
||||||
proxmox?: {
|
proxmox?: {
|
||||||
|
runtimeStatus?: string;
|
||||||
nodeName?: string;
|
nodeName?: string;
|
||||||
clusterName?: string;
|
clusterName?: string;
|
||||||
instance?: string;
|
instance?: string;
|
||||||
@@ -451,7 +452,9 @@ const mapResourceToWorkload = (resource: APIResource): WorkloadGuest | null => {
|
|||||||
node,
|
node,
|
||||||
instance,
|
instance,
|
||||||
status: normalizeWorkloadStatus(
|
status: normalizeWorkloadStatus(
|
||||||
resource.status || (platformType === 'vmware-vsphere' ? resource.vmware?.powerState : null),
|
resource.proxmox?.runtimeStatus ||
|
||||||
|
resource.status ||
|
||||||
|
(platformType === 'vmware-vsphere' ? resource.vmware?.powerState : null),
|
||||||
),
|
),
|
||||||
type:
|
type:
|
||||||
workloadType === 'vm'
|
workloadType === 'vm'
|
||||||
@@ -587,10 +590,9 @@ async function fetchWorkloads(): Promise<WorkloadGuest[]> {
|
|||||||
for (let page = 2; page <= totalPages; page++) {
|
for (let page = 2; page <= totalPages; page++) {
|
||||||
pageRequests.push(apiFetchJSON<unknown>(buildWorkloadsUrl(page), { cache: 'no-store' }));
|
pageRequests.push(apiFetchJSON<unknown>(buildWorkloadsUrl(page), { cache: 'no-store' }));
|
||||||
}
|
}
|
||||||
const settled = await Promise.allSettled(pageRequests);
|
const responses = await Promise.all(pageRequests);
|
||||||
for (const result of settled) {
|
for (const response of responses) {
|
||||||
if (result.status !== 'fulfilled') continue;
|
const pageData = resolveWorkloadsPayload(response);
|
||||||
const pageData = resolveWorkloadsPayload(result.value);
|
|
||||||
allResources.push(...pageData.data);
|
allResources.push(...pageData.data);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -775,8 +777,12 @@ export function useWorkloads(enabled: Accessor<boolean> = () => true) {
|
|||||||
}
|
}
|
||||||
applyWorkloads(data, scope);
|
applyWorkloads(data, scope);
|
||||||
setError(undefined);
|
setError(undefined);
|
||||||
} catch {
|
} catch (err) {
|
||||||
// Silently ignore poll errors; keep showing last data
|
// Keep the last coherent snapshot while making the failed refresh
|
||||||
|
// observable to the surface.
|
||||||
|
if (scope === resolveActiveOrgScope()) {
|
||||||
|
setError(err);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}, DEFAULT_POLL_INTERVAL_MS);
|
}, DEFAULT_POLL_INTERVAL_MS);
|
||||||
onCleanup(() => clearInterval(id));
|
onCleanup(() => clearInterval(id));
|
||||||
|
|||||||
@@ -240,6 +240,7 @@ describe('Resource Helper Functions', () => {
|
|||||||
vmid: 101,
|
vmid: 101,
|
||||||
nodeName: 'pve-a',
|
nodeName: 'pve-a',
|
||||||
instance: 'cluster-a',
|
instance: 'cluster-a',
|
||||||
|
runtimeStatus: 'running',
|
||||||
diskStatusReason: 'agent-not-running',
|
diskStatusReason: 'agent-not-running',
|
||||||
guestAgentStatus: 'expected-unreachable',
|
guestAgentStatus: 'expected-unreachable',
|
||||||
guestAgentExpected: true,
|
guestAgentExpected: true,
|
||||||
@@ -262,6 +263,7 @@ describe('Resource Helper Functions', () => {
|
|||||||
|
|
||||||
expect(resource.proxmox?.guestAgentStatus).toBe('expected-unreachable');
|
expect(resource.proxmox?.guestAgentStatus).toBe('expected-unreachable');
|
||||||
expect(resource.proxmox?.guestAgentExpected).toBe(true);
|
expect(resource.proxmox?.guestAgentExpected).toBe(true);
|
||||||
|
expect(resource.proxmox?.runtimeStatus).toBe('running');
|
||||||
expect(resource.proxmox?.diskStatusReason).toBe('agent-not-running');
|
expect(resource.proxmox?.diskStatusReason).toBe('agent-not-running');
|
||||||
expect(resource.incidents?.[0]?.source).toBe('qemu-guest-agent');
|
expect(resource.incidents?.[0]?.source).toBe('qemu-guest-agent');
|
||||||
});
|
});
|
||||||
|
|||||||
@@ -644,6 +644,8 @@ export interface ResourceAgentMeta {
|
|||||||
}
|
}
|
||||||
|
|
||||||
export interface ResourceProxmoxMeta {
|
export interface ResourceProxmoxMeta {
|
||||||
|
/** Proxmox-authored VM/LXC power state, kept separate from collection freshness. */
|
||||||
|
runtimeStatus?: string;
|
||||||
vmid?: number;
|
vmid?: number;
|
||||||
node?: string;
|
node?: string;
|
||||||
nodeName?: string;
|
nodeName?: string;
|
||||||
|
|||||||
@@ -779,11 +779,12 @@ func TestResourceListDerivesProxmoxWorkloadParentFromUnifiedSeed(t *testing.T) {
|
|||||||
Sources: []unified.DataSource{unified.SourceProxmox},
|
Sources: []unified.DataSource{unified.SourceProxmox},
|
||||||
Identity: unified.ResourceIdentity{Hostnames: []string{"cloudflared"}},
|
Identity: unified.ResourceIdentity{Hostnames: []string{"cloudflared"}},
|
||||||
Proxmox: &unified.ProxmoxData{
|
Proxmox: &unified.ProxmoxData{
|
||||||
SourceID: "delly:delly:104",
|
SourceID: "delly:delly:104",
|
||||||
NodeName: "delly",
|
RuntimeStatus: "running",
|
||||||
ClusterName: "homelab",
|
NodeName: "delly",
|
||||||
Instance: "delly",
|
ClusterName: "homelab",
|
||||||
VMID: 104,
|
Instance: "delly",
|
||||||
|
VMID: 104,
|
||||||
},
|
},
|
||||||
},
|
},
|
||||||
},
|
},
|
||||||
@@ -815,6 +816,9 @@ func TestResourceListDerivesProxmoxWorkloadParentFromUnifiedSeed(t *testing.T) {
|
|||||||
if resource.Proxmox == nil || resource.Proxmox.NodeName != "delly" || resource.Proxmox.ClusterName != "homelab" {
|
if resource.Proxmox == nil || resource.Proxmox.NodeName != "delly" || resource.Proxmox.ClusterName != "homelab" {
|
||||||
t.Fatalf("expected cloudflared Proxmox node metadata, got %+v", resource.Proxmox)
|
t.Fatalf("expected cloudflared Proxmox node metadata, got %+v", resource.Proxmox)
|
||||||
}
|
}
|
||||||
|
if resource.Proxmox.RuntimeStatus != "running" {
|
||||||
|
t.Fatalf("expected source-authored runtime status, got %+v", resource.Proxmox)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestStateEndpointDerivesProxmoxWorkloadParentFromSupplementalRecords(t *testing.T) {
|
func TestStateEndpointDerivesProxmoxWorkloadParentFromSupplementalRecords(t *testing.T) {
|
||||||
|
|||||||
@@ -0,0 +1,34 @@
|
|||||||
|
package models
|
||||||
|
|
||||||
|
import "time"
|
||||||
|
|
||||||
|
// UpdateGuestsForInstance replaces the VM and container projections for one
|
||||||
|
// Proxmox instance under a single state lock. Pollers collect and enrich both
|
||||||
|
// guest kinds before calling this method so readers cannot observe a mixed
|
||||||
|
// generation while a refresh is in flight.
|
||||||
|
func (s *State) UpdateGuestsForInstance(instanceName string, vms []VM, containers []Container) {
|
||||||
|
s.mu.Lock()
|
||||||
|
defer s.mu.Unlock()
|
||||||
|
|
||||||
|
s.VMs = updateSliceByInstanceWithBackup(
|
||||||
|
s.VMs, vms, instanceName,
|
||||||
|
func(vm VM) string { return vm.ID },
|
||||||
|
func(vm VM) string { return vm.Instance },
|
||||||
|
func(vm VM) int { return vm.VMID },
|
||||||
|
func(vm VM) time.Time { return vm.LastBackup },
|
||||||
|
func(vm VM, t time.Time) VM { vm.LastBackup = t; return vm },
|
||||||
|
cloneVM,
|
||||||
|
func(items []VM, i, j int) bool { return items[i].VMID < items[j].VMID },
|
||||||
|
)
|
||||||
|
s.Containers = updateSliceByInstanceWithBackup(
|
||||||
|
s.Containers, containers, instanceName,
|
||||||
|
func(ct Container) string { return ct.ID },
|
||||||
|
func(ct Container) string { return ct.Instance },
|
||||||
|
func(ct Container) int { return ct.VMID },
|
||||||
|
func(ct Container) time.Time { return ct.LastBackup },
|
||||||
|
func(ct Container, t time.Time) Container { ct.LastBackup = t; return ct },
|
||||||
|
cloneContainer,
|
||||||
|
func(items []Container, i, j int) bool { return items[i].VMID < items[j].VMID },
|
||||||
|
)
|
||||||
|
s.LastUpdate = time.Now()
|
||||||
|
}
|
||||||
@@ -1327,6 +1327,44 @@ func TestSyncGuestBackupTimesVMContainerCollision(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestUpdateGuestsForInstancePublishesCoherentGeneration(t *testing.T) {
|
||||||
|
state := NewState()
|
||||||
|
previousBackup := time.Now().Add(-2 * time.Hour).UTC()
|
||||||
|
|
||||||
|
state.UpdateVMs([]VM{
|
||||||
|
{ID: "lab-a:node-a:101", VMID: 101, Instance: "lab-a", Node: "node-a", LastBackup: previousBackup},
|
||||||
|
{ID: "lab-b:node-b:201", VMID: 201, Instance: "lab-b", Node: "node-b"},
|
||||||
|
})
|
||||||
|
state.UpdateContainers([]Container{
|
||||||
|
{ID: "lab-a:node-a:102", VMID: 102, Instance: "lab-a", Node: "node-a", LastBackup: previousBackup},
|
||||||
|
{ID: "lab-b:node-b:202", VMID: 202, Instance: "lab-b", Node: "node-b"},
|
||||||
|
})
|
||||||
|
before := state.GetSnapshot().LastUpdate
|
||||||
|
|
||||||
|
state.UpdateGuestsForInstance(
|
||||||
|
"lab-a",
|
||||||
|
[]VM{{ID: "lab-a:node-a:101", VMID: 101, Instance: "lab-a", Node: "node-a", Status: "running"}},
|
||||||
|
[]Container{{ID: "lab-a:node-a:103", VMID: 103, Instance: "lab-a", Node: "node-a", Status: "running"}},
|
||||||
|
)
|
||||||
|
|
||||||
|
snapshot := state.GetSnapshot()
|
||||||
|
if !snapshot.LastUpdate.After(before) {
|
||||||
|
t.Fatalf("LastUpdate did not advance: before=%v after=%v", before, snapshot.LastUpdate)
|
||||||
|
}
|
||||||
|
if len(snapshot.VMs) != 2 || len(snapshot.Containers) != 2 {
|
||||||
|
t.Fatalf("unexpected coherent generation sizes: vms=%d containers=%d", len(snapshot.VMs), len(snapshot.Containers))
|
||||||
|
}
|
||||||
|
if snapshot.VMs[0].ID != "lab-a:node-a:101" || !snapshot.VMs[0].LastBackup.Equal(previousBackup) {
|
||||||
|
t.Fatalf("updated VM did not retain backup state: %+v", snapshot.VMs[0])
|
||||||
|
}
|
||||||
|
if snapshot.Containers[0].ID != "lab-a:node-a:103" {
|
||||||
|
t.Fatalf("authoritatively deleted container remained in state: %+v", snapshot.Containers)
|
||||||
|
}
|
||||||
|
if snapshot.VMs[1].Instance != "lab-b" || snapshot.Containers[1].Instance != "lab-b" {
|
||||||
|
t.Fatalf("other instance was not isolated: vms=%+v containers=%+v", snapshot.VMs, snapshot.Containers)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func TestUpdateStorageBackupsForInstance(t *testing.T) {
|
func TestUpdateStorageBackupsForInstance(t *testing.T) {
|
||||||
state := NewState()
|
state := NewState()
|
||||||
|
|
||||||
|
|||||||
@@ -1110,11 +1110,14 @@ func TestProxmoxGuestDockerInventoryUsesCanonicalReportIngestPath(t *testing.T)
|
|||||||
},
|
},
|
||||||
"monitor_pve_guest_poll.go": {
|
"monitor_pve_guest_poll.go": {
|
||||||
"m.CollectProxmoxGuestDockerInventory(ctx, allContainers)",
|
"m.CollectProxmoxGuestDockerInventory(ctx, allContainers)",
|
||||||
"m.state.UpdateContainersForInstance(instanceName, allContainers)",
|
"m.state.UpdateGuestsForInstance(instanceName, allVMs, allContainers)",
|
||||||
},
|
},
|
||||||
"monitor_polling_containers.go": {
|
"monitor_polling_containers.go": {
|
||||||
"m.CollectProxmoxGuestDockerInventory(ctx, allContainers)",
|
"m.CollectProxmoxGuestDockerInventory(ctx, allContainers)",
|
||||||
"m.state.UpdateContainersForInstance(instanceName, allContainers)",
|
"return allContainers",
|
||||||
|
},
|
||||||
|
"monitor_pve.go": {
|
||||||
|
"m.state.UpdateGuestsForInstance(instanceName, vms, containers)",
|
||||||
},
|
},
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -22,6 +22,38 @@ type vmMemoryTrustStubClient struct {
|
|||||||
vmAgentMemCalls int
|
vmAgentMemCalls int
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestCollectVMsWithNodesRetainsFailedNodeRuntimeState(t *testing.T) {
|
||||||
|
monitor := newTestPVEMonitor("lab")
|
||||||
|
defer monitor.alertManager.Stop()
|
||||||
|
defer monitor.notificationMgr.Stop()
|
||||||
|
monitor.state.UpdateVMsForInstance("lab", []models.VM{{
|
||||||
|
ID: "lab:node-b:201",
|
||||||
|
VMID: 201,
|
||||||
|
Name: "database",
|
||||||
|
Node: "node-b",
|
||||||
|
Instance: "lab",
|
||||||
|
Status: "running",
|
||||||
|
}})
|
||||||
|
|
||||||
|
client := &partialNodeGuestClient{
|
||||||
|
stubPVEClient: &stubPVEClient{},
|
||||||
|
failedNodes: map[string]bool{"node-b": true},
|
||||||
|
}
|
||||||
|
vms := monitor.collectVMsWithNodes(
|
||||||
|
context.Background(),
|
||||||
|
"lab",
|
||||||
|
"",
|
||||||
|
false,
|
||||||
|
client,
|
||||||
|
[]proxmox.Node{{Node: "node-b", Status: "online"}},
|
||||||
|
map[string]string{"node-b": "online"},
|
||||||
|
)
|
||||||
|
|
||||||
|
if len(vms) != 1 || vms[0].ID != "lab:node-b:201" || vms[0].Status != "running" {
|
||||||
|
t.Fatalf("failed-node VM continuity = %+v, want retained source ID and running state", vms)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func (s *vmMemoryTrustStubClient) GetVMs(ctx context.Context, node string) ([]proxmox.VM, error) {
|
func (s *vmMemoryTrustStubClient) GetVMs(ctx context.Context, node string) ([]proxmox.VM, error) {
|
||||||
return s.vms, nil
|
return s.vms, nil
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1172,12 +1172,12 @@ func TestMonitor_PreviousGuestContextForInstance_Extra(t *testing.T) {
|
|||||||
if len(prev.vms) != 1 || prev.vms[0].VMID != 101 || prev.vms[0].Instance != "pve1" || prev.vms[0].Name != "vm1" {
|
if len(prev.vms) != 1 || prev.vms[0].VMID != 101 || prev.vms[0].Instance != "pve1" || prev.vms[0].Name != "vm1" {
|
||||||
t.Fatalf("expected only pve1 VMs, got %#v", prev.vms)
|
t.Fatalf("expected only pve1 VMs, got %#v", prev.vms)
|
||||||
}
|
}
|
||||||
canonicalID := prev.vms[0].ID
|
guestID := makeGuestID("pve1", "", 101)
|
||||||
if len(prev.vmsByID) != 2 || prev.vmsByID[canonicalID].VMID != 101 || prev.vmsByID[makeGuestID("pve1", "", 101)].VMID != 101 {
|
if prev.vms[0].ID != guestID || len(prev.vmsByID) != 1 || prev.vmsByID[guestID].VMID != 101 {
|
||||||
t.Fatalf("expected previous VM lookup to be indexed by canonical and runtime guest IDs, got %#v", prev.vmsByID)
|
t.Fatalf("expected previous VM lookup to retain the source-authored guest ID, got %#v", prev.vmsByID)
|
||||||
}
|
}
|
||||||
if prev.vmsByID[canonicalID].Disk.Total != 100 || prev.vmsByID[canonicalID].Disk.Used != 40 {
|
if prev.vmsByID[guestID].Disk.Total != 100 || prev.vmsByID[guestID].Disk.Used != 40 {
|
||||||
t.Fatalf("expected previous VM projection to preserve aggregate disk summary, got %#v", prev.vmsByID[canonicalID].Disk)
|
t.Fatalf("expected previous VM projection to preserve aggregate disk summary, got %#v", prev.vmsByID[guestID].Disk)
|
||||||
}
|
}
|
||||||
if len(prev.containers) != 2 {
|
if len(prev.containers) != 2 {
|
||||||
t.Fatalf("expected only pve1 containers, got %#v", prev.containers)
|
t.Fatalf("expected only pve1 containers, got %#v", prev.containers)
|
||||||
|
|||||||
@@ -13,7 +13,7 @@ import (
|
|||||||
"github.com/rs/zerolog/log"
|
"github.com/rs/zerolog/log"
|
||||||
)
|
)
|
||||||
|
|
||||||
func (m *Monitor) pollContainersWithNodes(ctx context.Context, instanceName string, clusterName string, isCluster bool, client PVEClientInterface, nodes []proxmox.Node, nodeEffectiveStatus map[string]string) {
|
func (m *Monitor) collectContainersWithNodes(ctx context.Context, instanceName string, clusterName string, isCluster bool, client PVEClientInterface, nodes []proxmox.Node, nodeEffectiveStatus map[string]string) []models.Container {
|
||||||
startTime := time.Now()
|
startTime := time.Now()
|
||||||
|
|
||||||
// Channel to collect container results from each node
|
// Channel to collect container results from each node
|
||||||
@@ -303,10 +303,12 @@ func (m *Monitor) pollContainersWithNodes(ctx context.Context, instanceName stri
|
|||||||
lxcTemplateSubjects := make(map[string]struct{})
|
lxcTemplateSubjects := make(map[string]struct{})
|
||||||
successfulNodes := 0
|
successfulNodes := 0
|
||||||
failedNodes := 0
|
failedNodes := 0
|
||||||
|
failedNodeNames := make(map[string]struct{})
|
||||||
|
|
||||||
for result := range resultChan {
|
for result := range resultChan {
|
||||||
if result.err != nil {
|
if result.err != nil {
|
||||||
failedNodes++
|
failedNodes++
|
||||||
|
failedNodeNames[result.node] = struct{}{}
|
||||||
} else {
|
} else {
|
||||||
successfulNodes++
|
successfulNodes++
|
||||||
allContainers = append(allContainers, result.containers...)
|
allContainers = append(allContainers, result.containers...)
|
||||||
@@ -319,28 +321,27 @@ func (m *Monitor) pollContainersWithNodes(ctx context.Context, instanceName stri
|
|||||||
m.updatePVEBackupTemplateSubjectsForType(instanceName, "lxc", lxcTemplateSubjects)
|
m.updatePVEBackupTemplateSubjectsForType(instanceName, "lxc", lxcTemplateSubjects)
|
||||||
}
|
}
|
||||||
|
|
||||||
// If we got ZERO containers but had containers before (likely cluster health issue),
|
preservedContainers := 0
|
||||||
// preserve previous containers instead of clearing them
|
if len(failedNodeNames) > 0 {
|
||||||
if len(allContainers) == 0 && len(nodes) > 0 {
|
for _, container := range prevGuests.containers {
|
||||||
allContainers = append(allContainers, prevGuests.containers...)
|
if _, failed := failedNodeNames[container.Node]; failed {
|
||||||
prevContainerCount := len(prevGuests.containers)
|
allContainers = append(allContainers, container)
|
||||||
if prevContainerCount > 0 {
|
preservedContainers++
|
||||||
log.Warn().
|
}
|
||||||
Str("instance", instanceName).
|
|
||||||
Int("prevContainers", prevContainerCount).
|
|
||||||
Int("successfulNodes", successfulNodes).
|
|
||||||
Int("totalNodes", len(nodes)).
|
|
||||||
Msg("Traditional polling returned zero containers but had containers before - preserving previous containers")
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
if preservedContainers > 0 {
|
||||||
|
log.Warn().
|
||||||
|
Str("instance", instanceName).
|
||||||
|
Int("preservedContainers", preservedContainers).
|
||||||
|
Int("failedNodes", failedNodes).
|
||||||
|
Msg("Preserved prior containers for nodes whose enumeration failed")
|
||||||
|
}
|
||||||
|
|
||||||
// Check Docker presence for containers that need it (new, restarted, started)
|
// Check Docker presence for containers that need it (new, restarted, started)
|
||||||
allContainers = m.CheckContainersForDocker(ctx, allContainers)
|
allContainers = m.CheckContainersForDocker(ctx, allContainers)
|
||||||
m.CollectProxmoxGuestDockerInventory(ctx, allContainers)
|
m.CollectProxmoxGuestDockerInventory(ctx, allContainers)
|
||||||
|
|
||||||
// Update state with all containers
|
|
||||||
m.state.UpdateContainersForInstance(instanceName, allContainers)
|
|
||||||
|
|
||||||
// Record guest metrics history for running containers (enables sparkline/trends view)
|
// Record guest metrics history for running containers (enables sparkline/trends view)
|
||||||
if !shouldSkipNativeMockStateMetricWrites() {
|
if !shouldSkipNativeMockStateMetricWrites() {
|
||||||
now := time.Now()
|
now := time.Now()
|
||||||
@@ -362,6 +363,16 @@ func (m *Monitor) pollContainersWithNodes(ctx context.Context, instanceName stri
|
|||||||
Int("failedNodes", failedNodes).
|
Int("failedNodes", failedNodes).
|
||||||
Dur("duration", duration).
|
Dur("duration", duration).
|
||||||
Msg("Parallel container polling completed")
|
Msg("Parallel container polling completed")
|
||||||
|
|
||||||
|
return allContainers
|
||||||
|
}
|
||||||
|
|
||||||
|
// pollContainersWithNodes retains the focused single-kind polling entry point
|
||||||
|
// used by tests and maintenance callers. The production guest cycle uses
|
||||||
|
// collectContainersWithNodes and publishes both guest kinds atomically.
|
||||||
|
func (m *Monitor) pollContainersWithNodes(ctx context.Context, instanceName string, clusterName string, isCluster bool, client PVEClientInterface, nodes []proxmox.Node, nodeEffectiveStatus map[string]string) {
|
||||||
|
containers := m.collectContainersWithNodes(ctx, instanceName, clusterName, isCluster, client, nodes, nodeEffectiveStatus)
|
||||||
|
m.state.UpdateContainersForInstance(instanceName, containers)
|
||||||
}
|
}
|
||||||
|
|
||||||
// pollStorageWithNodes polls storage from all nodes in parallel using goroutines
|
// pollStorageWithNodes polls storage from all nodes in parallel using goroutines
|
||||||
|
|||||||
@@ -12,7 +12,7 @@ import (
|
|||||||
"github.com/rs/zerolog/log"
|
"github.com/rs/zerolog/log"
|
||||||
)
|
)
|
||||||
|
|
||||||
func (m *Monitor) pollVMsWithNodes(ctx context.Context, instanceName string, clusterName string, isCluster bool, client PVEClientInterface, nodes []proxmox.Node, nodeEffectiveStatus map[string]string) {
|
func (m *Monitor) collectVMsWithNodes(ctx context.Context, instanceName string, clusterName string, isCluster bool, client PVEClientInterface, nodes []proxmox.Node, nodeEffectiveStatus map[string]string) []models.VM {
|
||||||
startTime := time.Now()
|
startTime := time.Now()
|
||||||
|
|
||||||
type nodeResult struct {
|
type nodeResult struct {
|
||||||
@@ -85,10 +85,12 @@ func (m *Monitor) pollVMsWithNodes(ctx context.Context, instanceName string, clu
|
|||||||
qemuTemplateSubjects := make(map[string]struct{})
|
qemuTemplateSubjects := make(map[string]struct{})
|
||||||
successfulNodes := 0
|
successfulNodes := 0
|
||||||
failedNodes := 0
|
failedNodes := 0
|
||||||
|
failedNodeNames := make(map[string]struct{})
|
||||||
|
|
||||||
for result := range resultChan {
|
for result := range resultChan {
|
||||||
if result.err != nil {
|
if result.err != nil {
|
||||||
failedNodes++
|
failedNodes++
|
||||||
|
failedNodeNames[result.node] = struct{}{}
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
successfulNodes++
|
successfulNodes++
|
||||||
@@ -101,20 +103,22 @@ func (m *Monitor) pollVMsWithNodes(ctx context.Context, instanceName string, clu
|
|||||||
m.updatePVEBackupTemplateSubjectsForType(instanceName, "qemu", qemuTemplateSubjects)
|
m.updatePVEBackupTemplateSubjectsForType(instanceName, "qemu", qemuTemplateSubjects)
|
||||||
}
|
}
|
||||||
|
|
||||||
if len(allVMs) == 0 && len(nodes) > 0 {
|
preservedVMs := 0
|
||||||
allVMs = append(allVMs, prevGuests.vms...)
|
if len(failedNodeNames) > 0 {
|
||||||
prevVMCount := len(prevGuests.vms)
|
for _, vm := range prevGuests.vms {
|
||||||
if prevVMCount > 0 {
|
if _, failed := failedNodeNames[vm.Node]; failed {
|
||||||
log.Warn().
|
allVMs = append(allVMs, vm)
|
||||||
Str("instance", instanceName).
|
preservedVMs++
|
||||||
Int("prevVMs", prevVMCount).
|
}
|
||||||
Int("successfulNodes", successfulNodes).
|
|
||||||
Int("totalNodes", len(nodes)).
|
|
||||||
Msg("Traditional polling returned zero VMs but had VMs before - preserving previous VMs")
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
if preservedVMs > 0 {
|
||||||
m.state.UpdateVMsForInstance(instanceName, allVMs)
|
log.Warn().
|
||||||
|
Str("instance", instanceName).
|
||||||
|
Int("preservedVMs", preservedVMs).
|
||||||
|
Int("failedNodes", failedNodes).
|
||||||
|
Msg("Preserved prior VMs for nodes whose enumeration failed")
|
||||||
|
}
|
||||||
|
|
||||||
if !shouldSkipNativeMockStateMetricWrites() {
|
if !shouldSkipNativeMockStateMetricWrites() {
|
||||||
now := time.Now()
|
now := time.Now()
|
||||||
@@ -136,4 +140,14 @@ func (m *Monitor) pollVMsWithNodes(ctx context.Context, instanceName string, clu
|
|||||||
Int("failedNodes", failedNodes).
|
Int("failedNodes", failedNodes).
|
||||||
Dur("duration", duration).
|
Dur("duration", duration).
|
||||||
Msg("Parallel VM polling completed")
|
Msg("Parallel VM polling completed")
|
||||||
|
|
||||||
|
return allVMs
|
||||||
|
}
|
||||||
|
|
||||||
|
// pollVMsWithNodes retains the focused single-kind polling entry point used by
|
||||||
|
// tests and maintenance callers. The production guest cycle uses
|
||||||
|
// collectVMsWithNodes and publishes both guest kinds atomically.
|
||||||
|
func (m *Monitor) pollVMsWithNodes(ctx context.Context, instanceName string, clusterName string, isCluster bool, client PVEClientInterface, nodes []proxmox.Node, nodeEffectiveStatus map[string]string) {
|
||||||
|
vms := m.collectVMsWithNodes(ctx, instanceName, clusterName, isCluster, client, nodes, nodeEffectiveStatus)
|
||||||
|
m.state.UpdateVMsForInstance(instanceName, vms)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -93,14 +93,17 @@ func previousVMFromView(vm *unifiedresources.VMView) models.VM {
|
|||||||
if vm == nil {
|
if vm == nil {
|
||||||
return models.VM{}
|
return models.VM{}
|
||||||
}
|
}
|
||||||
|
instance := vm.Instance()
|
||||||
|
node := vm.Node()
|
||||||
|
vmid := vm.VMID()
|
||||||
return models.VM{
|
return models.VM{
|
||||||
ID: vm.ID(),
|
ID: makeGuestID(instance, node, vmid),
|
||||||
Instance: vm.Instance(),
|
Instance: instance,
|
||||||
Node: vm.Node(),
|
Node: node,
|
||||||
VMID: vm.VMID(),
|
VMID: vmid,
|
||||||
Name: vm.Name(),
|
Name: vm.Name(),
|
||||||
Type: "qemu",
|
Type: "qemu",
|
||||||
Status: string(vm.Status()),
|
Status: vm.RuntimeStatus(),
|
||||||
IPAddresses: vm.IPAddresses(),
|
IPAddresses: vm.IPAddresses(),
|
||||||
OSName: vm.OSName(),
|
OSName: vm.OSName(),
|
||||||
OSVersion: vm.OSVersion(),
|
OSVersion: vm.OSVersion(),
|
||||||
@@ -122,13 +125,16 @@ func previousContainerFromView(ct *unifiedresources.ContainerView) models.Contai
|
|||||||
if ct == nil {
|
if ct == nil {
|
||||||
return models.Container{}
|
return models.Container{}
|
||||||
}
|
}
|
||||||
|
instance := ct.Instance()
|
||||||
|
node := ct.Node()
|
||||||
|
vmid := ct.VMID()
|
||||||
return models.Container{
|
return models.Container{
|
||||||
ID: ct.ID(),
|
ID: makeGuestID(instance, node, vmid),
|
||||||
Instance: ct.Instance(),
|
Instance: instance,
|
||||||
Node: ct.Node(),
|
Node: node,
|
||||||
VMID: ct.VMID(),
|
VMID: vmid,
|
||||||
Name: ct.Name(),
|
Name: ct.Name(),
|
||||||
Status: string(ct.Status()),
|
Status: ct.RuntimeStatus(),
|
||||||
Type: ct.ContainerType(),
|
Type: ct.ContainerType(),
|
||||||
IsOCI: ct.IsOCI(),
|
IsOCI: ct.IsOCI(),
|
||||||
LastSeen: ct.LastSeen(),
|
LastSeen: ct.LastSeen(),
|
||||||
|
|||||||
@@ -803,12 +803,16 @@ func (m *Monitor) pollGuestsWithFallback(
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Use optimized parallel polling for better performance
|
// Use optimized parallel polling for better performance
|
||||||
|
previous := m.previousGuestContextForInstance(instanceName)
|
||||||
|
vms := previous.vms
|
||||||
|
containers := previous.containers
|
||||||
if instanceCfg.MonitorVMs {
|
if instanceCfg.MonitorVMs {
|
||||||
m.pollVMsWithNodes(ctx, instanceName, instanceCfg.ClusterName, instanceCfg.IsCluster, client, nodes, nodeEffectiveStatus)
|
vms = m.collectVMsWithNodes(ctx, instanceName, instanceCfg.ClusterName, instanceCfg.IsCluster, client, nodes, nodeEffectiveStatus)
|
||||||
}
|
}
|
||||||
if instanceCfg.MonitorContainers {
|
if instanceCfg.MonitorContainers {
|
||||||
m.pollContainersWithNodes(ctx, instanceName, instanceCfg.ClusterName, instanceCfg.IsCluster, client, nodes, nodeEffectiveStatus)
|
containers = m.collectContainersWithNodes(ctx, instanceName, instanceCfg.ClusterName, instanceCfg.IsCluster, client, nodes, nodeEffectiveStatus)
|
||||||
}
|
}
|
||||||
|
m.state.UpdateGuestsForInstance(instanceName, vms, containers)
|
||||||
}
|
}
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
|
|||||||
@@ -44,15 +44,14 @@ func (m *Monitor) pollVMsAndContainersEfficient(ctx context.Context, instanceNam
|
|||||||
|
|
||||||
allVMs, allContainers = m.preserveGuestsForGracePeriod(instanceName, resources, prevGuests.vms, prevGuests.containers, nodeEffectiveStatus, allVMs, allContainers)
|
allVMs, allContainers = m.preserveGuestsForGracePeriod(instanceName, resources, prevGuests.vms, prevGuests.containers, nodeEffectiveStatus, allVMs, allContainers)
|
||||||
|
|
||||||
// Always update state when using efficient polling path
|
|
||||||
// Even if arrays are empty, we need to update to clear out VMs from genuinely offline nodes
|
|
||||||
m.state.UpdateVMsForInstance(instanceName, allVMs)
|
|
||||||
|
|
||||||
// Check Docker presence for containers that need it (new, restarted, started)
|
// Check Docker presence for containers that need it (new, restarted, started)
|
||||||
allContainers = m.CheckContainersForDocker(ctx, allContainers)
|
allContainers = m.CheckContainersForDocker(ctx, allContainers)
|
||||||
m.CollectProxmoxGuestDockerInventory(ctx, allContainers)
|
m.CollectProxmoxGuestDockerInventory(ctx, allContainers)
|
||||||
|
|
||||||
m.state.UpdateContainersForInstance(instanceName, allContainers)
|
// Publish the complete guest generation only after both VM and container
|
||||||
|
// collection/enrichment has finished. Empty authoritative results still
|
||||||
|
// remove genuinely deleted guests.
|
||||||
|
m.state.UpdateGuestsForInstance(instanceName, allVMs, allContainers)
|
||||||
|
|
||||||
m.recordGuestMetrics(allVMs, allContainers)
|
m.recordGuestMetrics(allVMs, allContainers)
|
||||||
|
|
||||||
|
|||||||
@@ -0,0 +1,116 @@
|
|||||||
|
package monitoring
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"errors"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"github.com/rcourtman/pulse-go-rewrite/internal/config"
|
||||||
|
"github.com/rcourtman/pulse-go-rewrite/internal/models"
|
||||||
|
"github.com/rcourtman/pulse-go-rewrite/pkg/proxmox"
|
||||||
|
)
|
||||||
|
|
||||||
|
type partialNodeGuestClient struct {
|
||||||
|
*stubPVEClient
|
||||||
|
failedNodes map[string]bool
|
||||||
|
vmsByNode map[string][]proxmox.VM
|
||||||
|
containersByNode map[string][]proxmox.Container
|
||||||
|
}
|
||||||
|
|
||||||
|
func (c *partialNodeGuestClient) GetClusterResources(context.Context, string) ([]proxmox.ClusterResource, error) {
|
||||||
|
return nil, errors.New("cluster resources unavailable")
|
||||||
|
}
|
||||||
|
|
||||||
|
func (c *partialNodeGuestClient) GetVMs(_ context.Context, node string) ([]proxmox.VM, error) {
|
||||||
|
if c.failedNodes[node] {
|
||||||
|
return nil, errors.New("VM enumeration unavailable")
|
||||||
|
}
|
||||||
|
return c.vmsByNode[node], nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (c *partialNodeGuestClient) GetContainers(_ context.Context, node string) ([]proxmox.Container, error) {
|
||||||
|
if c.failedNodes[node] {
|
||||||
|
return nil, errors.New("container enumeration unavailable")
|
||||||
|
}
|
||||||
|
return c.containersByNode[node], nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestPollGuestsWithFallbackRetainsOnlyFailedNodeGeneration(t *testing.T) {
|
||||||
|
monitor := newTestPVEMonitor("lab")
|
||||||
|
defer monitor.alertManager.Stop()
|
||||||
|
defer monitor.notificationMgr.Stop()
|
||||||
|
|
||||||
|
monitor.state.UpdateGuestsForInstance(
|
||||||
|
"lab",
|
||||||
|
[]models.VM{
|
||||||
|
{ID: "lab:node-a:101", VMID: 101, Name: "deleted-vm", Node: "node-a", Instance: "lab", Status: "running"},
|
||||||
|
{ID: "lab:node-b:201", VMID: 201, Name: "retained-vm", Node: "node-b", Instance: "lab", Status: "running"},
|
||||||
|
},
|
||||||
|
[]models.Container{
|
||||||
|
{ID: "lab:node-a:102", VMID: 102, Name: "deleted-ct", Node: "node-a", Instance: "lab", Status: "running"},
|
||||||
|
{ID: "lab:node-b:202", VMID: 202, Name: "retained-ct", Node: "node-b", Instance: "lab", Status: "running"},
|
||||||
|
},
|
||||||
|
)
|
||||||
|
monitor.state.UpdateGuestsForInstance(
|
||||||
|
"other",
|
||||||
|
[]models.VM{{ID: "other:node-c:301", VMID: 301, Node: "node-c", Instance: "other"}},
|
||||||
|
[]models.Container{{ID: "other:node-c:302", VMID: 302, Node: "node-c", Instance: "other"}},
|
||||||
|
)
|
||||||
|
|
||||||
|
client := &partialNodeGuestClient{
|
||||||
|
stubPVEClient: &stubPVEClient{},
|
||||||
|
failedNodes: map[string]bool{"node-b": true},
|
||||||
|
vmsByNode: map[string][]proxmox.VM{"node-a": {}},
|
||||||
|
containersByNode: map[string][]proxmox.Container{"node-a": {}},
|
||||||
|
}
|
||||||
|
nodes := []proxmox.Node{
|
||||||
|
{Node: "node-a", Status: "online"},
|
||||||
|
{Node: "node-b", Status: "online"},
|
||||||
|
}
|
||||||
|
nodeStatus := map[string]string{"node-a": "online", "node-b": "online"}
|
||||||
|
cfg := &config.PVEInstance{MonitorVMs: true, MonitorContainers: true}
|
||||||
|
|
||||||
|
if err := monitor.pollGuestsWithFallback(context.Background(), "lab", cfg, client, nodes, nodeStatus); err != nil {
|
||||||
|
t.Fatalf("partial poll failed: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
snapshot := monitor.GetState()
|
||||||
|
assertGuestIDs(t, snapshot.VMs, []string{"lab:node-b:201", "other:node-c:301"})
|
||||||
|
assertGuestIDs(t, snapshot.Containers, []string{"lab:node-b:202", "other:node-c:302"})
|
||||||
|
if snapshot.VMs[0].Status != "running" || snapshot.Containers[0].Status != "running" {
|
||||||
|
t.Fatalf("failed-node power state was not retained: vm=%q container=%q", snapshot.VMs[0].Status, snapshot.Containers[0].Status)
|
||||||
|
}
|
||||||
|
|
||||||
|
client.failedNodes["node-b"] = false
|
||||||
|
client.vmsByNode["node-b"] = []proxmox.VM{}
|
||||||
|
client.containersByNode["node-b"] = []proxmox.Container{}
|
||||||
|
|
||||||
|
if err := monitor.pollGuestsWithFallback(context.Background(), "lab", cfg, client, nodes, nodeStatus); err != nil {
|
||||||
|
t.Fatalf("recovery poll failed: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
snapshot = monitor.GetState()
|
||||||
|
assertGuestIDs(t, snapshot.VMs, []string{"other:node-c:301"})
|
||||||
|
assertGuestIDs(t, snapshot.Containers, []string{"other:node-c:302"})
|
||||||
|
}
|
||||||
|
|
||||||
|
func assertGuestIDs[T models.VM | models.Container](t *testing.T, guests []T, want []string) {
|
||||||
|
t.Helper()
|
||||||
|
got := make([]string, 0, len(guests))
|
||||||
|
for _, guest := range guests {
|
||||||
|
switch typed := any(guest).(type) {
|
||||||
|
case models.VM:
|
||||||
|
got = append(got, typed.ID)
|
||||||
|
case models.Container:
|
||||||
|
got = append(got, typed.ID)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if len(got) != len(want) {
|
||||||
|
t.Fatalf("guest ids = %v, want %v", got, want)
|
||||||
|
}
|
||||||
|
for i := range want {
|
||||||
|
if got[i] != want[i] {
|
||||||
|
t.Fatalf("guest ids = %v, want %v", got, want)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -200,6 +200,19 @@ func TestMonitorAdapterUsesConfiguredProxmoxStaleThreshold(t *testing.T) {
|
|||||||
if defaultVMs[0].Status() != StatusWarning {
|
if defaultVMs[0].Status() != StatusWarning {
|
||||||
t.Fatalf("default stale threshold status = %q, want warning", defaultVMs[0].Status())
|
t.Fatalf("default stale threshold status = %q, want warning", defaultVMs[0].Status())
|
||||||
}
|
}
|
||||||
|
if defaultVMs[0].RuntimeStatus() != "running" {
|
||||||
|
t.Fatalf("default stale runtime status = %q, want running", defaultVMs[0].RuntimeStatus())
|
||||||
|
}
|
||||||
|
defaultResources := defaultAdapter.GetByType(ResourceTypeVM)
|
||||||
|
if len(defaultResources) != 1 || defaultResources[0].Proxmox == nil {
|
||||||
|
t.Fatalf("expected one Proxmox VM resource, got %+v", defaultResources)
|
||||||
|
}
|
||||||
|
if got := defaultResources[0].Proxmox.RuntimeStatus; got != "running" {
|
||||||
|
t.Fatalf("Proxmox runtime status = %q, want running while collection is stale", got)
|
||||||
|
}
|
||||||
|
if got := defaultResources[0].SourceStatus[SourceProxmox].Status; got != "stale" {
|
||||||
|
t.Fatalf("Proxmox source status = %q, want stale", got)
|
||||||
|
}
|
||||||
|
|
||||||
adapter := NewMonitorAdapterWithStaleThresholds(NewRegistry(nil), map[DataSource]time.Duration{
|
adapter := NewMonitorAdapterWithStaleThresholds(NewRegistry(nil), map[DataSource]time.Duration{
|
||||||
SourceProxmox: 10 * time.Minute,
|
SourceProxmox: 10 * time.Minute,
|
||||||
|
|||||||
@@ -1676,6 +1676,7 @@ func resourceFromVM(vm models.VM) (Resource, ResourceIdentity) {
|
|||||||
metrics := metricsFromVM(vm)
|
metrics := metricsFromVM(vm)
|
||||||
proxmox := &ProxmoxData{
|
proxmox := &ProxmoxData{
|
||||||
SourceID: sourceID,
|
SourceID: sourceID,
|
||||||
|
RuntimeStatus: vm.Status,
|
||||||
NodeName: vm.Node,
|
NodeName: vm.Node,
|
||||||
Pool: vm.Pool,
|
Pool: vm.Pool,
|
||||||
Instance: vm.Instance,
|
Instance: vm.Instance,
|
||||||
@@ -1742,6 +1743,7 @@ func resourceFromContainer(ct models.Container) (Resource, ResourceIdentity) {
|
|||||||
metrics := metricsFromContainer(ct)
|
metrics := metricsFromContainer(ct)
|
||||||
proxmox := &ProxmoxData{
|
proxmox := &ProxmoxData{
|
||||||
SourceID: sourceID,
|
SourceID: sourceID,
|
||||||
|
RuntimeStatus: ct.Status,
|
||||||
NodeName: ct.Node,
|
NodeName: ct.Node,
|
||||||
Pool: ct.Pool,
|
Pool: ct.Pool,
|
||||||
Instance: ct.Instance,
|
Instance: ct.Instance,
|
||||||
|
|||||||
@@ -22,6 +22,23 @@ func TestCanonicalResourceTypeDoesNotAliasHost(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestProxmoxRuntimeStatusJSONContract(t *testing.T) {
|
||||||
|
payload := ProxmoxData{
|
||||||
|
SourceID: "lab:node-a:101",
|
||||||
|
RuntimeStatus: "running",
|
||||||
|
NodeName: "node-a",
|
||||||
|
VMID: 101,
|
||||||
|
}
|
||||||
|
|
||||||
|
data, err := json.Marshal(payload)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("marshal ProxmoxData: %v", err)
|
||||||
|
}
|
||||||
|
if !strings.Contains(string(data), `"runtimeStatus":"running"`) {
|
||||||
|
t.Fatalf("ProxmoxData JSON did not carry runtimeStatus: %s", data)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func TestHostSMARTMetaCarriesSizeBytesJSONContract(t *testing.T) {
|
func TestHostSMARTMetaCarriesSizeBytesJSONContract(t *testing.T) {
|
||||||
payload := HostSMARTMeta{
|
payload := HostSMARTMeta{
|
||||||
Device: "/dev/sda",
|
Device: "/dev/sda",
|
||||||
|
|||||||
@@ -14,6 +14,11 @@ import (
|
|||||||
type MonitorAdapter struct {
|
type MonitorAdapter struct {
|
||||||
registry *ResourceRegistry
|
registry *ResourceRegistry
|
||||||
|
|
||||||
|
// mutationMu serializes complete registry generations with incremental
|
||||||
|
// supplemental updates. Registry construction intentionally happens while
|
||||||
|
// readers keep using the prior pointer, but two writers must never publish
|
||||||
|
// out of order or mutate a registry after it has been replaced.
|
||||||
|
mutationMu sync.Mutex
|
||||||
mu sync.RWMutex
|
mu sync.RWMutex
|
||||||
activeAlerts []models.Alert
|
activeAlerts []models.Alert
|
||||||
lastRebuiltAt time.Time
|
lastRebuiltAt time.Time
|
||||||
@@ -142,6 +147,12 @@ func (a *MonitorAdapter) ResolveCanonicalResourceID(ref string) (string, bool) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (a *MonitorAdapter) replaceRegistry(snapshot models.StateSnapshot, recordsBySource map[DataSource][]IngestRecord) {
|
func (a *MonitorAdapter) replaceRegistry(snapshot models.StateSnapshot, recordsBySource map[DataSource][]IngestRecord) {
|
||||||
|
if a == nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
a.mutationMu.Lock()
|
||||||
|
defer a.mutationMu.Unlock()
|
||||||
|
|
||||||
registry := a.currentRegistry()
|
registry := a.currentRegistry()
|
||||||
if registry == nil {
|
if registry == nil {
|
||||||
return
|
return
|
||||||
@@ -360,6 +371,12 @@ func (a *MonitorAdapter) PopulateSnapshotAndSupplemental(snapshot models.StateSn
|
|||||||
// PopulateSupplementalRecords ingests source-native records emitted outside the
|
// PopulateSupplementalRecords ingests source-native records emitted outside the
|
||||||
// legacy state snapshot pipeline.
|
// legacy state snapshot pipeline.
|
||||||
func (a *MonitorAdapter) PopulateSupplementalRecords(source DataSource, records []IngestRecord) {
|
func (a *MonitorAdapter) PopulateSupplementalRecords(source DataSource, records []IngestRecord) {
|
||||||
|
if a == nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
a.mutationMu.Lock()
|
||||||
|
defer a.mutationMu.Unlock()
|
||||||
|
|
||||||
registry := a.currentRegistry()
|
registry := a.currentRegistry()
|
||||||
if registry == nil || len(records) == 0 || strings.TrimSpace(string(source)) == "" {
|
if registry == nil || len(records) == 0 || strings.TrimSpace(string(source)) == "" {
|
||||||
return
|
return
|
||||||
|
|||||||
@@ -0,0 +1,101 @@
|
|||||||
|
package unifiedresources
|
||||||
|
|
||||||
|
import (
|
||||||
|
"sync"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/rcourtman/pulse-go-rewrite/internal/models"
|
||||||
|
)
|
||||||
|
|
||||||
|
type blockingChangeStore struct {
|
||||||
|
*MemoryStore
|
||||||
|
entered chan struct{}
|
||||||
|
release chan struct{}
|
||||||
|
once sync.Once
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *blockingChangeStore) RecordChange(change ResourceChange) error {
|
||||||
|
s.once.Do(func() {
|
||||||
|
close(s.entered)
|
||||||
|
<-s.release
|
||||||
|
})
|
||||||
|
return s.MemoryStore.RecordChange(change)
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestMonitorAdapterSerializesSupplementalMutationAfterSnapshotPublication(t *testing.T) {
|
||||||
|
store := &blockingChangeStore{
|
||||||
|
MemoryStore: NewMemoryStore(),
|
||||||
|
entered: make(chan struct{}),
|
||||||
|
release: make(chan struct{}),
|
||||||
|
}
|
||||||
|
adapter := NewMonitorAdapter(NewRegistry(store))
|
||||||
|
|
||||||
|
rebuildDone := make(chan struct{})
|
||||||
|
go func() {
|
||||||
|
adapter.PopulateFromSnapshot(models.StateSnapshot{
|
||||||
|
LastUpdate: time.Now().UTC(),
|
||||||
|
VMs: []models.VM{{
|
||||||
|
ID: "lab:node-a:101",
|
||||||
|
VMID: 101,
|
||||||
|
Name: "database",
|
||||||
|
Node: "node-a",
|
||||||
|
Instance: "lab",
|
||||||
|
Status: "running",
|
||||||
|
LastSeen: time.Now().UTC(),
|
||||||
|
}},
|
||||||
|
})
|
||||||
|
close(rebuildDone)
|
||||||
|
}()
|
||||||
|
|
||||||
|
select {
|
||||||
|
case <-store.entered:
|
||||||
|
case <-time.After(time.Second):
|
||||||
|
t.Fatal("snapshot rebuild did not reach change publication")
|
||||||
|
}
|
||||||
|
|
||||||
|
supplementalDone := make(chan struct{})
|
||||||
|
go func() {
|
||||||
|
adapter.PopulateSupplementalRecords(SourceAgent, []IngestRecord{{
|
||||||
|
SourceID: "host-supplemental",
|
||||||
|
Resource: Resource{
|
||||||
|
Type: ResourceTypeAgent,
|
||||||
|
Name: "host-supplemental",
|
||||||
|
Status: StatusOnline,
|
||||||
|
LastSeen: time.Now().UTC(),
|
||||||
|
},
|
||||||
|
}})
|
||||||
|
close(supplementalDone)
|
||||||
|
}()
|
||||||
|
|
||||||
|
select {
|
||||||
|
case <-supplementalDone:
|
||||||
|
t.Fatal("supplemental mutation bypassed the in-flight snapshot generation")
|
||||||
|
case <-time.After(20 * time.Millisecond):
|
||||||
|
}
|
||||||
|
|
||||||
|
close(store.release)
|
||||||
|
select {
|
||||||
|
case <-rebuildDone:
|
||||||
|
case <-time.After(time.Second):
|
||||||
|
t.Fatal("snapshot rebuild did not complete")
|
||||||
|
}
|
||||||
|
select {
|
||||||
|
case <-supplementalDone:
|
||||||
|
case <-time.After(time.Second):
|
||||||
|
t.Fatal("supplemental mutation did not resume")
|
||||||
|
}
|
||||||
|
|
||||||
|
resources := adapter.GetAll()
|
||||||
|
if len(resources) != 2 {
|
||||||
|
t.Fatalf("final generation contains %d resources, want snapshot plus supplemental record: %+v", len(resources), resources)
|
||||||
|
}
|
||||||
|
var foundVM, foundSupplemental bool
|
||||||
|
for _, resource := range resources {
|
||||||
|
foundVM = foundVM || resource.Name == "database"
|
||||||
|
foundSupplemental = foundSupplemental || resource.Name == "host-supplemental"
|
||||||
|
}
|
||||||
|
if !foundVM || !foundSupplemental {
|
||||||
|
t.Fatalf("final generation lost a writer: %+v", resources)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -1,6 +1,7 @@
|
|||||||
package unifiedresources
|
package unifiedresources
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"fmt"
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
@@ -525,6 +526,89 @@ func TestMonitorAdapterIngestsAvailabilityAfterCorrelatableSupplementalSources(t
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestMonitorAdapterStalenessDoesNotEmitRemovalButAuthoritativeOmissionDoes(t *testing.T) {
|
||||||
|
store := NewMemoryStore()
|
||||||
|
adapter := NewMonitorAdapter(NewRegistry(store))
|
||||||
|
now := time.Now().UTC()
|
||||||
|
container := func(vmid int, name string, seen time.Time) models.Container {
|
||||||
|
return models.Container{
|
||||||
|
ID: fmt.Sprintf("lab:node-a:%d", vmid),
|
||||||
|
VMID: vmid,
|
||||||
|
Name: name,
|
||||||
|
Node: "node-a",
|
||||||
|
Instance: "lab",
|
||||||
|
Status: "running",
|
||||||
|
Type: "lxc",
|
||||||
|
LastSeen: seen,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
adapter.PopulateFromSnapshot(models.StateSnapshot{
|
||||||
|
LastUpdate: now,
|
||||||
|
Containers: []models.Container{
|
||||||
|
container(101, "alpha", now),
|
||||||
|
container(102, "beta", now),
|
||||||
|
},
|
||||||
|
})
|
||||||
|
initial := adapter.GetByType(ResourceTypeSystemContainer)
|
||||||
|
if len(initial) != 2 {
|
||||||
|
t.Fatalf("initial container count = %d, want 2", len(initial))
|
||||||
|
}
|
||||||
|
var removedID string
|
||||||
|
for _, resource := range initial {
|
||||||
|
if resource.Name == "beta" {
|
||||||
|
removedID = resource.ID
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if removedID == "" {
|
||||||
|
t.Fatal("beta canonical ID not found")
|
||||||
|
}
|
||||||
|
|
||||||
|
staleSeen := now.Add(-5 * time.Minute)
|
||||||
|
adapter.PopulateFromSnapshot(models.StateSnapshot{
|
||||||
|
LastUpdate: now.Add(time.Second),
|
||||||
|
Containers: []models.Container{
|
||||||
|
container(101, "alpha", staleSeen),
|
||||||
|
container(102, "beta", staleSeen),
|
||||||
|
},
|
||||||
|
})
|
||||||
|
if got := len(adapter.GetByType(ResourceTypeSystemContainer)); got != 2 {
|
||||||
|
t.Fatalf("stale refresh container count = %d, want 2", got)
|
||||||
|
}
|
||||||
|
if changes, err := store.GetRecentChanges(removedID, time.Time{}, 20); err != nil {
|
||||||
|
t.Fatalf("GetRecentChanges before deletion: %v", err)
|
||||||
|
} else {
|
||||||
|
for _, change := range changes {
|
||||||
|
if change.Metadata["changeType"] == "resource_removed" {
|
||||||
|
t.Fatalf("staleness emitted a removal: %+v", change)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
adapter.PopulateFromSnapshot(models.StateSnapshot{
|
||||||
|
LastUpdate: now.Add(2 * time.Second),
|
||||||
|
Containers: []models.Container{
|
||||||
|
container(101, "alpha", now.Add(2*time.Second)),
|
||||||
|
},
|
||||||
|
})
|
||||||
|
if got := len(adapter.GetByType(ResourceTypeSystemContainer)); got != 1 {
|
||||||
|
t.Fatalf("authoritative deletion container count = %d, want 1", got)
|
||||||
|
}
|
||||||
|
changes, err := store.GetRecentChanges(removedID, time.Time{}, 20)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("GetRecentChanges after deletion: %v", err)
|
||||||
|
}
|
||||||
|
removals := 0
|
||||||
|
for _, change := range changes {
|
||||||
|
if change.Metadata["changeType"] == "resource_removed" {
|
||||||
|
removals++
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if removals != 1 {
|
||||||
|
t.Fatalf("resource removal history count = %d, want 1: %+v", removals, changes)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func TestMonitorAdapterRecordChangeForwardsToStore(t *testing.T) {
|
func TestMonitorAdapterRecordChangeForwardsToStore(t *testing.T) {
|
||||||
store := NewMemoryStore()
|
store := NewMemoryStore()
|
||||||
adapter := NewMonitorAdapter(NewRegistry(store))
|
adapter := NewMonitorAdapter(NewRegistry(store))
|
||||||
|
|||||||
@@ -3372,6 +3372,9 @@ func mergeProxmoxData(existing *ProxmoxData, incoming *ProxmoxData) *ProxmoxData
|
|||||||
if incoming.HostURL != "" {
|
if incoming.HostURL != "" {
|
||||||
merged.HostURL = incoming.HostURL
|
merged.HostURL = incoming.HostURL
|
||||||
}
|
}
|
||||||
|
if incoming.RuntimeStatus != "" {
|
||||||
|
merged.RuntimeStatus = incoming.RuntimeStatus
|
||||||
|
}
|
||||||
if incoming.VMID != 0 {
|
if incoming.VMID != 0 {
|
||||||
merged.VMID = incoming.VMID
|
merged.VMID = incoming.VMID
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -5306,6 +5306,71 @@ func TestMarkStaleRecomputesFromRemainingFreshSources(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestMarkStaleKeepsProxmoxRuntimeStateIndependentOfAvailabilityFacet(t *testing.T) {
|
||||||
|
rr := NewRegistry(nil)
|
||||||
|
staleSeen := time.Now().UTC().Add(-5 * time.Minute)
|
||||||
|
freshSeen := time.Now().UTC()
|
||||||
|
|
||||||
|
for _, sourceID := range []string{"lab:node-a:101", "lab:node-a:102"} {
|
||||||
|
rr.IngestRecords(SourceProxmox, []IngestRecord{{
|
||||||
|
SourceID: sourceID,
|
||||||
|
Resource: Resource{
|
||||||
|
Type: ResourceTypeSystemContainer,
|
||||||
|
Name: sourceID,
|
||||||
|
Status: StatusOnline,
|
||||||
|
LastSeen: staleSeen,
|
||||||
|
Proxmox: &ProxmoxData{RuntimeStatus: "running", NodeName: "node-a"},
|
||||||
|
},
|
||||||
|
}})
|
||||||
|
}
|
||||||
|
|
||||||
|
resources := rr.ListByType(ResourceTypeSystemContainer)
|
||||||
|
if len(resources) != 2 {
|
||||||
|
t.Fatalf("expected two Proxmox containers, got %d", len(resources))
|
||||||
|
}
|
||||||
|
var checkedID string
|
||||||
|
for _, resource := range resources {
|
||||||
|
if resource.Name == "lab:node-a:102" {
|
||||||
|
checkedID = resource.ID
|
||||||
|
break
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if checkedID == "" {
|
||||||
|
t.Fatal("availability target container not found")
|
||||||
|
}
|
||||||
|
|
||||||
|
rr.IngestRecords(SourceAvailability, []IngestRecord{{
|
||||||
|
SourceID: "probe-102",
|
||||||
|
Resource: Resource{
|
||||||
|
Type: ResourceTypeNetworkEndpoint,
|
||||||
|
Name: "probe-102",
|
||||||
|
Status: StatusOnline,
|
||||||
|
LastSeen: freshSeen,
|
||||||
|
Availability: &AvailabilityData{
|
||||||
|
TargetID: "probe-102", LinkedResourceID: checkedID,
|
||||||
|
Address: "192.0.2.102", Protocol: "icmp", Enabled: true, Available: true,
|
||||||
|
},
|
||||||
|
},
|
||||||
|
}})
|
||||||
|
|
||||||
|
rr.MarkStale(freshSeen, nil)
|
||||||
|
|
||||||
|
resources = rr.ListByType(ResourceTypeSystemContainer)
|
||||||
|
statuses := make(map[string]ResourceStatus, len(resources))
|
||||||
|
for _, resource := range resources {
|
||||||
|
statuses[resource.Name] = resource.Status
|
||||||
|
if resource.Proxmox == nil || resource.Proxmox.RuntimeStatus != "running" {
|
||||||
|
t.Fatalf("runtime status changed with source freshness for %s: %+v", resource.Name, resource.Proxmox)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if got := statuses["lab:node-a:101"]; got != StatusWarning {
|
||||||
|
t.Fatalf("unfaceted stale container status = %q, want warning", got)
|
||||||
|
}
|
||||||
|
if got := statuses["lab:node-a:102"]; got != StatusOnline {
|
||||||
|
t.Fatalf("availability-faceted stale container status = %q, want online", got)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func TestResourceRegistryUsesConfiguredProxmoxStaleThresholds(t *testing.T) {
|
func TestResourceRegistryUsesConfiguredProxmoxStaleThresholds(t *testing.T) {
|
||||||
seen := time.Now().UTC().Add(-90 * time.Second).Truncate(time.Millisecond)
|
seen := time.Now().UTC().Add(-90 * time.Second).Truncate(time.Millisecond)
|
||||||
snapshot := models.StateSnapshot{
|
snapshot := models.StateSnapshot{
|
||||||
|
|||||||
@@ -331,7 +331,8 @@ type MetricValue struct {
|
|||||||
|
|
||||||
// ProxmoxData contains Proxmox-specific data for a resource.
|
// ProxmoxData contains Proxmox-specific data for a resource.
|
||||||
type ProxmoxData struct {
|
type ProxmoxData struct {
|
||||||
SourceID string `json:"sourceId,omitempty"` // raw model ID from source snapshot
|
SourceID string `json:"sourceId,omitempty"` // raw model ID from source snapshot
|
||||||
|
RuntimeStatus string `json:"runtimeStatus,omitempty"` // source-authored VM/LXC power state, independent of collection freshness
|
||||||
NodeName string `json:"nodeName,omitempty"`
|
NodeName string `json:"nodeName,omitempty"`
|
||||||
Pool string `json:"pool,omitempty"`
|
Pool string `json:"pool,omitempty"`
|
||||||
ClusterName string `json:"clusterName,omitempty"`
|
ClusterName string `json:"clusterName,omitempty"`
|
||||||
|
|||||||
@@ -97,6 +97,13 @@ func (v VMView) Status() ResourceStatus {
|
|||||||
return v.r.Status
|
return v.r.Status
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (v VMView) RuntimeStatus() string {
|
||||||
|
if v.r == nil || v.r.Proxmox == nil {
|
||||||
|
return ""
|
||||||
|
}
|
||||||
|
return strings.TrimSpace(v.r.Proxmox.RuntimeStatus)
|
||||||
|
}
|
||||||
|
|
||||||
func (v VMView) VMID() int {
|
func (v VMView) VMID() int {
|
||||||
if v.r == nil || v.r.Proxmox == nil {
|
if v.r == nil || v.r.Proxmox == nil {
|
||||||
return 0
|
return 0
|
||||||
@@ -375,6 +382,13 @@ func (v ContainerView) Status() ResourceStatus {
|
|||||||
return v.r.Status
|
return v.r.Status
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (v ContainerView) RuntimeStatus() string {
|
||||||
|
if v.r == nil || v.r.Proxmox == nil {
|
||||||
|
return ""
|
||||||
|
}
|
||||||
|
return strings.TrimSpace(v.r.Proxmox.RuntimeStatus)
|
||||||
|
}
|
||||||
|
|
||||||
func (v ContainerView) VMID() int {
|
func (v ContainerView) VMID() int {
|
||||||
if v.r == nil || v.r.Proxmox == nil {
|
if v.r == nil || v.r.Proxmox == nil {
|
||||||
return 0
|
return 0
|
||||||
|
|||||||
@@ -1,7 +1,12 @@
|
|||||||
import fs from "node:fs";
|
import fs from "node:fs";
|
||||||
import path from "node:path";
|
import path from "node:path";
|
||||||
import { fileURLToPath } from "node:url";
|
import { fileURLToPath } from "node:url";
|
||||||
import { expect, test as base, type Locator, type Page } from "@playwright/test";
|
import {
|
||||||
|
expect,
|
||||||
|
test as base,
|
||||||
|
type Locator,
|
||||||
|
type Page,
|
||||||
|
} from "@playwright/test";
|
||||||
|
|
||||||
import {
|
import {
|
||||||
createAuthenticatedStorageState,
|
createAuthenticatedStorageState,
|
||||||
@@ -113,6 +118,49 @@ async function readGuestDrawerActiveTab(detailRow: Locator): Promise<string> {
|
|||||||
return active;
|
return active;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
function issue1611Container(
|
||||||
|
vmid: number,
|
||||||
|
name: string,
|
||||||
|
status: "online" | "warning",
|
||||||
|
availability = false,
|
||||||
|
) {
|
||||||
|
return {
|
||||||
|
id: `lab-node-a-${vmid}`,
|
||||||
|
type: "system-container",
|
||||||
|
name,
|
||||||
|
status,
|
||||||
|
lastSeen: "2026-07-24T08:00:00Z",
|
||||||
|
vmid,
|
||||||
|
node: "node-a",
|
||||||
|
instance: "lab",
|
||||||
|
sources: ["proxmox", ...(availability ? ["availability"] : [])],
|
||||||
|
platformScopes: ["proxmox-pve"],
|
||||||
|
metrics: {
|
||||||
|
cpu: { percent: 0.12 },
|
||||||
|
memory: { used: 1024, total: 4096, percent: 25 },
|
||||||
|
disk: { used: 2048, total: 8192, percent: 25 },
|
||||||
|
},
|
||||||
|
proxmox: {
|
||||||
|
runtimeStatus: "running",
|
||||||
|
nodeName: "node-a",
|
||||||
|
instance: "lab",
|
||||||
|
vmid,
|
||||||
|
cpus: 2,
|
||||||
|
uptime: 3600,
|
||||||
|
},
|
||||||
|
...(availability
|
||||||
|
? {
|
||||||
|
availability: {
|
||||||
|
targetId: `probe-${vmid}`,
|
||||||
|
protocol: "icmp",
|
||||||
|
enabled: true,
|
||||||
|
available: true,
|
||||||
|
},
|
||||||
|
}
|
||||||
|
: {}),
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
test.describe.serial("Workloads Proxmox refresh stability", () => {
|
test.describe.serial("Workloads Proxmox refresh stability", () => {
|
||||||
test.setTimeout(180_000);
|
test.setTimeout(180_000);
|
||||||
|
|
||||||
@@ -172,7 +220,9 @@ test.describe.serial("Workloads Proxmox refresh stability", () => {
|
|||||||
await expect(discoveryButton).toBeVisible();
|
await expect(discoveryButton).toBeVisible();
|
||||||
await discoveryButton.click();
|
await discoveryButton.click();
|
||||||
|
|
||||||
await expect.poll(() => readGuestDrawerActiveTab(detailRow)).toBe("discovery");
|
await expect
|
||||||
|
.poll(() => readGuestDrawerActiveTab(detailRow))
|
||||||
|
.toBe("discovery");
|
||||||
|
|
||||||
const beforePollScrollTop = await readPrimaryViewportScrollTop(page);
|
const beforePollScrollTop = await readPrimaryViewportScrollTop(page);
|
||||||
|
|
||||||
@@ -188,4 +238,95 @@ test.describe.serial("Workloads Proxmox refresh stability", () => {
|
|||||||
Math.max(10, beforePollScrollTop - 80),
|
Math.max(10, beforePollScrollTop - 80),
|
||||||
);
|
);
|
||||||
});
|
});
|
||||||
|
|
||||||
|
test("retains running LXC rows through stale availability projections and removes a confirmed deletion", async ({
|
||||||
|
page,
|
||||||
|
}, testInfo) => {
|
||||||
|
test.skip(
|
||||||
|
testInfo.project.name.startsWith("mobile-"),
|
||||||
|
"Desktop-only workload refresh proof",
|
||||||
|
);
|
||||||
|
|
||||||
|
await ensureMockModeEnabled(page);
|
||||||
|
|
||||||
|
let workloadRequests = 0;
|
||||||
|
let staleProjectionResponses = 0;
|
||||||
|
let deletionResponses = 0;
|
||||||
|
let publishDeletion = false;
|
||||||
|
await page.route("**/api/resources?**", async (route) => {
|
||||||
|
const url = new URL(route.request().url());
|
||||||
|
if (
|
||||||
|
url.pathname !== "/api/resources" ||
|
||||||
|
url.searchParams.get("type") !== "vm,system-container,app-container,pod"
|
||||||
|
) {
|
||||||
|
await route.continue();
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
workloadRequests += 1;
|
||||||
|
let data;
|
||||||
|
if (workloadRequests === 1) {
|
||||||
|
data = [
|
||||||
|
issue1611Container(101, "lxc-alpha", "online"),
|
||||||
|
issue1611Container(102, "lxc-beta", "online", true),
|
||||||
|
issue1611Container(103, "lxc-gamma", "online"),
|
||||||
|
];
|
||||||
|
} else if (!publishDeletion) {
|
||||||
|
staleProjectionResponses += 1;
|
||||||
|
data = [
|
||||||
|
issue1611Container(101, "lxc-alpha", "warning"),
|
||||||
|
issue1611Container(102, "lxc-beta", "online", true),
|
||||||
|
issue1611Container(103, "lxc-gamma", "warning"),
|
||||||
|
];
|
||||||
|
} else {
|
||||||
|
deletionResponses += 1;
|
||||||
|
data = [
|
||||||
|
issue1611Container(101, "lxc-alpha", "online"),
|
||||||
|
issue1611Container(102, "lxc-beta", "online", true),
|
||||||
|
];
|
||||||
|
}
|
||||||
|
|
||||||
|
await route.fulfill({
|
||||||
|
status: 200,
|
||||||
|
contentType: "application/json",
|
||||||
|
body: JSON.stringify({
|
||||||
|
data,
|
||||||
|
meta: { page: 1, limit: 200, total: data.length, totalPages: 1 },
|
||||||
|
}),
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
|
await page.goto(
|
||||||
|
"/proxmox/workloads?type=system-container&platform=proxmox-pve&status=running",
|
||||||
|
{ waitUntil: "domcontentloaded" },
|
||||||
|
);
|
||||||
|
|
||||||
|
const rows = page.locator("tr[data-guest-id]");
|
||||||
|
await expect(rows).toHaveCount(3, { timeout: 60_000 });
|
||||||
|
await expect(rows.first()).toBeVisible();
|
||||||
|
|
||||||
|
await page.locator("th").filter({ hasText: "Name" }).last().click();
|
||||||
|
const retainedRow = rows.filter({ hasText: "lxc-beta" });
|
||||||
|
await retainedRow.click();
|
||||||
|
const detailRow = page.locator(
|
||||||
|
'tr[data-inline-detail-for="lab:node-a:102"]',
|
||||||
|
);
|
||||||
|
await expect(detailRow).toBeVisible();
|
||||||
|
|
||||||
|
await expect
|
||||||
|
.poll(() => staleProjectionResponses, { timeout: 15_000 })
|
||||||
|
.toBeGreaterThan(0);
|
||||||
|
await expect(rows).toHaveCount(3);
|
||||||
|
await expect(rows.filter({ hasText: "lxc-alpha" })).toBeVisible();
|
||||||
|
await expect(rows.filter({ hasText: "lxc-gamma" })).toBeVisible();
|
||||||
|
await expect(detailRow).toBeVisible();
|
||||||
|
|
||||||
|
publishDeletion = true;
|
||||||
|
await expect
|
||||||
|
.poll(() => deletionResponses, { timeout: 15_000 })
|
||||||
|
.toBeGreaterThan(0);
|
||||||
|
await expect(rows).toHaveCount(2);
|
||||||
|
await expect(rows.filter({ hasText: "lxc-gamma" })).toHaveCount(0);
|
||||||
|
await expect(detailRow).toBeVisible();
|
||||||
|
});
|
||||||
});
|
});
|
||||||
|
|||||||
Reference in New Issue
Block a user