mirror of
https://github.com/rcourtman/Pulse.git
synced 2026-08-04 21:33:23 +00:00
Forward-port Proxmox monitoring fixes from v5
This commit is contained in:
parent
1a0770f050
commit
c5f5af7abf
21 changed files with 989 additions and 99 deletions
|
|
@ -128,6 +128,8 @@ func buildCephClusterModel(instanceName string, status *proxmox.CephStatus, df *
|
|||
}
|
||||
|
||||
healthMsg := summarizeCephHealth(status)
|
||||
numMons := countCephMonitorDaemons(status)
|
||||
numMgrs := countCephManagerDaemons(status)
|
||||
|
||||
cluster := models.CephCluster{
|
||||
ID: clusterID,
|
||||
|
|
@ -140,8 +142,8 @@ func buildCephClusterModel(instanceName string, status *proxmox.CephStatus, df *
|
|||
UsedBytes: usedBytes,
|
||||
AvailableBytes: availBytes,
|
||||
UsagePercent: usagePercent,
|
||||
NumMons: countServiceDaemons(status.ServiceMap.Services, "mon"),
|
||||
NumMgrs: countServiceDaemons(status.ServiceMap.Services, "mgr"),
|
||||
NumMons: numMons,
|
||||
NumMgrs: numMgrs,
|
||||
NumOSDs: status.OSDMap.NumOSDs,
|
||||
NumOSDsUp: status.OSDMap.NumUpOSDs,
|
||||
NumOSDsIn: status.OSDMap.NumInOSDs,
|
||||
|
|
@ -154,6 +156,29 @@ func buildCephClusterModel(instanceName string, status *proxmox.CephStatus, df *
|
|||
return cluster
|
||||
}
|
||||
|
||||
func countCephMonitorDaemons(status *proxmox.CephStatus) int {
|
||||
if status == nil {
|
||||
return 0
|
||||
}
|
||||
if status.MonMap.NumMons > 0 {
|
||||
return status.MonMap.NumMons
|
||||
}
|
||||
return countServiceDaemons(status.ServiceMap.Services, "mon")
|
||||
}
|
||||
|
||||
func countCephManagerDaemons(status *proxmox.CephStatus) int {
|
||||
if status == nil {
|
||||
return 0
|
||||
}
|
||||
if status.MgrMap.NumMgrs > 0 {
|
||||
return status.MgrMap.NumMgrs
|
||||
}
|
||||
if status.MgrMap.ActiveName != "" {
|
||||
return 1 + len(status.MgrMap.Standbys)
|
||||
}
|
||||
return countServiceDaemons(status.ServiceMap.Services, "mgr")
|
||||
}
|
||||
|
||||
// summarizeCephHealth extracts human-readable messages from the Ceph health payload.
|
||||
func summarizeCephHealth(status *proxmox.CephStatus) string {
|
||||
if status == nil {
|
||||
|
|
|
|||
|
|
@ -263,6 +263,40 @@ func TestCountServiceDaemons(t *testing.T) {
|
|||
}
|
||||
}
|
||||
|
||||
func TestCountCephMonitorDaemons(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
status := &proxmox.CephStatus{
|
||||
MonMap: proxmox.CephMonMap{NumMons: 3},
|
||||
ServiceMap: proxmox.CephServiceMap{
|
||||
Services: map[string]proxmox.CephServiceDefinition{
|
||||
"mon": {
|
||||
Daemons: map[string]proxmox.CephServiceDaemon{
|
||||
"a": {Host: "node1", Status: "running"},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
if got := countCephMonitorDaemons(status); got != 3 {
|
||||
t.Fatalf("countCephMonitorDaemons() = %d, want 3", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCountCephManagerDaemons(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
status := &proxmox.CephStatus{
|
||||
MgrMap: proxmox.CephMgrMap{
|
||||
ActiveName: "mgr-a",
|
||||
Standbys: []string{"mgr-b", "mgr-c"},
|
||||
},
|
||||
}
|
||||
if got := countCephManagerDaemons(status); got != 3 {
|
||||
t.Fatalf("countCephManagerDaemons() = %d, want 3", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestExtractCephCheckSummary(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
|
|
|
|||
30
internal/monitoring/guest_host_agent_fallback.go
Normal file
30
internal/monitoring/guest_host_agent_fallback.go
Normal file
|
|
@ -0,0 +1,30 @@
|
|||
package monitoring
|
||||
|
||||
import "github.com/rcourtman/pulse-go-rewrite/internal/models"
|
||||
|
||||
func resolveGuestDiskFromLinkedHostAgent(guestID string, vmIDToHostAgent map[string]models.Host) (models.Disk, []models.Disk, bool) {
|
||||
if guestID == "" || len(vmIDToHostAgent) == 0 {
|
||||
return models.Disk{}, nil, false
|
||||
}
|
||||
|
||||
host, ok := vmIDToHostAgent[guestID]
|
||||
if !ok {
|
||||
return models.Disk{}, nil, false
|
||||
}
|
||||
|
||||
summary, ok := models.SummaryDisk(host.Disks)
|
||||
if !ok {
|
||||
return models.Disk{}, nil, false
|
||||
}
|
||||
|
||||
disks := append([]models.Disk(nil), host.Disks...)
|
||||
return models.Disk{
|
||||
Total: summary.Total,
|
||||
Used: summary.Used,
|
||||
Free: summary.Free,
|
||||
Usage: summary.Usage,
|
||||
Mountpoint: summary.Mountpoint,
|
||||
Type: summary.Type,
|
||||
Device: summary.Device,
|
||||
}, disks, true
|
||||
}
|
||||
|
|
@ -377,6 +377,7 @@ func processGuestNetworkInterfaces(raw []proxmox.VMNetworkInterface) ([]string,
|
|||
for _, iface := range raw {
|
||||
ifaceName := strings.TrimSpace(iface.Name)
|
||||
mac := strings.TrimSpace(iface.HardwareAddr)
|
||||
hadRawAddresses := len(iface.IPAddresses) > 0
|
||||
|
||||
addrSet := make(map[string]struct{})
|
||||
addresses := make([]string, 0, len(iface.IPAddresses))
|
||||
|
|
@ -410,7 +411,18 @@ func processGuestNetworkInterfaces(raw []proxmox.VMNetworkInterface) ([]string,
|
|||
txBytes := parseInterfaceStat(iface.Statistics, "tx-bytes")
|
||||
|
||||
if len(addresses) == 0 && rxBytes == 0 && txBytes == 0 {
|
||||
continue
|
||||
lowerName := strings.ToLower(ifaceName)
|
||||
if lowerName == "lo" || lowerName == "loopback" {
|
||||
continue
|
||||
}
|
||||
if ifaceName == "" && mac == "" {
|
||||
continue
|
||||
}
|
||||
if hadRawAddresses {
|
||||
// Preserve the interface identity even when every reported address
|
||||
// was filtered out. Early guest-agent payloads often surface only
|
||||
// link-local addresses first, and hiding the NIC entirely is worse.
|
||||
}
|
||||
}
|
||||
|
||||
guestIfaces = append(guestIfaces, models.GuestNetworkInterface{
|
||||
|
|
|
|||
|
|
@ -300,7 +300,7 @@ func TestProcessGuestNetworkInterfaces(t *testing.T) {
|
|||
wantIfaces: []models.GuestNetworkInterface{},
|
||||
},
|
||||
{
|
||||
name: "filter link-local fe80",
|
||||
name: "preserve named interface when only link-local addresses are reported",
|
||||
raw: []proxmox.VMNetworkInterface{
|
||||
{
|
||||
Name: "eth0",
|
||||
|
|
@ -311,8 +311,10 @@ func TestProcessGuestNetworkInterfaces(t *testing.T) {
|
|||
},
|
||||
},
|
||||
},
|
||||
wantIPs: []string{},
|
||||
wantIfaces: []models.GuestNetworkInterface{},
|
||||
wantIPs: []string{},
|
||||
wantIfaces: []models.GuestNetworkInterface{
|
||||
{Name: "eth0", MAC: "00:11:22:33:44:55", Addresses: nil},
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "filter IPv6 loopback ::1",
|
||||
|
|
@ -429,7 +431,7 @@ func TestProcessGuestNetworkInterfaces(t *testing.T) {
|
|||
},
|
||||
},
|
||||
{
|
||||
name: "interface with no IPs and no traffic is excluded",
|
||||
name: "named interface with no IPs and no traffic is preserved",
|
||||
raw: []proxmox.VMNetworkInterface{
|
||||
{
|
||||
Name: "eth0",
|
||||
|
|
@ -438,8 +440,10 @@ func TestProcessGuestNetworkInterfaces(t *testing.T) {
|
|||
Statistics: nil,
|
||||
},
|
||||
},
|
||||
wantIPs: []string{},
|
||||
wantIfaces: []models.GuestNetworkInterface{},
|
||||
wantIPs: []string{},
|
||||
wantIfaces: []models.GuestNetworkInterface{
|
||||
{Name: "eth0", MAC: "00:11:22:33:44:55", Addresses: nil},
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "whitespace trimmed from name and MAC",
|
||||
|
|
|
|||
|
|
@ -2,6 +2,7 @@ package monitoring
|
|||
|
||||
import (
|
||||
"context"
|
||||
"net"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
|
|
@ -276,3 +277,48 @@ func TestClusterEndpointEffectiveURL(t *testing.T) {
|
|||
t.Fatalf("empty endpoint = %q, want empty", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBuildClusterEndpointsForInit_RespectsDiscoveryPolicy(t *testing.T) {
|
||||
oldLookup := lookupIPFunc
|
||||
lookupIPFunc = func(host string) ([]net.IP, error) {
|
||||
switch host {
|
||||
case "allowed.local":
|
||||
return []net.IP{net.ParseIP("10.0.0.10")}, nil
|
||||
case "blocked.local":
|
||||
return []net.IP{net.ParseIP("192.168.1.10")}, nil
|
||||
default:
|
||||
return nil, nil
|
||||
}
|
||||
}
|
||||
t.Cleanup(func() {
|
||||
lookupIPFunc = oldLookup
|
||||
})
|
||||
|
||||
monitor := &Monitor{
|
||||
config: &config.Config{
|
||||
Discovery: config.DiscoveryConfig{
|
||||
SubnetAllowlist: []string{"10.0.0.0/8"},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
endpoints, _ := monitor.buildClusterEndpointsForInit(config.PVEInstance{
|
||||
Name: "cluster-a",
|
||||
Host: "https://main.local:8006",
|
||||
VerifySSL: true,
|
||||
ClusterEndpoints: []config.ClusterEndpoint{
|
||||
{NodeName: "node-a", Host: "allowed.local"},
|
||||
{NodeName: "node-b", Host: "blocked.local"},
|
||||
},
|
||||
})
|
||||
|
||||
if len(endpoints) != 2 {
|
||||
t.Fatalf("expected allowed endpoint plus main host fallback, got %#v", endpoints)
|
||||
}
|
||||
if endpoints[0] != "https://allowed.local:8006" {
|
||||
t.Fatalf("expected discovery-allowed endpoint first, got %#v", endpoints)
|
||||
}
|
||||
if endpoints[1] != "https://main.local:8006" {
|
||||
t.Fatalf("expected main host fallback retained, got %#v", endpoints)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -8,6 +8,8 @@ import (
|
|||
"github.com/rcourtman/pulse-go-rewrite/internal/config"
|
||||
)
|
||||
|
||||
var lookupIPFunc = net.LookupIP
|
||||
|
||||
func lookupClusterEndpointLabel(instance *config.PVEInstance, nodeName string) string {
|
||||
if instance == nil {
|
||||
return ""
|
||||
|
|
@ -134,3 +136,134 @@ func clusterEndpointEffectiveURL(endpoint config.ClusterEndpoint, verifySSL bool
|
|||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
func discoveryPolicyCIDRs(cidrs []string) []*net.IPNet {
|
||||
networks := make([]*net.IPNet, 0, len(cidrs))
|
||||
for _, cidr := range cidrs {
|
||||
cidr = strings.TrimSpace(cidr)
|
||||
if cidr == "" {
|
||||
continue
|
||||
}
|
||||
_, network, err := net.ParseCIDR(cidr)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
networks = append(networks, network)
|
||||
}
|
||||
return networks
|
||||
}
|
||||
|
||||
func discoveryPolicyBlockedIPs(ips []string) map[string]struct{} {
|
||||
blocked := make(map[string]struct{}, len(ips))
|
||||
for _, raw := range ips {
|
||||
ip := net.ParseIP(strings.TrimSpace(raw))
|
||||
if ip == nil {
|
||||
continue
|
||||
}
|
||||
blocked[ip.String()] = struct{}{}
|
||||
}
|
||||
return blocked
|
||||
}
|
||||
|
||||
func discoveryPolicyAllowsIP(ip net.IP, allowlist, blocklist []*net.IPNet, blockedIPs map[string]struct{}) bool {
|
||||
if ip == nil {
|
||||
return false
|
||||
}
|
||||
|
||||
if _, blocked := blockedIPs[ip.String()]; blocked {
|
||||
return false
|
||||
}
|
||||
|
||||
for _, network := range blocklist {
|
||||
if network.Contains(ip) {
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
if len(allowlist) == 0 {
|
||||
return true
|
||||
}
|
||||
|
||||
for _, network := range allowlist {
|
||||
if network.Contains(ip) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
||||
return false
|
||||
}
|
||||
|
||||
func discoveryPolicyIPsForEndpointHost(candidateURL string) []net.IP {
|
||||
if candidateURL == "" {
|
||||
return nil
|
||||
}
|
||||
|
||||
host := normalizeEndpointHost(candidateURL)
|
||||
if host == "" {
|
||||
return nil
|
||||
}
|
||||
|
||||
if ip := net.ParseIP(host); ip != nil {
|
||||
return []net.IP{ip}
|
||||
}
|
||||
|
||||
ips, err := lookupIPFunc(host)
|
||||
if err != nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
filtered := make([]net.IP, 0, len(ips))
|
||||
for _, ip := range ips {
|
||||
if ip == nil {
|
||||
continue
|
||||
}
|
||||
filtered = append(filtered, ip)
|
||||
}
|
||||
return filtered
|
||||
}
|
||||
|
||||
func clusterEndpointAllowedByDiscoveryPolicy(endpoint config.ClusterEndpoint, candidateURL string, discoveryCfg config.DiscoveryConfig) bool {
|
||||
if len(discoveryCfg.SubnetAllowlist) == 0 && len(discoveryCfg.SubnetBlocklist) == 0 && len(discoveryCfg.IPBlocklist) == 0 {
|
||||
return true
|
||||
}
|
||||
|
||||
allowlist := discoveryPolicyCIDRs(discoveryCfg.SubnetAllowlist)
|
||||
blocklist := discoveryPolicyCIDRs(discoveryCfg.SubnetBlocklist)
|
||||
blockedIPs := discoveryPolicyBlockedIPs(discoveryCfg.IPBlocklist)
|
||||
|
||||
resolvedIPs := discoveryPolicyIPsForEndpointHost(candidateURL)
|
||||
if len(resolvedIPs) == 0 {
|
||||
if ip := net.ParseIP(strings.TrimSpace(endpoint.EffectiveIP())); ip != nil {
|
||||
resolvedIPs = []net.IP{ip}
|
||||
}
|
||||
}
|
||||
if len(resolvedIPs) == 0 {
|
||||
return true
|
||||
}
|
||||
|
||||
for _, ip := range resolvedIPs {
|
||||
if !discoveryPolicyAllowsIP(ip, allowlist, blocklist, blockedIPs) {
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
return true
|
||||
}
|
||||
|
||||
func clusterEndpointRuntimeURL(endpoint config.ClusterEndpoint, verifySSL bool, hasFingerprint bool, discoveryCfg config.DiscoveryConfig) string {
|
||||
candidateURL := clusterEndpointEffectiveURL(endpoint, verifySSL, hasFingerprint)
|
||||
if candidateURL == "" {
|
||||
return ""
|
||||
}
|
||||
if !clusterEndpointAllowedByDiscoveryPolicy(endpoint, candidateURL, discoveryCfg) {
|
||||
return ""
|
||||
}
|
||||
return candidateURL
|
||||
}
|
||||
|
||||
func monitorDiscoveryConfig(m *Monitor) config.DiscoveryConfig {
|
||||
if m == nil || m.config == nil {
|
||||
return config.DiscoveryConfig{}
|
||||
}
|
||||
return m.config.Discovery
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1026,8 +1026,55 @@ func TestMonitor_PreviousGuestContextForInstance_Extra(t *testing.T) {
|
|||
if prev.containerOCIByVMID[112] || prev.containerOCIByVMID[211] {
|
||||
t.Fatalf("unexpected OCI classification leakage: %#v", prev.containerOCIByVMID)
|
||||
}
|
||||
if len(prev.hostAgentsByVMID) != 1 || prev.hostAgentsByVMID["vm-1"].LinkedVMID != "vm-1" {
|
||||
t.Fatalf("expected only online linked host with memory to be tracked, got %#v", prev.hostAgentsByVMID)
|
||||
if len(prev.hostAgentsByVMID) != 2 || prev.hostAgentsByVMID["vm-1"].LinkedVMID != "vm-1" || prev.hostAgentsByVMID["vm-3"].LinkedVMID != "vm-3" {
|
||||
t.Fatalf("expected all online linked hosts to be tracked for memory and disk fallback, got %#v", prev.hostAgentsByVMID)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBuildVMFromClusterResource_UsesLinkedHostAgentDiskFallback(t *testing.T) {
|
||||
monitor := &Monitor{rateTracker: NewRateTracker()}
|
||||
client := &mockPVEClientExtra{
|
||||
vmStatus: &proxmox.VMStatus{
|
||||
Status: "running",
|
||||
Agent: proxmox.VMAgentField{Value: 0},
|
||||
},
|
||||
}
|
||||
guestID := makeGuestID("cluster-a", "node-a", 101)
|
||||
|
||||
vm, _, _, _, ok := monitor.buildVMFromClusterResource(
|
||||
context.Background(),
|
||||
"cluster-a",
|
||||
proxmox.ClusterResource{
|
||||
Type: "qemu",
|
||||
Node: "node-a",
|
||||
Name: "app-vm",
|
||||
Status: "running",
|
||||
VMID: 101,
|
||||
MaxCPU: 4,
|
||||
MaxMem: 8192,
|
||||
MaxDisk: 1024,
|
||||
},
|
||||
client,
|
||||
guestID,
|
||||
map[string]models.Host{
|
||||
guestID: {
|
||||
ID: "host-1",
|
||||
LinkedVMID: guestID,
|
||||
Status: "online",
|
||||
Disks: []models.Disk{
|
||||
{Total: 500, Used: 200, Free: 300, Usage: 40, Mountpoint: "/", Type: "ext4", Device: "/dev/vda1"},
|
||||
},
|
||||
},
|
||||
},
|
||||
)
|
||||
if !ok {
|
||||
t.Fatal("expected VM to be built")
|
||||
}
|
||||
if vm.Disk.Total != 500 || vm.Disk.Used != 200 || vm.Disk.Usage != 40 {
|
||||
t.Fatalf("expected linked host agent disk summary to win, got %+v", vm.Disk)
|
||||
}
|
||||
if len(vm.Disks) != 1 || vm.Disks[0].Device != "/dev/vda1" {
|
||||
t.Fatalf("expected linked host agent disks to populate VM disks, got %+v", vm.Disks)
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -23,7 +23,7 @@ func (m *Monitor) pollPVENode(
|
|||
) (models.Node, string, string, error) {
|
||||
nodeStart := time.Now()
|
||||
displayName := getNodeDisplayName(instanceCfg, node.Node)
|
||||
connectionHost, guestURL := resolveNodeConnectionInfo(instanceCfg, node.Node)
|
||||
connectionHost, guestURL := resolveNodeConnectionInfo(instanceCfg, monitorDiscoveryConfig(m), node.Node)
|
||||
nodeID, effectiveStatus := m.determineNodeIDAndStatus(instanceName, instanceCfg, node)
|
||||
|
||||
modelNode := models.Node{
|
||||
|
|
|
|||
|
|
@ -13,14 +13,14 @@ import (
|
|||
"github.com/rs/zerolog/log"
|
||||
)
|
||||
|
||||
func resolveNodeConnectionInfo(instanceCfg *config.PVEInstance, nodeName string) (string, string) {
|
||||
func resolveNodeConnectionInfo(instanceCfg *config.PVEInstance, discoveryCfg config.DiscoveryConfig, nodeName string) (string, string) {
|
||||
connectionHost := instanceCfg.Host
|
||||
guestURL := instanceCfg.GuestURL
|
||||
if instanceCfg.IsCluster && len(instanceCfg.ClusterEndpoints) > 0 {
|
||||
hasFingerprint := instanceCfg.Fingerprint != ""
|
||||
for _, ep := range instanceCfg.ClusterEndpoints {
|
||||
if strings.EqualFold(ep.NodeName, nodeName) {
|
||||
if effective := clusterEndpointEffectiveURL(ep, instanceCfg.VerifySSL, hasFingerprint); effective != "" {
|
||||
if effective := clusterEndpointRuntimeURL(ep, instanceCfg.VerifySSL, hasFingerprint, discoveryCfg); effective != "" {
|
||||
connectionHost = effective
|
||||
}
|
||||
if ep.GuestURL != "" {
|
||||
|
|
|
|||
|
|
@ -497,6 +497,74 @@ func (m *Monitor) pollStorageWithNodes(ctx context.Context, instanceName string,
|
|||
allStorage = append(allStorage, entry.storage)
|
||||
}
|
||||
|
||||
// Some shared storages exist only in the cluster-wide storage config and do
|
||||
// not show up in per-node storage responses. Synthesize those entries so the
|
||||
// canonical storage surface still exposes them for alerts and filtering.
|
||||
existingSharedStorage := make(map[string]struct{}, len(allStorage))
|
||||
for _, storage := range allStorage {
|
||||
if storage.Shared {
|
||||
existingSharedStorage[storage.Instance+"/"+storage.Name] = struct{}{}
|
||||
}
|
||||
}
|
||||
|
||||
for _, clusterStorage := range clusterStorages {
|
||||
storageName := strings.TrimSpace(clusterStorage.Storage)
|
||||
if storageName == "" {
|
||||
continue
|
||||
}
|
||||
|
||||
shared := clusterStorage.Shared == 1 || isInherentlySharedStorageType(clusterStorage.Type)
|
||||
if !shared {
|
||||
continue
|
||||
}
|
||||
|
||||
key := storageInstanceName + "/" + storageName
|
||||
if _, exists := existingSharedStorage[key]; exists {
|
||||
continue
|
||||
}
|
||||
|
||||
nodesForStorage := parseClusterStorageNodes(clusterStorage.Nodes)
|
||||
if len(nodesForStorage) == 0 {
|
||||
nodesForStorage = make([]string, 0, len(nodes))
|
||||
for _, node := range nodes {
|
||||
nodeName := strings.TrimSpace(node.Node)
|
||||
if nodeName == "" {
|
||||
continue
|
||||
}
|
||||
nodesForStorage = append(nodesForStorage, nodeName)
|
||||
}
|
||||
}
|
||||
|
||||
nodeIDs := make([]string, 0, len(nodesForStorage))
|
||||
for _, nodeName := range nodesForStorage {
|
||||
nodeIDs = append(nodeIDs, fmt.Sprintf("%s-%s", storageInstanceName, nodeName))
|
||||
}
|
||||
|
||||
synthetic := models.Storage{
|
||||
ID: fmt.Sprintf("%s-cluster-%s", storageInstanceName, storageName),
|
||||
Name: storageName,
|
||||
Node: "cluster",
|
||||
Instance: storageInstanceName,
|
||||
Nodes: nodesForStorage,
|
||||
NodeIDs: nodeIDs,
|
||||
NodeCount: len(nodesForStorage),
|
||||
Type: clusterStorage.Type,
|
||||
Status: "available",
|
||||
Path: clusterStorage.Path,
|
||||
Total: int64(clusterStorage.Total),
|
||||
Used: int64(clusterStorage.Used),
|
||||
Free: int64(clusterStorage.Available),
|
||||
Usage: safePercentage(float64(clusterStorage.Used), float64(clusterStorage.Total)),
|
||||
Content: sortContent(clusterStorage.Content),
|
||||
Shared: true,
|
||||
Enabled: true,
|
||||
Active: true,
|
||||
}
|
||||
|
||||
allStorage = append(allStorage, synthetic)
|
||||
existingSharedStorage[key] = struct{}{}
|
||||
}
|
||||
|
||||
// Preserve existing storage data for nodes that weren't polled (offline or error)
|
||||
preservedCount := 0
|
||||
for _, existingStorage := range existingStorageMap {
|
||||
|
|
@ -586,11 +654,12 @@ func (m *Monitor) fetchNodeStorageFallback(ctx context.Context, instanceCfg *con
|
|||
|
||||
var target string
|
||||
hasFingerprint := strings.TrimSpace(instanceCfg.Fingerprint) != ""
|
||||
discoveryCfg := monitorDiscoveryConfig(m)
|
||||
for _, ep := range instanceCfg.ClusterEndpoints {
|
||||
if !strings.EqualFold(ep.NodeName, nodeName) {
|
||||
continue
|
||||
}
|
||||
target = clusterEndpointEffectiveURL(ep, instanceCfg.VerifySSL, hasFingerprint)
|
||||
target = clusterEndpointRuntimeURL(ep, instanceCfg.VerifySSL, hasFingerprint, discoveryCfg)
|
||||
if target != "" {
|
||||
break
|
||||
}
|
||||
|
|
|
|||
|
|
@ -206,6 +206,7 @@ func (m *Monitor) pollVMsWithNodes(ctx context.Context, instanceName string, clu
|
|||
diskTotal := vm.MaxDisk
|
||||
diskFree := diskTotal - diskUsed
|
||||
diskUsage := safePercentage(float64(diskUsed), float64(diskTotal))
|
||||
diskFromAgent := false
|
||||
diskStatusReason := ""
|
||||
var individualDisks []models.Disk
|
||||
|
||||
|
|
@ -428,6 +429,7 @@ func (m *Monitor) pollVMsWithNodes(ctx context.Context, instanceName string, clu
|
|||
diskUsed = usedBytes
|
||||
diskFree = totalBytes - usedBytes
|
||||
diskUsage = safePercentage(float64(usedBytes), float64(totalBytes))
|
||||
diskFromAgent = true
|
||||
diskStatusReason = "" // Clear reason on success
|
||||
|
||||
log.Info().
|
||||
|
|
@ -464,6 +466,24 @@ func (m *Monitor) pollVMsWithNodes(ctx context.Context, instanceName string, clu
|
|||
diskStatusReason = "no-status"
|
||||
}
|
||||
|
||||
if vm.Status == "running" && !diskFromAgent {
|
||||
if hostDisk, hostDisks, ok := resolveGuestDiskFromLinkedHostAgent(guestID, vmIDToHostAgent); ok && hostDisk.Total > 0 {
|
||||
diskTotal = uint64(hostDisk.Total)
|
||||
diskUsed = uint64(hostDisk.Used)
|
||||
diskFree = uint64(hostDisk.Free)
|
||||
diskUsage = hostDisk.Usage
|
||||
individualDisks = hostDisks
|
||||
diskStatusReason = ""
|
||||
log.Debug().
|
||||
Str("instance", instanceName).
|
||||
Str("vm", vm.Name).
|
||||
Str("node", n.Node).
|
||||
Int("vmid", vm.VMID).
|
||||
Float64("usage", hostDisk.Usage).
|
||||
Msg("QEMU disk: using linked Pulse host agent disk summary")
|
||||
}
|
||||
}
|
||||
|
||||
memTotalBytes := clampToInt64(memTotal)
|
||||
memUsedBytes := clampToInt64(memUsed)
|
||||
if memTotalBytes > 0 && memUsedBytes > memTotalBytes {
|
||||
|
|
|
|||
|
|
@ -50,7 +50,7 @@ func (m *Monitor) previousGuestContextForInstance(instanceName string) previousG
|
|||
continue
|
||||
}
|
||||
modelHost := previousHostFromView(host)
|
||||
if modelHost.LinkedVMID == "" || modelHost.Status != "online" || modelHost.Memory.Total <= 0 {
|
||||
if modelHost.LinkedVMID == "" || modelHost.Status != "online" {
|
||||
continue
|
||||
}
|
||||
ctx.hostAgentsByVMID[modelHost.LinkedVMID] = modelHost
|
||||
|
|
@ -126,6 +126,7 @@ func previousHostFromView(host *unifiedresources.HostView) models.Host {
|
|||
Status: string(host.Status()),
|
||||
LinkedVMID: host.LinkedVMID(),
|
||||
LastSeen: host.LastSeen(),
|
||||
Disks: guestDisksFromReadStateView(host.Disks()),
|
||||
Memory: models.Memory{
|
||||
Used: host.MemoryUsed(),
|
||||
Total: host.MemoryTotal(),
|
||||
|
|
|
|||
|
|
@ -1,7 +1,6 @@
|
|||
package monitoring
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"net"
|
||||
"net/url"
|
||||
"strings"
|
||||
|
|
@ -16,14 +15,15 @@ func (m *Monitor) buildClusterEndpointsForInit(pve config.PVEInstance) ([]string
|
|||
hasValidEndpoints := false
|
||||
endpoints := make([]string, 0, len(pve.ClusterEndpoints))
|
||||
endpointFingerprints := make(map[string]string)
|
||||
discoveryCfg := monitorDiscoveryConfig(m)
|
||||
|
||||
hasFingerprint := pve.Fingerprint != ""
|
||||
for _, ep := range pve.ClusterEndpoints {
|
||||
effectiveURL := clusterEndpointEffectiveURL(ep, pve.VerifySSL, hasFingerprint)
|
||||
effectiveURL := clusterEndpointRuntimeURL(ep, pve.VerifySSL, hasFingerprint, discoveryCfg)
|
||||
if effectiveURL == "" {
|
||||
log.Warn().
|
||||
Str("node", ep.NodeName).
|
||||
Msg("Skipping cluster endpoint with no host/IP")
|
||||
Msg("Skipping cluster endpoint with no allowed host/IP")
|
||||
continue
|
||||
}
|
||||
|
||||
|
|
@ -88,34 +88,55 @@ func (m *Monitor) buildClusterEndpointsForReconnect(pve config.PVEInstance) ([]s
|
|||
hasValidEndpoints := false
|
||||
endpoints := make([]string, 0, len(pve.ClusterEndpoints))
|
||||
endpointFingerprints := make(map[string]string)
|
||||
discoveryCfg := monitorDiscoveryConfig(m)
|
||||
hasFingerprint := pve.Fingerprint != ""
|
||||
|
||||
for _, ep := range pve.ClusterEndpoints {
|
||||
// Use EffectiveIP() which prefers IPOverride over auto-discovered IP
|
||||
host := ep.EffectiveIP()
|
||||
if host == "" {
|
||||
host = ep.Host
|
||||
}
|
||||
host := clusterEndpointRuntimeURL(ep, pve.VerifySSL, hasFingerprint, discoveryCfg)
|
||||
if host == "" {
|
||||
continue
|
||||
}
|
||||
if strings.Contains(host, ".") || net.ParseIP(host) != nil {
|
||||
hasValidEndpoints = true
|
||||
|
||||
if parsed, err := url.Parse(host); err == nil {
|
||||
hostname := parsed.Hostname()
|
||||
if hostname != "" && (strings.Contains(hostname, ".") || net.ParseIP(hostname) != nil) {
|
||||
hasValidEndpoints = true
|
||||
}
|
||||
} else {
|
||||
hostname := normalizeEndpointHost(host)
|
||||
if hostname != "" && (strings.Contains(hostname, ".") || net.ParseIP(hostname) != nil) {
|
||||
hasValidEndpoints = true
|
||||
}
|
||||
}
|
||||
|
||||
if !strings.HasPrefix(host, "http") {
|
||||
host = fmt.Sprintf("https://%s:8006", host)
|
||||
host = ensureClusterEndpointURL(host)
|
||||
}
|
||||
endpoints = append(endpoints, host)
|
||||
// Store per-endpoint fingerprint for TOFU
|
||||
if ep.Fingerprint != "" {
|
||||
endpointFingerprints[host] = ep.Fingerprint
|
||||
}
|
||||
}
|
||||
|
||||
if !hasValidEndpoints || len(endpoints) == 0 {
|
||||
endpoints = []string{pve.Host}
|
||||
if !strings.HasPrefix(endpoints[0], "http") {
|
||||
endpoints[0] = fmt.Sprintf("https://%s:8006", endpoints[0])
|
||||
fallback := ensureClusterEndpointURL(pve.Host)
|
||||
if fallback == "" {
|
||||
fallback = ensureClusterEndpointURL(pve.Host)
|
||||
}
|
||||
endpoints = []string{fallback}
|
||||
return endpoints, endpointFingerprints
|
||||
}
|
||||
|
||||
mainHostURL := ensureClusterEndpointURL(pve.Host)
|
||||
mainHostAlreadyIncluded := false
|
||||
for _, ep := range endpoints {
|
||||
if ep == mainHostURL {
|
||||
mainHostAlreadyIncluded = true
|
||||
break
|
||||
}
|
||||
}
|
||||
if !mainHostAlreadyIncluded && mainHostURL != "" {
|
||||
endpoints = append(endpoints, mainHostURL)
|
||||
}
|
||||
|
||||
return endpoints, endpointFingerprints
|
||||
|
|
|
|||
|
|
@ -25,6 +25,7 @@ type vmBuildState struct {
|
|||
diskUsed uint64
|
||||
diskFree uint64
|
||||
diskUsage float64
|
||||
diskFromAgent bool
|
||||
individualDisks []models.Disk
|
||||
ipAddresses []string
|
||||
networkInterfaces []models.GuestNetworkInterface
|
||||
|
|
@ -92,7 +93,7 @@ func (m *Monitor) applyVMStatusDetails(
|
|||
// Prefer guest agent data over cluster/resources data for accuracy
|
||||
if status.Agent.Value > 0 {
|
||||
var fsDisks []models.Disk
|
||||
state.diskTotal, state.diskUsed, state.diskFree, state.diskUsage, fsDisks = m.updateVMDisksFromGuestAgentFSInfo(
|
||||
state.diskTotal, state.diskUsed, state.diskFree, state.diskUsage, fsDisks, state.diskFromAgent = m.updateVMDisksFromGuestAgentFSInfo(
|
||||
ctx,
|
||||
instanceName,
|
||||
res,
|
||||
|
|
@ -116,6 +117,22 @@ func (m *Monitor) applyVMStatusDetails(
|
|||
Int("agent", status.Agent.Value).
|
||||
Msg("VM does not have guest agent enabled in config")
|
||||
}
|
||||
|
||||
if res.Status == "running" && !state.diskFromAgent {
|
||||
if hostDisk, hostDisks, ok := resolveGuestDiskFromLinkedHostAgent(guestID, vmIDToHostAgent); ok && hostDisk.Total > 0 {
|
||||
state.diskTotal = uint64(hostDisk.Total)
|
||||
state.diskUsed = uint64(hostDisk.Used)
|
||||
state.diskFree = uint64(hostDisk.Free)
|
||||
state.diskUsage = hostDisk.Usage
|
||||
state.individualDisks = hostDisks
|
||||
log.Debug().
|
||||
Str("instance", instanceName).
|
||||
Str("vm", res.Name).
|
||||
Int("vmid", res.VMID).
|
||||
Float64("usage", hostDisk.Usage).
|
||||
Msg("QEMU disk: using linked Pulse host agent disk summary")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (m *Monitor) buildVMFromClusterResource(
|
||||
|
|
@ -515,7 +532,7 @@ func (m *Monitor) updateVMDisksFromGuestAgentFSInfo(
|
|||
diskTotal uint64,
|
||||
diskUsed uint64,
|
||||
diskUsage float64,
|
||||
) (uint64, uint64, uint64, float64, []models.Disk) {
|
||||
) (uint64, uint64, uint64, float64, []models.Disk, bool) {
|
||||
log.Debug().
|
||||
Str("instance", instanceName).
|
||||
Str("vm", res.Name).
|
||||
|
|
@ -524,7 +541,7 @@ func (m *Monitor) updateVMDisksFromGuestAgentFSInfo(
|
|||
|
||||
fsInfo, ok := m.fetchVMFSInfo(ctx, instanceName, res, client)
|
||||
if !ok {
|
||||
return diskTotal, diskUsed, diskTotal - diskUsed, diskUsage, nil
|
||||
return diskTotal, diskUsed, diskTotal - diskUsed, diskUsage, nil, false
|
||||
}
|
||||
|
||||
log.Debug().
|
||||
|
|
@ -573,7 +590,7 @@ func (m *Monitor) updateVMDisksFromGuestAgentFSInfo(
|
|||
Uint64("old_disk", res.Disk).
|
||||
Uint64("old_maxdisk", res.MaxDisk).
|
||||
Msg("Using guest agent data for accurate disk usage (replacing cluster/resources data)")
|
||||
return diskTotal, diskUsed, diskTotal - diskUsed, diskUsage, summary.individualDisks
|
||||
return diskTotal, diskUsed, diskTotal - diskUsed, diskUsage, summary.individualDisks, true
|
||||
}
|
||||
|
||||
// Only special filesystems found - show allocated disk size instead
|
||||
|
|
@ -586,5 +603,5 @@ func (m *Monitor) updateVMDisksFromGuestAgentFSInfo(
|
|||
Int("filesystems_found", len(fsInfo)).
|
||||
Msg("Guest agent provided filesystem info but no usable filesystems found (all were special mounts)")
|
||||
|
||||
return diskTotal, diskUsed, diskTotal - diskUsed, diskUsage, nil
|
||||
return diskTotal, diskUsed, diskTotal - diskUsed, diskUsage, nil, false
|
||||
}
|
||||
|
|
|
|||
|
|
@ -8,6 +8,7 @@ import (
|
|||
"time"
|
||||
|
||||
"github.com/rcourtman/pulse-go-rewrite/internal/alerts"
|
||||
"github.com/rcourtman/pulse-go-rewrite/internal/config"
|
||||
"github.com/rcourtman/pulse-go-rewrite/internal/models"
|
||||
"github.com/rcourtman/pulse-go-rewrite/pkg/proxmox"
|
||||
)
|
||||
|
|
@ -229,3 +230,69 @@ func TestPollStorageWithNodesOptimizedRecordsMetricsAndAlerts(t *testing.T) {
|
|||
t.Fatalf("expected storage usage alert to be active, alerts=%+v", alerts)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPollStorageWithNodesSynthesizesSharedClusterOnlyStorage(t *testing.T) {
|
||||
monitor := &Monitor{
|
||||
state: models.NewState(),
|
||||
config: &config.Config{
|
||||
PVEInstances: []config.PVEInstance{
|
||||
{
|
||||
Name: "inst1",
|
||||
IsCluster: true,
|
||||
ClusterName: "cluster-a",
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
client := &fakeStorageClient{
|
||||
allStorage: []proxmox.Storage{
|
||||
{
|
||||
Storage: "cephfs",
|
||||
Type: "cephfs",
|
||||
Content: "images,backup",
|
||||
Shared: 1,
|
||||
Total: 1000,
|
||||
Used: 100,
|
||||
Available: 900,
|
||||
Nodes: "node1,node2",
|
||||
},
|
||||
},
|
||||
storageByNode: map[string][]proxmox.Storage{
|
||||
"node1": {
|
||||
{Storage: "local", Type: "dir", Content: "images", Active: 1, Enabled: 1, Total: 100, Used: 10, Available: 90},
|
||||
},
|
||||
"node2": {
|
||||
{Storage: "local", Type: "dir", Content: "images", Active: 1, Enabled: 1, Total: 100, Used: 20, Available: 80},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
nodes := []proxmox.Node{
|
||||
{Node: "node1", Status: "online"},
|
||||
{Node: "node2", Status: "online"},
|
||||
}
|
||||
|
||||
monitor.pollStorageWithNodes(context.Background(), "inst1", client, nodes)
|
||||
|
||||
var shared *models.Storage
|
||||
for _, storage := range monitor.state.GetSnapshot().Storage {
|
||||
if storage.Name == "cephfs" {
|
||||
storageCopy := storage
|
||||
shared = &storageCopy
|
||||
break
|
||||
}
|
||||
}
|
||||
if shared == nil {
|
||||
t.Fatalf("expected synthesized shared storage in state, got %+v", monitor.state.GetSnapshot().Storage)
|
||||
}
|
||||
if shared.ID != "cluster-a-cluster-cephfs" || shared.Node != "cluster" || !shared.Shared {
|
||||
t.Fatalf("unexpected synthesized shared storage identity: %+v", *shared)
|
||||
}
|
||||
if shared.NodeCount != 2 || len(shared.Nodes) != 2 {
|
||||
t.Fatalf("expected cluster storage node affinity, got %+v", *shared)
|
||||
}
|
||||
if shared.Total != 1000 || shared.Used != 100 || shared.Free != 900 {
|
||||
t.Fatalf("expected cluster storage capacity from config, got %+v", *shared)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -11,6 +11,8 @@ type CephStatus struct {
|
|||
FSID string `json:"fsid"`
|
||||
Health CephHealth `json:"health"`
|
||||
ServiceMap CephServiceMap `json:"servicemap"`
|
||||
MonMap CephMonMap `json:"monmap"`
|
||||
MgrMap CephMgrMap `json:"mgrmap"`
|
||||
OSDMap CephOSDMap `json:"osdmap"`
|
||||
PGMap CephPGMap `json:"pgmap"`
|
||||
}
|
||||
|
|
@ -57,6 +59,19 @@ type CephServiceDaemon struct {
|
|||
Status string `json:"status"`
|
||||
}
|
||||
|
||||
// CephMonMap captures monitor summary information.
|
||||
type CephMonMap struct {
|
||||
NumMons int `json:"num_mons"`
|
||||
}
|
||||
|
||||
// CephMgrMap captures manager summary information.
|
||||
type CephMgrMap struct {
|
||||
Available bool `json:"available"`
|
||||
NumMgrs int `json:"num_mgrs"`
|
||||
ActiveName string `json:"active_name"`
|
||||
Standbys []string `json:"standbys"`
|
||||
}
|
||||
|
||||
// CephOSDMap captures summary statistics about OSDs.
|
||||
type CephOSDMap struct {
|
||||
NumOSDs int `json:"num_osds"`
|
||||
|
|
|
|||
|
|
@ -27,6 +27,15 @@ func TestGetCephStatus(t *testing.T) {
|
|||
"servicemap": map[string]interface{}{
|
||||
"services": map[string]interface{}{},
|
||||
},
|
||||
"monmap": map[string]interface{}{
|
||||
"num_mons": 3,
|
||||
},
|
||||
"mgrmap": map[string]interface{}{
|
||||
"available": true,
|
||||
"num_mgrs": 2,
|
||||
"active_name": "mgr-a",
|
||||
"standbys": []string{"mgr-b"},
|
||||
},
|
||||
"osdmap": map[string]interface{}{
|
||||
"num_osds": 1,
|
||||
"num_up_osds": 1,
|
||||
|
|
@ -49,6 +58,9 @@ func TestGetCephStatus(t *testing.T) {
|
|||
if status.FSID != "fsid-1" || status.Health.Status != "HEALTH_OK" {
|
||||
t.Fatalf("unexpected status: %+v", status)
|
||||
}
|
||||
if status.MonMap.NumMons != 3 || status.MgrMap.NumMgrs != 2 || status.MgrMap.ActiveName != "mgr-a" {
|
||||
t.Fatalf("expected monmap/mgrmap to decode, got %+v", status)
|
||||
}
|
||||
}
|
||||
|
||||
func TestGetCephDF(t *testing.T) {
|
||||
|
|
|
|||
|
|
@ -1673,6 +1673,26 @@ type VMIPAddress struct {
|
|||
Prefix int `json:"prefix"`
|
||||
}
|
||||
|
||||
func (a *VMIPAddress) UnmarshalJSON(data []byte) error {
|
||||
var raw map[string]interface{}
|
||||
if err := json.Unmarshal(data, &raw); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
a.Address = coerceString(raw["ip-address"])
|
||||
|
||||
maxInt := uint64(^uint(0) >> 1)
|
||||
prefix, err := coerceUint64("prefix", raw["prefix"])
|
||||
if err != nil {
|
||||
prefix = 0
|
||||
}
|
||||
if prefix > maxInt {
|
||||
prefix = maxInt
|
||||
}
|
||||
a.Prefix = int(prefix)
|
||||
return nil
|
||||
}
|
||||
|
||||
type VMNetworkInterface struct {
|
||||
Name string `json:"name"`
|
||||
HardwareAddr string `json:"hardware-address"`
|
||||
|
|
@ -1682,6 +1702,134 @@ type VMNetworkInterface struct {
|
|||
HasIp6Gateway bool `json:"has-ipv6-synth-gateway,omitempty"`
|
||||
}
|
||||
|
||||
func (iface *VMNetworkInterface) UnmarshalJSON(data []byte) error {
|
||||
var raw map[string]json.RawMessage
|
||||
if err := json.Unmarshal(data, &raw); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
iface.Name = unmarshalRawString(raw["name"])
|
||||
iface.HardwareAddr = unmarshalRawString(raw["hardware-address"])
|
||||
iface.HasIp4Gateway = unmarshalRawBool(raw["has-ipv4-synth-gateway"])
|
||||
iface.HasIp6Gateway = unmarshalRawBool(raw["has-ipv6-synth-gateway"])
|
||||
|
||||
iface.Statistics = nil
|
||||
if statsRaw, ok := raw["statistics"]; ok && len(statsRaw) > 0 && string(statsRaw) != "null" {
|
||||
var stats interface{}
|
||||
if err := json.Unmarshal(statsRaw, &stats); err == nil {
|
||||
iface.Statistics = stats
|
||||
}
|
||||
}
|
||||
|
||||
iface.IPAddresses = nil
|
||||
if addressesRaw, ok := raw["ip-addresses"]; ok && len(addressesRaw) > 0 && string(addressesRaw) != "null" {
|
||||
var rawAddresses []json.RawMessage
|
||||
if err := json.Unmarshal(addressesRaw, &rawAddresses); err == nil {
|
||||
iface.IPAddresses = decodeVMIpAddresses(rawAddresses)
|
||||
} else {
|
||||
var rawAddress json.RawMessage
|
||||
if err := json.Unmarshal(addressesRaw, &rawAddress); err == nil && len(rawAddress) > 0 {
|
||||
iface.IPAddresses = decodeVMIpAddresses([]json.RawMessage{rawAddress})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func decodeVMIpAddresses(rawAddresses []json.RawMessage) []VMIPAddress {
|
||||
if len(rawAddresses) == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
addresses := make([]VMIPAddress, 0, len(rawAddresses))
|
||||
for _, rawAddr := range rawAddresses {
|
||||
var addr VMIPAddress
|
||||
if err := json.Unmarshal(rawAddr, &addr); err != nil {
|
||||
continue
|
||||
}
|
||||
if addr.Address == "" {
|
||||
continue
|
||||
}
|
||||
addresses = append(addresses, addr)
|
||||
}
|
||||
if len(addresses) == 0 {
|
||||
return nil
|
||||
}
|
||||
return addresses
|
||||
}
|
||||
|
||||
func unmarshalRawString(data json.RawMessage) string {
|
||||
if len(data) == 0 || string(data) == "null" {
|
||||
return ""
|
||||
}
|
||||
|
||||
var value interface{}
|
||||
if err := json.Unmarshal(data, &value); err != nil {
|
||||
return ""
|
||||
}
|
||||
return coerceString(value)
|
||||
}
|
||||
|
||||
func unmarshalRawBool(data json.RawMessage) bool {
|
||||
if len(data) == 0 || string(data) == "null" {
|
||||
return false
|
||||
}
|
||||
|
||||
var value bool
|
||||
if err := json.Unmarshal(data, &value); err == nil {
|
||||
return value
|
||||
}
|
||||
|
||||
var generic interface{}
|
||||
if err := json.Unmarshal(data, &generic); err != nil {
|
||||
return false
|
||||
}
|
||||
|
||||
switch v := generic.(type) {
|
||||
case string:
|
||||
lower := strings.ToLower(strings.TrimSpace(v))
|
||||
return lower == "true" || lower == "1" || lower == "yes"
|
||||
case float64:
|
||||
return v != 0
|
||||
case int:
|
||||
return v != 0
|
||||
case int64:
|
||||
return v != 0
|
||||
case uint64:
|
||||
return v != 0
|
||||
default:
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
func coerceString(value interface{}) string {
|
||||
switch v := value.(type) {
|
||||
case nil:
|
||||
return ""
|
||||
case string:
|
||||
return strings.TrimSpace(v)
|
||||
case json.Number:
|
||||
return strings.TrimSpace(v.String())
|
||||
case float64:
|
||||
return strings.TrimSpace(strconv.FormatFloat(v, 'f', -1, 64))
|
||||
case float32:
|
||||
return strings.TrimSpace(strconv.FormatFloat(float64(v), 'f', -1, 32))
|
||||
case int:
|
||||
return strconv.Itoa(v)
|
||||
case int32:
|
||||
return strconv.FormatInt(int64(v), 10)
|
||||
case int64:
|
||||
return strconv.FormatInt(v, 10)
|
||||
case uint32:
|
||||
return strconv.FormatUint(uint64(v), 10)
|
||||
case uint64:
|
||||
return strconv.FormatUint(v, 10)
|
||||
default:
|
||||
return ""
|
||||
}
|
||||
}
|
||||
|
||||
// GetVMFSInfo returns filesystem information from QEMU guest agent
|
||||
func (c *Client) GetVMFSInfo(ctx context.Context, node string, vmid int) ([]VMFileSystem, error) {
|
||||
resp, err := c.get(ctx, fmt.Sprintf("/nodes/%s/qemu/%d/agent/get-fsinfo", node, vmid))
|
||||
|
|
@ -1711,52 +1859,7 @@ func (c *Client) GetVMFSInfo(ctx context.Context, node string, vmid int) ([]VMFi
|
|||
} `json:"data"`
|
||||
}
|
||||
if err := json.Unmarshal(bodyBytes, &arrayResult); err == nil && arrayResult.Data.Result != nil {
|
||||
// Post-process to extract disk device names
|
||||
for i := range arrayResult.Data.Result {
|
||||
fs := &arrayResult.Data.Result[i]
|
||||
// Extract disk device name from the DiskRaw field
|
||||
if len(fs.DiskRaw) > 0 {
|
||||
// The disk field usually contains device info as a map
|
||||
if diskMap, ok := fs.DiskRaw[0].(map[string]interface{}); ok {
|
||||
// Try to get the device name from various possible fields
|
||||
if dev, ok := diskMap["dev"].(string); ok {
|
||||
fs.Disk = dev
|
||||
} else if serial, ok := diskMap["serial"].(string); ok {
|
||||
fs.Disk = serial
|
||||
} else if bus, ok := diskMap["bus-type"].(string); ok {
|
||||
if target, ok := diskMap["target"].(float64); ok {
|
||||
fs.Disk = fmt.Sprintf("%s-%d", bus, int(target))
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
// If we still don't have a disk identifier, use the mountpoint as a fallback
|
||||
if fs.Disk == "" && fs.Mountpoint != "" {
|
||||
// For root filesystem, use a special identifier
|
||||
if fs.Mountpoint == "/" {
|
||||
fs.Disk = "root-filesystem"
|
||||
} else {
|
||||
// For Windows, normalize drive letters to prevent duplicate counting
|
||||
// Windows guest agent can return multiple directory entries (C:\, C:\Users, C:\Windows)
|
||||
// all on the same physical drive. Without disk[] metadata, we must deduplicate by drive letter.
|
||||
isWindowsDrive := len(fs.Mountpoint) >= 2 && fs.Mountpoint[1] == ':' && strings.Contains(fs.Mountpoint, "\\")
|
||||
if isWindowsDrive {
|
||||
// Use drive letter as identifier (e.g., "C:" for C:\, C:\Users, etc.)
|
||||
driveLetter := strings.ToUpper(fs.Mountpoint[:2])
|
||||
fs.Disk = driveLetter
|
||||
log.Debug().
|
||||
Str("node", node).
|
||||
Int("vmid", vmid).
|
||||
Str("mountpoint", fs.Mountpoint).
|
||||
Str("synthesized_disk", driveLetter).
|
||||
Msg("Synthesized Windows drive identifier from mountpoint")
|
||||
} else {
|
||||
// Use mountpoint as unique identifier for non-Windows paths
|
||||
fs.Disk = fs.Mountpoint
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
postProcessVMFilesystems(node, vmid, arrayResult.Data.Result)
|
||||
return arrayResult.Data.Result, nil
|
||||
}
|
||||
|
||||
|
|
@ -1767,21 +1870,33 @@ func (c *Client) GetVMFSInfo(ctx context.Context, node string, vmid int) ([]VMFi
|
|||
} `json:"data"`
|
||||
}
|
||||
if err := json.Unmarshal(bodyBytes, &objectResult); err == nil {
|
||||
// If result is an object, it might be an error or empty response
|
||||
// Check if it's null or an error
|
||||
if objectResult.Data.Result == nil {
|
||||
log.Debug().
|
||||
Str("node", node).
|
||||
Int("vmid", vmid).
|
||||
Msg("GetVMFSInfo received null result - guest agent may not be providing disk info")
|
||||
} else {
|
||||
log.Debug().
|
||||
Str("node", node).
|
||||
Int("vmid", vmid).
|
||||
Interface("result", objectResult.Data.Result).
|
||||
Msg("GetVMFSInfo received object instead of array")
|
||||
return []VMFileSystem{}, nil
|
||||
}
|
||||
// Return empty array to indicate no filesystem info available
|
||||
|
||||
if fsMap, ok := objectResult.Data.Result.(map[string]interface{}); ok && looksLikeVMFilesystemResult(fsMap) {
|
||||
rawFS, marshalErr := json.Marshal(fsMap)
|
||||
if marshalErr != nil {
|
||||
return nil, fmt.Errorf("failed to marshal object-style guest filesystem result: %w", marshalErr)
|
||||
}
|
||||
var fs VMFileSystem
|
||||
if unmarshalErr := json.Unmarshal(rawFS, &fs); unmarshalErr != nil {
|
||||
return nil, fmt.Errorf("failed to parse object-style guest filesystem result: %w", unmarshalErr)
|
||||
}
|
||||
filesystems := []VMFileSystem{fs}
|
||||
postProcessVMFilesystems(node, vmid, filesystems)
|
||||
return filesystems, nil
|
||||
}
|
||||
|
||||
log.Debug().
|
||||
Str("node", node).
|
||||
Int("vmid", vmid).
|
||||
Interface("result", objectResult.Data.Result).
|
||||
Msg("GetVMFSInfo received object instead of array")
|
||||
return []VMFileSystem{}, nil
|
||||
}
|
||||
|
||||
|
|
@ -1789,6 +1904,61 @@ func (c *Client) GetVMFSInfo(ctx context.Context, node string, vmid int) ([]VMFi
|
|||
return nil, fmt.Errorf("unexpected response format from guest agent get-fsinfo")
|
||||
}
|
||||
|
||||
func looksLikeVMFilesystemResult(result map[string]interface{}) bool {
|
||||
if len(result) == 0 {
|
||||
return false
|
||||
}
|
||||
|
||||
for _, key := range []string{"mountpoint", "name", "type", "total-bytes", "total-bytes-privileged", "used-bytes", "disk"} {
|
||||
if _, ok := result[key]; ok {
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
||||
return false
|
||||
}
|
||||
|
||||
func postProcessVMFilesystems(node string, vmid int, filesystems []VMFileSystem) {
|
||||
for i := range filesystems {
|
||||
fs := &filesystems[i]
|
||||
if len(fs.DiskRaw) > 0 {
|
||||
if diskMap, ok := fs.DiskRaw[0].(map[string]interface{}); ok {
|
||||
if dev, ok := diskMap["dev"].(string); ok {
|
||||
fs.Disk = dev
|
||||
} else if serial, ok := diskMap["serial"].(string); ok {
|
||||
fs.Disk = serial
|
||||
} else if bus, ok := diskMap["bus-type"].(string); ok {
|
||||
if target, ok := diskMap["target"].(float64); ok {
|
||||
fs.Disk = fmt.Sprintf("%s-%d", bus, int(target))
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
if fs.Disk != "" || fs.Mountpoint == "" {
|
||||
continue
|
||||
}
|
||||
if fs.Mountpoint == "/" {
|
||||
fs.Disk = "root-filesystem"
|
||||
continue
|
||||
}
|
||||
|
||||
isWindowsDrive := len(fs.Mountpoint) >= 2 && fs.Mountpoint[1] == ':' && strings.Contains(fs.Mountpoint, "\\")
|
||||
if isWindowsDrive {
|
||||
driveLetter := strings.ToUpper(fs.Mountpoint[:2])
|
||||
fs.Disk = driveLetter
|
||||
log.Debug().
|
||||
Str("node", node).
|
||||
Int("vmid", vmid).
|
||||
Str("mountpoint", fs.Mountpoint).
|
||||
Str("synthesized_disk", driveLetter).
|
||||
Msg("Synthesized Windows drive identifier from mountpoint")
|
||||
continue
|
||||
}
|
||||
|
||||
fs.Disk = fs.Mountpoint
|
||||
}
|
||||
}
|
||||
|
||||
// GetVMNetworkInterfaces returns network interfaces reported by the guest agent
|
||||
func (c *Client) GetVMNetworkInterfaces(ctx context.Context, node string, vmid int) ([]VMNetworkInterface, error) {
|
||||
resp, err := c.get(ctx, fmt.Sprintf("/nodes/%s/qemu/%d/agent/network-get-interfaces", node, vmid))
|
||||
|
|
@ -1797,17 +1967,89 @@ func (c *Client) GetVMNetworkInterfaces(ctx context.Context, node string, vmid i
|
|||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
var result struct {
|
||||
Data struct {
|
||||
Result []VMNetworkInterface `json:"result"`
|
||||
} `json:"data"`
|
||||
}
|
||||
|
||||
if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
|
||||
bodyBytes, err := readResponseBodyLimited(resp.Body)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return result.Data.Result, nil
|
||||
var arrayResult struct {
|
||||
Data struct {
|
||||
Result []json.RawMessage `json:"result"`
|
||||
} `json:"data"`
|
||||
}
|
||||
if err := json.Unmarshal(bodyBytes, &arrayResult); err == nil && arrayResult.Data.Result != nil {
|
||||
interfaces := make([]VMNetworkInterface, 0, len(arrayResult.Data.Result))
|
||||
for idx, rawIface := range arrayResult.Data.Result {
|
||||
var iface VMNetworkInterface
|
||||
if err := json.Unmarshal(rawIface, &iface); err != nil {
|
||||
log.Warn().
|
||||
Err(err).
|
||||
Str("node", node).
|
||||
Int("vmid", vmid).
|
||||
Int("interface_index", idx).
|
||||
Msg("Skipping malformed guest agent network interface entry")
|
||||
continue
|
||||
}
|
||||
if !vmNetworkInterfaceHasUsefulData(iface) {
|
||||
continue
|
||||
}
|
||||
interfaces = append(interfaces, iface)
|
||||
}
|
||||
return interfaces, nil
|
||||
}
|
||||
|
||||
var objectResult struct {
|
||||
Data struct {
|
||||
Result interface{} `json:"result"`
|
||||
} `json:"data"`
|
||||
}
|
||||
if err := json.Unmarshal(bodyBytes, &objectResult); err == nil {
|
||||
if objectResult.Data.Result == nil {
|
||||
return []VMNetworkInterface{}, nil
|
||||
}
|
||||
|
||||
if ifaceMap, ok := objectResult.Data.Result.(map[string]interface{}); ok && looksLikeVMNetworkInterfaceResult(ifaceMap) {
|
||||
rawIface, marshalErr := json.Marshal(ifaceMap)
|
||||
if marshalErr != nil {
|
||||
return nil, fmt.Errorf("failed to marshal object-style guest network interface result: %w", marshalErr)
|
||||
}
|
||||
var iface VMNetworkInterface
|
||||
if unmarshalErr := json.Unmarshal(rawIface, &iface); unmarshalErr != nil {
|
||||
return nil, fmt.Errorf("failed to parse object-style guest network interface result: %w", unmarshalErr)
|
||||
}
|
||||
if !vmNetworkInterfaceHasUsefulData(iface) {
|
||||
return []VMNetworkInterface{}, nil
|
||||
}
|
||||
return []VMNetworkInterface{iface}, nil
|
||||
}
|
||||
|
||||
return []VMNetworkInterface{}, nil
|
||||
}
|
||||
|
||||
return nil, fmt.Errorf("unexpected response format from guest agent network-get-interfaces")
|
||||
}
|
||||
|
||||
func looksLikeVMNetworkInterfaceResult(result map[string]interface{}) bool {
|
||||
if len(result) == 0 {
|
||||
return false
|
||||
}
|
||||
|
||||
for _, key := range []string{"name", "hardware-address", "ip-addresses", "statistics"} {
|
||||
if _, ok := result[key]; ok {
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
||||
return false
|
||||
}
|
||||
|
||||
func vmNetworkInterfaceHasUsefulData(iface VMNetworkInterface) bool {
|
||||
return iface.Name != "" ||
|
||||
iface.HardwareAddr != "" ||
|
||||
len(iface.IPAddresses) > 0 ||
|
||||
iface.Statistics != nil ||
|
||||
iface.HasIp4Gateway ||
|
||||
iface.HasIp6Gateway
|
||||
}
|
||||
|
||||
// GetVMStatus returns detailed VM status including balloon info
|
||||
|
|
|
|||
|
|
@ -78,6 +78,39 @@ func TestClientVMFSInfoObjectResult(t *testing.T) {
|
|||
}
|
||||
}
|
||||
|
||||
func TestClientVMFSInfoObjectFilesystemResult(t *testing.T) {
|
||||
client := newTestClient(t, func(w http.ResponseWriter, r *http.Request) {
|
||||
switch r.URL.Path {
|
||||
case "/api2/json/nodes/node1/qemu/100/agent/get-fsinfo":
|
||||
writeJSON(t, w, map[string]interface{}{
|
||||
"data": map[string]interface{}{
|
||||
"result": map[string]interface{}{
|
||||
"name": "root",
|
||||
"type": "ext4",
|
||||
"mountpoint": "/",
|
||||
"total-bytes": 512,
|
||||
"used-bytes": 256,
|
||||
},
|
||||
},
|
||||
})
|
||||
default:
|
||||
http.NotFound(w, r)
|
||||
}
|
||||
})
|
||||
|
||||
ctx := context.Background()
|
||||
filesystems, err := client.GetVMFSInfo(ctx, "node1", 100)
|
||||
if err != nil {
|
||||
t.Fatalf("GetVMFSInfo error: %v", err)
|
||||
}
|
||||
if len(filesystems) != 1 {
|
||||
t.Fatalf("expected single filesystem, got %+v", filesystems)
|
||||
}
|
||||
if filesystems[0].Disk != "root-filesystem" {
|
||||
t.Fatalf("expected synthesized root disk identifier, got %+v", filesystems[0])
|
||||
}
|
||||
}
|
||||
|
||||
func TestClientContainerInterfacesError(t *testing.T) {
|
||||
client := newTestClient(t, func(w http.ResponseWriter, r *http.Request) {
|
||||
switch r.URL.Path {
|
||||
|
|
|
|||
|
|
@ -215,6 +215,68 @@ func TestClientClusterAndAgentInfo(t *testing.T) {
|
|||
}
|
||||
}
|
||||
|
||||
func TestClientGetVMNetworkInterfaces_ObjectAndPartialPayloads(t *testing.T) {
|
||||
client := newTestClient(t, func(w http.ResponseWriter, r *http.Request) {
|
||||
switch r.URL.Path {
|
||||
case "/api2/json/nodes/node1/qemu/100/agent/network-get-interfaces":
|
||||
writeJSON(t, w, map[string]interface{}{
|
||||
"data": map[string]interface{}{
|
||||
"result": []interface{}{
|
||||
map[string]interface{}{
|
||||
"name": "eth0",
|
||||
"hardware-address": "00:11:22:33:44:55",
|
||||
"ip-addresses": []interface{}{
|
||||
map[string]interface{}{"ip-address": "192.168.1.10", "prefix": 24},
|
||||
map[string]interface{}{"ip-address": "fe80::1", "prefix": "64"},
|
||||
},
|
||||
},
|
||||
map[string]interface{}{
|
||||
"name": 123,
|
||||
"ip-addresses": map[string]interface{}{"ip-address": "10.0.0.5", "prefix": "16"},
|
||||
},
|
||||
"malformed",
|
||||
},
|
||||
},
|
||||
})
|
||||
case "/api2/json/nodes/node1/qemu/101/agent/network-get-interfaces":
|
||||
writeJSON(t, w, map[string]interface{}{
|
||||
"data": map[string]interface{}{
|
||||
"result": map[string]interface{}{
|
||||
"name": "eth1",
|
||||
"hardware-address": "aa:bb:cc:dd:ee:ff",
|
||||
"ip-addresses": map[string]interface{}{"ip-address": "10.10.0.8", "prefix": 24},
|
||||
},
|
||||
},
|
||||
})
|
||||
default:
|
||||
http.NotFound(w, r)
|
||||
}
|
||||
})
|
||||
|
||||
ctx := context.Background()
|
||||
ifaces, err := client.GetVMNetworkInterfaces(ctx, "node1", 100)
|
||||
if err != nil {
|
||||
t.Fatalf("GetVMNetworkInterfaces array payload error: %v", err)
|
||||
}
|
||||
if len(ifaces) != 2 {
|
||||
t.Fatalf("expected 2 useful interfaces from partial payload, got %+v", ifaces)
|
||||
}
|
||||
if ifaces[0].Name != "eth0" || len(ifaces[0].IPAddresses) != 2 {
|
||||
t.Fatalf("expected eth0 addresses to be preserved, got %+v", ifaces[0])
|
||||
}
|
||||
if ifaces[1].Name != "123" || len(ifaces[1].IPAddresses) != 1 || ifaces[1].IPAddresses[0].Address != "10.0.0.5" {
|
||||
t.Fatalf("expected object-style address to be coerced, got %+v", ifaces[1])
|
||||
}
|
||||
|
||||
objectIfaces, err := client.GetVMNetworkInterfaces(ctx, "node1", 101)
|
||||
if err != nil {
|
||||
t.Fatalf("GetVMNetworkInterfaces object payload error: %v", err)
|
||||
}
|
||||
if len(objectIfaces) != 1 || objectIfaces[0].Name != "eth1" || objectIfaces[0].IPAddresses[0].Address != "10.10.0.8" {
|
||||
t.Fatalf("unexpected object-style interfaces: %+v", objectIfaces)
|
||||
}
|
||||
}
|
||||
|
||||
func TestClientStatusAndResources(t *testing.T) {
|
||||
client := newTestClient(t, func(w http.ResponseWriter, r *http.Request) {
|
||||
switch r.URL.Path {
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue