sdl-wkr Cluster Join

docs/DESIGN.md

@@ -8,6 +8,9 @@
 | **Version** | 0.3.2 |
 | **Date** | 2026-01-31 |
 
+## Overview
+
+The Software Defined Laboratory (SDL) project provides a generalized distributed computing platform for managing parallel computational workflows across a cluster of distributed nodes.  SDL is a minimalistic High Performance Compute (HPC) platform focused on extreme minimizaton of technical debt associated configuration, deployability, and maintenance of the cluster.
 
 # Roadmap
 
@@ -26,16 +29,16 @@ MVP - Minimum Viable Product Task List
 - [x] SDL_ID Node ID
 - [x] Version Updater Script
 - [x] Deploy sld-mgr to $SDL_HOME/sdl-mgr
-- [ ] Create Worker Join Process
-- [ ] Create Worker Heartbeat Process
 - [x] Create SDL Worker Dist Build Process
 - [x] Create SDL Worker Install to $SDL_HOME
 - [x] SDL Install Script
+- [ ] Create SDL Worker Auto Update Process
+- [ ] Create Worker Join Process
+- [ ] Create Worker Telementry Process
 - [ ] Data Storage Organizational Structure
 - [ ] MinIO S3 Storage Server
-- [ ] Worker Heartbeat Hardware Inventory (CPU, RAM, GPU, etc)
-- [ ] Create SDL Worker Auto Update Process
-- [ ] Worker Heartbeat Load Status (CPU, RAM, GPU, Disk IO, etc)
+- [ ] Worker Telemetry Hardware Inventory (CPU, RAM, GPU, etc)
+- [ ] Worker Telementry Load Status (CPU, RAM, GPU, Disk IO, etc)
 
 # Design
 

sdl-mgr/app/modules/sdl-mgr/index.js

