danvics/Pulse
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix(ai-chat): Display messages chronologically in AI chatbot

Browse source 
- Add 'content' type to StreamDisplayEvent for tracking text chunks
- Track content events in streamEvents array for chronological display
- Update render to use Switch/Match for cleaner conditional rendering
- Interleave thinking, tool calls, and content as they stream in
- Add fallback for old messages without streamEvents for backwards compat

Previously, tool/command outputs stayed at top while AI text responses
accumulated at the bottom. Now all events appear in order like a
normal chatbot.
This commit is contained in:
rcourtman 2025-12-11 23:02:59 +00:00
parent a082fd3fc1
commit d03b3d644c
4 changed files with 343 additions and 225 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

77
frontend-modern/src/components/AI/AIChat.tsx
View file

@ -1,4 +1,4 @@
import { Component, Show, createSignal, For, createEffect, createMemo, onMount } from 'solid-js'; import { Component, Show, createSignal, For, createEffect, createMemo, onMount, Switch, Match } from 'solid-js';
import { marked } from 'marked'; import { marked } from 'marked';
import { AIAPI } from '@/api/ai'; import { AIAPI } from '@/api/ai';
import { notificationStore } from '@/stores/notifications'; import { notificationStore } from '@/stores/notifications';
@ -116,9 +116,10 @@ interface PendingApproval {
// Unified event type for chronological display // Unified event type for chronological display
interface StreamDisplayEvent { interface StreamDisplayEvent {
type: 'thinking' | 'tool'; type: 'thinking' | 'tool' | 'content';
thinking?: string; thinking?: string;
tool?: AIToolExecution; tool?: AIToolExecution;
content?: string; // Text content chunk for chronological display
} }
interface Message { interface Message {
@ -548,13 +549,16 @@ export const AIChat: Component<AIChatProps> = (props) => {
} }
case 'content': { case 'content': {
const content = event.data as string; const content = event.data as string;
// Append content rather than replace - this allows intermediate AI responses if (!content.trim()) return msg; // Skip empty content
// during tool execution to accumulate, showing the user the full conversation flow // Track content in streamEvents for chronological display
const events = msg.streamEvents || [];
// Also accumulate in content for backwards compatibility / summary
const existingContent = msg.content || ''; const existingContent = msg.content || '';
const separator = existingContent && !existingContent.endsWith('\n') ? '\n\n' : ''; const separator = existingContent && !existingContent.endsWith('\n') ? '\n\n' : '';
return { return {
...msg, ...msg,
content: existingContent + separator + content, content: existingContent + separator + content,
streamEvents: [...events, { type: 'content' as const, content: content.trim() }],
}; };
} }
case 'complete': { case 'complete': {
@ -1089,38 +1093,45 @@ export const AIChat: Component<AIChatProps> = (props) => {
: 'bg-gray-100 dark:bg-gray-800 text-gray-900 dark:text-gray-100' : 'bg-gray-100 dark:bg-gray-800 text-gray-900 dark:text-gray-100'
}`} }`}
> >
{/* Render all events in chronological order - thinking and tools interleaved */} {/* Render all events in chronological order - thinking, tools, and content interleaved */}
<Show when={message.role === 'assistant' && message.streamEvents && message.streamEvents.length > 0}> <Show when={message.role === 'assistant' && message.streamEvents && message.streamEvents.length > 0}>
<div class="mb-3 space-y-2"> <div class="space-y-2">
<For each={message.streamEvents}> <For each={message.streamEvents}>
{(evt) => ( {(evt) => (
<Show <Switch>
when={evt.type === 'tool' && evt.tool} <Match when={evt.type === 'tool' && evt.tool}>
fallback={ {/* Tool call */}
// Thinking chunk <div class="rounded border border-gray-300 dark:border-gray-600 overflow-hidden">
<div class={`px-2 py-1 text-xs font-medium flex items-center gap-2 ${evt.tool!.success
? 'bg-green-100 dark:bg-green-900/30 text-green-800 dark:text-green-200'
: 'bg-red-100 dark:bg-red-900/30 text-red-800 dark:text-red-200'
}`}>
<svg class="w-3 h-3" fill="none" stroke="currentColor" viewBox="0 0 24 24">
<path stroke-linecap="round" stroke-linejoin="round" stroke-width="2" d="M8 9l3 3-3 3m5 0h3M5 20h14a2 2 0 002-2V6a2 2 0 00-2-2H5a2 2 0 00-2 2v12a2 2 0 002 2z" />
</svg>
<code class="font-mono">{evt.tool!.input}</code>
</div>
<Show when={evt.tool!.output}>
<pre class="px-2 py-1 text-xs font-mono bg-gray-50 dark:bg-gray-900 text-gray-700 dark:text-gray-300 overflow-x-auto max-h-32 overflow-y-auto whitespace-pre-wrap break-words">
{evt.tool!.output.length > 500 ? evt.tool!.output.substring(0, 500) + '...' : evt.tool!.output}
</pre>
</Show>
</div>
</Match>
<Match when={evt.type === 'thinking' && evt.thinking}>
{/* Thinking chunk */}
<div class="px-2 py-1.5 text-xs bg-blue-50 dark:bg-blue-900/20 text-gray-700 dark:text-gray-300 rounded border-l-2 border-blue-400 whitespace-pre-wrap"> <div class="px-2 py-1.5 text-xs bg-blue-50 dark:bg-blue-900/20 text-gray-700 dark:text-gray-300 rounded border-l-2 border-blue-400 whitespace-pre-wrap">
{sanitizeThinking(evt.thinking && evt.thinking.length > 500 ? evt.thinking.substring(0, 500) + '...' : evt.thinking || '')} {sanitizeThinking(evt.thinking!.length > 500 ? evt.thinking!.substring(0, 500) + '...' : evt.thinking!)}
</div> </div>
} </Match>
> <Match when={evt.type === 'content' && evt.content}>
{/* Tool call */} {/* Content chunk - rendered as markdown */}
<div class="rounded border border-gray-300 dark:border-gray-600 overflow-hidden"> <div
<div class={`px-2 py-1 text-xs font-medium flex items-center gap-2 ${evt.tool!.success class="text-sm prose prose-sm dark:prose-invert max-w-none prose-pre:bg-gray-800 prose-pre:text-gray-100 prose-code:text-purple-600 dark:prose-code:text-purple-400 prose-code:before:content-none prose-code:after:content-none"
? 'bg-green-100 dark:bg-green-900/30 text-green-800 dark:text-green-200' innerHTML={renderMarkdown(evt.content!)}
: 'bg-red-100 dark:bg-red-900/30 text-red-800 dark:text-red-200' />
}`}> </Match>
<svg class="w-3 h-3" fill="none" stroke="currentColor" viewBox="0 0 24 24"> </Switch>
<path stroke-linecap="round" stroke-linejoin="round" stroke-width="2" d="M8 9l3 3-3 3m5 0h3M5 20h14a2 2 0 002-2V6a2 2 0 00-2-2H5a2 2 0 00-2 2v12a2 2 0 002 2z" />
</svg>
<code class="font-mono">{evt.tool!.input}</code>
</div>
<Show when={evt.tool!.output}>
<pre class="px-2 py-1 text-xs font-mono bg-gray-50 dark:bg-gray-900 text-gray-700 dark:text-gray-300 overflow-x-auto max-h-32 overflow-y-auto whitespace-pre-wrap break-words">
{evt.tool!.output.length > 500 ? evt.tool!.output.substring(0, 500) + '...' : evt.tool!.output}
</pre>
</Show>
</div>
</Show>
)} )}
</For> </For>
</div> </div>
@ -1146,8 +1157,8 @@ export const AIChat: Component<AIChatProps> = (props) => {
</div> </div>
</Show> </Show>
{/* Show AI's response text AFTER tool calls */} {/* Show AI's response text - only if no streamEvents (fallback for old messages or messages without streaming) */}
<Show when={message.content}> <Show when={message.content && (!message.streamEvents || message.streamEvents.length === 0)}>
<div <div
class="text-sm prose prose-sm dark:prose-invert max-w-none prose-pre:bg-gray-800 prose-pre:text-gray-100 prose-code:text-purple-600 dark:prose-code:text-purple-400 prose-code:before:content-none prose-code:after:content-none" class="text-sm prose prose-sm dark:prose-invert max-w-none prose-pre:bg-gray-800 prose-pre:text-gray-100 prose-code:text-purple-600 dark:prose-code:text-purple-400 prose-code:before:content-none prose-code:after:content-none"
innerHTML={renderMarkdown(message.content)} innerHTML={renderMarkdown(message.content)}

475
internal/monitoring/monitor.go
View file

@ -4797,57 +4797,142 @@ func (m *Monitor) pollPVEInstance(ctx context.Context, instanceName string, clie
// Update state first so we have nodes available // Update state first so we have nodes available
m.state.UpdateNodesForInstance(instanceName, modelNodes) m.state.UpdateNodesForInstance(instanceName, modelNodes)
// Now get storage data to use as fallback for disk metrics if needed // Storage fallback is used to provide disk metrics when rootfs is not available.
// We run this asynchronously with a short timeout so it doesn't block VM/container polling.
// This addresses the issue where slow storage APIs (e.g., NFS mounts) can cause the entire
// polling task to timeout before reaching VM/container polling.
storageByNode := make(map[string]models.Disk) storageByNode := make(map[string]models.Disk)
var storageByNodeMu sync.Mutex
storageFallbackDone := make(chan struct{})
if instanceCfg.MonitorStorage { if instanceCfg.MonitorStorage {
_, err := client.GetAllStorage(ctx) go func() {
if err == nil { defer close(storageFallbackDone)
// Use a short timeout for storage fallback - it's an optimization, not critical
storageFallbackTimeout := 10 * time.Second
storageCtx, storageCancel := context.WithTimeout(context.Background(), storageFallbackTimeout)
defer storageCancel()
_, err := client.GetAllStorage(storageCtx)
if err != nil {
if storageCtx.Err() != nil {
log.Debug().
Str("instance", instanceName).
Dur("timeout", storageFallbackTimeout).
Msg("Storage fallback timed out - continuing without disk fallback data")
}
return
}
for _, node := range nodes { for _, node := range nodes {
// Check if context was cancelled
select {
case <-storageCtx.Done():
log.Debug().
Str("instance", instanceName).
Msg("Storage fallback cancelled - partial data collected")
return
default:
}
// Skip offline nodes to avoid 595 errors // Skip offline nodes to avoid 595 errors
if nodeEffectiveStatus[node.Node] != "online" { if nodeEffectiveStatus[node.Node] != "online" {
continue continue
} }
nodeStorages, err := client.GetStorage(ctx, node.Node) nodeStorages, err := client.GetStorage(storageCtx, node.Node)
if err == nil { if err != nil {
// Look for local or local-lvm storage as most stable disk metric continue
for _, storage := range nodeStorages { }
if reason, skip := readOnlyFilesystemReason(storage.Type, storage.Total, storage.Used); skip {
// Look for local or local-lvm storage as most stable disk metric
for _, storage := range nodeStorages {
if reason, skip := readOnlyFilesystemReason(storage.Type, storage.Total, storage.Used); skip {
log.Debug().
Str("node", node.Node).
Str("storage", storage.Storage).
Str("type", storage.Type).
Str("skipReason", reason).
Uint64("total", storage.Total).
Uint64("used", storage.Used).
Msg("Skipping read-only storage while building disk fallback")
continue
}
if storage.Storage == "local" || storage.Storage == "local-lvm" {
disk := models.Disk{
Total: int64(storage.Total),
Used: int64(storage.Used),
Free: int64(storage.Available),
Usage: safePercentage(float64(storage.Used), float64(storage.Total)),
}
// Prefer "local" over "local-lvm"
storageByNodeMu.Lock()
if _, exists := storageByNode[node.Node]; !exists || storage.Storage == "local" {
storageByNode[node.Node] = disk
log.Debug(). log.Debug().
Str("node", node.Node). Str("node", node.Node).
Str("storage", storage.Storage). Str("storage", storage.Storage).
Str("type", storage.Type). Float64("usage", disk.Usage).
Str("skipReason", reason). Msg("Using storage for disk metrics fallback")
Uint64("total", storage.Total).
Uint64("used", storage.Used).
Msg("Skipping read-only storage while building disk fallback")
continue
}
if storage.Storage == "local" || storage.Storage == "local-lvm" {
disk := models.Disk{
Total: int64(storage.Total),
Used: int64(storage.Used),
Free: int64(storage.Available),
Usage: safePercentage(float64(storage.Used), float64(storage.Total)),
}
// Prefer "local" over "local-lvm"
if _, exists := storageByNode[node.Node]; !exists || storage.Storage == "local" {
storageByNode[node.Node] = disk
log.Debug().
Str("node", node.Node).
Str("storage", storage.Storage).
Float64("usage", disk.Usage).
Msg("Using storage for disk metrics fallback")
}
} }
storageByNodeMu.Unlock()
} }
} }
} }
}()
} else {
// No storage monitoring, close channel immediately
close(storageFallbackDone)
}
// Poll VMs and containers FIRST - this is the most critical data.
// This happens immediately after starting the storage fallback goroutine,
// so VM/container polling runs in parallel with (and is not blocked by) storage operations.
if instanceCfg.MonitorVMs || instanceCfg.MonitorContainers {
select {
case <-ctx.Done():
pollErr = ctx.Err()
return
default:
// Always try the efficient cluster/resources endpoint first
// This endpoint works on both clustered and standalone nodes
// Testing confirmed it works on standalone nodes like pimox
useClusterEndpoint := m.pollVMsAndContainersEfficient(ctx, instanceName, client, nodeEffectiveStatus)
if !useClusterEndpoint {
// Fall back to traditional polling only if cluster/resources not available
// This should be rare - only for very old Proxmox versions
log.Debug().
Str("instance", instanceName).
Msg("cluster/resources endpoint not available, using traditional polling")
// Check if configuration needs updating
if instanceCfg.IsCluster {
isActuallyCluster, checkErr := client.IsClusterMember(ctx)
if checkErr == nil && !isActuallyCluster {
log.Warn().
Str("instance", instanceName).
Msg("Instance marked as cluster but is actually standalone - consider updating configuration")
instanceCfg.IsCluster = false
}
}
// Use optimized parallel polling for better performance
if instanceCfg.MonitorVMs {
m.pollVMsWithNodes(ctx, instanceName, client, nodes, nodeEffectiveStatus)
}
if instanceCfg.MonitorContainers {
m.pollContainersWithNodes(ctx, instanceName, client, nodes, nodeEffectiveStatus)
}
}
} }
} }
// Poll physical disks for health monitoring (enabled by default unless explicitly disabled) // Poll physical disks for health monitoring (enabled by default unless explicitly disabled)
// Skip if MonitorPhysicalDisks is explicitly set to false // Skip if MonitorPhysicalDisks is explicitly set to false
// Physical disk polling runs in a background goroutine since GetDisks can be slow
// and we don't want it to cause task timeouts. It has its own 5-minute interval anyway.
if instanceCfg.MonitorPhysicalDisks != nil && !*instanceCfg.MonitorPhysicalDisks { if instanceCfg.MonitorPhysicalDisks != nil && !*instanceCfg.MonitorPhysicalDisks {
log.Debug().Str("instance", instanceName).Msg("Physical disk monitoring explicitly disabled") log.Debug().Str("instance", instanceName).Msg("Physical disk monitoring explicitly disabled")
// Keep any existing disk data visible (don't clear it) // Keep any existing disk data visible (don't clear it)
@ -4887,152 +4972,192 @@ func (m *Monitor) pollPVEInstance(ctx context.Context, instanceName string, clie
m.state.UpdatePhysicalDisks(instanceName, updated) m.state.UpdatePhysicalDisks(instanceName, updated)
} }
} else { } else {
log.Debug(). // Run physical disk polling in background to avoid blocking the main task
Int("nodeCount", len(nodes)). go func(inst string, pveClient PVEClientInterface, nodeList []proxmox.Node, nodeStatus map[string]string, modelNodesCopy []models.Node) {
Dur("interval", pollingInterval). defer recoverFromPanic(fmt.Sprintf("pollPhysicalDisks-%s", inst))
Msg("Starting disk health polling")
// Use a generous timeout for disk polling
// Get existing disks from state to preserve data for offline nodes diskTimeout := 60 * time.Second
currentState := m.state.GetSnapshot() diskCtx, diskCancel := context.WithTimeout(context.Background(), diskTimeout)
existingDisksMap := make(map[string]models.PhysicalDisk) defer diskCancel()
for _, disk := range currentState.PhysicalDisks {
if disk.Instance == instanceName {
existingDisksMap[disk.ID] = disk
}
}
var allDisks []models.PhysicalDisk
polledNodes := make(map[string]bool) // Track which nodes we successfully polled
for _, node := range nodes {
// Skip offline nodes but preserve their existing disk data
if nodeEffectiveStatus[node.Node] != "online" {
log.Debug().Str("node", node.Node).Msg("Skipping disk poll for offline node - preserving existing data")
continue
}
// Get disk list for this node
log.Debug().Str("node", node.Node).Msg("Getting disk list for node")
disks, err := client.GetDisks(ctx, node.Node)
if err != nil {
// Check if it's a permission error or if the endpoint doesn't exist
errStr := err.Error()
if strings.Contains(errStr, "401") || strings.Contains(errStr, "403") {
log.Warn().
Str("node", node.Node).
Err(err).
Msg("Insufficient permissions to access disk information - check API token permissions")
} else if strings.Contains(errStr, "404") || strings.Contains(errStr, "501") {
log.Info().
Str("node", node.Node).
Msg("Disk monitoring not available on this node (may be using non-standard storage)")
} else {
log.Warn().
Str("node", node.Node).
Err(err).
Msg("Failed to get disk list")
}
continue
}
log.Debug(). log.Debug().
Str("node", node.Node). Int("nodeCount", len(nodeList)).
Int("diskCount", len(disks)). Dur("interval", pollingInterval).
Msg("Got disk list for node") Msg("Starting disk health polling")
// Mark this node as successfully polled // Get existing disks from state to preserve data for offline nodes
polledNodes[node.Node] = true currentState := m.state.GetSnapshot()
existingDisksMap := make(map[string]models.PhysicalDisk)
for _, disk := range currentState.PhysicalDisks {
if disk.Instance == inst {
existingDisksMap[disk.ID] = disk
}
}
// Check each disk for health issues and add to state var allDisks []models.PhysicalDisk
for _, disk := range disks { polledNodes := make(map[string]bool) // Track which nodes we successfully polled
// Create PhysicalDisk model
diskID := fmt.Sprintf("%s-%s-%s", instanceName, node.Node, strings.ReplaceAll(disk.DevPath, "/", "-")) for _, node := range nodeList {
physicalDisk := models.PhysicalDisk{ // Check if context timed out
ID: diskID, select {
Node: node.Node, case <-diskCtx.Done():
Instance: instanceName, log.Debug().
DevPath: disk.DevPath, Str("instance", inst).
Model: disk.Model, Msg("Physical disk polling timed out - preserving existing data")
Serial: disk.Serial, return
WWN: disk.WWN, default:
Type: disk.Type, }
Size: disk.Size,
Health: disk.Health, // Skip offline nodes but preserve their existing disk data
Wearout: disk.Wearout, if nodeStatus[node.Node] != "online" {
RPM: disk.RPM, log.Debug().Str("node", node.Node).Msg("Skipping disk poll for offline node - preserving existing data")
Used: disk.Used, continue
LastChecked: time.Now(),
} }
allDisks = append(allDisks, physicalDisk) // Get disk list for this node
log.Debug().Str("node", node.Node).Msg("Getting disk list for node")
disks, err := pveClient.GetDisks(diskCtx, node.Node)
if err != nil {
// Check if it's a permission error or if the endpoint doesn't exist
errStr := err.Error()
if strings.Contains(errStr, "401") || strings.Contains(errStr, "403") {
log.Warn().
Str("node", node.Node).
Err(err).
Msg("Insufficient permissions to access disk information - check API token permissions")
} else if strings.Contains(errStr, "404") || strings.Contains(errStr, "501") {
log.Info().
Str("node", node.Node).
Msg("Disk monitoring not available on this node (may be using non-standard storage)")
} else {
log.Warn().
Str("node", node.Node).
Err(err).
Msg("Failed to get disk list")
}
continue
}
log.Debug(). log.Debug().
Str("node", node.Node). Str("node", node.Node).
Str("disk", disk.DevPath). Int("diskCount", len(disks)).
Str("model", disk.Model). Msg("Got disk list for node")
Str("health", disk.Health).
Int("wearout", disk.Wearout).
Msg("Checking disk health")
normalizedHealth := strings.ToUpper(strings.TrimSpace(disk.Health)) // Mark this node as successfully polled
if normalizedHealth != "" && normalizedHealth != "UNKNOWN" && normalizedHealth != "PASSED" && normalizedHealth != "OK" { polledNodes[node.Node] = true
// Disk has failed or is failing - alert manager will handle this
log.Warn(). // Check each disk for health issues and add to state
for _, disk := range disks {
// Create PhysicalDisk model
diskID := fmt.Sprintf("%s-%s-%s", inst, node.Node, strings.ReplaceAll(disk.DevPath, "/", "-"))
physicalDisk := models.PhysicalDisk{
ID: diskID,
Node: node.Node,
Instance: inst,
DevPath: disk.DevPath,
Model: disk.Model,
Serial: disk.Serial,
WWN: disk.WWN,
Type: disk.Type,
Size: disk.Size,
Health: disk.Health,
Wearout: disk.Wearout,
RPM: disk.RPM,
Used: disk.Used,
LastChecked: time.Now(),
}
allDisks = append(allDisks, physicalDisk)
log.Debug().
Str("node", node.Node). Str("node", node.Node).
Str("disk", disk.DevPath). Str("disk", disk.DevPath).
Str("model", disk.Model). Str("model", disk.Model).
Str("health", disk.Health). Str("health", disk.Health).
Int("wearout", disk.Wearout). Int("wearout", disk.Wearout).
Msg("Disk health issue detected") Msg("Checking disk health")
// Pass disk info to alert manager normalizedHealth := strings.ToUpper(strings.TrimSpace(disk.Health))
m.alertManager.CheckDiskHealth(instanceName, node.Node, disk) if normalizedHealth != "" && normalizedHealth != "UNKNOWN" && normalizedHealth != "PASSED" && normalizedHealth != "OK" {
} else if disk.Wearout > 0 && disk.Wearout < 10 { // Disk has failed or is failing - alert manager will handle this
// Low wearout warning (less than 10% life remaining) log.Warn().
log.Warn(). Str("node", node.Node).
Str("node", node.Node). Str("disk", disk.DevPath).
Str("disk", disk.DevPath). Str("model", disk.Model).
Str("model", disk.Model). Str("health", disk.Health).
Int("wearout", disk.Wearout). Int("wearout", disk.Wearout).
Msg("SSD wearout critical - less than 10% life remaining") Msg("Disk health issue detected")
// Pass to alert manager for wearout alert // Pass disk info to alert manager
m.alertManager.CheckDiskHealth(instanceName, node.Node, disk) m.alertManager.CheckDiskHealth(inst, node.Node, disk)
} else if disk.Wearout > 0 && disk.Wearout < 10 {
// Low wearout warning (less than 10% life remaining)
log.Warn().
Str("node", node.Node).
Str("disk", disk.DevPath).
Str("model", disk.Model).
Int("wearout", disk.Wearout).
Msg("SSD wearout critical - less than 10% life remaining")
// Pass to alert manager for wearout alert
m.alertManager.CheckDiskHealth(inst, node.Node, disk)
}
} }
} }
}
// Preserve existing disk data for nodes that weren't polled (offline or error) // Preserve existing disk data for nodes that weren't polled (offline or error)
for _, existingDisk := range existingDisksMap { for _, existingDisk := range existingDisksMap {
// Only preserve if we didn't poll this node // Only preserve if we didn't poll this node
if !polledNodes[existingDisk.Node] { if !polledNodes[existingDisk.Node] {
// Keep the existing disk data but update the LastChecked to indicate it's stale // Keep the existing disk data but update the LastChecked to indicate it's stale
allDisks = append(allDisks, existingDisk) allDisks = append(allDisks, existingDisk)
log.Debug(). log.Debug().
Str("node", existingDisk.Node). Str("node", existingDisk.Node).
Str("disk", existingDisk.DevPath). Str("disk", existingDisk.DevPath).
Msg("Preserving existing disk data for unpolled node") Msg("Preserving existing disk data for unpolled node")
}
} }
}
allDisks = mergeNVMeTempsIntoDisks(allDisks, modelNodes) allDisks = mergeNVMeTempsIntoDisks(allDisks, modelNodesCopy)
// Update physical disks in state // Update physical disks in state
log.Debug(). log.Debug().
Str("instance", instanceName). Str("instance", inst).
Int("diskCount", len(allDisks)). Int("diskCount", len(allDisks)).
Int("preservedCount", len(existingDisksMap)-len(polledNodes)). Int("preservedCount", len(existingDisksMap)-len(polledNodes)).
Msg("Updating physical disks in state") Msg("Updating physical disks in state")
m.state.UpdatePhysicalDisks(instanceName, allDisks) m.state.UpdatePhysicalDisks(inst, allDisks)
}(instanceName, client, nodes, nodeEffectiveStatus, modelNodes)
} }
} }
// Note: Physical disk monitoring is now enabled by default with a 5-minute polling interval. // Note: Physical disk monitoring is now enabled by default with a 5-minute polling interval.
// Users can explicitly disable it in node settings. Disk data is preserved between polls. // Users can explicitly disable it in node settings. Disk data is preserved between polls.
// Wait for storage fallback to complete (with a short timeout) before using the data.
// This is non-blocking in the sense that VM/container polling has already completed by now.
// We give the storage fallback goroutine up to 2 additional seconds to finish if it's still running.
select {
case <-storageFallbackDone:
// Storage fallback completed normally
case <-time.After(2 * time.Second):
log.Debug().
Str("instance", instanceName).
Msg("Storage fallback still running - proceeding without waiting (disk fallback may be unavailable)")
}
// Update nodes with storage fallback if rootfs was not available // Update nodes with storage fallback if rootfs was not available
// Copy storageByNode under lock, then release to avoid holding during metric updates
storageByNodeMu.Lock()
localStorageByNode := make(map[string]models.Disk, len(storageByNode))
for k, v := range storageByNode {
localStorageByNode[k] = v
}
storageByNodeMu.Unlock()
for i := range modelNodes { for i := range modelNodes {
if modelNodes[i].Disk.Total == 0 { if modelNodes[i].Disk.Total == 0 {
if disk, exists := storageByNode[modelNodes[i].Name]; exists { if disk, exists := localStorageByNode[modelNodes[i].Name]; exists {
modelNodes[i].Disk = disk modelNodes[i].Disk = disk
log.Debug(). log.Debug().
Str("node", modelNodes[i].Name). Str("node", modelNodes[i].Name).
@ -5159,55 +5284,25 @@ func (m *Monitor) pollPVEInstance(ctx context.Context, instanceName string, clie
} }
} }
// Poll VMs and containers together using cluster/resources for efficiency // Poll storage in background if enabled - storage APIs can be slow (NFS mounts, etc.)
if instanceCfg.MonitorVMs || instanceCfg.MonitorContainers { // so we run this asynchronously to prevent it from causing task timeouts.
select { // This is similar to how backup polling runs in the background.
case <-ctx.Done():
pollErr = ctx.Err()
return
default:
// Always try the efficient cluster/resources endpoint first
// This endpoint works on both clustered and standalone nodes
// Testing confirmed it works on standalone nodes like pimox
useClusterEndpoint := m.pollVMsAndContainersEfficient(ctx, instanceName, client, nodeEffectiveStatus)
if !useClusterEndpoint {
// Fall back to traditional polling only if cluster/resources not available
// This should be rare - only for very old Proxmox versions
log.Debug().
Str("instance", instanceName).
Msg("cluster/resources endpoint not available, using traditional polling")
// Check if configuration needs updating
if instanceCfg.IsCluster {
isActuallyCluster, checkErr := client.IsClusterMember(ctx)
if checkErr == nil && !isActuallyCluster {
log.Warn().
Str("instance", instanceName).
Msg("Instance marked as cluster but is actually standalone - consider updating configuration")
instanceCfg.IsCluster = false
}
}
// Use optimized parallel polling for better performance
if instanceCfg.MonitorVMs {
m.pollVMsWithNodes(ctx, instanceName, client, nodes, nodeEffectiveStatus)
}
if instanceCfg.MonitorContainers {
m.pollContainersWithNodes(ctx, instanceName, client, nodes, nodeEffectiveStatus)
}
}
}
}
// Poll storage if enabled
if instanceCfg.MonitorStorage { if instanceCfg.MonitorStorage {
select { select {
case <-ctx.Done(): case <-ctx.Done():
pollErr = ctx.Err() pollErr = ctx.Err()
return return
default: default:
m.pollStorageWithNodes(ctx, instanceName, client, nodes) go func(inst string, pveClient PVEClientInterface, nodeList []proxmox.Node) {
defer recoverFromPanic(fmt.Sprintf("pollStorageWithNodes-%s", inst))
// Use a generous timeout for storage polling - it's not blocking the main task
storageTimeout := 60 * time.Second
storageCtx, storageCancel := context.WithTimeout(context.Background(), storageTimeout)
defer storageCancel()
m.pollStorageWithNodes(storageCtx, inst, pveClient, nodeList)
}(instanceName, client, nodes)
} }
} }

5
internal/monitoring/monitor_polling.go
View file

@ -2143,7 +2143,10 @@ func (m *Monitor) pollPVENode(
var proxyTemp *models.Temperature var proxyTemp *models.Temperature
var err error var err error
if m.tempCollector != nil { if m.tempCollector != nil {
tempCtx, tempCancel := context.WithTimeout(ctx, 30*time.Second) // Increased to accommodate SSH operations via proxy // Temperature collection is best-effort - use a short timeout to avoid blocking node polling
// Use context.Background() so the timeout is truly independent of the parent polling context
// If SSH is slow or unresponsive, we'll preserve previous temperature data
tempCtx, tempCancel := context.WithTimeout(context.Background(), 10*time.Second)
defer tempCancel() defer tempCancel()
// Determine SSH hostname to use (most robust approach): // Determine SSH hostname to use (most robust approach):

11
pkg/proxmox/cluster_client.go
View file

@ -632,6 +632,8 @@ func (cc *ClusterClient) executeWithFailover(ctx context.Context, fn func(*Clien
// Error 400 with "ds" parameter error means Proxmox 9.x doesn't support RRD data source filtering // Error 400 with "ds" parameter error means Proxmox 9.x doesn't support RRD data source filtering
// JSON unmarshal errors are data format issues, not connectivity problems // JSON unmarshal errors are data format issues, not connectivity problems
// Context deadline/timeout errors on storage endpoints mean storage issues, not node unreachability // Context deadline/timeout errors on storage endpoints mean storage issues, not node unreachability
// PBS (Proxmox Backup Server) errors are upstream storage issues, not node connectivity problems
// RRD data timeouts are secondary data fetch failures, not node unreachability
if isVMSpecificError(errStr) || if isVMSpecificError(errStr) ||
strings.Contains(errStr, "595") || strings.Contains(errStr, "595") ||
(strings.Contains(errStr, "500") && strings.Contains(errStr, "hostname lookup")) || (strings.Contains(errStr, "500") && strings.Contains(errStr, "hostname lookup")) ||
@ -643,7 +645,14 @@ func (cc *ClusterClient) executeWithFailover(ctx context.Context, fn func(*Clien
(strings.Contains(errStr, "storage '") && strings.Contains(errStr, "is not available on node")) || (strings.Contains(errStr, "storage '") && strings.Contains(errStr, "is not available on node")) ||
strings.Contains(errStr, "unexpected response format") || strings.Contains(errStr, "unexpected response format") ||
(strings.Contains(errStr, "context deadline exceeded") && strings.Contains(errStr, "/storage")) || (strings.Contains(errStr, "context deadline exceeded") && strings.Contains(errStr, "/storage")) ||
(strings.Contains(errStr, "Client.Timeout exceeded") && strings.Contains(errStr, "/storage")) { (strings.Contains(errStr, "Client.Timeout exceeded") && strings.Contains(errStr, "/storage")) ||
// PBS storage errors - Proxmox can't reach PBS, but node is still reachable
(strings.Contains(errStr, "500") && strings.Contains(errStr, "pbs-") && strings.Contains(errStr, "error fetching datastores")) ||
(strings.Contains(errStr, "500") && strings.Contains(errStr, "Can't connect to") && strings.Contains(errStr, ":8007")) ||
// RRD data timeouts - secondary metric fetch failures, node is still working
(strings.Contains(errStr, "context deadline exceeded") && strings.Contains(errStr, "/rrddata")) ||
(strings.Contains(errStr, "context deadline exceeded") && strings.Contains(errStr, "/lxc/") && strings.Contains(errStr, "rrd")) ||
(strings.Contains(errStr, "context deadline exceeded") && strings.Contains(errStr, "/qemu/") && strings.Contains(errStr, "rrd")) {
// This is likely a node-specific failure, not an endpoint failure // This is likely a node-specific failure, not an endpoint failure
// Return the error but don't mark the endpoint as unhealthy // Return the error but don't mark the endpoint as unhealthy
log.Debug(). log.Debug().