From 4918f710f8f23124a199b0679efaea2d2e01ef5b Mon Sep 17 00:00:00 2001 From: John Haverlack Date: Sun, 1 Feb 2026 18:33:08 -0900 Subject: [PATCH] Adding Cluster Telemetry --- sdl-mgr/app/modules/sdl-mgr/index.js | 119 +++++++++++++++++++++++-- sdl-mgr/conf/modules/sdl-mgr.json | 1 + sdl-wkr/app/modules/sdl-wkr/index.js | 124 ++++++++++++++++++++++++--- 3 files changed, 226 insertions(+), 18 deletions(-) diff --git a/sdl-mgr/app/modules/sdl-mgr/index.js b/sdl-mgr/app/modules/sdl-mgr/index.js index 715a07b..fb4f7f7 100644 --- a/sdl-mgr/app/modules/sdl-mgr/index.js +++ b/sdl-mgr/app/modules/sdl-mgr/index.js @@ -621,12 +621,121 @@ function startUdpBeacon() { } +// Update startHeartbeatListener to use telemetry topic +function startTelemetryListener() { // ✅ Renamed function + const sdlCfg = config.modules[module]; + const mqttCfg = config.modules.mqtt; -// ------------------------------- -// Entry point (ESM-safe) -// ------------------------------- + if (!sdlCfg?.enabled) { + log(`${module}: disabled, not listening for telemetry`); + return; + } + + if (!mqttCfg?.enabled) { + log(`${module}: MQTT disabled, cannot listen for telemetry`); + return; + } + + let clusterState = loadClusterState(config); + + const telemetryTopic = mqttCfg.topics?.[module]?.sub?.['sdl_cluster-telemetry']; // ✅ Changed + if (!telemetryTopic) { + log(`${module}: cluster-telemetry topic not configured`); + return; + } + + const mqttUrl = `mqtt://127.0.0.1:${mqttCfg.mqtt_port}`; + const client = mqtt.connect(mqttUrl); + + client.on('connect', () => { + log(`${module}: telemetry listener connected to MQTT at ${mqttUrl}`); + + client.subscribe(telemetryTopic, { qos: 1 }, err => { + if (err) { + log(`${module}: failed to subscribe to ${telemetryTopic}: ${err}`); + } else { + log(`${module}: subscribed to ${telemetryTopic}`); + } + }); + }); + + client.on('message', (topic, message) => { + if (topic !== telemetryTopic) return; + + try { + const telemetry = JSON.parse(message.toString()); + const sdl_id = telemetry.msg.sdl_id; + + // Update worker's last_seen timestamp + if (clusterState.workers[sdl_id]) { + clusterState.workers[sdl_id].last_seen = new Date().toISOString(); + clusterState.workers[sdl_id].status = 'active'; + + // Update resources if provided + if (telemetry.msg.resources) { + clusterState.workers[sdl_id].resources = telemetry.msg.resources; + } + + clusterState = computeStats(clusterState); + saveClusterState(config, clusterState); + } + } catch (err) { + log(`${module}: failed to process telemetry: ${err}`); + } + }); + + client.on('error', err => { + log(`${module}: telemetry listener MQTT error: ${err}`); + }); +} + +// Add Stale Worker Detection +function startStaleWorkerDetection() { + const sdlCfg = config.modules[module]; + + if (!sdlCfg?.enabled) { + log(`${module}: disabled, not starting stale worker detection`); + return; + } + + const checkInterval = 10000; // Check every 10 seconds + const staleThreshold = + Number.isInteger(sdlCfg.worker_stale_threshold) && + sdlCfg.worker_stale_threshold > 0 + ? sdlCfg.worker_stale_threshold + : 30000; // Default 30 seconds + + setInterval(() => { + let clusterState = loadClusterState(config); + const now = Date.now(); + let changed = false; + + for (const [sdl_id, worker] of Object.entries(clusterState.workers)) { + const lastSeenMs = Date.parse(worker.last_seen); + const ageMs = now - lastSeenMs; + + if (ageMs > staleThreshold && worker.status === 'active') { + log(`${module}: worker ${worker.hostname} (${sdl_id}) marked as inactive (last seen ${Math.round(ageMs / 1000)}s ago)`); + clusterState.workers[sdl_id].status = 'inactive'; + changed = true; + } + } + + if (changed) { + clusterState = computeStats(clusterState); + saveClusterState(config, clusterState); + } + }, checkInterval); + + log(`${module}: stale worker detection active (threshold: ${staleThreshold}ms, check interval: ${checkInterval}ms)`); +} + +// Update entry point startSDLStatusPub(); -startWorkersPub(); // ✅ Start workers publisher +startWorkersPub(); +startTelemetryListener(); startUdpBeacon(); loadClusterState(config); -startJoinHandler(); \ No newline at end of file +startJoinHandler(); +startStaleWorkerDetection(); + diff --git a/sdl-mgr/conf/modules/sdl-mgr.json b/sdl-mgr/conf/modules/sdl-mgr.json index 049e810..d4f7d6d 100644 --- a/sdl-mgr/conf/modules/sdl-mgr.json +++ b/sdl-mgr/conf/modules/sdl-mgr.json @@ -1,6 +1,7 @@ { "enabled": true, "beacon_udp_port":10101, + "worker_stale_threshold": 15000, "update_interval": { "udp_beacon": 2000, "cluster_status": 10000, diff --git a/sdl-wkr/app/modules/sdl-wkr/index.js b/sdl-wkr/app/modules/sdl-wkr/index.js index dab2045..94d47bd 100644 --- a/sdl-wkr/app/modules/sdl-wkr/index.js +++ b/sdl-wkr/app/modules/sdl-wkr/index.js @@ -18,6 +18,7 @@ let authorized = false; let mqttClient = null; let clusterConfig = null; let updating = false; +let telemetryTimer = null; // ------------------------------- // Start UDP discovery @@ -144,7 +145,8 @@ function joinCluster(mqttInfo, cluster) { distro_name: config.host.os.name, distro_version: config.host.os.version, cpus: os.cpus().length, - totalmem: os.totalmem() + totalmem: os.totalmem(), + gpus: Array.isArray(config.host.gpu) ? config.host.gpu.length : 0 // ✅ Fixed } } }; @@ -176,22 +178,21 @@ function joinCluster(mqttInfo, cluster) { // Handle MQTT messages // ------------------------------- function handleMqttMessage(topic, message) { - try { const payload = JSON.parse(message.toString()); const mqttTopics = clusterConfig.modules.mqtt.topics[module]; - if (topic === mqttTopics.sub['sdl_join-authz']) { - handleJoinAuthz(payload); - return; - } + if (topic === mqttTopics.sub['sdl_join-authz']) { + handleJoinAuthz(payload); + return; + } - if ( - authorized && - topic === mqttTopics.sub['sdl_cluster-status'] - ) { - handleClusterStatus(payload); - } + if ( + authorized && + topic === mqttTopics.sub['sdl_cluster-status'] + ) { + handleClusterStatus(payload); + } } catch (err) { log(`${module}: failed to parse MQTT message: ${err}`); @@ -221,11 +222,17 @@ function handleJoinAuthz(payload) { log(`${module}: subscribed to ${statusTopic}`); } }); + + // ✅ Start telemetry after authorization + startTelemetry(); } else { log(`${module}: join denied: ${payload.msg?.reason || 'unknown'}`); } } +// ------------------------------- +// Handle cluster status +// ------------------------------- function handleClusterStatus(payload) { const clusterVersion = payload.msg?.sdl?.version; const updateCmd = payload.msg?.sdl?.update_cmd; @@ -241,7 +248,9 @@ function handleClusterStatus(payload) { } } - +// ------------------------------- +// Trigger worker update +// ------------------------------- function triggerWorkerUpdate(targetVersion, updateCmd) { if (updating) return; updating = true; @@ -252,6 +261,9 @@ function triggerWorkerUpdate(targetVersion, updateCmd) { // Stop reacting to further control messages authorized = false; + + // ✅ Stop telemetry during update + stopTelemetry(); try { exec(updateCmd, { stdio: 'inherit' }); @@ -261,7 +273,93 @@ function triggerWorkerUpdate(targetVersion, updateCmd) { } } +// ------------------------------- +// ✅ Worker Telemetry +// ------------------------------- +function startTelemetry() { + if (telemetryTimer) { + clearInterval(telemetryTimer); + } + const telemetryInterval = + Number.isInteger(config.modules[module].update_interval?.worker_telemetry) && + config.modules[module].update_interval.worker_telemetry > 0 + ? config.modules[module].update_interval.worker_telemetry + : 5000; // Default 5 seconds + + const mqttTopics = clusterConfig.modules.mqtt.topics[module]; + const telemetryTopic = mqttTopics.pub['sdl_cluster-telemetry']; + + if (!telemetryTopic) { + log(`${module}: cluster-telemetry topic not configured`); + return; + } + + log(`${module}: starting telemetry (interval: ${telemetryInterval}ms)`); + + // Publish immediately + publishTelemetry(telemetryTopic); + + // Then publish periodically + telemetryTimer = setInterval(() => { + publishTelemetry(telemetryTopic); + }, telemetryInterval); +} + +function publishTelemetry(topic) { + if (!mqttClient || !authorized) return; + + const gpuCount = Array.isArray(config.host.gpu) ? config.host.gpu.length : 0; // ✅ Fixed + + const telemetry = { + ts: new Date().toISOString(), + sdl_id: config.identity.sdl_id, + role: 'sdl-wkr', + host: config.identity.hostname, + type: 'worker-telemetry', + msg: { + sdl_id: config.identity.sdl_id, + hostname: config.identity.hostname, + status: 'active', + resources: { + cpus: { + allocated: config.host.cpu.cores_logical, + available: config.host.cpu.cores_logical, + used: 0 + }, + memory: { + allocated: config.host.memory.total_bytes, + available: config.host.memory.total_bytes, + used: 0 + }, + gpus: { + allocated: gpuCount, + available: gpuCount, + used: 0 + } + } + } + }; + + mqttClient.publish( + topic, + JSON.stringify(telemetry), + { qos: 1 }, + err => { + if (err) { + log(`${module}: failed to publish telemetry: ${err}`); + } + } + ); +} + +function stopTelemetry() { + if (telemetryTimer) { + clearInterval(telemetryTimer); + telemetryTimer = null; + log(`${module}: telemetry stopped`); + } +} // ------------------------------- // Entry point