From cc3c0187a08dee5009337345f337e9a8187aedeb Mon Sep 17 00:00:00 2001 From: rcourtman Date: Fri, 5 Dec 2025 10:37:02 +0000 Subject: [PATCH] feat: AI features, agent improvements, and host monitoring enhancements AI Chat Integration: - Multi-provider support (Anthropic, OpenAI, Ollama) - Streaming responses with markdown rendering - Agent command execution for remote troubleshooting - Context-aware conversations with host/container metadata Agent Updates: - Add --enable-proxmox flag for automatic PVE/PBS token setup - Improve auto-update with semver comparison (prevents downgrades) - Add updatedFrom tracking to report previous version after update - Reduce initial update check delay from 30s to 5s - Add agent version column to Hosts page table Host Metrics: - Add DiskIO stats collection (read/write bytes, ops, time) - Improve disk filtering to exclude Docker overlay mounts - Add RAID array monitoring via mdadm - Enhanced temperature sensor parsing Frontend: - New Agent Version column on Hosts overview table - Improved node modal with agent-first installation flow - Add DiskIO display in host drawer - Better responsive handling for metric bars --- cmd/pulse-agent/main.go | 20 +- cmd/pulse-sensor-proxy/README.md | 5 + cmd/pulse/main.go | 14 +- docs/operations/SENSOR_PROXY_CONFIG.md | 4 + docs/operations/SENSOR_PROXY_LOGS.md | 4 + docs/security/SENSOR_PROXY_APPARMOR.md | 4 + docs/security/SENSOR_PROXY_HARDENING.md | 4 + docs/security/SENSOR_PROXY_NETWORK.md | 4 + docs/security/TEMPERATURE_MONITORING.md | 42 ++- frontend-modern/src/components/AI/AIChat.tsx | 84 +++-- .../src/components/Hosts/HostsOverview.tsx | 62 +++- .../Settings/ConfiguredNodeTables.tsx | 90 ++++- .../src/components/Settings/NodeModal.tsx | 249 +++++++++++-- .../src/components/Settings/Settings.tsx | 40 +- .../src/components/Settings/UnifiedAgents.tsx | 83 +++-- frontend-modern/src/types/ai.ts | 1 + frontend-modern/src/types/api.ts | 12 + frontend-modern/src/types/nodes.ts | 1 + frontend-modern/src/utils/format.ts | 5 + frontend-modern/vite.config.ts | 44 +++ internal/agentexec/server.go | 62 +++- internal/agentupdate/update.go | 52 ++- internal/ai/providers/anthropic.go | 94 +++-- internal/ai/service.go | 29 +- internal/alerts/alerts_test.go | 7 +- internal/api/config_handlers.go | 141 ++++++- internal/api/router.go | 3 + internal/config/config.go | 6 + internal/hostagent/agent.go | 62 ++++ internal/hostagent/commands.go | 11 +- internal/hostagent/proxmox_setup.go | 344 ++++++++++++++++++ internal/hostmetrics/collector.go | 94 ++++- internal/hostmetrics/collector_test.go | 146 +------- internal/models/converters.go | 4 + internal/models/models.go | 14 + internal/models/models_frontend.go | 1 + internal/monitoring/host_agent_temps.go | 266 ++++++++++++++ internal/monitoring/monitor.go | 18 + internal/monitoring/monitor_polling.go | 119 +++--- internal/sensors/parser.go | 17 + internal/utils/helpers.go | 45 +++ internal/utils/utils_test.go | 47 +++ pkg/agents/host/report.go | 15 + pkg/proxmox/cluster_client.go | 14 +- scripts/install.sh | 8 + 45 files changed, 2038 insertions(+), 353 deletions(-) create mode 100644 internal/hostagent/proxmox_setup.go create mode 100644 internal/monitoring/host_agent_temps.go diff --git a/cmd/pulse-agent/main.go b/cmd/pulse-agent/main.go index e6e221f..b9c90f6 100644 --- a/cmd/pulse-agent/main.go +++ b/cmd/pulse-agent/main.go @@ -78,6 +78,7 @@ func main() { Str("pulse_url", cfg.PulseURL). Bool("host_agent", cfg.EnableHost). Bool("docker_agent", cfg.EnableDocker). + Bool("proxmox_mode", cfg.EnableProxmox). Bool("auto_update", !cfg.DisableAutoUpdate). Msg("Starting Pulse Unified Agent") @@ -126,6 +127,8 @@ func main() { InsecureSkipVerify: cfg.InsecureSkipVerify, LogLevel: cfg.LogLevel, Logger: &logger, + EnableProxmox: cfg.EnableProxmox, + ProxmoxType: cfg.ProxmoxType, } agent, err := hostagent.New(hostCfg) @@ -259,8 +262,10 @@ type Config struct { Logger *zerolog.Logger // Module flags - EnableHost bool - EnableDocker bool + EnableHost bool + EnableDocker bool + EnableProxmox bool + ProxmoxType string // "pve", "pbs", or "" for auto-detect // Auto-update DisableAutoUpdate bool @@ -281,6 +286,8 @@ func loadConfig() Config { envLogLevel := utils.GetenvTrim("LOG_LEVEL") envEnableHost := utils.GetenvTrim("PULSE_ENABLE_HOST") envEnableDocker := utils.GetenvTrim("PULSE_ENABLE_DOCKER") + envEnableProxmox := utils.GetenvTrim("PULSE_ENABLE_PROXMOX") + envProxmoxType := utils.GetenvTrim("PULSE_PROXMOX_TYPE") envDisableAutoUpdate := utils.GetenvTrim("PULSE_DISABLE_AUTO_UPDATE") envHealthAddr := utils.GetenvTrim("PULSE_HEALTH_ADDR") @@ -302,6 +309,11 @@ func loadConfig() Config { defaultEnableDocker = utils.ParseBool(envEnableDocker) } + defaultEnableProxmox := false + if envEnableProxmox != "" { + defaultEnableProxmox = utils.ParseBool(envEnableProxmox) + } + defaultHealthAddr := envHealthAddr if defaultHealthAddr == "" { defaultHealthAddr = ":9191" @@ -318,6 +330,8 @@ func loadConfig() Config { enableHostFlag := flag.Bool("enable-host", defaultEnableHost, "Enable Host Agent module") enableDockerFlag := flag.Bool("enable-docker", defaultEnableDocker, "Enable Docker Agent module") + enableProxmoxFlag := flag.Bool("enable-proxmox", defaultEnableProxmox, "Enable Proxmox mode (creates API token, registers node)") + proxmoxTypeFlag := flag.String("proxmox-type", envProxmoxType, "Proxmox type: pve or pbs (auto-detected if not specified)") disableAutoUpdateFlag := flag.Bool("disable-auto-update", utils.ParseBool(envDisableAutoUpdate), "Disable automatic updates") healthAddrFlag := flag.String("health-addr", defaultHealthAddr, "Health/metrics server address (empty to disable)") showVersion := flag.Bool("version", false, "Print the agent version and exit") @@ -362,6 +376,8 @@ func loadConfig() Config { LogLevel: logLevel, EnableHost: *enableHostFlag, EnableDocker: *enableDockerFlag, + EnableProxmox: *enableProxmoxFlag, + ProxmoxType: strings.TrimSpace(*proxmoxTypeFlag), DisableAutoUpdate: *disableAutoUpdateFlag, HealthAddr: strings.TrimSpace(*healthAddrFlag), } diff --git a/cmd/pulse-sensor-proxy/README.md b/cmd/pulse-sensor-proxy/README.md index fb26e92..3753310 100644 --- a/cmd/pulse-sensor-proxy/README.md +++ b/cmd/pulse-sensor-proxy/README.md @@ -1,5 +1,10 @@ # pulse-sensor-proxy +> **⚠️ Deprecated:** The sensor-proxy is deprecated in favor of the unified Pulse agent. +> For new installations, use `install.sh --enable-proxmox` instead. The agent provides +> temperature monitoring plus additional features (AI, automatic Proxmox API setup). +> See [TEMPERATURE_MONITORING.md](/docs/security/TEMPERATURE_MONITORING.md) for details. + The sensor proxy keeps SSH identities and temperature polling logic on the Proxmox host while presenting a small RPC surface (Unix socket or HTTPS) to the Pulse server. It protects SSH keys from container breakouts, enforces per-UID diff --git a/cmd/pulse/main.go b/cmd/pulse/main.go index afb0c32..a6bb384 100644 --- a/cmd/pulse/main.go +++ b/cmd/pulse/main.go @@ -175,12 +175,16 @@ func runServer() { // Create HTTP server with unified configuration // In production, serve everything (frontend + API) on the frontend port + // NOTE: We use ReadHeaderTimeout instead of ReadTimeout to avoid affecting + // WebSocket connections. ReadTimeout sets a deadline on the underlying connection + // that persists even after WebSocket upgrade, causing premature disconnections. + // ReadHeaderTimeout only applies during header reading, not the full request body. srv := &http.Server{ - Addr: fmt.Sprintf("%s:%d", cfg.BackendHost, cfg.FrontendPort), - Handler: router.Handler(), - ReadTimeout: 15 * time.Second, - WriteTimeout: 60 * time.Second, // Increased from 15s to 60s to support large JSON responses (e.g., mock data) - IdleTimeout: 60 * time.Second, + Addr: fmt.Sprintf("%s:%d", cfg.BackendHost, cfg.FrontendPort), + Handler: router.Handler(), + ReadHeaderTimeout: 15 * time.Second, + WriteTimeout: 60 * time.Second, // Increased from 15s to 60s to support large JSON responses (e.g., mock data) + IdleTimeout: 60 * time.Second, } // Start config watcher for .env file changes diff --git a/docs/operations/SENSOR_PROXY_CONFIG.md b/docs/operations/SENSOR_PROXY_CONFIG.md index eae06ae..9305690 100644 --- a/docs/operations/SENSOR_PROXY_CONFIG.md +++ b/docs/operations/SENSOR_PROXY_CONFIG.md @@ -1,5 +1,9 @@ # ⚙️ Sensor Proxy Configuration +> **⚠️ Deprecated:** The sensor-proxy is deprecated in favor of the unified Pulse agent. +> For new installations, use `install.sh --enable-proxmox` instead. +> See [TEMPERATURE_MONITORING.md](/docs/security/TEMPERATURE_MONITORING.md). + Safe configuration management using the CLI (v4.31.1+). ## 📂 Files diff --git a/docs/operations/SENSOR_PROXY_LOGS.md b/docs/operations/SENSOR_PROXY_LOGS.md index 5c56b4d..06c5cf1 100644 --- a/docs/operations/SENSOR_PROXY_LOGS.md +++ b/docs/operations/SENSOR_PROXY_LOGS.md @@ -1,5 +1,9 @@ # 📝 Sensor Proxy Log Forwarding +> **⚠️ Deprecated:** The sensor-proxy is deprecated in favor of the unified Pulse agent. +> For new installations, use `install.sh --enable-proxmox` instead. +> See [TEMPERATURE_MONITORING.md](/docs/security/TEMPERATURE_MONITORING.md). + Forward `audit.log` and `proxy.log` to a central SIEM via RELP + TLS. ## 🚀 Quick Start diff --git a/docs/security/SENSOR_PROXY_APPARMOR.md b/docs/security/SENSOR_PROXY_APPARMOR.md index 44fb5a5..6d25c06 100644 --- a/docs/security/SENSOR_PROXY_APPARMOR.md +++ b/docs/security/SENSOR_PROXY_APPARMOR.md @@ -1,5 +1,9 @@ # 🛡️ Sensor Proxy Hardening +> **⚠️ Deprecated:** The sensor-proxy is deprecated in favor of the unified Pulse agent. +> For new installations, use `install.sh --enable-proxmox` instead. +> See [TEMPERATURE_MONITORING.md](/docs/security/TEMPERATURE_MONITORING.md). + Secure `pulse-sensor-proxy` with AppArmor and Seccomp. ## 🛡️ AppArmor diff --git a/docs/security/SENSOR_PROXY_HARDENING.md b/docs/security/SENSOR_PROXY_HARDENING.md index c03f99c..12f1c65 100644 --- a/docs/security/SENSOR_PROXY_HARDENING.md +++ b/docs/security/SENSOR_PROXY_HARDENING.md @@ -1,5 +1,9 @@ # 🛡️ Sensor Proxy Hardening +> **⚠️ Deprecated:** The sensor-proxy is deprecated in favor of the unified Pulse agent. +> For new installations, use `install.sh --enable-proxmox` instead. +> See [TEMPERATURE_MONITORING.md](/docs/security/TEMPERATURE_MONITORING.md). + The `pulse-sensor-proxy` runs on the host to securely collect temperatures, keeping SSH keys out of containers. ## 🏗️ Architecture diff --git a/docs/security/SENSOR_PROXY_NETWORK.md b/docs/security/SENSOR_PROXY_NETWORK.md index 55c6139..9a5d1fa 100644 --- a/docs/security/SENSOR_PROXY_NETWORK.md +++ b/docs/security/SENSOR_PROXY_NETWORK.md @@ -1,5 +1,9 @@ # 🌐 Sensor Proxy Network Segmentation +> **⚠️ Deprecated:** The sensor-proxy is deprecated in favor of the unified Pulse agent. +> For new installations, use `install.sh --enable-proxmox` instead. +> See [TEMPERATURE_MONITORING.md](/docs/security/TEMPERATURE_MONITORING.md). + Isolate the proxy to prevent lateral movement. ## 🚧 Zones diff --git a/docs/security/TEMPERATURE_MONITORING.md b/docs/security/TEMPERATURE_MONITORING.md index 760419c..0276393 100644 --- a/docs/security/TEMPERATURE_MONITORING.md +++ b/docs/security/TEMPERATURE_MONITORING.md @@ -1,29 +1,57 @@ -# 🌡️ Temperature Monitoring Security +# 🌡️ Temperature Monitoring -Secure architecture for collecting hardware temperatures. +Pulse supports two methods for collecting hardware temperatures from Proxmox nodes. -## 🛡️ Security Model +## Recommended: Pulse Agent + +The simplest and most feature-rich method is installing the Pulse agent on your Proxmox nodes: + +```bash +curl -fsSL http://your-pulse-server:7655/api/download/install.sh | bash -s -- \ + --url http://your-pulse-server:7655 \ + --token YOUR_TOKEN \ + --enable-proxmox +``` + +**Benefits:** +- ✅ One-command setup +- ✅ Automatic API token creation +- ✅ Temperature monitoring built-in +- ✅ Enables AI features for VM/container management +- ✅ No SSH keys or proxy configuration required + +The agent runs `sensors -j` locally and reports temperatures directly to Pulse. + +--- + +## Legacy: Sensor Proxy (SSH-based) + +For users who prefer not to install an agent on their hypervisor, the sensor-proxy method is still available. + +> **Note:** This method is deprecated and will be removed in a future release. Consider migrating to the agent-based approach. + +### 🛡️ Security Model * **Isolation**: SSH keys live on the host, not in the container. * **Least Privilege**: Proxy runs as `pulse-sensor-proxy` (no shell). * **Verification**: Container identity verified via `SO_PEERCRED`. -## 🏗️ Components +### 🏗️ Components 1. **Pulse Backend**: Connects to Unix socket `/mnt/pulse-proxy/pulse-sensor-proxy.sock`. 2. **Sensor Proxy**: Validates request, executes SSH to node. 3. **Target Node**: Accepts SSH key restricted to `sensors -j`. -## 🔒 Key Restrictions +### 🔒 Key Restrictions SSH keys deployed to nodes are locked down: ``` command="sensors -j",no-port-forwarding,no-X11-forwarding,no-agent-forwarding,no-pty ``` -## 🚦 Rate Limiting +### 🚦 Rate Limiting * **Per Peer**: ~12 req/min. * **Concurrency**: Max 2 parallel requests per peer. * **Global**: Max 8 concurrent requests. -## 📝 Auditing +### 📝 Auditing All requests logged to system journal: ```bash journalctl -u pulse-sensor-proxy diff --git a/frontend-modern/src/components/AI/AIChat.tsx b/frontend-modern/src/components/AI/AIChat.tsx index 5e305a6..4a7f1b6 100644 --- a/frontend-modern/src/components/AI/AIChat.tsx +++ b/frontend-modern/src/components/AI/AIChat.tsx @@ -272,12 +272,27 @@ export const AIChat: Component = (props) => { const previousMessages = messages(); // Build conversation history from previous messages (before we add new ones) + // Include tool call outputs so the AI remembers what commands were run const history = previousMessages - .filter((m) => m.content && !m.isStreaming) // Only include completed messages with content - .map((m) => ({ - role: m.role, - content: m.content, - })); + .filter((m) => !m.isStreaming) // Only include completed messages + .filter((m) => m.content || (m.toolCalls && m.toolCalls.length > 0)) // Must have content or tool calls + .map((m) => { + let content = m.content || ''; + + // For assistant messages, prepend tool call outputs so AI has full context + if (m.role === 'assistant' && m.toolCalls && m.toolCalls.length > 0) { + const toolSummary = m.toolCalls + .map((tc) => `[Executed: ${tc.input}]\n${tc.output}`) + .join('\n\n'); + content = toolSummary + (content ? '\n\n' + content : ''); + } + + return { + role: m.role, + content, + }; + }) + .filter((m) => m.content); // Filter out any empty messages // Add user message const userMessage: Message = { @@ -333,11 +348,16 @@ export const AIChat: Component = (props) => { } case 'tool_end': { const data = event.data as AIStreamToolEndData; - // Remove matching pending tool (by name since input won't match output) - const updatedPending = (msg.pendingTools || []).slice(0, -1); // Remove last pending + // Remove one pending tool with matching name + const pendingTools = msg.pendingTools || []; + const matchingIndex = pendingTools.findIndex((t) => t.name === data.name); + const updatedPending = matchingIndex >= 0 + ? [...pendingTools.slice(0, matchingIndex), ...pendingTools.slice(matchingIndex + 1)] + : pendingTools; + // Use input directly from tool_end event (authoritative) const newToolCall: AIToolExecution = { name: data.name, - input: (msg.pendingTools || []).find((t) => t.name === data.name)?.input || data.name, + input: data.input, output: data.output, success: data.success, }; @@ -656,7 +676,7 @@ export const AIChat: Component = (props) => { : 'bg-gray-100 dark:bg-gray-800 text-gray-900 dark:text-gray-100' }`} > - {/* Show completed tool calls */} + {/* Show completed tool calls FIRST - chronological order */} 0}>
@@ -683,20 +703,17 @@ export const AIChat: Component = (props) => {
- {/* Show subtle analyzing indicator when running commands */} - 0}> -
- - - - - Analyzing... -
+ {/* Show AI's response text AFTER tool calls */} + +
{/* Show commands awaiting approval */} 0}> -
+
{(approval) => (
@@ -756,24 +773,6 @@ export const AIChat: Component = (props) => {
- - {/* Show streaming indicator if no content yet but streaming */} - -
- - - - - Analyzing... -
- - - -
- {/* Minimal footer - no model/token info shown */}
@@ -783,6 +782,17 @@ export const AIChat: Component = (props) => {
+ {/* Processing indicator - sticky above input */} + +
+ + + + + Analyzing... +
+ + {/* Input Area */}
{/* Context section - always show with Add button */} diff --git a/frontend-modern/src/components/Hosts/HostsOverview.tsx b/frontend-modern/src/components/Hosts/HostsOverview.tsx index 8167739..2e8e3ce 100644 --- a/frontend-modern/src/components/Hosts/HostsOverview.tsx +++ b/frontend-modern/src/components/Hosts/HostsOverview.tsx @@ -2,7 +2,7 @@ import type { Component } from 'solid-js'; import { For, Show, createMemo, createSignal, createEffect, on, onMount, onCleanup } from 'solid-js'; import { useNavigate } from '@solidjs/router'; import type { Host } from '@/types/api'; -import { formatBytes, formatPercent, formatRelativeTime, formatUptime } from '@/utils/format'; +import { formatBytes, formatNumber, formatPercent, formatRelativeTime, formatUptime } from '@/utils/format'; import { Card } from '@/components/shared/Card'; import { EmptyState } from '@/components/shared/EmptyState'; import { MetricBar } from '@/components/Dashboard/MetricBar'; @@ -328,9 +328,12 @@ export const HostsOverview: Component = (props) => { handleSort('disk')}> Disk {renderSortIndicator('disk')} - handleSort('uptime')}> + handleSort('uptime')}> Uptime {renderSortIndicator('uptime')} + + Agent + @@ -413,6 +416,7 @@ export const HostsOverview: Component = (props) => { const hasDrawerContent = createMemo(() => { return ( (host.disks && host.disks.length > 0) || + (host.diskIO && host.diskIO.length > 0) || (host.networkInterfaces && host.networkInterfaces.length > 0) || (host.raid && host.raid.length > 0) || host.loadAverage || @@ -571,8 +575,8 @@ export const HostsOverview: Component = (props) => { {/* Uptime */} - -
+ +
—} @@ -583,12 +587,26 @@ export const HostsOverview: Component = (props) => {
+ + {/* Agent Version */} + +
+ —} + > + + {host.agentVersion} + + +
+ {/* Drawer - Additional Info */} - +
{/* System Info */} @@ -689,6 +707,40 @@ export const HostsOverview: Component = (props) => {
+ {/* Disk I/O */} + 0}> +
+
Disk I/O
+
+ + {(io) => ( +
+
{io.device}
+
+
+ Read: + {formatBytes(io.readBytes ?? 0, 1)} +
+
+ Write: + {formatBytes(io.writeBytes ?? 0, 1)} +
+
+ Read Ops: + {formatNumber(io.readOps ?? 0)} +
+
+ Write Ops: + {formatNumber(io.writeOps ?? 0)} +
+
+
+ )} + +
+
+ + {/* Temperature Sensors */} 0}>
diff --git a/frontend-modern/src/components/Settings/ConfiguredNodeTables.tsx b/frontend-modern/src/components/Settings/ConfiguredNodeTables.tsx index 7c9f06a..01148ea 100644 --- a/frontend-modern/src/components/Settings/ConfiguredNodeTables.tsx +++ b/frontend-modern/src/components/Settings/ConfiguredNodeTables.tsx @@ -20,9 +20,16 @@ type TemperatureSocketCooldownInfo = { lastError?: string; }; +// Host agent info passed from state +interface HostAgentInfo { + hostname: string; + status: string; +} + interface PveNodesTableProps { nodes: NodeConfigWithStatus[]; stateNodes: { instance: string; status?: string; connectionHealth?: string }[]; + stateHosts?: HostAgentInfo[]; globalTemperatureMonitoringEnabled?: boolean; temperatureTransports?: TemperatureTransportInfo | null; onTestConnection: (nodeId: string) => void; @@ -104,6 +111,7 @@ const resolveTemperatureTransport = ( node: NodeConfigWithStatus, info: TemperatureTransportInfo | null | undefined, globalEnabled: boolean, + hostAgent?: HostAgentInfo, ): TemperatureTransportBadge => { const monitoringEnabled = isTemperatureMonitoringEnabled(node, globalEnabled); const normalizedTransport = (node.temperatureTransport || '').toLowerCase(); @@ -125,6 +133,15 @@ const resolveTemperatureTransport = ( }; } + // If a host agent is connected and online, it provides temperatures directly + if (hostAgent?.status === 'online') { + return { + label: 'Via agent', + badgeClass: 'bg-green-100 dark:bg-green-900 text-green-700 dark:text-green-300', + description: 'Temperature data from Pulse agent', + }; + } + const key = nodeKey; const httpEntry = key ? info?.httpMap?.[key] : undefined; const socketStatus = info?.socketStatus; @@ -185,9 +202,9 @@ const resolveTemperatureTransport = ( return buildSocketBadge(); case 'ssh-blocked': return { - label: 'Proxy required', + label: 'Agent required', badgeClass: 'bg-amber-100 dark:bg-amber-900 text-amber-700 dark:text-amber-300', - description: 'Containerized Pulse requires pulse-sensor-proxy', + description: 'Install agent on node for temperature monitoring', }; case 'ssh': return { @@ -271,6 +288,16 @@ const resolvePveStatusMeta = ( } }; +// Helper to find matching host agent for a node by hostname matching +const findMatchingHostAgent = ( + nodeName: string, + hosts: HostAgentInfo[] | undefined, +): HostAgentInfo | undefined => { + if (!hosts || hosts.length === 0) return undefined; + const nodeNameLower = nodeName.toLowerCase().trim(); + return hosts.find((h) => h.hostname.toLowerCase().trim() === nodeNameLower); +}; + export const PveNodesTable: Component = (props) => { return ( @@ -304,11 +331,13 @@ export const PveNodesTable: Component = (props) => { const clusterName = createMemo(() => 'clusterName' in node && node.clusterName ? node.clusterName : 'Unknown', ); + const hostAgent = createMemo(() => findMatchingHostAgent(node.name, props.stateHosts)); const transportMeta = createMemo(() => resolveTemperatureTransport( node, props.temperatureTransports, props.globalTemperatureMonitoringEnabled ?? true, + hostAgent(), ), ); return ( @@ -410,9 +439,22 @@ export const PveNodesTable: Component = (props) => {
- - {node.user ? `User: ${node.user}` : `Token: ${node.tokenName}`} - +
+ + {node.user ? `User: ${node.user}` : `Token: ${node.tokenName}`} + + + + + Agent + + + + + API only + + +
@@ -603,9 +645,22 @@ export const PbsNodesTable: Component = (props) => {
- - {node.user ? `User: ${node.user}` : `Token: ${node.tokenName}`} - +
+ + {node.user ? `User: ${node.user}` : `Token: ${node.tokenName}`} + + + + + Agent + + + + + API only + + +
@@ -786,9 +841,22 @@ export const PmgNodesTable: Component = (props) => {
- - {node.user ? `User: ${node.user}` : `Token: ${node.tokenName}`} - +
+ + {node.user ? `User: ${node.user}` : `Token: ${node.tokenName}`} + + + + + Agent + + + + + API only + + +
diff --git a/frontend-modern/src/components/Settings/NodeModal.tsx b/frontend-modern/src/components/Settings/NodeModal.tsx index 6379349..fd6c2ff 100644 --- a/frontend-modern/src/components/Settings/NodeModal.tsx +++ b/frontend-modern/src/components/Settings/NodeModal.tsx @@ -80,7 +80,7 @@ export const NodeModal: Component = (props) => { host: '', guestURL: '', authType: nodeType === 'pmg' ? 'password' : ('token' as 'password' | 'token'), - setupMode: 'auto' as 'auto' | 'manual', + setupMode: 'agent' as 'agent' | 'auto' | 'manual', // Default to agent install (recommended) user: '', password: '', tokenName: '', @@ -101,6 +101,9 @@ export const NodeModal: Component = (props) => { const [proxyInstallCommand, setProxyInstallCommand] = createSignal(''); const [loadingProxyCommand, setLoadingProxyCommand] = createSignal(false); const [proxyCommandError, setProxyCommandError] = createSignal(null); + const [agentInstallCommand, setAgentInstallCommand] = createSignal(''); + const [loadingAgentCommand, setLoadingAgentCommand] = createSignal(false); + const [agentCommandError, setAgentCommandError] = createSignal(null); const showTemperatureMonitoringSection = () => typeof props.temperatureMonitoringEnabled === 'boolean'; const temperatureMonitoringEnabledValue = () => props.temperatureMonitoringEnabled ?? true; @@ -114,7 +117,7 @@ export const NodeModal: Component = (props) => { case 'socket-proxy': return { tone: 'success', - message: 'Temperatures flow through the host sensor proxy mounted at /run/pulse-sensor-proxy.', + message: 'Temperatures flow through the socket proxy mounted at /run/pulse-sensor-proxy.', }; case 'https-proxy': return { @@ -126,7 +129,7 @@ export const NodeModal: Component = (props) => { tone: 'danger', disable: true, message: - 'Pulse is running in a container without the pulse-sensor-proxy bind mount. Install the proxy on the host or register an HTTPS proxy before enabling temperatures.', + 'Pulse is running in a container. Install the Pulse agent on this node for temperature monitoring (Settings → Agents).', }; case 'ssh': return { @@ -400,8 +403,13 @@ export const NodeModal: Component = (props) => { setFormData((prev) => ({ ...prev, [field]: value })); - if (field === 'setupMode' && value !== 'auto') { - setQuickSetupCommand(''); + if (field === 'setupMode') { + if (value !== 'auto') { + setQuickSetupCommand(''); + } + if (value !== 'agent') { + setAgentInstallCommand(''); + } } }; @@ -764,7 +772,21 @@ export const NodeModal: Component = (props) => {
{/* Tab buttons */} -
+
+
- {/* Quick Setup Tab */} - -

- The command below creates the monitoring user, applies read-only - access, and adds the storage permissions Pulse needs to display - backups. -

+ {/* Agent Install Tab (Recommended) */} + +
+

+ Install the Pulse agent on your Proxmox node. This single command sets everything up: +

+
    +
  • Creates monitoring user and API token automatically
    • +
  • Registers the node with Pulse
    • +
  • Enables temperature monitoring (no SSH required)
    • +
  • Enables AI features for managing VMs/containers
    • +
+

+ Run this command on your Proxmox VE node: +

+
+ + 0} + fallback={ + + Click the button above to copy the install command + + } + > + + {agentInstallCommand()} + + +
+

+ The node will appear in Pulse automatically after the agent starts. +

+
+ + + {/* API Only Tab (formerly Quick Setup) */} + +
+

+ Limited functionality: API-only mode does not include temperature monitoring or AI features. + For full functionality, use the Agent Install tab instead. +

+

Just copy and run this one command on your Proxmox VE server:

@@ -1259,17 +1370,29 @@ export const NodeModal: Component = (props) => {
{/* Tab buttons for PBS */} -
+
+
- {/* Quick Setup Tab for PBS */} - + {/* Agent Install Tab for PBS */} + +
+

+ Install the Pulse agent on your Proxmox Backup Server. This is the recommended method as it provides: +

+
    +
  • One-command setup (creates API user and token automatically)
    • +
  • Built-in temperature monitoring (no SSH required)
    • +
  • AI features (execute commands via Pulse AI)
    • +
  • Automatic reconnection on network issues
    • +
+

+ Run this command on your PBS node: +

+
+ + + {agentInstallCommand() || 'Click the copy button to generate and copy the install command'} + +
+ +

{agentCommandError()}

+ +

+ The node will automatically appear in Pulse once the agent connects. +

+
+ + + {/* Quick Setup Tab for PBS (API Only) */} +

Just copy and run this one command on your Proxmox Backup Server:

@@ -1915,8 +2113,11 @@ export const NodeModal: Component = (props) => {
-
Install HTTPS proxy on this host
-
Generate a one-line installer command to run on the Proxmox host:
+
Temperature proxy (legacy)
+
+ Recommended: Install the Pulse agent instead (Settings → Agents) for temperatures + AI features. +
+
Or generate a one-line installer command for the standalone temperature proxy:
+
+
+
+ + {/* PVE Nodes Tab */}
@@ -2824,6 +2858,7 @@ const Settings: Component = (props) => { = (props) => { fallback={
Check proxy nodes is only available when the proxy socket - grants admin access. Run this diagnostic on the Proxmox host - or reinstall pulse-sensor-proxy in HTTP mode to manage nodes - remotely. + grants admin access. For best results, install the Pulse agent + on each Proxmox node (Settings → Agents).
} > diff --git a/frontend-modern/src/components/Settings/UnifiedAgents.tsx b/frontend-modern/src/components/Settings/UnifiedAgents.tsx index b8c84bf..ac9195c 100644 --- a/frontend-modern/src/components/Settings/UnifiedAgents.tsx +++ b/frontend-modern/src/components/Settings/UnifiedAgents.tsx @@ -108,6 +108,7 @@ export const UnifiedAgents: Component = () => { const [lookupError, setLookupError] = createSignal(null); const [lookupLoading, setLookupLoading] = createSignal(false); const [enableDocker, setEnableDocker] = createSignal(false); // Default to false - user must opt-in for Docker monitoring + const [enableProxmox, setEnableProxmox] = createSignal(false); // For Proxmox VE/PBS nodes - creates API token and auto-registers const [insecureMode, setInsecureMode] = createSignal(false); // For self-signed certificates (issue #806) createEffect(() => { @@ -236,6 +237,7 @@ export const UnifiedAgents: Component = () => { }; const getDockerFlag = () => enableDocker() ? ' --enable-docker' : ''; + const getProxmoxFlag = () => enableProxmox() ? ' --enable-proxmox' : ''; const getInsecureFlag = () => insecureMode() ? ' --insecure' : ''; const getCurlInsecureFlag = () => insecureMode() ? '-k' : ''; @@ -451,26 +453,48 @@ export const UnifiedAgents: Component = () => {
-
-

Installation commands

- - +
+
+

Installation commands

+
+ + + +
+
+ +
+

Proxmox auto-setup enabled

+

+ The agent will create a pulse-monitor user and API token on the Proxmox node, + then register it with Pulse automatically. Includes temperature monitoring. +

+
+
@@ -490,25 +514,22 @@ export const UnifiedAgents: Component = () => { if (insecureMode() && cmd.includes('curl -fsSL')) { cmd = cmd.replace('curl -fsSL', 'curl -kfsSL'); } + // For bash scripts (not PowerShell), append flags directly + const isBashScript = !cmd.includes('$env:') && !cmd.includes('irm'); // Append docker flag if enabled if (enableDocker()) { - // For PowerShell, we need to handle the env var or args differently if (cmd.includes('$env:PULSE_URL')) { - // Env var style: add $env:PULSE_ENABLE_DOCKER="true"; cmd = `$env:PULSE_ENABLE_DOCKER="true"; ` + cmd; - } else if (cmd.includes('irm')) { - // Simple irm style: no args passed to script directly in this snippet style - // Actually, the simple irm style relies on prompts, so flags don't apply directly unless we change the snippet - // But for the bash script, we append flags - if (!cmd.includes('irm')) { - cmd += getDockerFlag(); - } - } else { + } else if (isBashScript) { cmd += getDockerFlag(); } } + // Append proxmox flag if enabled (Linux only - Proxmox doesn't run on Windows/macOS) + if (enableProxmox() && isBashScript) { + cmd += getProxmoxFlag(); + } // Append insecure flag for agent if enabled - if (insecureMode() && !cmd.includes('$env:') && !cmd.includes('irm')) { + if (insecureMode() && isBashScript) { cmd += getInsecureFlag(); } return cmd; diff --git a/frontend-modern/src/types/ai.ts b/frontend-modern/src/types/ai.ts index 3b77bca..543d9e6 100644 --- a/frontend-modern/src/types/ai.ts +++ b/frontend-modern/src/types/ai.ts @@ -91,6 +91,7 @@ export interface AIStreamToolStartData { export interface AIStreamToolEndData { name: string; + input: string; output: string; success: boolean; } diff --git a/frontend-modern/src/types/api.ts b/frontend-modern/src/types/api.ts index 9a38ba0..5f10f10 100644 --- a/frontend-modern/src/types/api.ts +++ b/frontend-modern/src/types/api.ts @@ -329,6 +329,7 @@ export interface Host { loadAverage?: number[]; memory: Memory; disks?: Disk[]; + diskIO?: HostDiskIO[]; networkInterfaces?: HostNetworkInterface[]; sensors?: HostSensorSummary; raid?: HostRAIDArray[]; @@ -384,6 +385,17 @@ export interface HostRAIDDevice { slot: number; } +export interface HostDiskIO { + device: string; + readBytes?: number; + writeBytes?: number; + readOps?: number; + writeOps?: number; + readTimeMs?: number; + writeTimeMs?: number; + ioTimeMs?: number; +} + export interface HostLookupResponse { success: boolean; host: { diff --git a/frontend-modern/src/types/nodes.ts b/frontend-modern/src/types/nodes.ts index 82efd16..cf53e38 100644 --- a/frontend-modern/src/types/nodes.ts +++ b/frontend-modern/src/types/nodes.ts @@ -91,6 +91,7 @@ export type NodeConfig = (PVENodeConfig | PBSNodeConfig | PMGNodeConfig) & { temperature?: Temperature; displayName?: string; temperatureTransport?: TemperatureTransport; + source?: 'agent' | 'script' | ''; // How this node was registered }; export interface NodesResponse { diff --git a/frontend-modern/src/utils/format.ts b/frontend-modern/src/utils/format.ts index d3ab5c0..d57b5b2 100644 --- a/frontend-modern/src/utils/format.ts +++ b/frontend-modern/src/utils/format.ts @@ -25,6 +25,11 @@ export function formatPercent(value: number): string { return `${Math.round(value)}%`; } +export function formatNumber(value: number): string { + if (!Number.isFinite(value)) return '0'; + return value.toLocaleString(); +} + export function formatUptime(seconds: number, condensed = false): string { if (!seconds || seconds < 0) return '0s'; diff --git a/frontend-modern/vite.config.ts b/frontend-modern/vite.config.ts index 39f925a..e8ce005 100644 --- a/frontend-modern/vite.config.ts +++ b/frontend-modern/vite.config.ts @@ -51,6 +51,24 @@ export default defineConfig({ target: backendWsUrl, ws: true, changeOrigin: true, + // Browser WebSocket connections are long-lived, disable timeouts + proxyTimeout: 0, + timeout: 0, + configure: (proxy, _options) => { + proxy.options.timeout = 0; + proxy.options.proxyTimeout = 0; + + proxy.on('proxyReqWs', (proxyReq, req, socket) => { + socket.setTimeout(0); + socket.setNoDelay(true); + socket.setKeepAlive(true, 30000); + }); + proxy.on('open', (proxySocket) => { + proxySocket.setTimeout(0); + proxySocket.setNoDelay(true); + proxySocket.setKeepAlive(true, 30000); + }); + }, }, '/api/ai/execute/stream': { target: backendUrl, @@ -86,6 +104,32 @@ export default defineConfig({ target: backendWsUrl, ws: true, changeOrigin: true, + // Agent WebSocket connections are long-lived, disable timeouts + // proxyTimeout: 0 disables the proxy-to-target timeout + // timeout: 0 disables the client-to-proxy timeout + proxyTimeout: 0, + timeout: 0, + configure: (proxy, _options) => { + // Disable http-proxy's internal timeout (default is 2 minutes but seems to be 10s) + proxy.options.timeout = 0; + proxy.options.proxyTimeout = 0; + + proxy.on('proxyReqWs', (proxyReq, req, socket) => { + // Disable socket timeouts for WebSocket connections + socket.setTimeout(0); + socket.setNoDelay(true); + socket.setKeepAlive(true, 30000); + }); + proxy.on('open', (proxySocket) => { + // Also disable timeout on the proxy socket + proxySocket.setTimeout(0); + proxySocket.setNoDelay(true); + proxySocket.setKeepAlive(true, 30000); + }); + proxy.on('error', (err, req, res) => { + console.error('[Agent WS Proxy Error]', err.message); + }); + }, }, '/api': { target: backendUrl, diff --git a/internal/agentexec/server.go b/internal/agentexec/server.go index 82be036..05d288c 100644 --- a/internal/agentexec/server.go +++ b/internal/agentexec/server.go @@ -46,12 +46,31 @@ func NewServer(validateToken func(token string) bool) *Server { // HandleWebSocket handles incoming WebSocket connections from agents func (s *Server) HandleWebSocket(w http.ResponseWriter, r *http.Request) { + // CRITICAL: Clear http.Server deadlines BEFORE WebSocket upgrade. + // The http.Server.ReadTimeout sets a deadline on the underlying connection when + // the request starts. We must clear it before the upgrade or the connection will + // be closed when that deadline fires (typically ~15 seconds after connection). + // Use http.ResponseController (Go 1.20+) to clear the deadline. + rc := http.NewResponseController(w) + if err := rc.SetReadDeadline(time.Time{}); err != nil { + log.Debug().Err(err).Msg("Failed to clear read deadline via ResponseController") + } + if err := rc.SetWriteDeadline(time.Time{}); err != nil { + log.Debug().Err(err).Msg("Failed to clear write deadline via ResponseController") + } + conn, err := upgrader.Upgrade(w, r, nil) if err != nil { log.Error().Err(err).Msg("Failed to upgrade WebSocket connection") return } + // Also clear on the WebSocket's underlying connection as a safety net + if netConn := conn.NetConn(); netConn != nil { + netConn.SetReadDeadline(time.Time{}) + netConn.SetWriteDeadline(time.Time{}) + } + // Read first message (must be agent_register) conn.SetReadDeadline(time.Now().Add(30 * time.Second)) _, msgBytes, err := conn.ReadMessage() @@ -139,8 +158,27 @@ func (s *Server) HandleWebSocket(w http.ResponseWriter, r *http.Request) { Payload: RegisteredPayload{Success: true, Message: "Registered"}, }) - // Clear deadline for normal operation + // Clear deadline for normal operation - both on the WebSocket and underlying connection + // This is critical because Go's http.Server.ReadTimeout may have set a deadline + // on the underlying connection before the WebSocket upgrade occurred. conn.SetReadDeadline(time.Time{}) + conn.SetWriteDeadline(time.Time{}) + if netConn := conn.NetConn(); netConn != nil { + netConn.SetReadDeadline(time.Time{}) + netConn.SetWriteDeadline(time.Time{}) + } + + // Set up ping/pong handlers to keep connection alive + conn.SetPongHandler(func(appData string) error { + // Reset read deadline on pong received + conn.SetReadDeadline(time.Time{}) + return nil + }) + + // Start server-side ping loop to keep connection alive + pingDone := make(chan struct{}) + go s.pingLoop(ac, pingDone) + defer close(pingDone) // Run read loop (blocking) - don't use goroutine, or HTTP handler will close connection s.readLoop(ac) @@ -214,6 +252,28 @@ func (s *Server) readLoop(ac *agentConn) { } } +func (s *Server) pingLoop(ac *agentConn, done chan struct{}) { + ticker := time.NewTicker(5 * time.Second) + defer ticker.Stop() + + for { + select { + case <-done: + return + case <-ac.done: + return + case <-ticker.C: + ac.writeMu.Lock() + err := ac.conn.WriteControl(websocket.PingMessage, []byte{}, time.Now().Add(5*time.Second)) + ac.writeMu.Unlock() + if err != nil { + log.Debug().Err(err).Str("agent_id", ac.agent.AgentID).Msg("Failed to send ping to agent") + return + } + } + } +} + func (s *Server) sendMessage(conn *websocket.Conn, msg Message) error { msgBytes, err := json.Marshal(msg) if err != nil { diff --git a/internal/agentupdate/update.go b/internal/agentupdate/update.go index 37d5235..2808550 100644 --- a/internal/agentupdate/update.go +++ b/internal/agentupdate/update.go @@ -106,11 +106,11 @@ func (u *Updater) RunLoop(ctx context.Context) { return } - // Initial check after a short delay + // Initial check after a short delay (5s to quickly update outdated agents) select { case <-ctx.Done(): return - case <-time.After(30 * time.Second): + case <-time.After(5 * time.Second): u.CheckAndUpdate(ctx) } @@ -156,13 +156,23 @@ func (u *Updater) CheckAndUpdate(ctx context.Context) { return } - // Normalize both versions by stripping "v" prefix for comparison. - // Server returns version without prefix (e.g., "4.33.1"), but agent's - // CurrentVersion may include it (e.g., "v4.33.1") depending on build. - if utils.NormalizeVersion(serverVersion) == utils.NormalizeVersion(u.cfg.CurrentVersion) { + // Compare versions using semver comparison to only update to newer versions. + // This prevents accidental downgrades if the server temporarily has an older version. + currentNorm := utils.NormalizeVersion(u.cfg.CurrentVersion) + serverNorm := utils.NormalizeVersion(serverVersion) + + cmp := utils.CompareVersions(serverNorm, currentNorm) + if cmp == 0 { u.logger.Debug().Str("version", u.cfg.CurrentVersion).Msg("Agent is up to date") return } + if cmp < 0 { + u.logger.Debug(). + Str("currentVersion", currentNorm). + Str("serverVersion", serverNorm). + Msg("Server has older version, skipping downgrade") + return + } u.logger.Info(). Str("currentVersion", u.cfg.CurrentVersion). @@ -402,6 +412,10 @@ func (u *Updater) performUpdate(ctx context.Context) error { // Remove backup on success os.Remove(backupPath) + // Write previous version to a file so the agent can report "updated from X" on next start + updateInfoPath := filepath.Join(targetDir, ".pulse-update-info") + _ = os.WriteFile(updateInfoPath, []byte(u.cfg.CurrentVersion), 0644) + // On Unraid, also update the persistent copy on the flash drive // This ensures the update survives reboots if isUnraid() { @@ -433,6 +447,32 @@ func (u *Updater) performUpdate(ctx context.Context) error { return restartProcess(execPath) } +// GetUpdatedFromVersion checks if the agent was recently updated and returns the previous version. +// Returns empty string if no update info exists. Clears the info file after reading. +func GetUpdatedFromVersion() string { + execPath, err := os.Executable() + if err != nil { + return "" + } + + // Resolve symlinks to get the real path + realExecPath, err := filepath.EvalSymlinks(execPath) + if err != nil { + realExecPath = execPath + } + + updateInfoPath := filepath.Join(filepath.Dir(realExecPath), ".pulse-update-info") + data, err := os.ReadFile(updateInfoPath) + if err != nil { + return "" + } + + // Clear the file after reading (only report once) + os.Remove(updateInfoPath) + + return strings.TrimSpace(string(data)) +} + // determineArch returns the architecture string for download URLs (e.g., "linux-amd64", "darwin-arm64"). func determineArch() string { os := runtime.GOOS diff --git a/internal/ai/providers/anthropic.go b/internal/ai/providers/anthropic.go index ea1afd0..a5d1971 100644 --- a/internal/ai/providers/anthropic.go +++ b/internal/ai/providers/anthropic.go @@ -8,11 +8,15 @@ import ( "io" "net/http" "time" + + "github.com/rs/zerolog/log" ) const ( anthropicAPIURL = "https://api.anthropic.com/v1/messages" anthropicAPIVersion = "2023-06-01" + maxRetries = 3 + initialBackoff = 2 * time.Second ) // AnthropicClient implements the Provider interface for Anthropic's Claude API @@ -195,32 +199,76 @@ func (c *AnthropicClient) Chat(ctx context.Context, req ChatRequest) (*ChatRespo return nil, fmt.Errorf("failed to marshal request: %w", err) } - httpReq, err := http.NewRequestWithContext(ctx, "POST", anthropicAPIURL, bytes.NewReader(body)) - if err != nil { - return nil, fmt.Errorf("failed to create request: %w", err) - } + // Retry loop for transient errors (429, 529, 5xx) + var respBody []byte + var lastErr error - httpReq.Header.Set("Content-Type", "application/json") - httpReq.Header.Set("x-api-key", c.apiKey) - httpReq.Header.Set("anthropic-version", anthropicAPIVersion) + for attempt := 0; attempt <= maxRetries; attempt++ { + if attempt > 0 { + // Exponential backoff: 2s, 4s, 8s + backoff := initialBackoff * time.Duration(1<<(attempt-1)) + log.Warn(). + Int("attempt", attempt). + Dur("backoff", backoff). + Str("last_error", lastErr.Error()). + Msg("Retrying Anthropic API request after transient error") - resp, err := c.client.Do(httpReq) - if err != nil { - return nil, fmt.Errorf("request failed: %w", err) - } - defer resp.Body.Close() - - respBody, err := io.ReadAll(resp.Body) - if err != nil { - return nil, fmt.Errorf("failed to read response: %w", err) - } - - if resp.StatusCode != http.StatusOK { - var errResp anthropicError - if err := json.Unmarshal(respBody, &errResp); err == nil && errResp.Error.Message != "" { - return nil, fmt.Errorf("API error (%d): %s", resp.StatusCode, errResp.Error.Message) + select { + case <-ctx.Done(): + return nil, ctx.Err() + case <-time.After(backoff): + } } - return nil, fmt.Errorf("API error (%d): %s", resp.StatusCode, string(respBody)) + + httpReq, err := http.NewRequestWithContext(ctx, "POST", anthropicAPIURL, bytes.NewReader(body)) + if err != nil { + return nil, fmt.Errorf("failed to create request: %w", err) + } + + httpReq.Header.Set("Content-Type", "application/json") + httpReq.Header.Set("x-api-key", c.apiKey) + httpReq.Header.Set("anthropic-version", anthropicAPIVersion) + + resp, err := c.client.Do(httpReq) + if err != nil { + lastErr = fmt.Errorf("request failed: %w", err) + continue + } + + respBody, err = io.ReadAll(resp.Body) + resp.Body.Close() + if err != nil { + lastErr = fmt.Errorf("failed to read response: %w", err) + continue + } + + // Check if this is a retryable error + if resp.StatusCode == 429 || resp.StatusCode == 529 || resp.StatusCode >= 500 { + var errResp anthropicError + errMsg := string(respBody) + if err := json.Unmarshal(respBody, &errResp); err == nil && errResp.Error.Message != "" { + errMsg = errResp.Error.Message + } + lastErr = fmt.Errorf("API error (%d): %s", resp.StatusCode, errMsg) + continue + } + + // Non-retryable error + if resp.StatusCode != http.StatusOK { + var errResp anthropicError + if err := json.Unmarshal(respBody, &errResp); err == nil && errResp.Error.Message != "" { + return nil, fmt.Errorf("API error (%d): %s", resp.StatusCode, errResp.Error.Message) + } + return nil, fmt.Errorf("API error (%d): %s", resp.StatusCode, string(respBody)) + } + + // Success - break out of retry loop + lastErr = nil + break + } + + if lastErr != nil { + return nil, fmt.Errorf("failed after %d retries: %w", maxRetries, lastErr) } var anthropicResp anthropicResponse diff --git a/internal/ai/service.go b/internal/ai/service.go index e50af36..f1f0964 100644 --- a/internal/ai/service.go +++ b/internal/ai/service.go @@ -369,6 +369,7 @@ type ToolStartData struct { // ToolEndData is sent when a tool execution completes type ToolEndData struct { Name string `json:"name"` + Input string `json:"input"` Output string `json:"output"` Success bool `json:"success"` } @@ -663,7 +664,7 @@ Always execute the commands rather than telling the user how to do it.` // Stream tool end event callback(StreamEvent{ Type: "tool_end", - Data: ToolEndData{Name: tc.Name, Output: result, Success: execution.Success}, + Data: ToolEndData{Name: tc.Name, Input: toolInput, Output: result, Success: execution.Success}, }) } @@ -931,14 +932,26 @@ func (s *Service) executeOnAgent(ctx context.Context, req ExecuteRequest, comman // Fall back to context-based routing if VMID lookup didn't find anything if targetNode == "" { - // Extract node info from target ID (e.g., "delly-135" -> "delly") + // For host targets, use the hostname directly from context + if req.TargetType == "host" { + if hostname, ok := req.Context["hostname"].(string); ok && hostname != "" { + targetNode = strings.ToLower(hostname) + log.Debug(). + Str("hostname", hostname). + Str("target_type", req.TargetType). + Msg("Using hostname from context for host target routing") + } + } + // For VMs/containers, extract node info from target ID (e.g., "delly-135" -> "delly") // or from context (guest_node field) - if node, ok := req.Context["guest_node"].(string); ok && node != "" { - targetNode = strings.ToLower(node) - } else if req.TargetID != "" { - parts := strings.Split(req.TargetID, "-") - if len(parts) >= 2 { - targetNode = strings.ToLower(parts[0]) + if targetNode == "" { + if node, ok := req.Context["guest_node"].(string); ok && node != "" { + targetNode = strings.ToLower(node) + } else if req.TargetID != "" { + parts := strings.Split(req.TargetID, "-") + if len(parts) >= 2 { + targetNode = strings.ToLower(parts[0]) + } } } } diff --git a/internal/alerts/alerts_test.go b/internal/alerts/alerts_test.go index 3bcbedc..2e15fcc 100644 --- a/internal/alerts/alerts_test.go +++ b/internal/alerts/alerts_test.go @@ -14688,7 +14688,6 @@ func TestLoadActiveAlerts(t *testing.T) { t.Run("loads alerts from valid file", func(t *testing.T) { m := newTestManager(t) - m.ClearActiveAlerts() // Create an alert and save it startTime := time.Now().Add(-30 * time.Minute) @@ -14714,8 +14713,10 @@ func TestLoadActiveAlerts(t *testing.T) { // Save to disk _ = m.SaveActiveAlerts() - // Clear and reload - m.ClearActiveAlerts() + // Clear in-memory map only (don't use ClearActiveAlerts which triggers async save) + m.mu.Lock() + m.activeAlerts = make(map[string]*Alert) + m.mu.Unlock() err := m.LoadActiveAlerts() if err != nil { diff --git a/internal/api/config_handlers.go b/internal/api/config_handlers.go index 4b0c6f3..2571c0f 100644 --- a/internal/api/config_handlers.go +++ b/internal/api/config_handlers.go @@ -430,6 +430,7 @@ type NodeResponse struct { IsCluster bool `json:"isCluster,omitempty"` ClusterName string `json:"clusterName,omitempty"` ClusterEndpoints []config.ClusterEndpoint `json:"clusterEndpoints,omitempty"` + Source string `json:"source,omitempty"` // "agent" or "script" - how this node was registered } func determineTemperatureTransport(enabled bool, proxyURL, proxyToken string, socketAvailable bool, containerSSHBlocked bool) string { @@ -834,6 +835,7 @@ func (h *ConfigHandlers) GetAllNodesForAPI() []NodeResponse { IsCluster: pve.IsCluster, ClusterName: pve.ClusterName, ClusterEndpoints: pve.ClusterEndpoints, + Source: pve.Source, } nodes = append(nodes, node) } @@ -859,6 +861,7 @@ func (h *ConfigHandlers) GetAllNodesForAPI() []NodeResponse { MonitorPruneJobs: pbs.MonitorPruneJobs, MonitorGarbageJobs: pbs.MonitorGarbageJobs, Status: h.getNodeStatus("pbs", pbs.Name), + Source: pbs.Source, } nodes = append(nodes, node) } @@ -5539,7 +5542,7 @@ func (h *ConfigHandlers) HandleUpdateMockMode(w http.ResponseWriter, r *http.Req } } -// AutoRegisterRequest represents a request from the setup script to auto-register a node +// AutoRegisterRequest represents a request from the setup script or agent to auto-register a node type AutoRegisterRequest struct { Type string `json:"type"` // "pve" or "pbs" Host string `json:"host"` // The host URL @@ -5548,6 +5551,7 @@ type AutoRegisterRequest struct { ServerName string `json:"serverName"` // Hostname or IP SetupCode string `json:"setupCode,omitempty"` // One-time setup code for authentication (deprecated) AuthToken string `json:"authToken,omitempty"` // Direct auth token from URL (new approach) + Source string `json:"source,omitempty"` // "agent" or "script" - indicates how the node was registered // New secure fields RequestToken bool `json:"requestToken,omitempty"` // If true, Pulse will generate and return a token Username string `json:"username,omitempty"` // Username for creating token (e.g., "root@pam") @@ -5752,18 +5756,29 @@ func (h *ConfigHandlers) HandleAutoRegister(w http.ResponseWriter, r *http.Reque MonitorGarbageJobs: &boolFalse, } - // Check if a node with this host already exists + // Check if a node with this host or name already exists + // Match by host URL or by node name (to handle hostname vs IP differences) existingIndex := -1 if req.Type == "pve" { for i, node := range h.config.PVEInstances { - if node.Host == host { // Use normalized host for comparison + if node.Host == host { + existingIndex = i + break + } + // Also match by name if ServerName matches existing node name + if req.ServerName != "" && strings.EqualFold(node.Name, req.ServerName) { existingIndex = i break } } } else { for i, node := range h.config.PBSInstances { - if node.Host == host { // Use normalized host for comparison + if node.Host == host { + existingIndex = i + break + } + // Also match by name if ServerName matches existing node name + if req.ServerName != "" && strings.EqualFold(node.Name, req.ServerName) { existingIndex = i break } @@ -5780,6 +5795,10 @@ func (h *ConfigHandlers) HandleAutoRegister(w http.ResponseWriter, r *http.Reque instance.Password = "" instance.TokenName = nodeConfig.TokenName instance.TokenValue = nodeConfig.TokenValue + // Update source if provided (allows upgrade from script to agent) + if req.Source != "" { + instance.Source = req.Source + } // Check for cluster if not already detected if !instance.IsCluster { @@ -5809,6 +5828,10 @@ func (h *ConfigHandlers) HandleAutoRegister(w http.ResponseWriter, r *http.Reque instance.Password = "" instance.TokenName = nodeConfig.TokenName instance.TokenValue = nodeConfig.TokenValue + // Update source if provided (allows upgrade from script to agent) + if req.Source != "" { + instance.Source = req.Source + } // Keep other settings as they were } log.Info(). @@ -5864,6 +5887,7 @@ func (h *ConfigHandlers) HandleAutoRegister(w http.ResponseWriter, r *http.Reque IsCluster: isCluster, ClusterName: clusterName, ClusterEndpoints: clusterEndpoints, + Source: req.Source, // Track how this node was registered } h.config.PVEInstances = append(h.config.PVEInstances, newInstance) @@ -5911,6 +5935,7 @@ func (h *ConfigHandlers) HandleAutoRegister(w http.ResponseWriter, r *http.Reque MonitorVerifyJobs: monitorVerifyJobs, MonitorPruneJobs: monitorPruneJobs, MonitorGarbageJobs: monitorGarbageJobs, + Source: req.Source, // Track how this node was registered } h.config.PBSInstances = append(h.config.PBSInstances, newInstance) } @@ -6272,3 +6297,111 @@ func (h *ConfigHandlers) generateOrLoadSSHKey(sshDir, privateKeyPath, publicKeyP return publicKeyString } + +// AgentInstallCommandRequest represents a request for an agent install command +type AgentInstallCommandRequest struct { + Type string `json:"type"` // "pve" or "pbs" +} + +// AgentInstallCommandResponse contains the generated install command +type AgentInstallCommandResponse struct { + Command string `json:"command"` + Token string `json:"token"` +} + +// HandleAgentInstallCommand generates an API token and install command for agent-based Proxmox setup +func (h *ConfigHandlers) HandleAgentInstallCommand(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) + return + } + + var req AgentInstallCommandRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + http.Error(w, "Invalid request body", http.StatusBadRequest) + return + } + + // Validate type + if req.Type != "pve" && req.Type != "pbs" { + http.Error(w, "Type must be 'pve' or 'pbs'", http.StatusBadRequest) + return + } + + // Generate a new API token with host report and host manage scopes + rawToken, err := internalauth.GenerateAPIToken() + if err != nil { + log.Error().Err(err).Msg("Failed to generate API token for agent install") + http.Error(w, "Failed to generate API token", http.StatusInternalServerError) + return + } + + tokenName := fmt.Sprintf("proxmox-agent-%s-%d", req.Type, time.Now().Unix()) + scopes := []string{ + config.ScopeHostReport, + config.ScopeHostManage, + } + + record, err := config.NewAPITokenRecord(rawToken, tokenName, scopes) + if err != nil { + log.Error().Err(err).Str("token_name", tokenName).Msg("Failed to construct API token record") + http.Error(w, "Failed to generate token", http.StatusInternalServerError) + return + } + + // Persist the token + config.Mu.Lock() + h.config.APITokens = append(h.config.APITokens, *record) + h.config.SortAPITokens() + h.config.APITokenEnabled = true + + if h.persistence != nil { + if err := h.persistence.SaveAPITokens(h.config.APITokens); err != nil { + // Rollback the in-memory addition + h.config.APITokens = h.config.APITokens[:len(h.config.APITokens)-1] + config.Mu.Unlock() + log.Error().Err(err).Msg("Failed to persist API tokens after creation") + http.Error(w, "Failed to save token to disk: "+err.Error(), http.StatusInternalServerError) + return + } + } + config.Mu.Unlock() + + // Derive Pulse URL from the request + host := r.Host + if parsedHost, parsedPort, err := net.SplitHostPort(host); err == nil { + if (parsedHost == "127.0.0.1" || parsedHost == "localhost") && parsedPort == strconv.Itoa(h.config.FrontendPort) { + // Prefer a user-configured public URL when we're running on loopback + if publicURL := strings.TrimSpace(h.config.PublicURL); publicURL != "" { + if parsedURL, err := url.Parse(publicURL); err == nil && parsedURL.Host != "" { + host = parsedURL.Host + } + } + } + } + + // Detect protocol - check both TLS and proxy headers + scheme := "http" + if r.TLS != nil || r.Header.Get("X-Forwarded-Proto") == "https" { + scheme = "https" + } + pulseURL := fmt.Sprintf("%s://%s", scheme, host) + + // Generate the install command + command := fmt.Sprintf(`curl -fsSL %s/install.sh | bash -s -- \ + --url %s \ + --token %s \ + --enable-proxmox`, + pulseURL, pulseURL, rawToken) + + log.Info(). + Str("token_name", tokenName). + Str("type", req.Type). + Msg("Generated agent install command with API token") + + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(AgentInstallCommandResponse{ + Command: command, + Token: rawToken, + }) +} diff --git a/internal/api/router.go b/internal/api/router.go index 2cd5a76..b3a582b 100644 --- a/internal/api/router.go +++ b/internal/api/router.go @@ -923,6 +923,9 @@ func (r *Router) setupRoutes() { // Generate setup script URL with temporary token (for authenticated users) r.mux.HandleFunc("/api/setup-script-url", r.configHandlers.HandleSetupScriptURL) + // Generate agent install command with API token (for authenticated users) + r.mux.HandleFunc("/api/agent-install-command", RequireAuth(r.config, r.configHandlers.HandleAgentInstallCommand)) + // Auto-register route for setup scripts r.mux.HandleFunc("/api/auto-register", r.configHandlers.HandleAutoRegister) // Discovery endpoint diff --git a/internal/config/config.go b/internal/config/config.go index 1308630..976bda2 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -424,6 +424,9 @@ type PVEInstance struct { IsCluster bool // True if this is a cluster ClusterName string // Cluster name if applicable ClusterEndpoints []ClusterEndpoint // All discovered cluster nodes + + // Agent tracking + Source string // "agent" or "script" - how this node was registered (empty = legacy/manual) } // ClusterEndpoint represents a single node in a cluster @@ -461,6 +464,9 @@ type PBSInstance struct { MonitorGarbageJobs bool TemperatureMonitoringEnabled *bool // Monitor temperature via SSH (nil = use global setting, true/false = override) SSHPort int // SSH port for temperature monitoring (0 = use global default) + + // Agent tracking + Source string // "agent" or "script" - how this node was registered (empty = legacy/manual) } // PMGInstance represents a Proxmox Mail Gateway connection diff --git a/internal/hostagent/agent.go b/internal/hostagent/agent.go index 74372c1..d01160e 100644 --- a/internal/hostagent/agent.go +++ b/internal/hostagent/agent.go @@ -12,6 +12,7 @@ import ( "strings" "time" + "github.com/rcourtman/pulse-go-rewrite/internal/agentupdate" "github.com/rcourtman/pulse-go-rewrite/internal/buffer" "github.com/rcourtman/pulse-go-rewrite/internal/hostmetrics" "github.com/rcourtman/pulse-go-rewrite/internal/mdadm" @@ -35,6 +36,10 @@ type Config struct { RunOnce bool LogLevel zerolog.Level Logger *zerolog.Logger + + // Proxmox integration + EnableProxmox bool // If true, creates Proxmox API token and registers node on startup + ProxmoxType string // "pve", "pbs", or "" for auto-detect } // Agent is responsible for collecting host metrics and shipping them to Pulse. @@ -54,6 +59,7 @@ type Agent struct { machineID string agentID string agentVersion string + updatedFrom string // Previous version if recently auto-updated (reported once) interval time.Duration trimmedPulseURL string reportBuffer *buffer.Queue[agentshost.Report] @@ -169,6 +175,15 @@ func New(cfg Config) (*Agent, error) { const bufferCapacity = 60 + // Check if agent was recently auto-updated (only reported once per restart) + updatedFrom := agentupdate.GetUpdatedFromVersion() + if updatedFrom != "" { + logger.Info(). + Str("previousVersion", updatedFrom). + Str("currentVersion", agentVersion). + Msg("Agent was auto-updated") + } + agent := &Agent{ cfg: cfg, logger: logger, @@ -184,6 +199,7 @@ func New(cfg Config) (*Agent, error) { machineID: machineID, agentID: agentID, agentVersion: agentVersion, + updatedFrom: updatedFrom, interval: cfg.Interval, trimmedPulseURL: pulseURL, reportBuffer: buffer.New[agentshost.Report](bufferCapacity), @@ -201,6 +217,11 @@ func (a *Agent) Run(ctx context.Context) error { return a.runOnce(ctx) } + // Proxmox setup (if enabled) + if a.cfg.EnableProxmox { + a.runProxmoxSetup(ctx) + } + // Start command client in background for AI command execution if a.commandClient != nil { go func() { @@ -304,6 +325,7 @@ func (a *Agent) buildReport(ctx context.Context) (agentshost.Report, error) { Type: a.cfg.AgentType, IntervalSeconds: int(a.interval / time.Second), Hostname: a.hostname, + UpdatedFrom: a.updatedFrom, }, Host: agentshost.HostInfo{ ID: a.machineID, @@ -325,6 +347,7 @@ func (a *Agent) buildReport(ctx context.Context) (agentshost.Report, error) { Memory: snapshot.Memory, }, Disks: append([]agentshost.Disk(nil), snapshot.Disks...), + DiskIO: append([]agentshost.DiskIO(nil), snapshot.DiskIO...), Network: append([]agentshost.NetworkInterface(nil), snapshot.Network...), Sensors: sensorData, RAID: raidData, @@ -458,3 +481,42 @@ func (a *Agent) collectRAIDArrays(ctx context.Context) []agentshost.RAIDArray { return arrays } + +// runProxmoxSetup performs one-time Proxmox API token setup and node registration. +func (a *Agent) runProxmoxSetup(ctx context.Context) { + a.logger.Info().Msg("Proxmox mode enabled, checking setup...") + + setup := NewProxmoxSetup( + a.logger, + a.httpClient, + a.trimmedPulseURL, + a.cfg.APIToken, + a.cfg.ProxmoxType, + a.hostname, + a.cfg.InsecureSkipVerify, + ) + + result, err := setup.Run(ctx) + if err != nil { + a.logger.Error().Err(err).Msg("Proxmox setup failed") + return + } + + if result == nil { + // Already registered + return + } + + if result.Registered { + a.logger.Info(). + Str("type", result.ProxmoxType). + Str("host", result.NodeHost). + Str("token_id", result.TokenID). + Msg("Proxmox node registered successfully") + } else { + a.logger.Warn(). + Str("type", result.ProxmoxType). + Str("host", result.NodeHost). + Msg("Proxmox token created but registration failed (node may need manual configuration)") + } +} diff --git a/internal/hostagent/commands.go b/internal/hostagent/commands.go index 5b77d57..53e82d6 100644 --- a/internal/hostagent/commands.go +++ b/internal/hostagent/commands.go @@ -140,7 +140,7 @@ func (c *CommandClient) connectAndHandle(ctx context.Context) error { MinVersion: tls.VersionTLS12, InsecureSkipVerify: c.insecureSkipVerify, }, - HandshakeTimeout: 10 * time.Second, + HandshakeTimeout: 45 * time.Second, } // Connect @@ -172,6 +172,15 @@ func (c *CommandClient) connectAndHandle(ctx context.Context) error { c.logger.Info().Msg("Connected and registered with Pulse command server") + // Clear any deadlines that may have been set during handshake + // The HandshakeTimeout in the Dialer may have set a deadline on the underlying connection + conn.SetReadDeadline(time.Time{}) + conn.SetWriteDeadline(time.Time{}) + if netConn := conn.NetConn(); netConn != nil { + netConn.SetReadDeadline(time.Time{}) + netConn.SetWriteDeadline(time.Time{}) + } + // Start ping loop pingDone := make(chan struct{}) go c.pingLoop(ctx, conn, pingDone) diff --git a/internal/hostagent/proxmox_setup.go b/internal/hostagent/proxmox_setup.go new file mode 100644 index 0000000..5d021ac --- /dev/null +++ b/internal/hostagent/proxmox_setup.go @@ -0,0 +1,344 @@ +package hostagent + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "net/http" + "os" + "os/exec" + "regexp" + "strings" + "time" + + "github.com/rs/zerolog" +) + +// ProxmoxSetup handles one-time Proxmox API token creation and node registration. +type ProxmoxSetup struct { + logger zerolog.Logger + httpClient *http.Client + pulseURL string + apiToken string + proxmoxType string // "pve", "pbs", or "" for auto-detect + hostname string + insecureSkipVerify bool +} + +// ProxmoxSetupResult contains the result of a successful Proxmox setup. +type ProxmoxSetupResult struct { + ProxmoxType string // "pve" or "pbs" + TokenID string // Full token ID (e.g., "pulse-monitor@pam!pulse-1234567890") + TokenValue string // The secret token value + NodeHost string // The host URL for auto-registration + Registered bool // Whether the node was successfully registered with Pulse +} + +const ( + proxmoxUser = "pulse-monitor" + proxmoxUserPVE = "pulse-monitor@pam" + proxmoxUserPBS = "pulse-monitor@pbs" + proxmoxComment = "Pulse monitoring service" + stateFilePath = "/var/lib/pulse-agent/proxmox-registered" + stateFileDir = "/var/lib/pulse-agent" +) + +// NewProxmoxSetup creates a new ProxmoxSetup instance. +func NewProxmoxSetup(logger zerolog.Logger, httpClient *http.Client, pulseURL, apiToken, proxmoxType, hostname string, insecure bool) *ProxmoxSetup { + return &ProxmoxSetup{ + logger: logger, + httpClient: httpClient, + pulseURL: strings.TrimRight(pulseURL, "/"), + apiToken: apiToken, + proxmoxType: proxmoxType, + hostname: hostname, + insecureSkipVerify: insecure, + } +} + +// Run executes the Proxmox setup process: +// 1. Detects Proxmox type (if not specified) +// 2. Creates the monitoring user and API token +// 3. Registers the node with Pulse via auto-register +func (p *ProxmoxSetup) Run(ctx context.Context) (*ProxmoxSetupResult, error) { + // Check if already registered (idempotency) + if p.isAlreadyRegistered() { + p.logger.Info().Msg("Proxmox node already registered, skipping setup") + return nil, nil + } + + // Detect Proxmox type + ptype := p.proxmoxType + if ptype == "" { + detected := p.detectProxmoxType() + if detected == "" { + return nil, fmt.Errorf("this system does not appear to be a Proxmox VE or PBS node") + } + ptype = detected + p.logger.Info().Str("type", ptype).Msg("Auto-detected Proxmox type") + } + + // Create monitoring user and token + tokenID, tokenValue, err := p.setupToken(ctx, ptype) + if err != nil { + return nil, fmt.Errorf("failed to create Proxmox API token: %w", err) + } + + p.logger.Info().Str("token_id", tokenID).Msg("Created Proxmox API token") + + // Get the host URL for registration + hostURL := p.getHostURL(ptype) + + // Register with Pulse + registered := false + if err := p.registerWithPulse(ctx, ptype, hostURL, tokenID, tokenValue); err != nil { + p.logger.Warn().Err(err).Msg("Failed to register with Pulse (node may already exist)") + } else { + registered = true + p.markAsRegistered() + p.logger.Info().Str("host", hostURL).Msg("Successfully registered Proxmox node with Pulse") + } + + return &ProxmoxSetupResult{ + ProxmoxType: ptype, + TokenID: tokenID, + TokenValue: tokenValue, + NodeHost: hostURL, + Registered: registered, + }, nil +} + +// detectProxmoxType checks for pvesh (PVE) or proxmox-backup-manager (PBS). +func (p *ProxmoxSetup) detectProxmoxType() string { + // Check for PVE first + if _, err := exec.LookPath("pvesh"); err == nil { + return "pve" + } + // Check for PBS + if _, err := exec.LookPath("proxmox-backup-manager"); err == nil { + return "pbs" + } + return "" +} + +// setupToken creates the monitoring user and API token. +func (p *ProxmoxSetup) setupToken(ctx context.Context, ptype string) (string, string, error) { + tokenName := fmt.Sprintf("pulse-%d", time.Now().Unix()) + + if ptype == "pve" { + return p.setupPVEToken(ctx, tokenName) + } + return p.setupPBSToken(ctx, tokenName) +} + +// setupPVEToken creates a PVE monitoring user and token. +func (p *ProxmoxSetup) setupPVEToken(ctx context.Context, tokenName string) (string, string, error) { + // Create user (ignore error if already exists) + _ = runCommand(ctx, "pveum", "user", "add", proxmoxUserPVE, "--comment", proxmoxComment) + + // Add PVEAuditor role + if err := runCommand(ctx, "pveum", "aclmod", "/", "-user", proxmoxUserPVE, "-role", "PVEAuditor"); err != nil { + p.logger.Warn().Err(err).Msg("Failed to add PVEAuditor role (may already exist)") + } + + // Try to create PulseMonitor role with additional privileges + _ = runCommand(ctx, "pveum", "role", "add", "PulseMonitor", "-privs", "Sys.Audit,VM.Monitor,Datastore.Audit") + _ = runCommand(ctx, "pveum", "aclmod", "/", "-user", proxmoxUserPVE, "-role", "PulseMonitor") + + // Create token with privilege separation disabled + output, err := runCommandOutput(ctx, "pveum", "user", "token", "add", proxmoxUserPVE, tokenName, "--privsep", "0") + if err != nil { + return "", "", fmt.Errorf("failed to create token: %w", err) + } + + // Parse token value from output + tokenValue := p.parseTokenValue(output) + if tokenValue == "" { + return "", "", fmt.Errorf("failed to parse token value from pveum output") + } + + tokenID := fmt.Sprintf("%s!%s", proxmoxUserPVE, tokenName) + return tokenID, tokenValue, nil +} + +// setupPBSToken creates a PBS monitoring user and token. +func (p *ProxmoxSetup) setupPBSToken(ctx context.Context, tokenName string) (string, string, error) { + // Create user (ignore error if already exists) + _ = runCommand(ctx, "proxmox-backup-manager", "user", "create", proxmoxUserPBS) + + // Add Audit role + _ = runCommand(ctx, "proxmox-backup-manager", "acl", "update", "/", "Audit", "--auth-id", proxmoxUserPBS) + + // Create token + output, err := runCommandOutput(ctx, "proxmox-backup-manager", "user", "generate-token", proxmoxUserPBS, tokenName) + if err != nil { + return "", "", fmt.Errorf("failed to create token: %w", err) + } + + // Parse token value from JSON output + tokenValue := p.parsePBSTokenValue(output) + if tokenValue == "" { + return "", "", fmt.Errorf("failed to parse token value from PBS output") + } + + // Add Audit role for the token itself + tokenID := fmt.Sprintf("%s!%s", proxmoxUserPBS, tokenName) + _ = runCommand(ctx, "proxmox-backup-manager", "acl", "update", "/", "Audit", "--auth-id", tokenID) + + return tokenID, tokenValue, nil +} + +// parseTokenValue extracts the token value from PVE pveum output. +// The output format is typically a table with columns: │ key │ value │ +// Example output: +// ┌──────────────┬──────────────────────────────────────┐ +// │ key │ value │ +// ╞══════════════╪══════════════════════════════════════╡ +// │ full-tokenid │ pulse-monitor@pam!pulse-token │ +// ├──────────────┼──────────────────────────────────────┤ +// │ info │ {"privsep":"0"} │ +// ├──────────────┼──────────────────────────────────────┤ +// │ value │ xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx │ +// └──────────────┴──────────────────────────────────────┘ +func (p *ProxmoxSetup) parseTokenValue(output string) string { + // Look for the "value" row in the table output - where "value" is the KEY column + // We need to find the row where parts[1] (key column) is "value", not any row + // containing the word "value" (which would match the header row) + lines := strings.Split(output, "\n") + for _, line := range lines { + // Split by the box drawing character │ (U+2502) + parts := strings.Split(line, "│") + if len(parts) >= 3 { + key := strings.TrimSpace(parts[1]) + if key == "value" { + return strings.TrimSpace(parts[2]) + } + } + } + + // Fallback: try regex for UUID-like token + re := regexp.MustCompile(`[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12}`) + if match := re.FindString(output); match != "" { + return match + } + + return "" +} + +// parsePBSTokenValue extracts the token value from PBS JSON output. +func (p *ProxmoxSetup) parsePBSTokenValue(output string) string { + // PBS outputs JSON like: {"tokenid": "...", "value": "..."} + var result struct { + Value string `json:"value"` + } + if err := json.Unmarshal([]byte(output), &result); err == nil && result.Value != "" { + return result.Value + } + + // Fallback: extract from "value": "..." pattern + re := regexp.MustCompile(`"value"\s*:\s*"([^"]+)"`) + if matches := re.FindStringSubmatch(output); len(matches) >= 2 { + return matches[1] + } + + return "" +} + +// getHostURL constructs the host URL for this Proxmox node. +// Prefers IP address over hostname since hostnames are often not DNS-resolvable. +func (p *ProxmoxSetup) getHostURL(ptype string) string { + port := "8006" + if ptype == "pbs" { + port = "8007" + } + + // Always prefer IP address since hostnames often can't be resolved by Pulse + // (e.g., "pi" hostname won't resolve via DNS) + if out, err := exec.Command("hostname", "-I").Output(); err == nil { + ips := strings.Fields(string(out)) + if len(ips) > 0 { + // Use first non-loopback IP + for _, ip := range ips { + if ip != "127.0.0.1" && ip != "::1" { + return fmt.Sprintf("https://%s:%s", ip, port) + } + } + } + } + + // Fallback to hostname if IP detection failed + hostname := p.hostname + if hostname == "" { + hostname = "localhost" + } + return fmt.Sprintf("https://%s:%s", hostname, port) +} + +// registerWithPulse calls the auto-register endpoint to add the node. +func (p *ProxmoxSetup) registerWithPulse(ctx context.Context, ptype, hostURL, tokenID, tokenValue string) error { + payload := map[string]interface{}{ + "type": ptype, + "host": hostURL, + "serverName": p.hostname, + "tokenId": tokenID, + "tokenValue": tokenValue, + "source": "agent", // Indicates this was registered via agent + } + + body, err := json.Marshal(payload) + if err != nil { + return fmt.Errorf("marshal payload: %w", err) + } + + req, err := http.NewRequestWithContext(ctx, http.MethodPost, p.pulseURL+"/api/auto-register", bytes.NewReader(body)) + if err != nil { + return fmt.Errorf("create request: %w", err) + } + + req.Header.Set("Content-Type", "application/json") + req.Header.Set("X-API-Token", p.apiToken) + + resp, err := p.httpClient.Do(req) + if err != nil { + return fmt.Errorf("request failed: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode >= 400 { + return fmt.Errorf("auto-register returned %d", resp.StatusCode) + } + + return nil +} + +// isAlreadyRegistered checks if we've already done Proxmox setup. +func (p *ProxmoxSetup) isAlreadyRegistered() bool { + _, err := os.Stat(stateFilePath) + return err == nil +} + +// markAsRegistered creates a state file to indicate setup is complete. +func (p *ProxmoxSetup) markAsRegistered() { + if err := os.MkdirAll(stateFileDir, 0755); err != nil { + p.logger.Warn().Err(err).Msg("Failed to create state directory") + return + } + + if err := os.WriteFile(stateFilePath, []byte(time.Now().Format(time.RFC3339)), 0644); err != nil { + p.logger.Warn().Err(err).Msg("Failed to write state file") + } +} + +// runCommand executes a command and returns any error. +func runCommand(ctx context.Context, name string, args ...string) error { + cmd := exec.CommandContext(ctx, name, args...) + return cmd.Run() +} + +// runCommandOutput executes a command and returns the output. +func runCommandOutput(ctx context.Context, name string, args ...string) (string, error) { + cmd := exec.CommandContext(ctx, name, args...) + output, err := cmd.CombinedOutput() + return string(output), err +} diff --git a/internal/hostmetrics/collector.go b/internal/hostmetrics/collector.go index 2f2f418..6f1bd38 100644 --- a/internal/hostmetrics/collector.go +++ b/internal/hostmetrics/collector.go @@ -18,14 +18,15 @@ import ( // System call wrappers for testing var ( - cpuCounts = gocpu.CountsWithContext - cpuPercent = gocpu.PercentWithContext - loadAvg = goload.AvgWithContext - virtualMemory = gomem.VirtualMemoryWithContext - diskPartitions = godisk.PartitionsWithContext - diskUsage = godisk.UsageWithContext - netInterfaces = gonet.InterfacesWithContext - netIOCounters = gonet.IOCountersWithContext + cpuCounts = gocpu.CountsWithContext + cpuPercent = gocpu.PercentWithContext + loadAvg = goload.AvgWithContext + virtualMemory = gomem.VirtualMemoryWithContext + diskPartitions = godisk.PartitionsWithContext + diskUsage = godisk.UsageWithContext + diskIOCounters = godisk.IOCountersWithContext + netInterfaces = gonet.InterfacesWithContext + netIOCounters = gonet.IOCountersWithContext ) // Snapshot represents a host resource utilisation sample. @@ -35,6 +36,7 @@ type Snapshot struct { LoadAverage []float64 Memory agentshost.MemoryMetric Disks []agentshost.Disk + DiskIO []agentshost.DiskIO Network []agentshost.NetworkInterface } @@ -77,6 +79,7 @@ func Collect(ctx context.Context) (Snapshot, error) { } snapshot.Disks = collectDisks(collectCtx) + snapshot.DiskIO = collectDiskIO(collectCtx) snapshot.Network = collectNetwork(collectCtx) return snapshot, nil @@ -234,3 +237,78 @@ func isLoopback(flags []string) bool { } return false } + +// collectDiskIO gathers I/O statistics for physical block devices. +// Only reports whole disks (nvme0n1, sda), not partitions (nvme0n1p1, sda1). +func collectDiskIO(ctx context.Context) []agentshost.DiskIO { + counters, err := diskIOCounters(ctx) + if err != nil { + return nil + } + + devices := make([]agentshost.DiskIO, 0, len(counters)) + for name, stats := range counters { + // Skip partitions - only report whole devices + if isPartition(name) { + continue + } + // Skip loop devices and ram disks + if strings.HasPrefix(name, "loop") || strings.HasPrefix(name, "ram") { + continue + } + // Skip device-mapper and md devices (report at physical level) + if strings.HasPrefix(name, "dm-") { + continue + } + + devices = append(devices, agentshost.DiskIO{ + Device: name, + ReadBytes: stats.ReadBytes, + WriteBytes: stats.WriteBytes, + ReadOps: stats.ReadCount, + WriteOps: stats.WriteCount, + ReadTime: stats.ReadTime, + WriteTime: stats.WriteTime, + IOTime: stats.IoTime, + }) + } + + sort.Slice(devices, func(i, j int) bool { return devices[i].Device < devices[j].Device }) + return devices +} + +// isPartition returns true if the device name looks like a partition +// e.g., sda1, nvme0n1p1, vda2 +func isPartition(name string) bool { + // NVMe partitions: nvme0n1p1, nvme0n1p2 + if strings.Contains(name, "n") && strings.Contains(name, "p") { + // Check if it ends with pN where N is a digit + idx := strings.LastIndex(name, "p") + if idx > 0 && idx < len(name)-1 { + rest := name[idx+1:] + if len(rest) > 0 && rest[0] >= '0' && rest[0] <= '9' { + return true + } + } + } + // Traditional partitions: sda1, vda2, hda1 + if len(name) > 2 { + last := name[len(name)-1] + if last >= '0' && last <= '9' { + // Check if second-to-last is a letter (sda1) or also a digit (sda10) + secondLast := name[len(name)-2] + if (secondLast >= 'a' && secondLast <= 'z') || (secondLast >= '0' && secondLast <= '9') { + // Exclude things like "md0" (whole device) - check for common prefixes + if strings.HasPrefix(name, "sd") || strings.HasPrefix(name, "vd") || + strings.HasPrefix(name, "hd") || strings.HasPrefix(name, "xvd") { + return true + } + } + } + } + // ZFS devices: zd0p1, zd16p1 + if strings.HasPrefix(name, "zd") && strings.Contains(name, "p") { + return true + } + return false +} diff --git a/internal/hostmetrics/collector_test.go b/internal/hostmetrics/collector_test.go index e912b75..453c334 100644 --- a/internal/hostmetrics/collector_test.go +++ b/internal/hostmetrics/collector_test.go @@ -1,133 +1,25 @@ package hostmetrics import ( - "context" - "testing" - - godisk "github.com/shirou/gopsutil/v4/disk" - "github.com/stretchr/testify/assert" + "context" + "encoding/json" + "testing" ) -func TestCollectDisks_ZFS_Deduplication(t *testing.T) { - // Save original functions and restore after test - origDiskPartitions := diskPartitions - origDiskUsage := diskUsage - defer func() { - diskPartitions = origDiskPartitions - diskUsage = origDiskUsage - }() - - // Mock partitions - diskPartitions = func(ctx context.Context, all bool) ([]godisk.PartitionStat, error) { - return []godisk.PartitionStat{ - {Device: "tank/dataset1", Mountpoint: "/mnt/dataset1", Fstype: "zfs"}, - {Device: "tank/dataset2", Mountpoint: "/mnt/dataset2", Fstype: "zfs"}, - {Device: "tank/dataset3", Mountpoint: "/mnt/dataset3", Fstype: "zfs"}, - }, nil - } - - // Mock usage - diskUsage = func(ctx context.Context, path string) (*godisk.UsageStat, error) { - // All datasets on the same pool share the same free space - return &godisk.UsageStat{ - Total: 1000, - Used: 500, - Free: 500, - UsedPercent: 50.0, - }, nil - } - - // Mock zpool stats query to fail so we test the fallback logic (which uses the datasets) - // We can't easily mock queryZpoolStats because it's in zfs.go and not exported/variable-ized in the same way, - // but we can rely on the fact that `exec.LookPath("zpool")` will likely fail or `zpool list` will fail in the test environment. - // However, to be sure, we should probably mock it. - // But wait, `summarizeZFSPools` calls `queryZpoolStats`. - // `queryZpoolStats` is a variable in zfs.go! `var queryZpoolStats = fetchZpoolStats` - - // Let's mock queryZpoolStats too. - origQueryZpoolStats := queryZpoolStats - defer func() { queryZpoolStats = origQueryZpoolStats }() - - queryZpoolStats = func(ctx context.Context, pools []string) (map[string]zpoolStats, error) { - return nil, assert.AnError // Simulate failure - } - - ctx := context.Background() - disks := collectDisks(ctx) - - // Should return 1 disk (the pool), not 3 - assert.Len(t, disks, 1) - assert.Equal(t, "tank", disks[0].Device) - assert.Equal(t, int64(1000), disks[0].TotalBytes) -} - -func TestCollectDisks_ZFS_CaseInsensitive(t *testing.T) { - // Save original functions - origDiskPartitions := diskPartitions - origDiskUsage := diskUsage - defer func() { - diskPartitions = origDiskPartitions - diskUsage = origDiskUsage - }() - - // Mock partitions with "ZFS" (uppercase) - diskPartitions = func(ctx context.Context, all bool) ([]godisk.PartitionStat, error) { - return []godisk.PartitionStat{ - {Device: "tank/dataset1", Mountpoint: "/mnt/dataset1", Fstype: "ZFS"}, - {Device: "tank/dataset2", Mountpoint: "/mnt/dataset2", Fstype: "ZFS"}, - }, nil - } - - diskUsage = func(ctx context.Context, path string) (*godisk.UsageStat, error) { - return &godisk.UsageStat{Total: 1000, Used: 500, Free: 500}, nil - } - - // Mock queryZpoolStats to fail - origQueryZpoolStats := queryZpoolStats - defer func() { queryZpoolStats = origQueryZpoolStats }() - queryZpoolStats = func(ctx context.Context, pools []string) (map[string]zpoolStats, error) { - return nil, assert.AnError - } - - ctx := context.Background() - disks := collectDisks(ctx) - - // If case-sensitive check fails, it will return 2 disks - // If we fix it, it should return 1 - assert.Len(t, disks, 1) -} - -func TestCollectDisks_ZFS_Fuse(t *testing.T) { - // Save original functions - origDiskPartitions := diskPartitions - origDiskUsage := diskUsage - defer func() { - diskPartitions = origDiskPartitions - diskUsage = origDiskUsage - }() - - // Mock partitions with "fuse.zfs" - diskPartitions = func(ctx context.Context, all bool) ([]godisk.PartitionStat, error) { - return []godisk.PartitionStat{ - {Device: "tank/dataset1", Mountpoint: "/mnt/dataset1", Fstype: "fuse.zfs"}, - {Device: "tank/dataset2", Mountpoint: "/mnt/dataset2", Fstype: "fuse.zfs"}, - }, nil - } - - diskUsage = func(ctx context.Context, path string) (*godisk.UsageStat, error) { - return &godisk.UsageStat{Total: 1000, Used: 500, Free: 500}, nil - } - - // Mock queryZpoolStats to fail - origQueryZpoolStats := queryZpoolStats - defer func() { queryZpoolStats = origQueryZpoolStats }() - queryZpoolStats = func(ctx context.Context, pools []string) (map[string]zpoolStats, error) { - return nil, assert.AnError - } - - ctx := context.Background() - disks := collectDisks(ctx) - - // If check fails, it will return 2 disks - assert.Len(t, disks, 1) +func TestCollectDiskIO(t *testing.T) { + ctx := context.Background() + + snapshot, err := Collect(ctx) + if err != nil { + t.Fatalf("Collect failed: %v", err) + } + + t.Logf("DiskIO count: %d", len(snapshot.DiskIO)) + + if len(snapshot.DiskIO) == 0 { + t.Error("Expected disk IO data but got none") + } + + data, _ := json.MarshalIndent(snapshot.DiskIO, "", " ") + t.Logf("DiskIO data:\n%s", string(data)) } diff --git a/internal/models/converters.go b/internal/models/converters.go index b8ac272..5001378 100644 --- a/internal/models/converters.go +++ b/internal/models/converters.go @@ -336,6 +336,10 @@ func (h Host) ToFrontend() HostFrontend { host.Disks = append([]Disk(nil), h.Disks...) } + if len(h.DiskIO) > 0 { + host.DiskIO = append([]DiskIO(nil), h.DiskIO...) + } + if len(h.NetworkInterfaces) > 0 { host.NetworkInterfaces = make([]HostNetworkInterface, len(h.NetworkInterfaces)) copy(host.NetworkInterfaces, h.NetworkInterfaces) diff --git a/internal/models/models.go b/internal/models/models.go index af20f17..cb13780 100644 --- a/internal/models/models.go +++ b/internal/models/models.go @@ -162,6 +162,7 @@ type Host struct { Memory Memory `json:"memory"` LoadAverage []float64 `json:"loadAverage,omitempty"` Disks []Disk `json:"disks,omitempty"` + DiskIO []DiskIO `json:"diskIO,omitempty"` NetworkInterfaces []HostNetworkInterface `json:"networkInterfaces,omitempty"` Sensors HostSensorSummary `json:"sensors,omitempty"` RAID []HostRAIDArray `json:"raid,omitempty"` @@ -219,6 +220,19 @@ type HostRAIDDevice struct { Slot int `json:"slot"` } +// DiskIO represents I/O statistics for a block device. +// Counters are cumulative since boot. +type DiskIO struct { + Device string `json:"device"` + ReadBytes uint64 `json:"readBytes,omitempty"` + WriteBytes uint64 `json:"writeBytes,omitempty"` + ReadOps uint64 `json:"readOps,omitempty"` + WriteOps uint64 `json:"writeOps,omitempty"` + ReadTime uint64 `json:"readTimeMs,omitempty"` + WriteTime uint64 `json:"writeTimeMs,omitempty"` + IOTime uint64 `json:"ioTimeMs,omitempty"` +} + // DockerHost represents a Docker host reporting metrics via the external agent. type DockerHost struct { ID string `json:"id"` diff --git a/internal/models/models_frontend.go b/internal/models/models_frontend.go index cfbe29d..0384845 100644 --- a/internal/models/models_frontend.go +++ b/internal/models/models_frontend.go @@ -315,6 +315,7 @@ type HostFrontend struct { LoadAverage []float64 `json:"loadAverage,omitempty"` Memory *Memory `json:"memory,omitempty"` Disks []Disk `json:"disks,omitempty"` + DiskIO []DiskIO `json:"diskIO,omitempty"` NetworkInterfaces []HostNetworkInterface `json:"networkInterfaces,omitempty"` Sensors *HostSensorSummaryFrontend `json:"sensors,omitempty"` Status string `json:"status"` diff --git a/internal/monitoring/host_agent_temps.go b/internal/monitoring/host_agent_temps.go new file mode 100644 index 0000000..52c8ac8 --- /dev/null +++ b/internal/monitoring/host_agent_temps.go @@ -0,0 +1,266 @@ +package monitoring + +import ( + "math" + "regexp" + "sort" + "strconv" + "strings" + "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/models" + "github.com/rs/zerolog/log" +) + +// getHostAgentTemperature looks for a matching host agent by hostname and converts +// its sensor data to the Temperature model used by Proxmox nodes. +// Returns nil if no matching host agent is found or if no temperature data is available. +func (m *Monitor) getHostAgentTemperature(nodeName string) *models.Temperature { + if m.state == nil { + return nil + } + + hosts := m.state.GetHosts() + if len(hosts) == 0 { + return nil + } + + // Find a host agent whose hostname matches the Proxmox node name + // The agent on a Proxmox node will have the same hostname as the node + nodeLower := strings.ToLower(strings.TrimSpace(nodeName)) + var matchedHost *models.Host + for i := range hosts { + hostnameLower := strings.ToLower(strings.TrimSpace(hosts[i].Hostname)) + if hostnameLower == nodeLower { + matchedHost = &hosts[i] + break + } + } + + if matchedHost == nil { + return nil + } + + // Check if the host agent has temperature data + if len(matchedHost.Sensors.TemperatureCelsius) == 0 { + return nil + } + + // Convert host agent sensor data to Temperature model + return convertHostSensorsToTemperature(matchedHost.Sensors, matchedHost.LastSeen) +} + +// convertHostSensorsToTemperature converts HostSensorSummary to the Temperature model. +// The host agent reports temperatures in a flat map with keys like: +// - "cpu_package" -> CPU package temperature +// - "cpu_core_0", "cpu_core_1", etc. -> individual core temperatures +// - "nvme0", "nvme1", etc. -> NVMe temperatures +// - "gpu_edge", "gpu_junction", etc. -> GPU temperatures +func convertHostSensorsToTemperature(sensors models.HostSensorSummary, lastSeen time.Time) *models.Temperature { + if len(sensors.TemperatureCelsius) == 0 { + return nil + } + + temp := &models.Temperature{ + Available: true, + LastUpdate: lastSeen, + Cores: []models.CoreTemp{}, + NVMe: []models.NVMeTemp{}, + GPU: []models.GPUTemp{}, + } + + var coreTemps []float64 + corePattern := regexp.MustCompile(`^cpu_core_(\d+)$`) + nvmePattern := regexp.MustCompile(`^(nvme\d+)$`) + gpuPattern := regexp.MustCompile(`^gpu_(.+)$`) + + for key, value := range sensors.TemperatureCelsius { + keyLower := strings.ToLower(key) + + // CPU package temperature + if keyLower == "cpu_package" { + temp.CPUPackage = value + temp.HasCPU = true + continue + } + + // CPU core temperatures + if matches := corePattern.FindStringSubmatch(keyLower); len(matches) == 2 { + coreNum, err := strconv.Atoi(matches[1]) + if err == nil { + temp.Cores = append(temp.Cores, models.CoreTemp{ + Core: coreNum, + Temp: value, + }) + coreTemps = append(coreTemps, value) + temp.HasCPU = true + } + continue + } + + // NVMe temperatures + if matches := nvmePattern.FindStringSubmatch(keyLower); len(matches) == 2 { + temp.NVMe = append(temp.NVMe, models.NVMeTemp{ + Device: matches[1], + Temp: value, + }) + temp.HasNVMe = true + continue + } + + // GPU temperatures (gpu_edge, gpu_junction, gpu_mem, or generic gpu_) + if matches := gpuPattern.FindStringSubmatch(keyLower); len(matches) == 2 { + gpuKey := matches[1] + // Handle specific GPU temp types + if gpuKey == "edge" || gpuKey == "junction" || gpuKey == "mem" { + // Find or create GPU entry for the default device + found := false + for i := range temp.GPU { + if temp.GPU[i].Device == "gpu0" { + switch gpuKey { + case "edge": + temp.GPU[i].Edge = value + case "junction": + temp.GPU[i].Junction = value + case "mem": + temp.GPU[i].Mem = value + } + found = true + break + } + } + if !found { + gpu := models.GPUTemp{Device: "gpu0"} + switch gpuKey { + case "edge": + gpu.Edge = value + case "junction": + gpu.Junction = value + case "mem": + gpu.Mem = value + } + temp.GPU = append(temp.GPU, gpu) + } + } else { + // Generic GPU entry with edge temp + temp.GPU = append(temp.GPU, models.GPUTemp{ + Device: gpuKey, + Edge: value, + }) + } + temp.HasGPU = true + continue + } + } + + // Sort cores by core number + sort.Slice(temp.Cores, func(i, j int) bool { + return temp.Cores[i].Core < temp.Cores[j].Core + }) + + // Sort NVMe by device name + sort.Slice(temp.NVMe, func(i, j int) bool { + return temp.NVMe[i].Device < temp.NVMe[j].Device + }) + + // Calculate CPUMax from core temperatures if package temp wasn't available + if len(coreTemps) > 0 { + maxTemp := 0.0 + for _, t := range coreTemps { + if t > maxTemp { + maxTemp = t + } + } + temp.CPUMax = maxTemp + + // If no package temp, use max core temp as package + if temp.CPUPackage == 0 { + temp.CPUPackage = maxTemp + } + } + + // Validate we have at least some data + if !temp.HasCPU && !temp.HasGPU && !temp.HasNVMe { + return nil + } + + log.Debug(). + Str("source", "host-agent"). + Float64("cpuPackage", temp.CPUPackage). + Float64("cpuMax", temp.CPUMax). + Int("coreCount", len(temp.Cores)). + Int("nvmeCount", len(temp.NVMe)). + Int("gpuCount", len(temp.GPU)). + Msg("Converted host agent sensors to temperature data") + + return temp +} + +// isHostAgentTemperatureRecent checks if the host agent temperature data is recent enough to use. +// We consider data stale if the host hasn't reported in more than 2 minutes. +func isHostAgentTemperatureRecent(lastSeen time.Time) bool { + const staleDuration = 2 * time.Minute + return time.Since(lastSeen) < staleDuration +} + +// mergeTemperatureData merges host agent temperature with existing/proxy temperature data. +// Host agent data takes priority for CPU temperatures since it's more reliable (no SSH required). +// NVMe/SMART data is merged - host agent NVMe data supplements proxy SMART data. +func mergeTemperatureData(hostAgentTemp, proxyTemp *models.Temperature) *models.Temperature { + if hostAgentTemp == nil { + return proxyTemp + } + if proxyTemp == nil { + return hostAgentTemp + } + + // Start with host agent data as base since it's more reliable + result := &models.Temperature{ + CPUPackage: hostAgentTemp.CPUPackage, + CPUMax: hostAgentTemp.CPUMax, + CPUMin: proxyTemp.CPUMin, // Preserve historical min + CPUMaxRecord: math.Max(hostAgentTemp.CPUPackage, proxyTemp.CPUMaxRecord), // Update historical max + MinRecorded: proxyTemp.MinRecorded, + MaxRecorded: proxyTemp.MaxRecorded, + Cores: hostAgentTemp.Cores, + GPU: hostAgentTemp.GPU, + NVMe: hostAgentTemp.NVMe, + Available: true, + HasCPU: hostAgentTemp.HasCPU, + HasGPU: hostAgentTemp.HasGPU, + HasNVMe: hostAgentTemp.HasNVMe, + HasSMART: proxyTemp.HasSMART, // SMART data only comes from proxy + LastUpdate: hostAgentTemp.LastUpdate, + } + + // Use host agent CPU data if available, fall back to proxy + if !hostAgentTemp.HasCPU && proxyTemp.HasCPU { + result.CPUPackage = proxyTemp.CPUPackage + result.CPUMax = proxyTemp.CPUMax + result.Cores = proxyTemp.Cores + result.HasCPU = true + } + + // Keep proxy SMART data (host agent doesn't have smartctl access currently) + if proxyTemp.HasSMART { + result.SMART = proxyTemp.SMART + } + + // Merge GPU data - prefer host agent if available + if !hostAgentTemp.HasGPU && proxyTemp.HasGPU { + result.GPU = proxyTemp.GPU + result.HasGPU = true + } + + // Update historical max if current is higher + currentTemp := result.CPUPackage + if currentTemp == 0 && result.CPUMax > 0 { + currentTemp = result.CPUMax + } + if currentTemp > result.CPUMaxRecord { + result.CPUMaxRecord = currentTemp + result.MaxRecorded = time.Now() + } + + return result +} diff --git a/internal/monitoring/monitor.go b/internal/monitoring/monitor.go index e931415..e8f992f 100644 --- a/internal/monitoring/monitor.go +++ b/internal/monitoring/monitor.go @@ -1875,6 +1875,20 @@ func (m *Monitor) ApplyHostReport(report agentshost.Report, tokenRecord *config. }) } + diskIO := make([]models.DiskIO, 0, len(report.DiskIO)) + for _, io := range report.DiskIO { + diskIO = append(diskIO, models.DiskIO{ + Device: io.Device, + ReadBytes: io.ReadBytes, + WriteBytes: io.WriteBytes, + ReadOps: io.ReadOps, + WriteOps: io.WriteOps, + ReadTime: io.ReadTime, + WriteTime: io.WriteTime, + IOTime: io.IOTime, + }) + } + network := make([]models.HostNetworkInterface, 0, len(report.Network)) for _, nic := range report.Network { network = append(network, models.HostNetworkInterface{ @@ -1928,6 +1942,7 @@ func (m *Monitor) ApplyHostReport(report agentshost.Report, tokenRecord *config. LoadAverage: append([]float64(nil), report.Host.LoadAverage...), Memory: memory, Disks: disks, + DiskIO: diskIO, NetworkInterfaces: network, Sensors: models.HostSensorSummary{ TemperatureCelsius: cloneStringFloatMap(report.Sensors.TemperatureCelsius), @@ -1950,6 +1965,9 @@ func (m *Monitor) ApplyHostReport(report agentshost.Report, tokenRecord *config. if len(host.Disks) == 0 { host.Disks = nil } + if len(host.DiskIO) == 0 { + host.DiskIO = nil + } if len(host.NetworkInterfaces) == 0 { host.NetworkInterfaces = nil } diff --git a/internal/monitoring/monitor_polling.go b/internal/monitoring/monitor_polling.go index 51f2b22..928611e 100644 --- a/internal/monitoring/monitor_polling.go +++ b/internal/monitoring/monitor_polling.go @@ -2086,48 +2086,84 @@ func (m *Monitor) pollPVENode( if instanceCfg.TemperatureMonitoringEnabled != nil { tempMonitoringEnabled = *instanceCfg.TemperatureMonitoringEnabled } - if effectiveStatus == "online" && m.tempCollector != nil && tempMonitoringEnabled { - tempCtx, tempCancel := context.WithTimeout(ctx, 30*time.Second) // Increased to accommodate SSH operations via proxy - defer tempCancel() + if effectiveStatus == "online" && tempMonitoringEnabled { + // First, check if there's a matching host agent with temperature data. + // Host agent temperatures are preferred because they don't require SSH access. + hostAgentTemp := m.getHostAgentTemperature(node.Node) + if hostAgentTemp != nil { + log.Debug(). + Str("node", node.Node). + Float64("cpuPackage", hostAgentTemp.CPUPackage). + Float64("cpuMax", hostAgentTemp.CPUMax). + Int("nvmeCount", len(hostAgentTemp.NVMe)). + Msg("Using temperature data from host agent") + } - // Determine SSH hostname to use (most robust approach): - // Prefer the resolved host for this node, with cluster overrides when available. - sshHost := modelNode.Host - foundNodeEndpoint := false + // If no host agent temp or we need additional data (SMART), try SSH/proxy collection + var proxyTemp *models.Temperature + var err error + if m.tempCollector != nil { + tempCtx, tempCancel := context.WithTimeout(ctx, 30*time.Second) // Increased to accommodate SSH operations via proxy + defer tempCancel() - if modelNode.IsClusterMember && instanceCfg.IsCluster { - // Try to find specific endpoint configuration for this node - if len(instanceCfg.ClusterEndpoints) > 0 { - hasFingerprint := instanceCfg.Fingerprint != "" - for _, ep := range instanceCfg.ClusterEndpoints { - if strings.EqualFold(ep.NodeName, node.Node) { - if effective := clusterEndpointEffectiveURL(ep, instanceCfg.VerifySSL, hasFingerprint); effective != "" { - sshHost = effective - foundNodeEndpoint = true + // Determine SSH hostname to use (most robust approach): + // Prefer the resolved host for this node, with cluster overrides when available. + sshHost := modelNode.Host + foundNodeEndpoint := false + + if modelNode.IsClusterMember && instanceCfg.IsCluster { + // Try to find specific endpoint configuration for this node + if len(instanceCfg.ClusterEndpoints) > 0 { + hasFingerprint := instanceCfg.Fingerprint != "" + for _, ep := range instanceCfg.ClusterEndpoints { + if strings.EqualFold(ep.NodeName, node.Node) { + if effective := clusterEndpointEffectiveURL(ep, instanceCfg.VerifySSL, hasFingerprint); effective != "" { + sshHost = effective + foundNodeEndpoint = true + } + break } - break } } + + // If no specific endpoint found, fall back to node name + if !foundNodeEndpoint { + sshHost = node.Node + log.Debug(). + Str("node", node.Node). + Str("instance", instanceCfg.Name). + Msg("Node endpoint not found in cluster metadata - falling back to node name for temperature collection") + } } - // If no specific endpoint found, fall back to node name - if !foundNodeEndpoint { + if strings.TrimSpace(sshHost) == "" { sshHost = node.Node - log.Debug(). - Str("node", node.Node). - Str("instance", instanceCfg.Name). - Msg("Node endpoint not found in cluster metadata - falling back to node name for temperature collection") + } + + // Skip SSH/proxy collection if we already have host agent data and no proxy is configured + // (proxy might provide additional SMART data that host agent doesn't have) + skipProxyCollection := hostAgentTemp != nil && + strings.TrimSpace(instanceCfg.TemperatureProxyURL) == "" && + !m.HasSocketTemperatureProxy() + + if !skipProxyCollection { + // Use HTTP proxy if configured for this instance, otherwise fall back to socket/SSH + proxyTemp, err = m.tempCollector.CollectTemperatureWithProxy(tempCtx, sshHost, node.Node, instanceCfg.TemperatureProxyURL, instanceCfg.TemperatureProxyToken) + if err != nil && hostAgentTemp == nil { + log.Debug(). + Str("node", node.Node). + Str("sshHost", sshHost). + Bool("isCluster", modelNode.IsClusterMember). + Int("endpointCount", len(instanceCfg.ClusterEndpoints)). + Msg("Temperature collection failed - check SSH access") + } } } - if strings.TrimSpace(sshHost) == "" { - sshHost = node.Node - } + // Merge host agent and proxy temperatures + temp := mergeTemperatureData(hostAgentTemp, proxyTemp) - // Use HTTP proxy if configured for this instance, otherwise fall back to socket/SSH - temp, err := m.tempCollector.CollectTemperatureWithProxy(tempCtx, sshHost, node.Node, instanceCfg.TemperatureProxyURL, instanceCfg.TemperatureProxyToken) - - if err == nil && temp != nil && temp.Available { + if temp != nil && temp.Available { // Get the current CPU temperature (prefer package, fall back to max) currentTemp := temp.CPUPackage if currentTemp == 0 && temp.CPUMax > 0 { @@ -2171,28 +2207,29 @@ func (m *Monitor) pollPVENode( } modelNode.Temperature = temp + + // Determine source for logging + tempSource := "proxy/ssh" + if hostAgentTemp != nil && proxyTemp == nil { + tempSource = "host-agent" + } else if hostAgentTemp != nil && proxyTemp != nil { + tempSource = "host-agent+proxy" + } + log.Debug(). Str("node", node.Node). - Str("sshHost", sshHost). + Str("source", tempSource). Float64("cpuPackage", temp.CPUPackage). Float64("cpuMax", temp.CPUMax). Float64("cpuMin", temp.CPUMin). Float64("cpuMaxRecord", temp.CPUMaxRecord). Int("nvmeCount", len(temp.NVMe)). Msg("Collected temperature data") - } else if err != nil { + } else if hostAgentTemp == nil && proxyTemp == nil { log.Debug(). Str("node", node.Node). - Str("sshHost", sshHost). Bool("isCluster", modelNode.IsClusterMember). - Int("endpointCount", len(instanceCfg.ClusterEndpoints)). - Msg("Temperature collection failed - check SSH access") - } else if temp != nil { - log.Debug(). - Str("node", node.Node). - Str("sshHost", sshHost). - Bool("available", temp.Available). - Msg("Temperature data unavailable after collection") + Msg("No temperature data available (no host agent or proxy/SSH)") } } diff --git a/internal/sensors/parser.go b/internal/sensors/parser.go index d60299e..1f31ab2 100644 --- a/internal/sensors/parser.go +++ b/internal/sensors/parser.go @@ -109,6 +109,7 @@ func isCPUChip(chipLower string) bool { func parseCPUTemps(chipMap map[string]interface{}, data *TemperatureData) { foundPackageTemp := false var chipletTemps []float64 + var genericTemp float64 // For chips that only report temp1 for sensorName, sensorData := range chipMap { sensorMap, ok := sensorData.(map[string]interface{}) @@ -131,6 +132,14 @@ func parseCPUTemps(chipMap map[string]interface{}, data *TemperatureData) { } } + // Capture generic temp1 for chips like cpu_thermal (RPi, ARM SoCs) + // that don't have labeled sensors + if sensorNameLower == "temp1" { + if tempVal := extractTempInput(sensorMap); !math.IsNaN(tempVal) && tempVal > 0 { + genericTemp = tempVal + } + } + // Look for AMD chiplet temperatures if strings.HasPrefix(sensorName, "Tccd") { if tempVal := extractTempInput(sensorMap); !math.IsNaN(tempVal) && tempVal > 0 { @@ -175,6 +184,14 @@ func parseCPUTemps(chipMap map[string]interface{}, data *TemperatureData) { } } } + + // Fallback: use generic temp1 for chips like cpu_thermal (RPi, ARM SoCs) + if !foundPackageTemp && data.CPUPackage == 0 && genericTemp > 0 { + data.CPUPackage = genericTemp + if genericTemp > data.CPUMax { + data.CPUMax = genericTemp + } + } } func parseNVMeTemps(chipName string, chipMap map[string]interface{}, data *TemperatureData) { diff --git a/internal/utils/helpers.go b/internal/utils/helpers.go index 2e1d7c0..823a279 100644 --- a/internal/utils/helpers.go +++ b/internal/utils/helpers.go @@ -48,3 +48,48 @@ func GetenvTrim(key string) string { func NormalizeVersion(version string) string { return strings.TrimPrefix(strings.TrimSpace(version), "v") } + +// CompareVersions compares two semver-like version strings. +// Returns: +// +// 1 if a > b (a is newer) +// 0 if a == b +// +// -1 if a < b (b is newer) +// +// Handles versions like "4.33.1", "v4.33.1", "4.33" gracefully. +func CompareVersions(a, b string) int { + // Normalize both versions + a = NormalizeVersion(a) + b = NormalizeVersion(b) + + // Split into parts + partsA := strings.Split(a, ".") + partsB := strings.Split(b, ".") + + // Compare each part numerically + maxLen := len(partsA) + if len(partsB) > maxLen { + maxLen = len(partsB) + } + + for i := 0; i < maxLen; i++ { + var numA, numB int + + if i < len(partsA) { + fmt.Sscanf(partsA[i], "%d", &numA) + } + if i < len(partsB) { + fmt.Sscanf(partsB[i], "%d", &numB) + } + + if numA > numB { + return 1 + } + if numA < numB { + return -1 + } + } + + return 0 +} diff --git a/internal/utils/utils_test.go b/internal/utils/utils_test.go index f772db1..123e324 100644 --- a/internal/utils/utils_test.go +++ b/internal/utils/utils_test.go @@ -318,3 +318,50 @@ func TestNormalizeVersion(t *testing.T) { }) } } + +func TestCompareVersions(t *testing.T) { + tests := []struct { + a string + b string + expected int + }{ + // Equal versions + {"4.33.1", "4.33.1", 0}, + {"v4.33.1", "4.33.1", 0}, + {"4.33.1", "v4.33.1", 0}, + {"1.0.0", "1.0.0", 0}, + + // a > b (a is newer) + {"4.33.2", "4.33.1", 1}, + {"4.34.0", "4.33.1", 1}, + {"5.0.0", "4.33.1", 1}, + {"4.33.10", "4.33.9", 1}, + {"4.33.1", "4.33", 1}, // Missing patch = 0 + + // a < b (b is newer) + {"4.33.1", "4.33.2", -1}, + {"4.33.1", "4.34.0", -1}, + {"4.33.1", "5.0.0", -1}, + {"4.33.9", "4.33.10", -1}, + {"4.33", "4.33.1", -1}, + + // With v prefix + {"v4.34.0", "v4.33.1", 1}, + {"v4.33.1", "v4.34.0", -1}, + + // Edge cases + {"0.0.1", "0.0.0", 1}, + {"0.0.0", "0.0.1", -1}, + {"1.0", "0.9.9", 1}, + } + + for _, tc := range tests { + name := tc.a + "_vs_" + tc.b + t.Run(name, func(t *testing.T) { + result := CompareVersions(tc.a, tc.b) + if result != tc.expected { + t.Errorf("CompareVersions(%q, %q) = %d, want %d", tc.a, tc.b, result, tc.expected) + } + }) + } +} diff --git a/pkg/agents/host/report.go b/pkg/agents/host/report.go index 8242d3a..1452cc8 100644 --- a/pkg/agents/host/report.go +++ b/pkg/agents/host/report.go @@ -8,6 +8,7 @@ type Report struct { Host HostInfo `json:"host"` Metrics Metrics `json:"metrics"` Disks []Disk `json:"disks,omitempty"` + DiskIO []DiskIO `json:"diskIO,omitempty"` Network []NetworkInterface `json:"network,omitempty"` Sensors Sensors `json:"sensors,omitempty"` RAID []RAIDArray `json:"raid,omitempty"` @@ -23,6 +24,7 @@ type AgentInfo struct { Type string `json:"type,omitempty"` // "unified", "host", or "docker" - empty means legacy IntervalSeconds int `json:"intervalSeconds,omitempty"` Hostname string `json:"hostname,omitempty"` + UpdatedFrom string `json:"updatedFrom,omitempty"` // Previous version if recently auto-updated } // HostInfo contains platform and identification details about the monitored host. @@ -70,6 +72,19 @@ type Disk struct { Usage float64 `json:"usage,omitempty"` } +// DiskIO represents disk I/O statistics for a block device. +// These are cumulative counters since boot - the backend calculates rates. +type DiskIO struct { + Device string `json:"device"` // e.g., "nvme0n1", "sda" + ReadBytes uint64 `json:"readBytes,omitempty"` // Total bytes read + WriteBytes uint64 `json:"writeBytes,omitempty"` // Total bytes written + ReadOps uint64 `json:"readOps,omitempty"` // Total read operations + WriteOps uint64 `json:"writeOps,omitempty"` // Total write operations + ReadTime uint64 `json:"readTimeMs,omitempty"` // Total time spent reading (ms) + WriteTime uint64 `json:"writeTimeMs,omitempty"`// Total time spent writing (ms) + IOTime uint64 `json:"ioTimeMs,omitempty"` // Total time spent doing I/O (ms) +} + // NetworkInterface summarises network adapter statistics. type NetworkInterface struct { Name string `json:"name"` diff --git a/pkg/proxmox/cluster_client.go b/pkg/proxmox/cluster_client.go index d87fa62..f726e98 100644 --- a/pkg/proxmox/cluster_client.go +++ b/pkg/proxmox/cluster_client.go @@ -343,8 +343,8 @@ func (cc *ClusterClient) getHealthyClient(ctx context.Context) (*Client, error) return nil, fmt.Errorf("failed to create client for %s: %w", selectedEndpoint, err) } - // Quick connectivity test - testCtx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + // Connectivity test - 5 seconds to allow for TLS handshake (~3s typical) + testCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) testNodes, testErr := testClient.GetNodes(testCtx) cancel() @@ -489,10 +489,12 @@ func (cc *ClusterClient) recoverUnhealthyNodes(ctx context.Context) { cc.lastHealthCheck[ep] = now cc.mu.Unlock() - // Try to create a client and test connection with shorter timeout + // Try to create a client and test connection + // Note: 5-second timeout needed because TLS handshake to Proxmox API + // typically takes ~3 seconds on local networks cfg := cc.config cfg.Host = ep - cfg.Timeout = 2 * time.Second // Use shorter timeout for recovery attempts + cfg.Timeout = 5 * time.Second testClient, err := NewClient(cfg) if err != nil { @@ -504,8 +506,8 @@ func (cc *ClusterClient) recoverUnhealthyNodes(ctx context.Context) { return } - // Try a simple API call with short timeout - testCtx, cancel := context.WithTimeout(ctx, 2*time.Second) + // Try a simple API call + testCtx, cancel := context.WithTimeout(ctx, 5*time.Second) _, err = testClient.GetNodes(testCtx) cancel() diff --git a/scripts/install.sh b/scripts/install.sh index 087d8d4..c4e97d0 100644 --- a/scripts/install.sh +++ b/scripts/install.sh @@ -54,6 +54,8 @@ PULSE_TOKEN="" INTERVAL="30s" ENABLE_HOST="true" ENABLE_DOCKER="false" +ENABLE_PROXMOX="false" +PROXMOX_TYPE="" UNINSTALL="false" INSECURE="false" AGENT_ID="" @@ -83,6 +85,8 @@ build_exec_args() { EXEC_ARGS="$EXEC_ARGS --enable-host=false" fi if [[ "$ENABLE_DOCKER" == "true" ]]; then EXEC_ARGS="$EXEC_ARGS --enable-docker"; fi + if [[ "$ENABLE_PROXMOX" == "true" ]]; then EXEC_ARGS="$EXEC_ARGS --enable-proxmox"; fi + if [[ -n "$PROXMOX_TYPE" ]]; then EXEC_ARGS="$EXEC_ARGS --proxmox-type ${PROXMOX_TYPE}"; fi if [[ "$INSECURE" == "true" ]]; then EXEC_ARGS="$EXEC_ARGS --insecure"; fi if [[ -n "$AGENT_ID" ]]; then EXEC_ARGS="$EXEC_ARGS --agent-id ${AGENT_ID}"; fi } @@ -98,6 +102,8 @@ build_exec_args_array() { EXEC_ARGS_ARRAY+=(--enable-host=false) fi if [[ "$ENABLE_DOCKER" == "true" ]]; then EXEC_ARGS_ARRAY+=(--enable-docker); fi + if [[ "$ENABLE_PROXMOX" == "true" ]]; then EXEC_ARGS_ARRAY+=(--enable-proxmox); fi + if [[ -n "$PROXMOX_TYPE" ]]; then EXEC_ARGS_ARRAY+=(--proxmox-type "$PROXMOX_TYPE"); fi if [[ "$INSECURE" == "true" ]]; then EXEC_ARGS_ARRAY+=(--insecure); fi if [[ -n "$AGENT_ID" ]]; then EXEC_ARGS_ARRAY+=(--agent-id "$AGENT_ID"); fi } @@ -112,6 +118,8 @@ while [[ $# -gt 0 ]]; do --disable-host) ENABLE_HOST="false"; shift ;; --enable-docker) ENABLE_DOCKER="true"; shift ;; --disable-docker) ENABLE_DOCKER="false"; shift ;; + --enable-proxmox) ENABLE_PROXMOX="true"; shift ;; + --proxmox-type) PROXMOX_TYPE="$2"; shift 2 ;; --insecure) INSECURE="true"; shift ;; --uninstall) UNINSTALL="true"; shift ;; --agent-id) AGENT_ID="$2"; shift 2 ;;