From 57429900a630d3e4d6a7ca8715016293594a589e Mon Sep 17 00:00:00 2001 From: rcourtman Date: Mon, 20 Oct 2025 11:10:18 +0000 Subject: [PATCH] feat: add adaptive polling scheduler infrastructure (Phase 2 Tasks 1-3) Implements adaptive scheduling foundation for Phase 2: - Poll cycle metrics: duration, staleness, queue depth, in-flight counters - Adaptive scheduler with pluggable staleness/interval/enqueue interfaces - Config support: ADAPTIVE_POLLING_ENABLED flag + min/max/base intervals - Feature flag defaults to disabled for safe rollout - Scheduler wiring into Monitor with conditional instantiation Tasks 1-3 of 10 complete. Ready for staleness tracker implementation. --- cmd/pulse/main.go | 5 + cmd/pulse/metrics_server.go | 40 ++ docker-compose.parallel.yml | 43 ++ .../src/components/Backups/UnifiedBackups.tsx | 72 +-- .../src/components/Dashboard/Dashboard.tsx | 61 +- .../src/components/Storage/Storage.tsx | 60 +- .../src/hooks/usePersistentSignal.ts | 75 +++ internal/api/config_handlers.go | 2 +- internal/config/config.go | 115 +++- internal/config/persistence.go | 4 + internal/monitoring/metrics.go | 263 +++++++++ internal/monitoring/monitor.go | 553 +++++++++++++++--- internal/monitoring/monitor_polling.go | 96 +++ internal/monitoring/monitor_storage_test.go | 2 +- internal/monitoring/scheduler.go | 289 +++++++++ 15 files changed, 1468 insertions(+), 212 deletions(-) create mode 100644 cmd/pulse/metrics_server.go create mode 100644 docker-compose.parallel.yml create mode 100644 frontend-modern/src/hooks/usePersistentSignal.ts create mode 100644 internal/monitoring/metrics.go create mode 100644 internal/monitoring/scheduler.go diff --git a/cmd/pulse/main.go b/cmd/pulse/main.go index 6af4701..1b7d58f 100644 --- a/cmd/pulse/main.go +++ b/cmd/pulse/main.go @@ -29,6 +29,8 @@ var ( GitCommit = "unknown" ) +const metricsPort = 9091 + var rootCmd = &cobra.Command{ Use: "pulse", Short: "Pulse - Proxmox VE and PBS monitoring system", @@ -91,6 +93,9 @@ func runServer() { ctx, cancel := context.WithCancel(context.Background()) defer cancel() + metricsAddr := fmt.Sprintf("%s:%d", cfg.BackendHost, metricsPort) + startMetricsServer(ctx, metricsAddr) + // Initialize WebSocket hub first wsHub := websocket.NewHub(nil) // Set allowed origins from configuration diff --git a/cmd/pulse/metrics_server.go b/cmd/pulse/metrics_server.go new file mode 100644 index 0000000..4ee5de5 --- /dev/null +++ b/cmd/pulse/metrics_server.go @@ -0,0 +1,40 @@ +package main + +import ( + "context" + "net/http" + "time" + + "github.com/prometheus/client_golang/prometheus/promhttp" + "github.com/rs/zerolog/log" +) + +func startMetricsServer(ctx context.Context, addr string) { + mux := http.NewServeMux() + mux.Handle("/metrics", promhttp.Handler()) + + srv := &http.Server{ + Addr: addr, + Handler: mux, + ReadTimeout: 5 * time.Second, + WriteTimeout: 10 * time.Second, + IdleTimeout: 30 * time.Second, + } + + go func() { + <-ctx.Done() + shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := srv.Shutdown(shutdownCtx); err != nil && err != http.ErrServerClosed { + log.Warn().Err(err).Msg("Failed to shut down metrics server cleanly") + } + }() + + go func() { + log.Info().Str("addr", addr).Msg("Metrics endpoint listening") + if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed { + log.Warn().Err(err).Msg("Metrics server stopped unexpectedly") + } + }() +} + diff --git a/docker-compose.parallel.yml b/docker-compose.parallel.yml new file mode 100644 index 0000000..6744eed --- /dev/null +++ b/docker-compose.parallel.yml @@ -0,0 +1,43 @@ +version: "3.8" + +x-codex-mcp-http: &codex_mcp_http + build: . + image: codex-mcp-server:latest + entrypoint: /bin/sh + command: > + -c 'pip install --quiet --disable-pip-version-check uvicorn starlette sse-starlette && + mkdir -p /root/.config && + cp -r /gh-config-source /root/.config/gh && + exec python /app/server.py' + volumes: + - /usr/local/bin/codex:/usr/local/bin/codex:ro + - /usr/local/lib/node_modules/@openai/codex:/usr/local/lib/node_modules/@openai/codex:ro + - /usr/local/lib/node_modules/@openai/codex/vendor:/usr/local/vendor:ro + - /tmp:/tmp + - /home/pulse/.codex:/root/.codex + - /home/pulse/.config/gh:/gh-config-source:ro + - /opt/pulse:/opt/pulse + - /opt/pulse/.mcp-servers/codex/server_http_fixed.py:/app/server.py:ro + environment: + HOME: /root + labels: + com.pulse.project: "codex-mcp" + +services: + codex-parallel-1: + <<: *codex_mcp_http + container_name: codex-parallel-1 + ports: + - "18765:8765" + + codex-parallel-2: + <<: *codex_mcp_http + container_name: codex-parallel-2 + ports: + - "18766:8765" + + codex-parallel-3: + <<: *codex_mcp_http + container_name: codex-parallel-3 + ports: + - "18767:8765" diff --git a/frontend-modern/src/components/Backups/UnifiedBackups.tsx b/frontend-modern/src/components/Backups/UnifiedBackups.tsx index 6ab83ce..e848be1 100644 --- a/frontend-modern/src/components/Backups/UnifiedBackups.tsx +++ b/frontend-modern/src/components/Backups/UnifiedBackups.tsx @@ -1,4 +1,4 @@ -import { Component, createSignal, Show, For, createMemo, createEffect, onMount } from 'solid-js'; +import { Component, createSignal, Show, For, createMemo, createEffect } from 'solid-js'; import { useNavigate } from '@solidjs/router'; import { useWebSocket } from '@/App'; import { formatBytes, formatAbsoluteTime, formatRelativeTime, formatUptime } from '@/utils/format'; @@ -12,6 +12,24 @@ import { EmptyState } from '@/components/shared/EmptyState'; import { SectionHeader } from '@/components/shared/SectionHeader'; import { showTooltip, hideTooltip } from '@/components/shared/Tooltip'; import type { BackupType, GuestType, UnifiedBackup } from '@/types/backups'; +import { usePersistentSignal } from '@/hooks/usePersistentSignal'; + +type BackupSortKey = keyof Pick< + UnifiedBackup, + 'backupTime' | 'name' | 'node' | 'vmid' | 'backupType' | 'size' | 'storage' | 'verified' | 'type' | 'owner' +>; +const BACKUP_SORT_KEY_VALUES: readonly BackupSortKey[] = [ + 'backupTime', + 'name', + 'node', + 'vmid', + 'backupType', + 'size', + 'storage', + 'verified', + 'type', + 'owner', +] as const; type FilterableGuestType = 'VM' | 'LXC' | 'Host'; @@ -52,21 +70,23 @@ const UnifiedBackups: Component = () => { else if (value === 'pve') setBackupTypeFilter('local'); else if (value === 'pbs') setBackupTypeFilter('remote'); }; - type BackupSortKey = keyof Pick< - UnifiedBackup, - | 'backupTime' - | 'name' - | 'node' - | 'vmid' - | 'backupType' - | 'size' - | 'storage' - | 'verified' - | 'type' - | 'owner' - >; - const [sortKey, setSortKey] = createSignal('backupTime'); - const [sortDirection, setSortDirection] = createSignal<'asc' | 'desc'>('desc'); + const [sortKey, setSortKey] = usePersistentSignal( + 'backupsSortKey', + 'backupTime', + { + deserialize: (raw) => + BACKUP_SORT_KEY_VALUES.includes(raw as BackupSortKey) + ? (raw as BackupSortKey) + : 'backupTime', + }, + ); + const [sortDirection, setSortDirection] = usePersistentSignal<'asc' | 'desc'>( + 'backupsSortDirection', + 'desc', + { + deserialize: (raw) => (raw === 'asc' || raw === 'desc' ? raw : 'desc'), + }, + ); const [selectedDateRange, setSelectedDateRange] = createSignal<{ start: number; end: number; @@ -89,26 +109,6 @@ const UnifiedBackups: Component = () => { { value: 'owner', label: 'Owner' }, ]; - onMount(() => { - const savedSortKey = localStorage.getItem('backupsSortKey') as BackupSortKey | null; - if (savedSortKey && sortKeyOptions.some((option) => option.value === savedSortKey)) { - setSortKey(savedSortKey); - } - - const savedSortDirection = localStorage.getItem('backupsSortDirection'); - if (savedSortDirection === 'asc' || savedSortDirection === 'desc') { - setSortDirection(savedSortDirection); - } - }); - - createEffect(() => { - localStorage.setItem('backupsSortKey', sortKey()); - }); - - createEffect(() => { - localStorage.setItem('backupsSortDirection', sortDirection()); - }); - // Extract PBS instance from search term const selectedPBSInstance = createMemo(() => { const search = searchTerm(); diff --git a/frontend-modern/src/components/Dashboard/Dashboard.tsx b/frontend-modern/src/components/Dashboard/Dashboard.tsx index 84610bd..6bb32ef 100644 --- a/frontend-modern/src/components/Dashboard/Dashboard.tsx +++ b/frontend-modern/src/components/Dashboard/Dashboard.tsx @@ -17,6 +17,7 @@ import { NodeGroupHeader } from '@/components/shared/NodeGroupHeader'; import { ProxmoxSectionNav } from '@/components/Proxmox/ProxmoxSectionNav'; import { isNodeOnline } from '@/utils/status'; import { getNodeDisplayName } from '@/utils/nodes'; +import { usePersistentSignal } from '@/hooks/usePersistentSignal'; interface DashboardProps { vms: VM[]; @@ -38,32 +39,31 @@ export function Dashboard(props: DashboardProps) { const [guestMetadata, setGuestMetadata] = createSignal>({}); // Initialize from localStorage with proper type checking - const storedViewMode = localStorage.getItem('dashboardViewMode'); - const [viewMode, setViewMode] = createSignal( - storedViewMode === 'all' || storedViewMode === 'vm' || storedViewMode === 'lxc' - ? storedViewMode - : 'all', - ); + const [viewMode, setViewMode] = usePersistentSignal('dashboardViewMode', 'all', { + deserialize: (raw) => (raw === 'all' || raw === 'vm' || raw === 'lxc' ? raw : 'all'), + }); - const storedStatusMode = localStorage.getItem('dashboardStatusMode'); - const [statusMode, setStatusMode] = createSignal( - storedStatusMode === 'all' || storedStatusMode === 'running' || storedStatusMode === 'stopped' - ? storedStatusMode - : 'all', - ); + const [statusMode, setStatusMode] = usePersistentSignal('dashboardStatusMode', 'all', { + deserialize: (raw) => + raw === 'all' || raw === 'running' || raw === 'stopped' ? raw : ('all' as StatusMode), + }); // Grouping mode - grouped by node or flat list - const storedGroupingMode = localStorage.getItem('dashboardGroupingMode'); - const [groupingMode, setGroupingMode] = createSignal( - storedGroupingMode === 'grouped' || storedGroupingMode === 'flat' - ? storedGroupingMode - : 'grouped', + const [groupingMode, setGroupingMode] = usePersistentSignal( + 'dashboardGroupingMode', + 'grouped', + { + deserialize: (raw) => (raw === 'grouped' || raw === 'flat' ? raw : 'grouped'), + }, ); - const [showFilters, setShowFilters] = createSignal( - localStorage.getItem('dashboardShowFilters') !== null - ? localStorage.getItem('dashboardShowFilters') === 'true' - : false, // Default to collapsed + const [showFilters, setShowFilters] = usePersistentSignal( + 'dashboardShowFilters', + false, + { + deserialize: (raw) => raw === 'true', + serialize: (value) => String(value), + }, ); // Sorting state - default to VMID ascending (matches Proxmox order) @@ -112,25 +112,6 @@ export function Dashboard(props: DashboardProps) { return undefined; }; - - - // Persist filter states to localStorage - createEffect(() => { - localStorage.setItem('dashboardViewMode', viewMode()); - }); - - createEffect(() => { - localStorage.setItem('dashboardStatusMode', statusMode()); - }); - - createEffect(() => { - localStorage.setItem('dashboardGroupingMode', groupingMode()); - }); - - createEffect(() => { - localStorage.setItem('dashboardShowFilters', showFilters().toString()); - }); - // Sort handler const handleSort = (key: keyof (VM | Container)) => { if (sortKey() === key) { diff --git a/frontend-modern/src/components/Storage/Storage.tsx b/frontend-modern/src/components/Storage/Storage.tsx index c80a2b3..210dc40 100644 --- a/frontend-modern/src/components/Storage/Storage.tsx +++ b/frontend-modern/src/components/Storage/Storage.tsx @@ -1,4 +1,4 @@ -import { Component, For, Show, createSignal, createMemo, createEffect, onMount } from 'solid-js'; +import { Component, For, Show, createSignal, createMemo, createEffect } from 'solid-js'; import { useNavigate } from '@solidjs/router'; import { useWebSocket } from '@/App'; import { getAlertStyles } from '@/utils/alerts'; @@ -13,18 +13,39 @@ import { EmptyState } from '@/components/shared/EmptyState'; import { NodeGroupHeader } from '@/components/shared/NodeGroupHeader'; import { ProxmoxSectionNav } from '@/components/Proxmox/ProxmoxSectionNav'; import { getNodeDisplayName } from '@/utils/nodes'; +import { usePersistentSignal } from '@/hooks/usePersistentSignal'; + +type StorageSortKey = 'name' | 'node' | 'type' | 'status' | 'usage' | 'free' | 'total'; const Storage: Component = () => { const navigate = useNavigate(); const { state, connected, activeAlerts, initialDataReceived } = useWebSocket(); - const [viewMode, setViewMode] = createSignal<'node' | 'storage'>('node'); + const [viewMode, setViewMode] = usePersistentSignal<'node' | 'storage'>( + 'storageViewMode', + 'node', + { + deserialize: (raw) => (raw === 'storage' ? 'storage' : 'node'), + }, + ); const [tabView, setTabView] = createSignal<'pools' | 'disks'>('pools'); const [searchTerm, setSearchTerm] = createSignal(''); const [selectedNode, setSelectedNode] = createSignal(null); const [expandedStorage, setExpandedStorage] = createSignal(null); - type StorageSortKey = 'name' | 'node' | 'type' | 'status' | 'usage' | 'free' | 'total'; - const [sortKey, setSortKey] = createSignal('name'); - const [sortDirection, setSortDirection] = createSignal<'asc' | 'desc'>('asc'); + const [sortKey, setSortKey] = usePersistentSignal('storageSortKey', 'name', { + deserialize: (raw) => + (['name', 'node', 'type', 'status', 'usage', 'free', 'total'] as const).includes( + raw as StorageSortKey, + ) + ? (raw as StorageSortKey) + : 'name', + }); + const [sortDirection, setSortDirection] = usePersistentSignal<'asc' | 'desc'>( + 'storageSortDirection', + 'asc', + { + deserialize: (raw) => (raw === 'desc' ? 'desc' : 'asc'), + }, + ); // Create a mapping from node instance ID to node object const nodeByInstance = createMemo(() => { @@ -158,35 +179,6 @@ const Storage: Component = () => { { value: 'total', label: 'Total Capacity' }, ]; - // Load preferences from localStorage - onMount(() => { - const savedViewMode = localStorage.getItem('storageViewMode'); - if (savedViewMode === 'storage') setViewMode('storage'); - - const savedSortKey = localStorage.getItem('storageSortKey') as StorageSortKey | null; - if (savedSortKey && sortKeyOptions.some((option) => option.value === savedSortKey)) { - setSortKey(savedSortKey); - } - - const savedSortDirection = localStorage.getItem('storageSortDirection'); - if (savedSortDirection === 'desc' || savedSortDirection === 'asc') { - setSortDirection(savedSortDirection); - } - }); - - // Persist preferences - createEffect(() => { - localStorage.setItem('storageViewMode', viewMode()); - }); - - createEffect(() => { - localStorage.setItem('storageSortKey', sortKey()); - }); - - createEffect(() => { - localStorage.setItem('storageSortDirection', sortDirection()); - }); - // Filter storage - in storage view, filter out 0 capacity and deduplicate const filteredStorage = createMemo(() => { let storage = state.storage || []; diff --git a/frontend-modern/src/hooks/usePersistentSignal.ts b/frontend-modern/src/hooks/usePersistentSignal.ts new file mode 100644 index 0000000..7f3a0e9 --- /dev/null +++ b/frontend-modern/src/hooks/usePersistentSignal.ts @@ -0,0 +1,75 @@ +import { Accessor, Setter, createEffect, createSignal } from 'solid-js'; + +export type PersistentSignalOptions = { + /** + * Custom serialization function. Defaults to `String(value)`. + */ + serialize?: (value: T) => string; + /** + * Custom deserialization function. Defaults to casting the stored string. + */ + deserialize?: (value: string) => T; + /** + * Optional equality comparison passed to Solid's `createSignal`. + */ + equals?: (prev: T, next: T) => boolean; + /** + * Alternate storage implementation (defaults to `window.localStorage`). + */ + storage?: Storage; +}; + +/** + * Creates a Solid signal that persists its value to localStorage (or a custom storage). + * The signal reads the initial value synchronously from storage when available. + */ +export function usePersistentSignal( + key: string, + defaultValue: T, + options: PersistentSignalOptions = {}, +): [Accessor, Setter] { + const storage = + options.storage ?? (typeof window !== 'undefined' ? window.localStorage : undefined); + const serialize = options.serialize ?? ((value: T) => String(value)); + const deserialize = options.deserialize ?? ((value: string) => value as unknown as T); + + const initialValue = (() => { + if (!storage) { + return defaultValue; + } + + try { + const raw = storage.getItem(key); + if (raw === null) { + return defaultValue; + } + return deserialize(raw); + } catch (err) { + console.warn(`[usePersistentSignal] Failed to read "${key}" from storage`, err); + return defaultValue; + } + })(); + + const signalOptions = options.equals ? { equals: options.equals } : undefined; + const [value, setValue] = createSignal(initialValue, signalOptions); + + createEffect(() => { + if (!storage) { + return; + } + + const current = value(); + try { + if (current === undefined || current === null) { + storage.removeItem(key); + } else { + storage.setItem(key, serialize(current)); + } + } catch (err) { + console.warn(`[usePersistentSignal] Failed to persist "${key}"`, err); + } + }); + + return [value, setValue]; +} + diff --git a/internal/api/config_handlers.go b/internal/api/config_handlers.go index f49bd06..7584bf5 100644 --- a/internal/api/config_handlers.go +++ b/internal/api/config_handlers.go @@ -4318,7 +4318,7 @@ if [ "$AUTO_REG_SUCCESS" != true ]; then fi `, serverName, time.Now().Format("2006-01-02 15:04:05"), pulseIP, tokenName, tokenName, tokenName, tokenName, tokenName, tokenName, - authToken, pulseURL, serverHost, tokenName, tokenName, storagePerms, sshKeys.ProxyPublicKey, sshKeys.SensorsPublicKey, pulseURL, pulseURL, pulseURL, pulseURL, authToken, pulseURL, authToken) + authToken, pulseURL, serverHost, tokenName, tokenName, storagePerms, sshKeys.ProxyPublicKey, sshKeys.SensorsPublicKey, pulseURL, pulseURL, pulseURL, pulseURL, pulseURL, pulseURL, authToken, pulseURL, authToken, pulseURL, tokenName) } else { // PBS script = fmt.Sprintf(`#!/bin/bash diff --git a/internal/config/config.go b/internal/config/config.go index 914b67d..d610185 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -83,6 +83,10 @@ type Config struct { BackupPollingInterval time.Duration `envconfig:"BACKUP_POLLING_INTERVAL"` EnableBackupPolling bool `envconfig:"ENABLE_BACKUP_POLLING" default:"true"` WebhookBatchDelay time.Duration `envconfig:"WEBHOOK_BATCH_DELAY" default:"10s"` + AdaptivePollingEnabled bool `envconfig:"ADAPTIVE_POLLING_ENABLED" default:"false"` + AdaptivePollingBaseInterval time.Duration `envconfig:"ADAPTIVE_POLLING_BASE_INTERVAL" default:"10s"` + AdaptivePollingMinInterval time.Duration `envconfig:"ADAPTIVE_POLLING_MIN_INTERVAL" default:"5s"` + AdaptivePollingMaxInterval time.Duration `envconfig:"ADAPTIVE_POLLING_MAX_INTERVAL" default:"5m"` // Logging settings LogLevel string `envconfig:"LOG_LEVEL" default:"info"` @@ -262,6 +266,10 @@ func Load() (*Config, error) { BackupPollingInterval: 0, EnableBackupPolling: true, WebhookBatchDelay: 10 * time.Second, + AdaptivePollingEnabled: false, + AdaptivePollingBaseInterval: 10 * time.Second, + AdaptivePollingMinInterval: 5 * time.Second, + AdaptivePollingMaxInterval: 5 * time.Minute, LogLevel: "info", LogMaxSize: 100, LogMaxAge: 30, @@ -305,14 +313,26 @@ func Load() (*Config, error) { cfg.PMGPollingInterval = time.Duration(systemSettings.PMGPollingInterval) * time.Second } - if systemSettings.BackupPollingInterval > 0 { - cfg.BackupPollingInterval = time.Duration(systemSettings.BackupPollingInterval) * time.Second - } else if systemSettings.BackupPollingInterval == 0 { - cfg.BackupPollingInterval = 0 - } - if systemSettings.BackupPollingEnabled != nil { - cfg.EnableBackupPolling = *systemSettings.BackupPollingEnabled - } + if systemSettings.BackupPollingInterval > 0 { + cfg.BackupPollingInterval = time.Duration(systemSettings.BackupPollingInterval) * time.Second + } else if systemSettings.BackupPollingInterval == 0 { + cfg.BackupPollingInterval = 0 + } + if systemSettings.BackupPollingEnabled != nil { + cfg.EnableBackupPolling = *systemSettings.BackupPollingEnabled + } + if systemSettings.AdaptivePollingEnabled != nil { + cfg.AdaptivePollingEnabled = *systemSettings.AdaptivePollingEnabled + } + if systemSettings.AdaptivePollingBaseInterval > 0 { + cfg.AdaptivePollingBaseInterval = time.Duration(systemSettings.AdaptivePollingBaseInterval) * time.Second + } + if systemSettings.AdaptivePollingMinInterval > 0 { + cfg.AdaptivePollingMinInterval = time.Duration(systemSettings.AdaptivePollingMinInterval) * time.Second + } + if systemSettings.AdaptivePollingMaxInterval > 0 { + cfg.AdaptivePollingMaxInterval = time.Duration(systemSettings.AdaptivePollingMaxInterval) * time.Second + } if systemSettings.UpdateChannel != "" { cfg.UpdateChannel = systemSettings.UpdateChannel @@ -430,6 +450,47 @@ func Load() (*Config, error) { log.Info().Bool("enabled", cfg.EnableBackupPolling).Msg("Overriding backup polling enabled flag from environment") } + if adaptiveEnabled := strings.TrimSpace(os.Getenv("ADAPTIVE_POLLING_ENABLED")); adaptiveEnabled != "" { + switch strings.ToLower(adaptiveEnabled) { + case "0", "false", "no", "off": + cfg.AdaptivePollingEnabled = false + default: + cfg.AdaptivePollingEnabled = true + } + cfg.EnvOverrides["ADAPTIVE_POLLING_ENABLED"] = true + log.Info().Bool("enabled", cfg.AdaptivePollingEnabled).Msg("Adaptive polling feature flag overridden by environment") + } + + if baseInterval := strings.TrimSpace(os.Getenv("ADAPTIVE_POLLING_BASE_INTERVAL")); baseInterval != "" { + if dur, err := time.ParseDuration(baseInterval); err == nil { + cfg.AdaptivePollingBaseInterval = dur + cfg.EnvOverrides["ADAPTIVE_POLLING_BASE_INTERVAL"] = true + log.Info().Dur("interval", dur).Msg("Adaptive polling base interval overridden by environment") + } else { + log.Warn().Str("value", baseInterval).Msg("Invalid ADAPTIVE_POLLING_BASE_INTERVAL value, expected duration string") + } + } + + if minInterval := strings.TrimSpace(os.Getenv("ADAPTIVE_POLLING_MIN_INTERVAL")); minInterval != "" { + if dur, err := time.ParseDuration(minInterval); err == nil { + cfg.AdaptivePollingMinInterval = dur + cfg.EnvOverrides["ADAPTIVE_POLLING_MIN_INTERVAL"] = true + log.Info().Dur("interval", dur).Msg("Adaptive polling min interval overridden by environment") + } else { + log.Warn().Str("value", minInterval).Msg("Invalid ADAPTIVE_POLLING_MIN_INTERVAL value, expected duration string") + } + } + + if maxInterval := strings.TrimSpace(os.Getenv("ADAPTIVE_POLLING_MAX_INTERVAL")); maxInterval != "" { + if dur, err := time.ParseDuration(maxInterval); err == nil { + cfg.AdaptivePollingMaxInterval = dur + cfg.EnvOverrides["ADAPTIVE_POLLING_MAX_INTERVAL"] = true + log.Info().Dur("interval", dur).Msg("Adaptive polling max interval overridden by environment") + } else { + log.Warn().Str("value", maxInterval).Msg("Invalid ADAPTIVE_POLLING_MAX_INTERVAL value, expected duration string") + } + } + // Support both FRONTEND_PORT (preferred) and PORT (legacy) env vars if frontendPort := os.Getenv("FRONTEND_PORT"); frontendPort != "" { if p, err := strconv.Atoi(frontendPort); err == nil { @@ -744,17 +805,22 @@ func SaveConfig(cfg *Config) error { } // Save system configuration + adaptiveEnabled := cfg.AdaptivePollingEnabled systemSettings := SystemSettings{ // Note: PVE polling is hardcoded to 10s - UpdateChannel: cfg.UpdateChannel, - AutoUpdateEnabled: cfg.AutoUpdateEnabled, - AutoUpdateCheckInterval: int(cfg.AutoUpdateCheckInterval.Hours()), - AutoUpdateTime: cfg.AutoUpdateTime, - AllowedOrigins: cfg.AllowedOrigins, - ConnectionTimeout: int(cfg.ConnectionTimeout.Seconds()), - LogLevel: cfg.LogLevel, - DiscoveryEnabled: cfg.DiscoveryEnabled, - DiscoverySubnet: cfg.DiscoverySubnet, + UpdateChannel: cfg.UpdateChannel, + AutoUpdateEnabled: cfg.AutoUpdateEnabled, + AutoUpdateCheckInterval: int(cfg.AutoUpdateCheckInterval.Hours()), + AutoUpdateTime: cfg.AutoUpdateTime, + AllowedOrigins: cfg.AllowedOrigins, + ConnectionTimeout: int(cfg.ConnectionTimeout.Seconds()), + LogLevel: cfg.LogLevel, + DiscoveryEnabled: cfg.DiscoveryEnabled, + DiscoverySubnet: cfg.DiscoverySubnet, + AdaptivePollingEnabled: &adaptiveEnabled, + AdaptivePollingBaseInterval: int(cfg.AdaptivePollingBaseInterval / time.Second), + AdaptivePollingMinInterval: int(cfg.AdaptivePollingMinInterval / time.Second), + AdaptivePollingMaxInterval: int(cfg.AdaptivePollingMaxInterval / time.Second), // APIToken removed - now handled via .env only } if err := globalPersistence.SaveSystemSettings(systemSettings); err != nil { @@ -796,6 +862,21 @@ func (c *Config) Validate() error { if c.ConnectionTimeout < time.Second { return fmt.Errorf("connection timeout must be at least 1 second") } + if c.AdaptivePollingMinInterval <= 0 { + return fmt.Errorf("adaptive polling min interval must be greater than 0") + } + if c.AdaptivePollingBaseInterval <= 0 { + return fmt.Errorf("adaptive polling base interval must be greater than 0") + } + if c.AdaptivePollingMaxInterval <= 0 { + return fmt.Errorf("adaptive polling max interval must be greater than 0") + } + if c.AdaptivePollingMinInterval > c.AdaptivePollingMaxInterval { + return fmt.Errorf("adaptive polling min interval cannot exceed max interval") + } + if c.AdaptivePollingBaseInterval < c.AdaptivePollingMinInterval || c.AdaptivePollingBaseInterval > c.AdaptivePollingMaxInterval { + return fmt.Errorf("adaptive polling base interval must be between min and max intervals") + } // Validate PVE instances for i, pve := range c.PVEInstances { diff --git a/internal/config/persistence.go b/internal/config/persistence.go index f2a87ad..d72f5dc 100644 --- a/internal/config/persistence.go +++ b/internal/config/persistence.go @@ -667,6 +667,10 @@ type SystemSettings struct { PMGPollingInterval int `json:"pmgPollingInterval"` // PMG polling interval in seconds BackupPollingInterval int `json:"backupPollingInterval,omitempty"` BackupPollingEnabled *bool `json:"backupPollingEnabled,omitempty"` + AdaptivePollingEnabled *bool `json:"adaptivePollingEnabled,omitempty"` + AdaptivePollingBaseInterval int `json:"adaptivePollingBaseInterval,omitempty"` + AdaptivePollingMinInterval int `json:"adaptivePollingMinInterval,omitempty"` + AdaptivePollingMaxInterval int `json:"adaptivePollingMaxInterval,omitempty"` BackendPort int `json:"backendPort,omitempty"` FrontendPort int `json:"frontendPort,omitempty"` AllowedOrigins string `json:"allowedOrigins,omitempty"` diff --git a/internal/monitoring/metrics.go b/internal/monitoring/metrics.go new file mode 100644 index 0000000..51d8710 --- /dev/null +++ b/internal/monitoring/metrics.go @@ -0,0 +1,263 @@ +package monitoring + +import ( + stdErrors "errors" + "fmt" + "sync" + "time" + + internalerrors "github.com/rcourtman/pulse-go-rewrite/internal/errors" + "github.com/prometheus/client_golang/prometheus" +) + +// PollMetrics manages Prometheus instrumentation for polling activity. +type PollMetrics struct { + pollDuration *prometheus.HistogramVec + pollResults *prometheus.CounterVec + pollErrors *prometheus.CounterVec + lastSuccess *prometheus.GaugeVec + staleness *prometheus.GaugeVec + queueDepth prometheus.Gauge + inflight *prometheus.GaugeVec + + mu sync.RWMutex + lastSuccessByKey map[string]time.Time + pending int +} + +var ( + pollMetricsInstance *PollMetrics + pollMetricsOnce sync.Once +) + +func getPollMetrics() *PollMetrics { + pollMetricsOnce.Do(func() { + pollMetricsInstance = newPollMetrics() + }) + return pollMetricsInstance +} + +func newPollMetrics() *PollMetrics { + pm := &PollMetrics{ + pollDuration: prometheus.NewHistogramVec( + prometheus.HistogramOpts{ + Namespace: "pulse", + Subsystem: "monitor", + Name: "poll_duration_seconds", + Help: "Duration of polling operations per instance.", + Buckets: []float64{0.1, 0.25, 0.5, 1, 2.5, 5, 10, 15, 20, 30}, + }, + []string{"instance_type", "instance"}, + ), + pollResults: prometheus.NewCounterVec( + prometheus.CounterOpts{ + Namespace: "pulse", + Subsystem: "monitor", + Name: "poll_total", + Help: "Total polling attempts partitioned by result.", + }, + []string{"instance_type", "instance", "result"}, + ), + pollErrors: prometheus.NewCounterVec( + prometheus.CounterOpts{ + Namespace: "pulse", + Subsystem: "monitor", + Name: "poll_errors_total", + Help: "Polling failures grouped by error type.", + }, + []string{"instance_type", "instance", "error_type"}, + ), + lastSuccess: prometheus.NewGaugeVec( + prometheus.GaugeOpts{ + Namespace: "pulse", + Subsystem: "monitor", + Name: "poll_last_success_timestamp", + Help: "Unix timestamp of the last successful poll.", + }, + []string{"instance_type", "instance"}, + ), + staleness: prometheus.NewGaugeVec( + prometheus.GaugeOpts{ + Namespace: "pulse", + Subsystem: "monitor", + Name: "poll_staleness_seconds", + Help: "Seconds since the last successful poll. -1 indicates no successes yet.", + }, + []string{"instance_type", "instance"}, + ), + queueDepth: prometheus.NewGauge( + prometheus.GaugeOpts{ + Namespace: "pulse", + Subsystem: "monitor", + Name: "poll_queue_depth", + Help: "Approximate number of poll tasks waiting to complete in the current cycle.", + }, + ), + inflight: prometheus.NewGaugeVec( + prometheus.GaugeOpts{ + Namespace: "pulse", + Subsystem: "monitor", + Name: "poll_inflight", + Help: "Current number of poll operations executing per instance type.", + }, + []string{"instance_type"}, + ), + lastSuccessByKey: make(map[string]time.Time), + } + + prometheus.MustRegister( + pm.pollDuration, + pm.pollResults, + pm.pollErrors, + pm.lastSuccess, + pm.staleness, + pm.queueDepth, + pm.inflight, + ) + + return pm +} + +// RecordResult records metrics for a polling result. +func (pm *PollMetrics) RecordResult(result PollResult) { + if pm == nil { + return + } + + labels := prometheus.Labels{ + "instance_type": result.InstanceType, + "instance": result.InstanceName, + } + + duration := result.EndTime.Sub(result.StartTime).Seconds() + if duration < 0 { + duration = 0 + } + pm.pollDuration.With(labels).Observe(duration) + + resultValue := "success" + if !result.Success { + resultValue = "error" + } + pm.pollResults.With(prometheus.Labels{ + "instance_type": result.InstanceType, + "instance": result.InstanceName, + "result": resultValue, + }).Inc() + + if result.Success { + pm.lastSuccess.With(labels).Set(float64(result.EndTime.Unix())) + pm.storeLastSuccess(result.InstanceType, result.InstanceName, result.EndTime) + pm.updateStaleness(result.InstanceType, result.InstanceName, 0) + } else { + errType := pm.classifyError(result.Error) + pm.pollErrors.With(prometheus.Labels{ + "instance_type": result.InstanceType, + "instance": result.InstanceName, + "error_type": errType, + }).Inc() + + if last, ok := pm.lastSuccessFor(result.InstanceType, result.InstanceName); ok && !last.IsZero() { + staleness := result.EndTime.Sub(last).Seconds() + if staleness < 0 { + staleness = 0 + } + pm.updateStaleness(result.InstanceType, result.InstanceName, staleness) + } else { + pm.updateStaleness(result.InstanceType, result.InstanceName, -1) + } + } + + pm.decrementPending() +} + +// ResetQueueDepth sets the pending queue depth for the next polling cycle. +func (pm *PollMetrics) ResetQueueDepth(total int) { + if pm == nil { + return + } + if total < 0 { + total = 0 + } + + pm.mu.Lock() + pm.pending = total + pm.mu.Unlock() + pm.queueDepth.Set(float64(total)) +} + +// SetQueueDepth allows direct gauge control when needed. +func (pm *PollMetrics) SetQueueDepth(depth int) { + if pm == nil { + return + } + if depth < 0 { + depth = 0 + } + pm.queueDepth.Set(float64(depth)) +} + +// IncInFlight increments the in-flight gauge for the given instance type. +func (pm *PollMetrics) IncInFlight(instanceType string) { + if pm == nil { + return + } + pm.inflight.WithLabelValues(instanceType).Inc() +} + +// DecInFlight decrements the in-flight gauge for the given instance type. +func (pm *PollMetrics) DecInFlight(instanceType string) { + if pm == nil { + return + } + pm.inflight.WithLabelValues(instanceType).Dec() +} + +func (pm *PollMetrics) decrementPending() { + if pm == nil { + return + } + + pm.mu.Lock() + if pm.pending > 0 { + pm.pending-- + } + current := pm.pending + pm.mu.Unlock() + + pm.queueDepth.Set(float64(current)) +} + +func (pm *PollMetrics) storeLastSuccess(instanceType, instance string, ts time.Time) { + pm.mu.Lock() + pm.lastSuccessByKey[pm.key(instanceType, instance)] = ts + pm.mu.Unlock() +} + +func (pm *PollMetrics) lastSuccessFor(instanceType, instance string) (time.Time, bool) { + pm.mu.RLock() + ts, ok := pm.lastSuccessByKey[pm.key(instanceType, instance)] + pm.mu.RUnlock() + return ts, ok +} + +func (pm *PollMetrics) updateStaleness(instanceType, instance string, value float64) { + pm.staleness.WithLabelValues(instanceType, instance).Set(value) +} + +func (pm *PollMetrics) key(instanceType, instance string) string { + return fmt.Sprintf("%s::%s", instanceType, instance) +} + +func (pm *PollMetrics) classifyError(err error) string { + if err == nil { + return "none" + } + + var monitorErr *internalerrors.MonitorError + if stdErrors.As(err, &monitorErr) { + return string(monitorErr.Type) + } + + return "unknown" +} diff --git a/internal/monitoring/monitor.go b/internal/monitoring/monitor.go index bc9b288..39e7074 100644 --- a/internal/monitoring/monitor.go +++ b/internal/monitoring/monitor.go @@ -255,6 +255,8 @@ type Monitor struct { pveClients map[string]PVEClientInterface pbsClients map[string]*pbs.Client pmgClients map[string]*pmg.Client + pollMetrics *PollMetrics + scheduler *AdaptiveScheduler tempCollector *TemperatureCollector // SSH-based temperature collector mu sync.RWMutex startTime time.Time @@ -1312,12 +1314,23 @@ func New(cfg *config.Config) (*Monitor, error) { // Security warning if running in container with SSH temperature monitoring checkContainerizedTempMonitoring() + var scheduler *AdaptiveScheduler + if cfg.AdaptivePollingEnabled { + scheduler = NewAdaptiveScheduler(SchedulerConfig{ + BaseInterval: cfg.AdaptivePollingBaseInterval, + MinInterval: cfg.AdaptivePollingMinInterval, + MaxInterval: cfg.AdaptivePollingMaxInterval, + }, nil, nil, nil) + } + m := &Monitor{ config: cfg, state: models.NewState(), pveClients: make(map[string]PVEClientInterface), pbsClients: make(map[string]*pbs.Client), pmgClients: make(map[string]*pmg.Client), + pollMetrics: getPollMetrics(), + scheduler: scheduler, tempCollector: tempCollector, startTime: time.Now(), rateTracker: NewRateTracker(), @@ -1343,6 +1356,10 @@ func New(cfg *config.Config) (*Monitor, error) { guestMetadataCache: make(map[string]guestMetadataCacheEntry), } + if m.pollMetrics != nil { + m.pollMetrics.ResetQueueDepth(0) + } + // Load saved configurations if alertConfig, err := m.configPersist.LoadAlertConfig(); err == nil { m.alertManager.UpdateConfig(*alertConfig) @@ -1909,12 +1926,26 @@ func (m *Monitor) poll(ctx context.Context, wsHub *websocket.Hub) { log.Debug().Msg("Starting polling cycle") startTime := time.Now() + now := startTime + + plannedTasks := m.buildScheduledTasks(now) + dueTasks := plannedTasks + if m.scheduler != nil { + due := m.scheduler.DispatchDue(ctx, now, plannedTasks) + if len(due) > 0 { + dueTasks = due + } + } + + if m.pollMetrics != nil { + m.pollMetrics.ResetQueueDepth(len(dueTasks)) + } if m.config.ConcurrentPolling { // Use concurrent polling - m.pollConcurrent(ctx) + m.pollConcurrent(ctx, dueTasks) } else { - m.pollSequential(ctx) + m.pollSequential(ctx, dueTasks) } // Update performance metrics @@ -2052,62 +2083,59 @@ func (m *Monitor) pruneStaleDockerAlerts() bool { return cleared } -// pollConcurrent polls all instances concurrently -func (m *Monitor) pollConcurrent(ctx context.Context) { +// pollConcurrent polls instances concurrently based on scheduled tasks. +func (m *Monitor) pollConcurrent(ctx context.Context, tasks []ScheduledTask) { + if len(tasks) == 0 { + return + } + var wg sync.WaitGroup ctx, cancel := context.WithCancel(ctx) defer cancel() - // Poll PVE instances - for name, client := range m.pveClients { - // Check if context is already cancelled before starting + for _, task := range tasks { select { case <-ctx.Done(): return default: } - wg.Add(1) - go func(instanceName string, c PVEClientInterface) { - defer wg.Done() - // Pass context to ensure cancellation propagates - m.pollPVEInstance(ctx, instanceName, c) - }(name, client) - } - - // Poll PBS instances - for name, client := range m.pbsClients { - // Check if context is already cancelled before starting - select { - case <-ctx.Done(): - return + switch task.InstanceType { + case InstanceTypePVE: + client, ok := m.pveClients[task.InstanceName] + if !ok || client == nil { + continue + } + wg.Add(1) + go func(name string, c PVEClientInterface) { + defer wg.Done() + m.pollPVEInstance(ctx, name, c) + }(task.InstanceName, client) + case InstanceTypePBS: + client, ok := m.pbsClients[task.InstanceName] + if !ok || client == nil { + continue + } + wg.Add(1) + go func(name string, c *pbs.Client) { + defer wg.Done() + m.pollPBSInstance(ctx, name, c) + }(task.InstanceName, client) + case InstanceTypePMG: + client, ok := m.pmgClients[task.InstanceName] + if !ok || client == nil { + continue + } + wg.Add(1) + go func(name string, c *pmg.Client) { + defer wg.Done() + m.pollPMGInstance(ctx, name, c) + }(task.InstanceName, client) default: + log.Debug().Str("instance", task.InstanceName).Str("type", string(task.InstanceType)).Msg("Skipping unsupported task type") } - - wg.Add(1) - go func(instanceName string, c *pbs.Client) { - defer wg.Done() - // Pass context to ensure cancellation propagates - m.pollPBSInstance(ctx, instanceName, c) - }(name, client) } - // Poll PMG instances - for name, client := range m.pmgClients { - select { - case <-ctx.Done(): - return - default: - } - - wg.Add(1) - go func(instanceName string, c *pmg.Client) { - defer wg.Done() - m.pollPMGInstance(ctx, instanceName, c) - }(name, client) - } - - // Wait for all goroutines to complete or context cancellation done := make(chan struct{}) go func() { wg.Wait() @@ -2116,55 +2144,63 @@ func (m *Monitor) pollConcurrent(ctx context.Context) { select { case <-done: - // All goroutines completed normally case <-ctx.Done(): - // Context cancelled, cancel all operations cancel() - // Still wait for goroutines to finish gracefully wg.Wait() } } -// pollSequential polls all instances sequentially -func (m *Monitor) pollSequential(ctx context.Context) { - // Poll PVE instances - for name, client := range m.pveClients { - // Check context before each instance +// pollSequential polls instances sequentially based on scheduled tasks. +func (m *Monitor) pollSequential(ctx context.Context, tasks []ScheduledTask) { + for _, task := range tasks { select { case <-ctx.Done(): return default: } - m.pollPVEInstance(ctx, name, client) - } - // Poll PBS instances - for name, client := range m.pbsClients { - // Check context before each instance - select { - case <-ctx.Done(): - return + switch task.InstanceType { + case InstanceTypePVE: + if client, ok := m.pveClients[task.InstanceName]; ok && client != nil { + m.pollPVEInstance(ctx, task.InstanceName, client) + } + case InstanceTypePBS: + if client, ok := m.pbsClients[task.InstanceName]; ok && client != nil { + m.pollPBSInstance(ctx, task.InstanceName, client) + } + case InstanceTypePMG: + if client, ok := m.pmgClients[task.InstanceName]; ok && client != nil { + m.pollPMGInstance(ctx, task.InstanceName, client) + } default: + log.Debug().Str("instance", task.InstanceName).Str("type", string(task.InstanceType)).Msg("Skipping unsupported task type") } - m.pollPBSInstance(ctx, name, client) - } - - // Poll PMG instances - for name, client := range m.pmgClients { - select { - case <-ctx.Done(): - return - default: - } - m.pollPMGInstance(ctx, name, client) } } // pollPVEInstance polls a single PVE instance func (m *Monitor) pollPVEInstance(ctx context.Context, instanceName string, client PVEClientInterface) { + start := time.Now() + var pollErr error + if m.pollMetrics != nil { + m.pollMetrics.IncInFlight("pve") + defer m.pollMetrics.DecInFlight("pve") + defer func() { + m.pollMetrics.RecordResult(PollResult{ + InstanceName: instanceName, + InstanceType: "pve", + Success: pollErr == nil, + Error: pollErr, + StartTime: start, + EndTime: time.Now(), + }) + }() + } + // Check if context is cancelled select { case <-ctx.Done(): + pollErr = ctx.Err() log.Debug().Str("instance", instanceName).Msg("Polling cancelled") return default: @@ -2181,6 +2217,7 @@ func (m *Monitor) pollPVEInstance(ctx context.Context, instanceName string, clie } } if instanceCfg == nil { + pollErr = fmt.Errorf("pve instance config not found for %s", instanceName) return } @@ -2188,6 +2225,7 @@ func (m *Monitor) pollPVEInstance(ctx context.Context, instanceName string, clie nodes, err := client.GetNodes(ctx) if err != nil { monErr := errors.WrapConnectionError("poll_nodes", instanceName, err) + pollErr = monErr log.Error().Err(monErr).Str("instance", instanceName).Msg("Failed to get nodes") m.state.SetConnectionHealth(instanceName, false) @@ -3068,6 +3106,7 @@ func (m *Monitor) pollPVEInstance(ctx context.Context, instanceName string, clie if instanceCfg.MonitorVMs || instanceCfg.MonitorContainers { select { case <-ctx.Done(): + pollErr = ctx.Err() return default: // Always try the efficient cluster/resources endpoint first @@ -3108,6 +3147,7 @@ func (m *Monitor) pollPVEInstance(ctx context.Context, instanceName string, clie if instanceCfg.MonitorStorage { select { case <-ctx.Done(): + pollErr = ctx.Err() return default: m.pollStorageWithNodes(ctx, instanceName, client, nodes) @@ -3128,18 +3168,19 @@ func (m *Monitor) pollPVEInstance(ctx context.Context, instanceName string, clie m.mu.RUnlock() shouldPoll, reason, newLast := m.shouldRunBackupPoll(lastPoll, now) - if !shouldPoll { - if reason != "" { - log.Debug(). - Str("instance", instanceName). - Str("reason", reason). - Msg("Skipping PVE backup polling this cycle") - } - } else { - select { - case <-ctx.Done(): - return - default: + if !shouldPoll { + if reason != "" { + log.Debug(). + Str("instance", instanceName). + Str("reason", reason). + Msg("Skipping PVE backup polling this cycle") + } + } else { + select { + case <-ctx.Done(): + pollErr = ctx.Err() + return + default: m.mu.Lock() m.lastPVEBackupPoll[instanceName] = newLast m.mu.Unlock() @@ -3855,10 +3896,354 @@ func (m *Monitor) pollVMsAndContainersEfficient(ctx context.Context, instanceNam return true } +// pollBackupTasks polls backup tasks from a PVE instance +func (m *Monitor) pollBackupTasks(ctx context.Context, instanceName string, client PVEClientInterface) { + log.Debug().Str("instance", instanceName).Msg("Polling backup tasks") + + tasks, err := client.GetBackupTasks(ctx) + if err != nil { + monErr := errors.WrapAPIError("get_backup_tasks", instanceName, err, 0) + log.Error().Err(monErr).Str("instance", instanceName).Msg("Failed to get backup tasks") + return + } + + var backupTasks []models.BackupTask + for _, task := range tasks { + // Extract VMID from task ID (format: "UPID:node:pid:starttime:type:vmid:user@realm:") + vmid := 0 + if task.ID != "" { + if vmidInt, err := strconv.Atoi(task.ID); err == nil { + vmid = vmidInt + } + } + + taskID := fmt.Sprintf("%s-%s", instanceName, task.UPID) + + backupTask := models.BackupTask{ + ID: taskID, + Node: task.Node, + Type: task.Type, + VMID: vmid, + Status: task.Status, + StartTime: time.Unix(task.StartTime, 0), + } + + if task.EndTime > 0 { + backupTask.EndTime = time.Unix(task.EndTime, 0) + } + + backupTasks = append(backupTasks, backupTask) + } + + // Update state with new backup tasks for this instance + m.state.UpdateBackupTasksForInstance(instanceName, backupTasks) +} + +// pollPBSInstance polls a single PBS instance +func (m *Monitor) pollPBSInstance(ctx context.Context, instanceName string, client *pbs.Client) { + // Check if context is cancelled + select { + case <-ctx.Done(): + log.Debug().Str("instance", instanceName).Msg("Polling cancelled") + return + default: + } + + log.Debug().Str("instance", instanceName).Msg("Polling PBS instance") + + // Get instance config + var instanceCfg *config.PBSInstance + for _, cfg := range m.config.PBSInstances { + if cfg.Name == instanceName { + instanceCfg = &cfg + log.Debug(). + Str("instance", instanceName). + Bool("monitorDatastores", cfg.MonitorDatastores). + Msg("Found PBS instance config") + break + } + } + if instanceCfg == nil { + log.Error().Str("instance", instanceName).Msg("PBS instance config not found") + return + } + + // Initialize PBS instance with default values + pbsInst := models.PBSInstance{ + ID: "pbs-" + instanceName, + Name: instanceName, + Host: instanceCfg.Host, + Status: "offline", + Version: "unknown", + ConnectionHealth: "unhealthy", + LastSeen: time.Now(), + } + + // Try to get version first + version, versionErr := client.GetVersion(ctx) + if versionErr == nil { + pbsInst.Status = "online" + pbsInst.Version = version.Version + pbsInst.ConnectionHealth = "healthy" + m.resetAuthFailures(instanceName, "pbs") + m.state.SetConnectionHealth("pbs-"+instanceName, true) + + log.Debug(). + Str("instance", instanceName). + Str("version", version.Version). + Bool("monitorDatastores", instanceCfg.MonitorDatastores). + Msg("PBS version retrieved successfully") + } else { + log.Debug().Err(versionErr).Str("instance", instanceName).Msg("Failed to get PBS version, trying fallback") + + ctx2, cancel2 := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel2() + _, datastoreErr := client.GetDatastores(ctx2) + if datastoreErr == nil { + pbsInst.Status = "online" + pbsInst.Version = "connected" + pbsInst.ConnectionHealth = "healthy" + m.resetAuthFailures(instanceName, "pbs") + m.state.SetConnectionHealth("pbs-"+instanceName, true) + + log.Info(). + Str("instance", instanceName). + Msg("PBS connected (version unavailable but datastores accessible)") + } else { + pbsInst.Status = "offline" + pbsInst.ConnectionHealth = "error" + monErr := errors.WrapConnectionError("get_pbs_version", instanceName, versionErr) + log.Error().Err(monErr).Str("instance", instanceName).Msg("Failed to connect to PBS") + m.state.SetConnectionHealth("pbs-"+instanceName, false) + + if errors.IsAuthError(versionErr) || errors.IsAuthError(datastoreErr) { + m.recordAuthFailure(instanceName, "pbs") + return + } + } + } + + // Get node status (CPU, memory, etc.) + nodeStatus, err := client.GetNodeStatus(ctx) + if err != nil { + log.Debug().Err(err).Str("instance", instanceName).Msg("Could not get PBS node status (may need Sys.Audit permission)") + } else if nodeStatus != nil { + pbsInst.CPU = nodeStatus.CPU + if nodeStatus.Memory.Total > 0 { + pbsInst.Memory = float64(nodeStatus.Memory.Used) / float64(nodeStatus.Memory.Total) * 100 + pbsInst.MemoryUsed = nodeStatus.Memory.Used + pbsInst.MemoryTotal = nodeStatus.Memory.Total + } + pbsInst.Uptime = nodeStatus.Uptime + + log.Debug(). + Str("instance", instanceName). + Float64("cpu", pbsInst.CPU). + Float64("memory", pbsInst.Memory). + Int64("uptime", pbsInst.Uptime). + Msg("PBS node status retrieved") + } + + // Poll datastores if enabled + if instanceCfg.MonitorDatastores { + datastores, err := client.GetDatastores(ctx) + if err != nil { + monErr := errors.WrapAPIError("get_datastores", instanceName, err, 0) + log.Error().Err(monErr).Str("instance", instanceName).Msg("Failed to get datastores") + } else { + log.Info(). + Str("instance", instanceName). + Int("count", len(datastores)). + Msg("Got PBS datastores") + + for _, ds := range datastores { + total := ds.Total + if total == 0 && ds.TotalSpace > 0 { + total = ds.TotalSpace + } + used := ds.Used + if used == 0 && ds.UsedSpace > 0 { + used = ds.UsedSpace + } + avail := ds.Avail + if avail == 0 && ds.AvailSpace > 0 { + avail = ds.AvailSpace + } + if total == 0 && used > 0 && avail > 0 { + total = used + avail + } + + log.Debug(). + Str("store", ds.Store). + Int64("total", total). + Int64("used", used). + Int64("avail", avail). + Int64("orig_total", ds.Total). + Int64("orig_total_space", ds.TotalSpace). + Msg("PBS datastore details") + + modelDS := models.PBSDatastore{ + Name: ds.Store, + Total: total, + Used: used, + Free: avail, + Usage: safePercentage(float64(used), float64(total)), + Status: "available", + DeduplicationFactor: ds.DeduplicationFactor, + } + + namespaces, err := client.ListNamespaces(ctx, ds.Store, "", 0) + if err != nil { + log.Warn().Err(err). + Str("instance", instanceName). + Str("datastore", ds.Store). + Msg("Failed to list namespaces") + } else { + for _, ns := range namespaces { + nsPath := ns.NS + if nsPath == "" { + nsPath = ns.Path + } + if nsPath == "" { + nsPath = ns.Name + } + + modelNS := models.PBSNamespace{ + Path: nsPath, + Parent: ns.Parent, + Depth: strings.Count(nsPath, "/"), + } + modelDS.Namespaces = append(modelDS.Namespaces, modelNS) + } + + hasRoot := false + for _, ns := range modelDS.Namespaces { + if ns.Path == "" { + hasRoot = true + break + } + } + if !hasRoot { + modelDS.Namespaces = append([]models.PBSNamespace{{Path: "", Depth: 0}}, modelDS.Namespaces...) + } + } + + pbsInst.Datastores = append(pbsInst.Datastores, modelDS) + } + } + } + + // Update state and run alerts + m.state.UpdatePBSInstance(pbsInst) + log.Info(). + Str("instance", instanceName). + Str("id", pbsInst.ID). + Int("datastores", len(pbsInst.Datastores)). + Msg("PBS instance updated in state") + + if m.alertManager != nil { + m.alertManager.CheckPBS(pbsInst) + } + + // Poll backups if enabled + if instanceCfg.MonitorBackups { + if len(pbsInst.Datastores) == 0 { + log.Debug(). + Str("instance", instanceName). + Msg("No PBS datastores available for backup polling") + } else if !m.config.EnableBackupPolling { + log.Debug(). + Str("instance", instanceName). + Msg("Skipping PBS backup polling - globally disabled") + } else { + now := time.Now() + + m.mu.RLock() + lastPoll := m.lastPBSBackupPoll[instanceName] + inProgress := m.pbsBackupPollers[instanceName] + m.mu.RUnlock() + + shouldPoll, reason, newLast := m.shouldRunBackupPoll(lastPoll, now) + if !shouldPoll { + if reason != "" { + log.Debug(). + Str("instance", instanceName). + Str("reason", reason). + Msg("Skipping PBS backup polling this cycle") + } + } else if inProgress { + log.Debug(). + Str("instance", instanceName). + Msg("PBS backup polling already in progress") + } else { + datastoreSnapshot := make([]models.PBSDatastore, len(pbsInst.Datastores)) + copy(datastoreSnapshot, pbsInst.Datastores) + + m.mu.Lock() + if m.pbsBackupPollers == nil { + m.pbsBackupPollers = make(map[string]bool) + } + if m.pbsBackupPollers[instanceName] { + m.mu.Unlock() + } else { + m.pbsBackupPollers[instanceName] = true + m.lastPBSBackupPoll[instanceName] = newLast + m.mu.Unlock() + + go func(ds []models.PBSDatastore, inst string, start time.Time, pbsClient *pbs.Client) { + defer func() { + m.mu.Lock() + delete(m.pbsBackupPollers, inst) + m.lastPBSBackupPoll[inst] = time.Now() + m.mu.Unlock() + }() + + log.Info(). + Str("instance", inst). + Int("datastores", len(ds)). + Msg("Starting background PBS backup polling") + + backupCtx, cancel := context.WithTimeout(context.Background(), 5*time.Minute) + defer cancel() + + m.pollPBSBackups(backupCtx, inst, pbsClient, ds) + + log.Info(). + Str("instance", inst). + Dur("duration", time.Since(start)). + Msg("Completed background PBS backup polling") + }(datastoreSnapshot, instanceName, now, client) + } + } + } + } else { + log.Debug(). + Str("instance", instanceName). + Msg("PBS backup monitoring disabled") + } +} // pollPMGInstance polls a single Proxmox Mail Gateway instance func (m *Monitor) pollPMGInstance(ctx context.Context, instanceName string, client *pmg.Client) { + start := time.Now() + var pollErr error + if m.pollMetrics != nil { + m.pollMetrics.IncInFlight("pmg") + defer m.pollMetrics.DecInFlight("pmg") + defer func() { + m.pollMetrics.RecordResult(PollResult{ + InstanceName: instanceName, + InstanceType: "pmg", + Success: pollErr == nil, + Error: pollErr, + StartTime: start, + EndTime: time.Now(), + }) + }() + } + select { case <-ctx.Done(): + pollErr = ctx.Err() log.Debug().Str("instance", instanceName).Msg("PMG polling cancelled by context") return default: @@ -3876,6 +4261,7 @@ func (m *Monitor) pollPMGInstance(ctx context.Context, instanceName string, clie if instanceCfg == nil { log.Error().Str("instance", instanceName).Msg("PMG instance config not found") + pollErr = fmt.Errorf("pmg instance config not found for %s", instanceName) return } @@ -3893,6 +4279,7 @@ func (m *Monitor) pollPMGInstance(ctx context.Context, instanceName string, clie version, err := client.GetVersion(ctx) if err != nil { monErr := errors.WrapConnectionError("pmg_get_version", instanceName, err) + pollErr = monErr log.Error().Err(monErr).Str("instance", instanceName).Msg("Failed to connect to PMG instance") m.state.SetConnectionHealth("pmg-"+instanceName, false) m.state.UpdatePMGInstance(pmgInst) diff --git a/internal/monitoring/monitor_polling.go b/internal/monitoring/monitor_polling.go index 5ac04f9..9d155ee 100644 --- a/internal/monitoring/monitor_polling.go +++ b/internal/monitoring/monitor_polling.go @@ -16,6 +16,102 @@ import ( "github.com/rs/zerolog/log" ) +func (m *Monitor) describeInstancesForScheduler() []InstanceDescriptor { + total := len(m.pveClients) + len(m.pbsClients) + len(m.pmgClients) + if total == 0 { + return nil + } + + descriptors := make([]InstanceDescriptor, 0, total) + + if len(m.pveClients) > 0 { + names := make([]string, 0, len(m.pveClients)) + for name := range m.pveClients { + names = append(names, name) + } + sort.Strings(names) + for _, name := range names { + desc := InstanceDescriptor{ + Name: name, + Type: InstanceTypePVE, + } + if m.scheduler != nil { + if last, ok := m.scheduler.LastScheduled(InstanceTypePVE, name); ok { + desc.LastScheduled = last.NextRun + desc.LastInterval = last.Interval + } + } + descriptors = append(descriptors, desc) + } + } + + if len(m.pbsClients) > 0 { + names := make([]string, 0, len(m.pbsClients)) + for name := range m.pbsClients { + names = append(names, name) + } + sort.Strings(names) + for _, name := range names { + desc := InstanceDescriptor{ + Name: name, + Type: InstanceTypePBS, + } + if m.scheduler != nil { + if last, ok := m.scheduler.LastScheduled(InstanceTypePBS, name); ok { + desc.LastScheduled = last.NextRun + desc.LastInterval = last.Interval + } + } + descriptors = append(descriptors, desc) + } + } + + if len(m.pmgClients) > 0 { + names := make([]string, 0, len(m.pmgClients)) + for name := range m.pmgClients { + names = append(names, name) + } + sort.Strings(names) + for _, name := range names { + desc := InstanceDescriptor{ + Name: name, + Type: InstanceTypePMG, + } + if m.scheduler != nil { + if last, ok := m.scheduler.LastScheduled(InstanceTypePMG, name); ok { + desc.LastScheduled = last.NextRun + desc.LastInterval = last.Interval + } + } + descriptors = append(descriptors, desc) + } + } + + return descriptors +} + +func (m *Monitor) buildScheduledTasks(now time.Time) []ScheduledTask { + descriptors := m.describeInstancesForScheduler() + if len(descriptors) == 0 { + return nil + } + + if m.scheduler == nil { + tasks := make([]ScheduledTask, 0, len(descriptors)) + for _, desc := range descriptors { + tasks = append(tasks, ScheduledTask{ + InstanceName: desc.Name, + InstanceType: desc.Type, + NextRun: now, + Interval: DefaultSchedulerConfig().BaseInterval, + }) + } + return tasks + } + + return m.scheduler.BuildPlan(now, descriptors) +} + // convertPoolInfoToModel converts Proxmox ZFS pool info to our model func convertPoolInfoToModel(poolInfo *proxmox.ZFSPoolInfo) *models.ZFSPool { if poolInfo == nil { diff --git a/internal/monitoring/monitor_storage_test.go b/internal/monitoring/monitor_storage_test.go index 537efc0..9f68ed6 100644 --- a/internal/monitoring/monitor_storage_test.go +++ b/internal/monitoring/monitor_storage_test.go @@ -157,7 +157,7 @@ func TestPollStorageWithNodesOptimizedRecordsMetricsAndAlerts(t *testing.T) { {Node: "node1", Status: "online"}, } - monitor.pollStorageWithNodesOptimized(context.Background(), "inst1", client, nodes) + monitor.pollStorageWithNodes(context.Background(), "inst1", client, nodes) metrics := monitor.metricsHistory.GetAllStorageMetrics("inst1-node1-local", time.Minute) if len(metrics["usage"]) != 1 { diff --git a/internal/monitoring/scheduler.go b/internal/monitoring/scheduler.go new file mode 100644 index 0000000..49d8a43 --- /dev/null +++ b/internal/monitoring/scheduler.go @@ -0,0 +1,289 @@ +package monitoring + +import ( + "context" + "sort" + "sync" + "time" + + "github.com/rs/zerolog/log" +) + +// InstanceType represents a polling target category. +type InstanceType string + +const ( + InstanceTypePVE InstanceType = "pve" + InstanceTypePBS InstanceType = "pbs" + InstanceTypePMG InstanceType = "pmg" +) + +// StalenessSource provides normalized freshness hints for an instance. +type StalenessSource interface { + StalenessScore(instanceType InstanceType, instanceName string) (float64, bool) +} + +// IntervalSelector chooses the next polling cadence for an instance. +type IntervalSelector interface { + SelectInterval(req IntervalRequest) time.Duration +} + +// TaskEnqueuer receives scheduled tasks for downstream execution. +type TaskEnqueuer interface { + Enqueue(ctx context.Context, task ScheduledTask) error +} + +// IntervalRequest bundles the context required to compute the next polling interval. +type IntervalRequest struct { + Now time.Time + BaseInterval time.Duration + MinInterval time.Duration + MaxInterval time.Duration + LastInterval time.Duration + LastSuccess time.Time + LastScheduled time.Time + StalenessScore float64 + ErrorCount int +} + +// InstanceDescriptor describes a monitored endpoint for scheduling purposes. +type InstanceDescriptor struct { + Name string + Type InstanceType + LastSuccess time.Time + LastFailure time.Time + LastScheduled time.Time + LastInterval time.Duration + ErrorCount int + Metadata map[string]any +} + +// ScheduledTask represents a single polling opportunity planned by the scheduler. +type ScheduledTask struct { + InstanceName string + InstanceType InstanceType + NextRun time.Time + Interval time.Duration + Priority float64 + Metadata map[string]any +} + +// SchedulerConfig contains tunables for the adaptive scheduler. +type SchedulerConfig struct { + BaseInterval time.Duration + MinInterval time.Duration + MaxInterval time.Duration +} + +// DefaultSchedulerConfig returns conservative defaults that preserve current behaviour. +func DefaultSchedulerConfig() SchedulerConfig { + return SchedulerConfig{ + BaseInterval: 10 * time.Second, + MinInterval: 5 * time.Second, + MaxInterval: 5 * time.Minute, + } +} + +// AdaptiveScheduler orchestrates poll execution plans using pluggable scoring strategies. +type AdaptiveScheduler struct { + cfg SchedulerConfig + staleness StalenessSource + interval IntervalSelector + enqueuer TaskEnqueuer + + mu sync.RWMutex + lastPlan map[string]ScheduledTask +} + +// NewAdaptiveScheduler constructs a scheduler with safe defaults. +func NewAdaptiveScheduler(cfg SchedulerConfig, staleness StalenessSource, interval IntervalSelector, enqueuer TaskEnqueuer) *AdaptiveScheduler { + if cfg.BaseInterval <= 0 { + cfg.BaseInterval = DefaultSchedulerConfig().BaseInterval + } + if cfg.MinInterval <= 0 { + cfg.MinInterval = DefaultSchedulerConfig().MinInterval + } + if cfg.MaxInterval <= 0 || cfg.MaxInterval < cfg.MinInterval { + cfg.MaxInterval = DefaultSchedulerConfig().MaxInterval + } + if staleness == nil { + staleness = noopStalenessSource{} + } + if interval == nil { + interval = &fixedIntervalSelector{interval: cfg.BaseInterval} + } + if enqueuer == nil { + enqueuer = noopTaskEnqueuer{} + } + + return &AdaptiveScheduler{ + cfg: cfg, + staleness: staleness, + interval: interval, + enqueuer: enqueuer, + lastPlan: make(map[string]ScheduledTask), + } +} + +// BuildPlan produces an ordered set of scheduled tasks for the supplied inventory. +func (s *AdaptiveScheduler) BuildPlan(now time.Time, inventory []InstanceDescriptor) []ScheduledTask { + if len(inventory) == 0 { + return nil + } + + s.mu.Lock() + defer s.mu.Unlock() + + tasks := make([]ScheduledTask, 0, len(inventory)) + for _, inst := range inventory { + score, ok := s.staleness.StalenessScore(inst.Type, inst.Name) + if !ok { + score = 0 + } + + lastScheduled := inst.LastScheduled + lastInterval := inst.LastInterval + if cached, exists := s.lastPlan[schedulerKey(inst.Type, inst.Name)]; exists { + if lastScheduled.IsZero() { + lastScheduled = cached.NextRun + } + if lastInterval == 0 { + lastInterval = cached.Interval + } + } + if lastInterval == 0 { + lastInterval = s.cfg.BaseInterval + } + + req := IntervalRequest{ + Now: now, + BaseInterval: s.cfg.BaseInterval, + MinInterval: s.cfg.MinInterval, + MaxInterval: s.cfg.MaxInterval, + LastInterval: lastInterval, + LastSuccess: inst.LastSuccess, + LastScheduled: lastScheduled, + StalenessScore: score, + ErrorCount: inst.ErrorCount, + } + + nextInterval := s.interval.SelectInterval(req) + if nextInterval <= 0 { + nextInterval = s.cfg.BaseInterval + } + if nextInterval < s.cfg.MinInterval { + nextInterval = s.cfg.MinInterval + } + if nextInterval > s.cfg.MaxInterval { + nextInterval = s.cfg.MaxInterval + } + + nextRun := now + if !lastScheduled.IsZero() { + nextRun = lastScheduled.Add(nextInterval) + } else if !inst.LastSuccess.IsZero() { + nextRun = inst.LastSuccess.Add(nextInterval) + } + if nextRun.Before(now) { + nextRun = now + } + + task := ScheduledTask{ + InstanceName: inst.Name, + InstanceType: inst.Type, + NextRun: nextRun, + Interval: nextInterval, + Priority: score, + Metadata: inst.Metadata, + } + + s.lastPlan[schedulerKey(inst.Type, inst.Name)] = task + tasks = append(tasks, task) + } + + sort.Slice(tasks, func(i, j int) bool { + if tasks[i].NextRun.Equal(tasks[j].NextRun) { + if tasks[i].Priority == tasks[j].Priority { + return tasks[i].InstanceName < tasks[j].InstanceName + } + return tasks[i].Priority > tasks[j].Priority + } + return tasks[i].NextRun.Before(tasks[j].NextRun) + }) + + return tasks +} + +// FilterDue returns tasks whose NextRun is at or before now. +func (s *AdaptiveScheduler) FilterDue(now time.Time, tasks []ScheduledTask) []ScheduledTask { + if len(tasks) == 0 { + return nil + } + + due := make([]ScheduledTask, 0, len(tasks)) + for _, task := range tasks { + if !task.NextRun.After(now) { + due = append(due, task) + } + } + return due +} + +// DispatchDue enqueues due tasks using the configured sink for tracking purposes. +func (s *AdaptiveScheduler) DispatchDue(ctx context.Context, now time.Time, tasks []ScheduledTask) []ScheduledTask { + if s == nil { + return tasks + } + due := s.FilterDue(now, tasks) + if len(due) == 0 { + return due + } + for _, task := range due { + if err := s.enqueuer.Enqueue(ctx, task); err != nil { + log.Warn(). + Err(err). + Str("instance", task.InstanceName). + Str("type", string(task.InstanceType)). + Msg("Failed to enqueue scheduled task") + } + } + return due +} + +// LastScheduled returns the last recorded task for the given instance, if any. +func (s *AdaptiveScheduler) LastScheduled(instanceType InstanceType, instanceName string) (ScheduledTask, bool) { + if s == nil { + return ScheduledTask{}, false + } + s.mu.RLock() + defer s.mu.RUnlock() + task, ok := s.lastPlan[schedulerKey(instanceType, instanceName)] + return task, ok +} + +func schedulerKey(instanceType InstanceType, name string) string { + return string(instanceType) + "::" + name +} + +type noopStalenessSource struct{} + +func (noopStalenessSource) StalenessScore(instanceType InstanceType, instanceName string) (float64, bool) { + return 0, false +} + +type fixedIntervalSelector struct { + interval time.Duration +} + +func (f *fixedIntervalSelector) SelectInterval(req IntervalRequest) time.Duration { + if f.interval > 0 { + return f.interval + } + return req.BaseInterval +} + +type noopTaskEnqueuer struct{} + +func (noopTaskEnqueuer) Enqueue(ctx context.Context, task ScheduledTask) error { + return nil +}