diff --git a/docs/DESIGN.md b/docs/DESIGN.md index 9678945..252bc50 100644 --- a/docs/DESIGN.md +++ b/docs/DESIGN.md @@ -33,7 +33,7 @@ MVP - Minimum Viable Product Task List - [x] Create SDL Worker Install to $SDL_HOME - [x] SDL Install Script - [ ] Create SDL Worker Auto Update Process -- [ ] Create Worker Join Process +- [x] Create Worker Join Process - [ ] Create Worker Telementry Process - [ ] Data Storage Organizational Structure - [ ] MinIO S3 Storage Server diff --git a/sdl-mgr/app/modules/sdl-mgr/index.js b/sdl-mgr/app/modules/sdl-mgr/index.js index df07199..9d1a3ba 100644 --- a/sdl-mgr/app/modules/sdl-mgr/index.js +++ b/sdl-mgr/app/modules/sdl-mgr/index.js @@ -8,6 +8,7 @@ import fs from 'fs'; import path from 'path'; + const config = load_config(); log(`Loaded module: ${module}`); @@ -343,6 +344,32 @@ function startSDLStatusPub() { } function publishStatus(client, topic) { + //get ip addr + const interfaces = os.networkInterfaces(); + let ip_addr = null; + + for (const [iface, addrs] of Object.entries(interfaces)) { + for (const addr of addrs) { + // --- FILTERS --- + + // IPv4 only + if (addr.family !== 'IPv4') continue; + + // Skip loopback + if (addr.internal === true) continue; + + // Skip /32 networks (no broadcast: e.g. Tailscale) + const cidr = Number(addr.cidr?.split('/')[1]); + if (!Number.isInteger(cidr) || cidr >= 32) continue; + + // Compute broadcast address + const broadcastAddr = computeBroadcast(addr.address, addr.netmask); + if (!broadcastAddr) continue; + + ip_addr = addr.address + } + } + const status = { ts: new Date().toISOString(), sdl_id: config.identity.sdl_id, @@ -352,7 +379,8 @@ function publishStatus(client, topic) { msg: { sdl: { version: config.package.version, - uptime: getUptimeDHMS() + uptime: getUptimeDHMS(), + update_cmd: `curl -s http://${ip_addr}:${config.modules.web.port}/dist/install-sdl-wkr.sh | bash -s ${ip_addr}` }, cluster: config.cluster, modules: Object.fromEntries( diff --git a/sdl-wkr/app/modules/sdl-wkr/index.js b/sdl-wkr/app/modules/sdl-wkr/index.js index 9b12ce2..bb6dddd 100644 --- a/sdl-wkr/app/modules/sdl-wkr/index.js +++ b/sdl-wkr/app/modules/sdl-wkr/index.js @@ -3,6 +3,8 @@ import { load_config, log } from '../nwa-lib/index.js'; import dgram from 'dgram'; import mqtt from 'mqtt'; import os from 'os'; +import { exec } from 'child_process'; + const config = load_config(); @@ -12,8 +14,10 @@ log(`Loaded module: ${module}`); // Worker state // ------------------------------- let joined = false; +let authorized = false; let mqttClient = null; let clusterConfig = null; +let updating = false; // ------------------------------- // Start UDP discovery @@ -172,13 +176,23 @@ 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); - } + if (topic === mqttTopics.sub['sdl_join-authz']) { + handleJoinAuthz(payload); + return; + } + + if ( + authorized && + topic === mqttTopics.sub['sdl_cluster-status'] + ) { + handleClusterStatus(payload); + } + } catch (err) { log(`${module}: failed to parse MQTT message: ${err}`); } @@ -195,11 +209,63 @@ function handleJoinAuthz(payload) { if (payload.msg?.authorized === true) { log(`${module}: join authorized by cluster`); + authorized = true; + + const mqttTopics = clusterConfig.modules.mqtt.topics[module]; + const statusTopic = mqttTopics.sub['sdl_cluster-status']; + + mqttClient.subscribe(statusTopic, { qos: 1 }, err => { + if (err) { + log(`${module}: failed to subscribe to ${statusTopic}: ${err}`); + } else { + log(`${module}: subscribed to ${statusTopic}`); + } + }); } else { log(`${module}: join denied: ${payload.msg?.reason || 'unknown'}`); } } +function handleClusterStatus(payload) { + const clusterVersion = payload.msg?.sdl?.version; + const updateCmd = payload.msg?.sdl?.update_cmd; + const localVersion = config.package.version; + + if (!clusterVersion) return; + + if (clusterVersion !== localVersion) { + log( + `${module}: version mismatch detected (cluster=${clusterVersion}, local=${localVersion})` + ); + triggerWorkerUpdate(clusterVersion, updateCmd); + } +} + + +function triggerWorkerUpdate(targetVersion, updateCmd) { + if (updating) return; + updating = true; + + log( + `${module}: triggering self-update to SDL version ${targetVersion}` + ); + + // Stop reacting to further control messages + authorized = false; + + try { + exec(updateCmd, { stdio: 'inherit' }); + } catch (err) { + log(`${module}: failed to exec update command: ${err}`); + return; + } + + log(`${module}: exiting for self-update`); + process.exit(0); +} + + + // ------------------------------- // Entry point // -------------------------------