@@ -191,7 +191,8 @@ function startUdpBeacon() {
               host: addr.address,
               port: config.modules.web.port,
               api_config: '/api/config',
-              config_url: `http://${addr.address}:${config.modules.web.port}/api/config` 
+              config_url: `http://${addr.address}:${config.modules.web.port}/api/config`,
+              web_ui_url: `http://${addr.address}:${config.modules.web.port}`, 
             },
             sdl_wkr_install_cmd: {
               curl: `curl -s http://${addr.address}:${config.modules.web.port}/dist/install-sdl-wkr.sh | bash -s ${addr.address}`,

sdl-mgr/conf/modules/mqtt.json

@@ -6,7 +6,7 @@
     "topics":{
         "sdl-mgr":{
             "pub": {
-                "sdl_join-auth": "sdl/join/auth",
+                "sdl_join-authz": "sdl/join/authz",
                 "sdl_cluster-status": "sdl/cluster/status",
                 "sdl_cluster-workers": "sdl/cluster/workers"
             },
@@ -21,7 +21,7 @@
                 "sdl_cluster-telemetry": "sdl/cluster/telemetry"
             },
             "sub": {
-                "sdl_join-auth": "sdl/join/auth",
+                "sdl_join-authz": "sdl/join/authz",
                 "sdl_cluster-status": "sdl/cluster/status"
             }
         },

sdl-wkr/app/modules/nwa-lib/index.js

@@ -1,3 +1,4 @@
+// Node Web App (NWA) Library Module
 import fs from 'fs';
 import path from 'path';
 import os from 'os';
@@ -6,12 +7,16 @@ import { hideBin } from 'yargs/helpers';
 import { fileURLToPath } from 'url';
 import crypto from 'crypto';
 import { execFileSync } from 'child_process';
+import { get } from 'http';
 
 const __filename = fileURLToPath(import.meta.url);
 const __dirname = path.dirname(__filename);
 
 // console.log("DEBUG: __dirname: " + __dirname);
 
+// Process start time (set once per Node.js process)
+const PROCESS_START_TS = Date.now();
+
 const config = load_config();
 
 function load_config() {
@@ -572,4 +577,26 @@ function pciShortId(id) {
   return m ? m[1].toLowerCase() : null;
 }
 
-export { load_config, log, ipToInt, intToIp, computeBroadcast };
+function getProcessStartTs() {
+  return PROCESS_START_TS;
+}
+
+function getUptimeMs() {
+  return Date.now() - PROCESS_START_TS;
+}
+
+function getUptimeSec() {
+  return Math.floor((Date.now() - PROCESS_START_TS) / 1000);
+}
+
+function getUptimeDHMS() {
+  const uptimeSec = getUptimeSec();
+  const d = Math.floor(uptimeSec / 86400);
+  const h = Math.floor((uptimeSec % 86400) / 3600);
+  const m = Math.floor((uptimeSec % 3600) / 60);
+  const s = Math.floor(uptimeSec % 60);
+  return `${d}.${h}.${m}.${s}`;
+}
+
+
+export { load_config, log, ipToInt, intToIp, computeBroadcast, getProcessStartTs, getUptimeMs, getUptimeSec, getUptimeDHMS };

sdl-wkr/app/modules/sdl-wkr/index.js

@@ -8,13 +8,12 @@ const config = load_config();
 
 log(`Loaded module: ${module}`);
 
-// log(`${module}: CONFIG: ${JSON.stringify(config, null, 2)}`, true);
-
 // -------------------------------
 // Worker state
 // -------------------------------
 let joined = false;
 let mqttClient = null;
+let clusterConfig = null;
 
 // -------------------------------
 // Start UDP discovery
@@ -43,11 +42,11 @@ function startUdpListener() {
     try {
       const beacon = JSON.parse(msg.toString());
 
-      if (beacon?.sdl?.cluster?.id == null) return;
-      if (beacon?.sdl?.mqtt?.host == null) return;
+      if (beacon?.type !== 'udp-beacon') return;
+      if (beacon?.msg?.cluster?.id == null) return;
+      if (beacon?.msg?.mqtt?.host == null) return;
 
       log(`${module}: received SDL beacon from ${rinfo.address}`);
-      log(`${module}: UDP beacon: ${JSON.stringify(beacon)}`, false);
 
       handleBeacon(beacon);
     } catch {
@@ -64,26 +63,37 @@ function startUdpListener() {
 // -------------------------------
 // Handle beacon
 // -------------------------------
-function handleBeacon(beacon) {
-  const sdl = beacon.sdl;
-  const mqttInfo = sdl.mqtt;
+async function handleBeacon(beacon) {
+  const beaconMsg = beacon.msg;
+  const mqttInfo = beaconMsg.mqtt;
+  const cluster = beaconMsg.cluster;
 
-  // Cluster match (strict)
-  const expectedClusterId =
-    config.modules[module].cluster_id || sdl.cluster.id;
-
-  if (sdl.cluster.id !== expectedClusterId) {
-    log(
-      `${module}: ignoring beacon for cluster ${sdl.cluster.id}`
-    );
+  // Optional: If cluster_id is configured, validate it. Otherwise auto-join first beacon.
+  const expectedClusterId = config.modules[module].cluster_id;
+  if (expectedClusterId && cluster.id !== expectedClusterId) {
+    log(`${module}: ignoring beacon for cluster ${cluster.id}`);
     return;
   }
 
-  log(
-    `${module}: joining cluster '${sdl.cluster.name}' via MQTT ${mqttInfo.host}:${mqttInfo.port}`
-  );
+  log(`${module}: cluster '${cluster.name}' discovered via ${beacon.host}`);
 
-  joinCluster(mqttInfo, sdl.cluster);
+  // Fetch full cluster config from API
+  const configUrl = beaconMsg.web.config_url;
+  try {
+    const response = await fetch(configUrl);
+    if (!response.ok) {
+      throw new Error(`HTTP ${response.status}`);
+    }
+    clusterConfig = await response.json();
+    log(`${module}: fetched cluster config from ${configUrl}`);
+  } catch (err) {
+    log(`${module}: failed to fetch cluster config: ${err}`);
+    return;
+  }
+
+  log(`${module}: joining cluster '${cluster.name}' via MQTT ${mqttInfo.host}:${mqttInfo.port}`);
+
+  joinCluster(mqttInfo, cluster);
 }
 
 // -------------------------------
@@ -98,33 +108,96 @@ function joinCluster(mqttInfo, cluster) {
   mqttClient.on('connect', () => {
     log(`${module}: connected to MQTT broker at ${mqttUrl}`);
 
+    const mqttTopics = clusterConfig.modules.mqtt.topics[module];
+    
+    // Subscribe to authz topic
+    const authzTopic = mqttTopics.sub['sdl_join-authz'];
+    mqttClient.subscribe(authzTopic, { qos: 1 }, err => {
+      if (err) {
+        log(`${module}: failed to subscribe to ${authzTopic}: ${err}`);
+      } else {
+        log(`${module}: subscribed to ${authzTopic}`);
+      }
+    });
+
+    const joinTopic = mqttTopics.pub['sdl_join-req'];
+    
     const joinPayload = {
-      type: 'sdl_worker_join',
-      timestamp: new Date().toISOString(),
-      sdl_wkr: config.identity.hostname,
-      sdl_id: config.identity.sdl_id
+      ts: new Date().toISOString(),
+      sdl_id: config.identity.sdl_id,
+      role: 'sdl-wkr',
+      host: config.identity.hostname,
+      type: 'join-request',
+      msg: {
+        cluster_id: cluster.id,
+        worker: {
+          sdl_id: config.identity.sdl_id,
+          hostname: config.identity.hostname,
+          version: config.package.version,
+          platform: os.platform(),
+          arch: os.arch(),
+          cpus: os.cpus().length,
+          totalmem: os.totalmem()
+        }
+      }
     };
 
     mqttClient.publish(
-      mqttInfo.join_topic,
+      joinTopic,
       JSON.stringify(joinPayload),
       { qos: 1 },
       err => {
         if (err) {
-          log(`${module}: failed to publish join message: ${err}`);
+          log(`${module}: failed to publish join request: ${err}`);
         } else {
-          log(`${module}: published join request to ${mqttInfo.join_topic}`);
+          log(`${module}: published join request to ${joinTopic}`);
         }
       }
     );
   });
 
+  mqttClient.on('message', (topic, message) => {
+    handleMqttMessage(topic, message);
+  });
+
   mqttClient.on('error', err => {
     log(`${module}: MQTT error: ${err}`);
   });
 }
 
+// -------------------------------
+// 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);
+    }
+  } catch (err) {
+    log(`${module}: failed to parse MQTT message: ${err}`);
+  }
+}
+
+// -------------------------------
+// Handle join authorization
+// -------------------------------
+function handleJoinAuthz(payload) {
+  // Check if this authz is for us
+  if (payload.msg?.sdl_id !== config.identity.sdl_id) {
+    return;
+  }
+
+  if (payload.msg?.authorized === true) {
+    log(`${module}: join authorized by cluster`);
+  } else {
+    log(`${module}: join denied: ${payload.msg?.reason || 'unknown'}`);
+  }
+}
+
 // -------------------------------
 // Entry point
 // -------------------------------
-startUdpListener();
+startUdpListener();