@@ -654,9 +665,11 @@ async function initializeMainApp() {
await loadMemoryStats();
checkOllamaStatus();
checkSearchStatus();
+ updateClusterStatus();
updateSystemStats();
setInterval(checkOllamaStatus, 30000);
setInterval(checkSearchStatus, 60000);
+ setInterval(updateClusterStatus, 15000);
setInterval(updateSystemStats, 2000);
document.getElementById('userInput').addEventListener('input', updateTokenThermometer);
updateTokenThermometer();
@@ -730,6 +743,37 @@ async function checkSearchStatus() {
}
}
+async function updateClusterStatus() {
+ try {
+ const resp = await authFetch('/api/cluster');
+ const data = await resp.json();
+ const statusEl = document.getElementById('clusterStatus');
+ const panel = document.getElementById('clusterPanel');
+ const count = data.node_count || 0;
+ statusEl.innerHTML = `
cluster: ${count} node${count !== 1 ? 's' : ''}`;
+ if (count === 0) {
+ panel.style.display = 'none';
+ return;
+ }
+ panel.style.display = 'block';
+ const now = Date.now();
+ let html = '';
+ for (const [name, node] of Object.entries(data.nodes)) {
+ const lastSeen = node.last_seen ? new Date(node.last_seen).getTime() : 0;
+ const ageSec = (now - lastSeen) / 1000;
+ let dotClass = 'active';
+ if (node.status === 'swapping') dotClass = 'swapping';
+ else if (node.status === 'error') dotClass = 'error';
+ else if (ageSec > 60) dotClass = 'offline';
+ const modelName = node.active_model ? (node.active_model.name || 'unknown') : 'idle';
+ html += `
${escapeHtml(name)}${escapeHtml(modelName)}
`;
+ }
+ panel.innerHTML = html;
+ } catch(e) {
+ document.getElementById('clusterStatus').innerHTML = '
cluster: error';
+ }
+}
+
async function loadModels() {
try {
const resp = await authFetch('/api/models');
diff --git a/tests/test_cluster_heartbeat.py b/tests/test_cluster_heartbeat.py
new file mode 100644
index 0000000..4cf96cb
--- /dev/null
+++ b/tests/test_cluster_heartbeat.py
@@ -0,0 +1,49 @@
+"""Tests for cluster.py heartbeat handler."""
+import asyncio
+
+import cluster
+from config import AMQP_EXCHANGE_SYSTEM
+
+
+def _reset():
+ cluster.CLUSTER_NODES.clear()
+ cluster.CLUSTER_EVENTS.clear()
+ cluster.CLUSTER_COORDINATOR = None
+ cluster._pending_pings.clear()
+
+
+# ---------- 1. handle_heartbeat() for known node updates last_seen ----------
+
+
+def test_heartbeat_updates_last_seen(monkeypatch):
+ _reset()
+
+ cluster.CLUSTER_NODES["jarvis"] = {
+ "name": "jarvis", "type": "worker", "status": "active",
+ "last_seen": "2020-01-01T00:00:00Z",
+ }
+ original = cluster.CLUSTER_NODES["jarvis"]["last_seen"]
+
+ asyncio.run(cluster.handle_heartbeat(
+ AMQP_EXCHANGE_SYSTEM, "node.jarvis.heartbeat",
+ {"node_name": "jarvis"},
+ ))
+
+ assert cluster.CLUSTER_NODES["jarvis"]["last_seen"] != original
+ assert cluster.CLUSTER_NODES["jarvis"]["status"] == "active" # unchanged
+
+
+# ---------- 2. handle_heartbeat() for unknown node logs warning, no add ----------
+
+
+def test_heartbeat_unknown_node_logs_warning(caplog, monkeypatch):
+ _reset()
+ caplog.set_level("WARNING")
+
+ asyncio.run(cluster.handle_heartbeat(
+ AMQP_EXCHANGE_SYSTEM, "node.stranger.heartbeat",
+ {"node_name": "stranger"},
+ ))
+
+ assert "stranger" not in cluster.CLUSTER_NODES
+ assert any("unknown node" in rec.message and "stranger" in rec.message for rec in caplog.records)