From b95c01066e52b4720f30fa921c73f94dd9a88473 Mon Sep 17 00:00:00 2001 From: rcourtman Date: Thu, 23 Oct 2025 07:50:45 +0000 Subject: [PATCH] Capture dynamic LXC IP metrics (#596) --- internal/monitoring/harness_integration.go | 3 + internal/monitoring/monitor.go | 131 ++++++++++++++++ internal/monitoring/monitor_memory_test.go | 3 + internal/monitoring/monitor_snapshots_test.go | 3 + internal/monitoring/monitor_storage_test.go | 3 + pkg/proxmox/client.go | 142 ++++++++++++++++++ pkg/proxmox/cluster_client.go | 14 ++ 7 files changed, 299 insertions(+) diff --git a/internal/monitoring/harness_integration.go b/internal/monitoring/harness_integration.go index 4cb4dc1..80935a6 100644 --- a/internal/monitoring/harness_integration.go +++ b/internal/monitoring/harness_integration.go @@ -414,6 +414,9 @@ func (noopPVEClient) GetContainerStatus(ctx context.Context, node string, vmid i func (noopPVEClient) GetContainerConfig(ctx context.Context, node string, vmid int) (map[string]interface{}, error) { return nil, nil } +func (noopPVEClient) ExecContainerCommand(ctx context.Context, node string, vmid int, command []string) (string, error) { + return "", nil +} func (noopPVEClient) GetClusterResources(ctx context.Context, resourceType string) ([]proxmox.ClusterResource, error) { return nil, nil } diff --git a/internal/monitoring/monitor.go b/internal/monitoring/monitor.go index 97a9a55..ee2363f 100644 --- a/internal/monitoring/monitor.go +++ b/internal/monitoring/monitor.go @@ -56,6 +56,7 @@ type PVEClientInterface interface { GetVMStatus(ctx context.Context, node string, vmid int) (*proxmox.VMStatus, error) GetContainerStatus(ctx context.Context, node string, vmid int) (*proxmox.Container, error) GetContainerConfig(ctx context.Context, node string, vmid int) (map[string]interface{}, error) + ExecContainerCommand(ctx context.Context, node string, vmid int, command []string) (string, error) GetClusterResources(ctx context.Context, resourceType string) ([]proxmox.ClusterResource, error) IsClusterMember(ctx context.Context) (bool, error) GetVMFSInfo(ctx context.Context, node string, vmid int) ([]proxmox.VMFileSystem, error) @@ -467,6 +468,8 @@ type Monitor struct { dockerCommandIndex map[string]string guestMetadataMu sync.RWMutex guestMetadataCache map[string]guestMetadataCacheEntry + containerIPCacheMu sync.RWMutex + containerIPCache map[string]containerIPCacheEntry executor PollExecutor breakerBaseRetry time.Duration breakerMaxDelay time.Duration @@ -483,6 +486,13 @@ type rrdMemCacheEntry struct { fetchedAt time.Time } +const containerIPCacheTTL = 90 * time.Second + +type containerIPCacheEntry struct { + addresses []string + fetchedAt time.Time +} + // safePercentage calculates percentage safely, returning 0 if divisor is 0 func safePercentage(used, total float64) float64 { if total == 0 { @@ -1958,6 +1968,18 @@ func (m *Monitor) enrichContainerMetadata(ctx context.Context, client PVEClientI } } + if len(addressOrder) == 0 { + dynamicIPs := m.lookupDynamicContainerIPs(ctx, client, instanceName, nodeName, container.VMID) + if len(dynamicIPs) > 0 { + for _, addr := range dynamicIPs { + addAddress(addr) + } + if len(networkIfaces) == 1 && len(networkIfaces[0].Addresses) == 0 { + networkIfaces[0].Addresses = dedupeStringsPreserveOrder(dynamicIPs) + } + } + } + if len(networkIfaces) > 1 { sort.SliceStable(networkIfaces, func(i, j int) bool { left := strings.TrimSpace(networkIfaces[i].Name) @@ -2331,6 +2353,114 @@ func extractContainerRootDeviceFromConfig(config map[string]interface{}) string return device } +func (m *Monitor) lookupDynamicContainerIPs(ctx context.Context, client PVEClientInterface, instanceName, nodeName string, vmid int) []string { + if m == nil || client == nil { + return nil + } + + cacheKey := fmt.Sprintf("%s|%s|%d", instanceName, nodeName, vmid) + now := time.Now() + + m.containerIPCacheMu.RLock() + if entry, ok := m.containerIPCache[cacheKey]; ok && now.Sub(entry.fetchedAt) < containerIPCacheTTL { + addresses := cloneStringSlice(entry.addresses) + m.containerIPCacheMu.RUnlock() + return addresses + } + m.containerIPCacheMu.RUnlock() + + execCtx, cancel := context.WithTimeout(ctx, 3*time.Second) + output, err := client.ExecContainerCommand(execCtx, nodeName, vmid, []string{"ip", "-j", "addr", "show", "scope", "global"}) + cancel() + if err != nil { + log.Debug(). + Err(err). + Str("instance", instanceName). + Str("node", nodeName). + Int("vmid", vmid). + Msg("Failed to execute container IP command") + } + + addresses := parseContainerIPsFromJSON(output) + if len(addresses) == 0 { + fallbackCtx, fallbackCancel := context.WithTimeout(ctx, 3*time.Second) + fallbackOutput, fallbackErr := client.ExecContainerCommand(fallbackCtx, nodeName, vmid, []string{"hostname", "-I"}) + fallbackCancel() + if fallbackErr != nil { + log.Debug(). + Err(fallbackErr). + Str("instance", instanceName). + Str("node", nodeName). + Int("vmid", vmid). + Msg("Failed to execute hostname -I for container") + } else { + addresses = parseContainerIPsFromWhitespaceList(fallbackOutput) + } + } + + addresses = dedupeStringsPreserveOrder(addresses) + + m.containerIPCacheMu.Lock() + m.containerIPCache[cacheKey] = containerIPCacheEntry{ + addresses: cloneStringSlice(addresses), + fetchedAt: time.Now(), + } + m.containerIPCacheMu.Unlock() + + return addresses +} + +func parseContainerIPsFromJSON(output string) []string { + output = strings.TrimSpace(output) + if output == "" { + return nil + } + + type addrInfo struct { + Local string `json:"local"` + } + type iface struct { + Name string `json:"ifname"` + AddrInfo []addrInfo `json:"addr_info"` + } + + var interfaces []iface + if err := json.Unmarshal([]byte(output), &interfaces); err != nil { + return nil + } + + var addresses []string + for _, network := range interfaces { + for _, info := range network.AddrInfo { + sanitized := sanitizeGuestAddressStrings(info.Local) + if len(sanitized) > 0 { + addresses = append(addresses, sanitized...) + } + } + } + + return addresses +} + +func parseContainerIPsFromWhitespaceList(output string) []string { + output = strings.TrimSpace(output) + if output == "" { + return nil + } + + fields := strings.Fields(output) + if len(fields) == 0 { + return nil + } + + var addresses []string + for _, field := range fields { + addresses = append(addresses, sanitizeGuestAddressStrings(field)...) + } + + return addresses +} + // GetConnectionStatuses returns the current connection status for all nodes func (m *Monitor) GetConnectionStatuses() map[string]bool { if mock.IsMockEnabled() { @@ -2531,6 +2661,7 @@ func New(cfg *config.Config) (*Monitor, error) { dockerCommands: make(map[string]*dockerHostCommand), dockerCommandIndex: make(map[string]string), guestMetadataCache: make(map[string]guestMetadataCacheEntry), + containerIPCache: make(map[string]containerIPCacheEntry), instanceInfoCache: make(map[string]*instanceInfo), pollStatusMap: make(map[string]*pollStatus), dlqInsightMap: make(map[string]*dlqInsight), diff --git a/internal/monitoring/monitor_memory_test.go b/internal/monitoring/monitor_memory_test.go index e7d8d33..0e0b08a 100644 --- a/internal/monitoring/monitor_memory_test.go +++ b/internal/monitoring/monitor_memory_test.go @@ -80,6 +80,9 @@ func (s *stubPVEClient) GetContainerStatus(ctx context.Context, node string, vmi func (s *stubPVEClient) GetContainerConfig(ctx context.Context, node string, vmid int) (map[string]interface{}, error) { return nil, nil } +func (s *stubPVEClient) ExecContainerCommand(ctx context.Context, node string, vmid int, command []string) (string, error) { + return "", nil +} func (s *stubPVEClient) GetClusterResources(ctx context.Context, resourceType string) ([]proxmox.ClusterResource, error) { return nil, nil diff --git a/internal/monitoring/monitor_snapshots_test.go b/internal/monitoring/monitor_snapshots_test.go index 3949cd4..07542bd 100644 --- a/internal/monitoring/monitor_snapshots_test.go +++ b/internal/monitoring/monitor_snapshots_test.go @@ -60,6 +60,9 @@ func (f fakeSnapshotClient) GetContainerStatus(ctx context.Context, node string, func (f fakeSnapshotClient) GetContainerConfig(ctx context.Context, node string, vmid int) (map[string]interface{}, error) { return nil, nil } +func (f fakeSnapshotClient) ExecContainerCommand(ctx context.Context, node string, vmid int, command []string) (string, error) { + return "", nil +} func (f fakeSnapshotClient) GetClusterResources(ctx context.Context, resourceType string) ([]proxmox.ClusterResource, error) { return nil, nil } diff --git a/internal/monitoring/monitor_storage_test.go b/internal/monitoring/monitor_storage_test.go index 4ea1c8d..94d088f 100644 --- a/internal/monitoring/monitor_storage_test.go +++ b/internal/monitoring/monitor_storage_test.go @@ -80,6 +80,9 @@ func (f *fakeStorageClient) GetContainerStatus(ctx context.Context, node string, func (f *fakeStorageClient) GetContainerConfig(ctx context.Context, node string, vmid int) (map[string]interface{}, error) { return nil, nil } +func (f *fakeStorageClient) ExecContainerCommand(ctx context.Context, node string, vmid int, command []string) (string, error) { + return "", nil +} func (f *fakeStorageClient) GetClusterResources(ctx context.Context, resourceType string) ([]proxmox.ClusterResource, error) { return nil, nil diff --git a/pkg/proxmox/client.go b/pkg/proxmox/client.go index 8e06d89..3cd2d1b 100644 --- a/pkg/proxmox/client.go +++ b/pkg/proxmox/client.go @@ -1008,6 +1008,148 @@ func (c *Client) GetBackupTasks(ctx context.Context) ([]Task, error) { return allTasks, nil } +type taskStatusResponse struct { + Status string `json:"status"` + ExitStatus string `json:"exitstatus"` + Type string `json:"type"` + UPID string `json:"upid"` + ID string `json:"id"` +} + +type taskLogEntry struct { + LineNumber int `json:"n"` + Text string `json:"t"` +} + +func (c *Client) getTaskStatus(ctx context.Context, node, upid string) (*taskStatusResponse, error) { + encodedUPID := url.PathEscape(upid) + resp, err := c.get(ctx, fmt.Sprintf("/nodes/%s/tasks/%s/status", node, encodedUPID)) + if err != nil { + return nil, err + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + body, _ := io.ReadAll(resp.Body) + return nil, fmt.Errorf("failed to get task status (status %d): %s", resp.StatusCode, strings.TrimSpace(string(body))) + } + + var result struct { + Data taskStatusResponse `json:"data"` + } + + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + return nil, err + } + + return &result.Data, nil +} + +func (c *Client) waitForTaskCompletion(ctx context.Context, node, upid string) (*taskStatusResponse, error) { + ticker := time.NewTicker(500 * time.Millisecond) + defer ticker.Stop() + + for { + status, err := c.getTaskStatus(ctx, node, upid) + if err != nil { + return nil, err + } + + if status != nil && strings.ToLower(status.Status) != "running" && strings.ToLower(status.Status) != "active" { + return status, nil + } + + select { + case <-ctx.Done(): + return nil, ctx.Err() + case <-ticker.C: + } + } +} + +func (c *Client) getTaskLog(ctx context.Context, node, upid string) ([]string, error) { + encodedUPID := url.PathEscape(upid) + resp, err := c.get(ctx, fmt.Sprintf("/nodes/%s/tasks/%s/log?start=0", node, encodedUPID)) + if err != nil { + return nil, err + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + body, _ := io.ReadAll(resp.Body) + return nil, fmt.Errorf("failed to get task log (status %d): %s", resp.StatusCode, strings.TrimSpace(string(body))) + } + + var result struct { + Data []taskLogEntry `json:"data"` + } + + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + return nil, err + } + + lines := make([]string, 0, len(result.Data)) + for _, entry := range result.Data { + lines = append(lines, entry.Text) + } + return lines, nil +} + +// ExecContainerCommand executes a command within an LXC container and returns the stdout output. +func (c *Client) ExecContainerCommand(ctx context.Context, node string, vmid int, command []string) (string, error) { + if len(command) == 0 { + return "", fmt.Errorf("command is required") + } + + params := url.Values{} + params.Set("command", command[0]) + for _, arg := range command[1:] { + params.Add("extra-args", arg) + } + + resp, err := c.post(ctx, fmt.Sprintf("/nodes/%s/lxc/%d/exec", node, vmid), params) + if err != nil { + return "", err + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + body, _ := io.ReadAll(resp.Body) + return "", fmt.Errorf("failed to execute container command (status %d): %s", resp.StatusCode, strings.TrimSpace(string(body))) + } + + var result struct { + Data struct { + UPID string `json:"upid"` + } `json:"data"` + } + + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + return "", err + } + + if result.Data.UPID == "" { + return "", fmt.Errorf("exec call did not return a UPID") + } + + status, err := c.waitForTaskCompletion(ctx, node, result.Data.UPID) + if err != nil { + return "", err + } + + if status != nil && status.ExitStatus != "" && !strings.EqualFold(status.ExitStatus, "ok") { + return "", fmt.Errorf("container command failed: %s", status.ExitStatus) + } + + lines, err := c.getTaskLog(ctx, node, result.Data.UPID) + if err != nil { + return "", err + } + + output := strings.Join(lines, "\n") + return strings.TrimSpace(output), nil +} + // GetStorageContent returns the content of a specific storage func (c *Client) GetStorageContent(ctx context.Context, node, storage string) ([]StorageContent, error) { // Storage content queries can take longer on large storages diff --git a/pkg/proxmox/cluster_client.go b/pkg/proxmox/cluster_client.go index 983f0b0..d873281 100644 --- a/pkg/proxmox/cluster_client.go +++ b/pkg/proxmox/cluster_client.go @@ -1017,6 +1017,20 @@ func (cc *ClusterClient) GetContainerConfig(ctx context.Context, node string, vm return result, err } +// ExecContainerCommand executes a command within an LXC container using cluster failover +func (cc *ClusterClient) ExecContainerCommand(ctx context.Context, node string, vmid int, command []string) (string, error) { + var output string + err := cc.executeWithFailover(ctx, func(client *Client) error { + result, err := client.ExecContainerCommand(ctx, node, vmid, command) + if err != nil { + return err + } + output = result + return nil + }) + return output, err +} + // IsClusterMember checks if this node is part of a cluster func (cc *ClusterClient) IsClusterMember(ctx context.Context) (bool, error) { var result bool