Adding Cluster Telemetry

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();
+startStaleWorkerDetection();
+

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,

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,7 +178,6 @@ 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];
@@ -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;
@@ -253,6 +262,9 @@ function triggerWorkerUpdate(targetVersion, updateCmd) {
   // Stop reacting to further control messages
   authorized = false;
   
+  // ✅ Stop telemetry during update
+  stopTelemetry();
+
   try {
     exec(updateCmd, { stdio: 'inherit' });
   } catch (err) {
@@ -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