gsadmin/pulse
1
0
Fork 0
mirror of https://github.com/rcourtman/Pulse.git synced 2026-09-24 04:07:16 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: improve WebSocket connection reliability in dev environment

Browse Source 
- Increase WebSocket buffer sizes from 64KB to 4MB to handle large mock data
- Add robust reconnection logic with exponential backoff
- Implement heartbeat mechanism to detect stale connections faster
- Add manual reconnect button in UI when connection fails
- Fix unused variable warnings in monitor code
- Add debug logging to trace WebSocket state initialization

This resolves the issue where the frontend would hang after code changes
during hot-reload, especially when using mock mode with many nodes.
This commit is contained in:
Pulse Monitor
2025-09-10 21:35:20 +00:00
parent 6b62bc64ee
commit 6e9cc27ae1
5 changed files with 134 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files
frontend-modern/src/components/Dashboard/Dashboard.tsx
+23 -3
View File
@@ -26,7 +26,7 @@ type GroupingMode = 'grouped' | 'flat';
export function Dashboard(props: DashboardProps) {
const { connected, activeAlerts, initialDataReceived } = useWebSocket();
const { connected, activeAlerts, initialDataReceived, reconnecting, reconnect } = useWebSocket();
const [search, setSearch] = createSignal('');
const [isSearchLocked, setIsSearchLocked] = createSignal(false);
const [selectedNode, setSelectedNode] = createSignal<string | null>(null);
@@ -647,7 +647,17 @@ export function Dashboard(props: DashboardProps) {
<path class="opacity-75" fill="currentColor" d="M4 12a8 8 0 018-8V0C5.373 0 0 5.373 0 12h4zm2 5.291A7.962 7.962 0 014 12H0c0 3.042 1.135 5.824 3 7.938l3-2.647z"></path>
</svg>
<h3 class="text-sm font-medium text-gray-900 dark:text-gray-100 mb-2">Loading dashboard data...</h3>
<p class="text-xs text-gray-600 dark:text-gray-400">Connecting to monitoring service</p>
<p class="text-xs text-gray-600 dark:text-gray-400">
{reconnecting() ? 'Reconnecting to monitoring service...' : 'Connecting to monitoring service'}
</p>
<Show when={!connected() && !reconnecting()}>
<button
onClick={() => reconnect()}
class="mt-3 px-4 py-2 text-xs bg-blue-500 text-white rounded hover:bg-blue-600 transition-colors"
>
Retry Connection
</button>
</Show>
</div>
</div>
</Show>
@@ -682,7 +692,17 @@ export function Dashboard(props: DashboardProps) {
<path stroke-linecap="round" stroke-linejoin="round" stroke-width="2" d="M12 8v4m0 4h.01M21 12a9 9 0 11-18 0 9 9 0 0118 0z" />
</svg>
<h3 class="text-sm font-medium text-red-800 dark:text-red-200 mb-2">Connection Lost</h3>
<p class="text-xs text-red-700 dark:text-red-300">Unable to connect to the backend server. Attempting to reconnect...</p>
<p class="text-xs text-red-700 dark:text-red-300">
{reconnecting() ? 'Attempting to reconnect...' : 'Unable to connect to the backend server'}
</p>
<Show when={!reconnecting()}>
<button
onClick={() => reconnect()}
class="mt-3 px-4 py-2 text-xs bg-red-600 text-white rounded hover:bg-red-700 transition-colors"
>
Reconnect Now
</button>
</Show>
</div>
</div>
</Show>
frontend-modern/src/stores/websocket.ts
+98 -72
View File
@@ -55,28 +55,90 @@ export function createWebSocketStore(url: string) {
let ws: WebSocket | null = null;
let reconnectTimeout: number;
let heartbeatInterval: number;
let reconnectAttempt = 0;
let isReconnecting = false;
const maxReconnectDelay = POLLING_INTERVALS.RECONNECT_MAX;
const initialReconnectDelay = POLLING_INTERVALS.RECONNECT_BASE;
const heartbeatIntervalMs = 30000; // Send heartbeat every 30 seconds
const connect = () => {
try {
// Close existing connection if any
if (ws && ws.readyState !== WebSocket.CLOSED) {
ws.close();
if (ws) {
if (ws.readyState === WebSocket.OPEN || ws.readyState === WebSocket.CONNECTING) {
ws.close(1000, 'Reconnecting');
}
ws = null;
}
// Add a small delay before reconnecting to avoid rapid reconnect loops
if (reconnectAttempt > 0) {
const delay = Math.min(100 * reconnectAttempt, 1000);
setTimeout(() => {
ws = new WebSocket(url);
setupWebSocket();
}, delay);
return;
}
ws = new WebSocket(url);
setupWebSocket();
} catch (err) {
logger.error('Failed to create WebSocket', err);
handleReconnect();
}
};
ws.onopen = () => {
logger.debug('connect');
setConnected(true);
setReconnecting(false); // Clear reconnecting state
reconnectAttempt = 0; // Reset reconnect attempts on successful connection
// Alerts will come with the initial state broadcast
};
const handleReconnect = () => {
if (isReconnecting) return;
isReconnecting = true;
setReconnecting(true);
// Clear any existing timeout
if (reconnectTimeout) {
window.clearTimeout(reconnectTimeout);
reconnectTimeout = 0;
}
// Calculate exponential backoff delay
const delay = Math.min(
initialReconnectDelay * Math.pow(2, reconnectAttempt),
maxReconnectDelay
);
logger.info(`Reconnecting in ${delay}ms (attempt ${reconnectAttempt + 1})`);
reconnectAttempt++;
reconnectTimeout = window.setTimeout(() => {
isReconnecting = false;
connect();
}, delay);
};
const setupWebSocket = () => {
if (!ws) return;
ws.onopen = () => {
logger.debug('connect');
setConnected(true);
setReconnecting(false); // Clear reconnecting state
reconnectAttempt = 0; // Reset reconnect attempts on successful connection
isReconnecting = false;
// Start heartbeat to keep connection alive
if (heartbeatInterval) {
window.clearInterval(heartbeatInterval);
}
heartbeatInterval = window.setInterval(() => {
if (ws && ws.readyState === WebSocket.OPEN) {
ws.send(JSON.stringify({ type: 'ping', data: { timestamp: Date.now() } }));
}
}, heartbeatIntervalMs);
// Alerts will come with the initial state broadcast
};
ws.onmessage = (event) => {
let data;
@@ -303,71 +365,32 @@ export function createWebSocketStore(url: string) {
}
};
ws.onclose = (event) => {
logger.debug('disconnect', { code: event.code, reason: event.reason });
setConnected(false);
setInitialDataReceived(false);
// Don't reconnect if we're already trying
if (isReconnecting) {
return;
}
// Clear any existing timeout to prevent multiple reconnections
if (reconnectTimeout) {
window.clearTimeout(reconnectTimeout);
reconnectTimeout = 0;
}
isReconnecting = true;
setReconnecting(true);
// Calculate exponential backoff delay
const delay = Math.min(
initialReconnectDelay * Math.pow(2, reconnectAttempt),
maxReconnectDelay
);
logger.info(`Reconnecting in ${delay}ms (attempt ${reconnectAttempt + 1})`);
reconnectAttempt++;
reconnectTimeout = window.setTimeout(() => {
isReconnecting = false;
setReconnecting(false);
connect();
}, delay);
};
ws.onerror = (error) => {
// Don't log connection errors if we're already connected
// Browser may show errors for initial connection attempts even after success
if (!connected()) {
logger.debug('error', error);
}
};
} catch (err) {
logger.error('Failed to connect', err);
ws.onclose = (event) => {
logger.debug('disconnect', { code: event.code, reason: event.reason });
setConnected(false);
setInitialDataReceived(false);
// Don't reconnect if we're already trying
if (isReconnecting) return;
// Clear heartbeat interval
if (heartbeatInterval) {
window.clearInterval(heartbeatInterval);
heartbeatInterval = 0;
}
isReconnecting = true;
setReconnecting(true);
// Don't try to reconnect if the close was intentional (code 1000)
if (event.code === 1000 && event.reason === 'Reconnecting') {
return;
}
// Use exponential backoff for connection errors too
const delay = Math.min(
initialReconnectDelay * Math.pow(2, reconnectAttempt),
maxReconnectDelay
);
reconnectAttempt++;
reconnectTimeout = window.setTimeout(() => {
isReconnecting = false;
setReconnecting(false);
connect();
}, delay);
}
handleReconnect();
};
ws.onerror = (error) => {
// Don't log connection errors if we're already connected
// Browser may show errors for initial connection attempts even after success
if (!connected()) {
logger.debug('error', error);
}
};
};
// Connect immediately
@@ -376,7 +399,10 @@ export function createWebSocketStore(url: string) {
// Cleanup on unmount
onCleanup(() => {
window.clearTimeout(reconnectTimeout);
ws?.close();
window.clearInterval(heartbeatInterval);
if (ws) {
ws.close(1000, 'Component unmounting');
}
});
return {
internal/monitoring/monitor.go
+1 -1
View File
@@ -1041,7 +1041,7 @@ func (m *Monitor) pollPVEInstance(ctx context.Context, instanceName string, clie
}
// Preserve existing disk data for nodes that weren't polled (offline or error)
for diskID, existingDisk := range existingDisksMap {
for _, existingDisk := range existingDisksMap {
// Only preserve if we didn't poll this node
if !polledNodes[existingDisk.Node] {
// Keep the existing disk data but update the LastChecked to indicate it's stale
internal/monitoring/monitor_optimized.go
+1 -1
View File
@@ -941,7 +941,7 @@ func (m *Monitor) pollStorageWithNodesOptimized(ctx context.Context, instanceNam
// Preserve existing storage data for nodes that weren't polled (offline or error)
preservedCount := 0
for storageID, existingStorage := range existingStorageMap {
for _, existingStorage := range existingStorageMap {
// Only preserve if we didn't poll this node
if !polledNodes[existingStorage.Node] && existingStorage.Node != "cluster" {
allStorage = append(allStorage, existingStorage)
internal/websocket/hub.go
+11 -3
View File
@@ -2,6 +2,7 @@ package websocket
import (
"encoding/json"
"fmt"
"math"
"net"
"net/http"
@@ -198,9 +199,11 @@ func (h *Hub) Run() {
log.Info().Str("client", client.id).Msg("WebSocket client connected")
// Send initial state to the new client immediately
log.Debug().Bool("hasGetState", h.getState != nil).Msg("Checking getState function for new client")
if h.getState != nil {
// Add a small delay to ensure client is ready
go func() {
log.Debug().Str("client", client.id).Msg("Starting initial state goroutine")
time.Sleep(500 * time.Millisecond)
// First send a small welcome message
@@ -225,10 +228,15 @@ func (h *Hub) Run() {
// Then send the initial state after another delay
time.Sleep(100 * time.Millisecond)
log.Debug().Str("client", client.id).Msg("About to get state")
// Get the state
stateData := h.getState()
log.Debug().Str("client", client.id).Interface("stateType", fmt.Sprintf("%T", stateData)).Msg("Got state for initial message")
initialMsg := Message{
Type: "initialState",
Data: sanitizeData(h.getState()),
Data: sanitizeData(stateData),
}
if data, err := json.Marshal(initialMsg); err == nil {
// Check if client is still registered before sending
@@ -299,8 +307,8 @@ func (h *Hub) HandleWebSocket(w http.ResponseWriter, r *http.Request) {
// Create upgrader with our origin check
upgrader := websocket.Upgrader{
ReadBufferSize: 1024 * 64, // 64KB to handle large state messages
WriteBufferSize: 1024 * 64, // 64KB to handle large state messages
ReadBufferSize: 1024 * 1024 * 4, // 4MB to handle large state messages
WriteBufferSize: 1024 * 1024 * 4, // 4MB to handle large state messages
CheckOrigin: h.checkOrigin,
}