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.
This commit is contained in:
rcourtman
parent
524f42cc28
commit
57429900a6
15 changed files with 1468 additions and 212 deletions
|
|
@ -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
|
||||
|
|
|
|||
40
cmd/pulse/metrics_server.go
Normal file
40
cmd/pulse/metrics_server.go
Normal file
|
|
@ -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")
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
43
docker-compose.parallel.yml
Normal file
43
docker-compose.parallel.yml
Normal file
|
|
@ -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"
|
||||
|
|
@ -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<BackupSortKey>('backupTime');
|
||||
const [sortDirection, setSortDirection] = createSignal<'asc' | 'desc'>('desc');
|
||||
const [sortKey, setSortKey] = usePersistentSignal<BackupSortKey>(
|
||||
'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();
|
||||
|
|
|
|||
|
|
@ -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<Record<string, GuestMetadata>>({});
|
||||
|
||||
// Initialize from localStorage with proper type checking
|
||||
const storedViewMode = localStorage.getItem('dashboardViewMode');
|
||||
const [viewMode, setViewMode] = createSignal<ViewMode>(
|
||||
storedViewMode === 'all' || storedViewMode === 'vm' || storedViewMode === 'lxc'
|
||||
? storedViewMode
|
||||
: 'all',
|
||||
);
|
||||
const [viewMode, setViewMode] = usePersistentSignal<ViewMode>('dashboardViewMode', 'all', {
|
||||
deserialize: (raw) => (raw === 'all' || raw === 'vm' || raw === 'lxc' ? raw : 'all'),
|
||||
});
|
||||
|
||||
const storedStatusMode = localStorage.getItem('dashboardStatusMode');
|
||||
const [statusMode, setStatusMode] = createSignal<StatusMode>(
|
||||
storedStatusMode === 'all' || storedStatusMode === 'running' || storedStatusMode === 'stopped'
|
||||
? storedStatusMode
|
||||
: 'all',
|
||||
);
|
||||
const [statusMode, setStatusMode] = usePersistentSignal<StatusMode>('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<GroupingMode>(
|
||||
storedGroupingMode === 'grouped' || storedGroupingMode === 'flat'
|
||||
? storedGroupingMode
|
||||
: 'grouped',
|
||||
const [groupingMode, setGroupingMode] = usePersistentSignal<GroupingMode>(
|
||||
'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<boolean>(
|
||||
'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) {
|
||||
|
|
|
|||
|
|
@ -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<string | null>(null);
|
||||
const [expandedStorage, setExpandedStorage] = createSignal<string | null>(null);
|
||||
type StorageSortKey = 'name' | 'node' | 'type' | 'status' | 'usage' | 'free' | 'total';
|
||||
const [sortKey, setSortKey] = createSignal<StorageSortKey>('name');
|
||||
const [sortDirection, setSortDirection] = createSignal<'asc' | 'desc'>('asc');
|
||||
const [sortKey, setSortKey] = usePersistentSignal<StorageSortKey>('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 || [];
|
||||
|
|
|
|||
75
frontend-modern/src/hooks/usePersistentSignal.ts
Normal file
75
frontend-modern/src/hooks/usePersistentSignal.ts
Normal file
|
|
@ -0,0 +1,75 @@
|
|||
import { Accessor, Setter, createEffect, createSignal } from 'solid-js';
|
||||
|
||||
export type PersistentSignalOptions<T> = {
|
||||
/**
|
||||
* 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<T>(
|
||||
key: string,
|
||||
defaultValue: T,
|
||||
options: PersistentSignalOptions<T> = {},
|
||||
): [Accessor<T>, Setter<T>] {
|
||||
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<T>(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];
|
||||
}
|
||||
|
||||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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 {
|
||||
|
|
|
|||
|
|
@ -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"`
|
||||
|
|
|
|||
263
internal/monitoring/metrics.go
Normal file
263
internal/monitoring/metrics.go
Normal file
|
|
@ -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"
|
||||
}
|
||||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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 {
|
||||
|
|
|
|||
|
|
@ -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 {
|
||||
|
|
|
|||
289
internal/monitoring/scheduler.go
Normal file
289
internal/monitoring/scheduler.go
Normal file
|
|
@ -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
|
||||
}
|
||||
Loading…
Reference in a new issue