diff --git a/sdl-mgr/app/modules/nwa-lib/index.js b/sdl-mgr/app/modules/nwa-lib/index.js index a027e66..df7665f 100644 --- a/sdl-mgr/app/modules/nwa-lib/index.js +++ b/sdl-mgr/app/modules/nwa-lib/index.js @@ -185,6 +185,10 @@ function load_config() { }; + config.sdl = { + version: config.package.version + } + config.identity = { sdl_id: null, created_at: null, diff --git a/sdl-mgr/app/modules/sdl-mgr/index.js b/sdl-mgr/app/modules/sdl-mgr/index.js index 00d2a87..33172a6 100644 --- a/sdl-mgr/app/modules/sdl-mgr/index.js +++ b/sdl-mgr/app/modules/sdl-mgr/index.js @@ -77,6 +77,224 @@ function loadClusterState(config) { return JSON.parse(data); } +function saveClusterState(config, state) { + const clusterDir = config.dirs.clstr; + const clusterFile = path.join(clusterDir, 'cluster.json'); + + // Update metadata timestamp and uptime + state.meta.updated = new Date().toISOString(); + state.meta.uptime = getUptimeDHMS(); + + // Write to disk + try { + fs.writeFileSync(clusterFile, JSON.stringify(state, null, 2)); + log(`${module}: cluster state saved to ${clusterFile}`); + } catch (err) { + log(`${module}: failed to save cluster state: ${err}`); + throw err; + } +} + +function addWorker(state, workerData) { + const sdl_id = workerData.sdl_id; + + // Check if worker already exists + const isNewWorker = !state.workers[sdl_id]; + + // Add or update worker + state.workers[sdl_id] = { + sdl_id: workerData.sdl_id, + hostname: workerData.hostname, + version: workerData.version, + platform: workerData.platform, + arch: workerData.arch, + distro: workerData.distro || "Unknown", + distro_name: workerData.distro_name || "Unknown", + distro_version: workerData.distro_version || "Unknown", + resources: { + cpus: { + allocated: workerData.cpus || 0, + available: workerData.cpus || 0, + used: 0 + }, + memory: { + allocated: workerData.totalmem || 0, + available: workerData.totalmem || 0, + used: 0 + }, + gpus: { + allocated: workerData.gpus || 0, + available: workerData.gpus || 0, + used: 0 + } + }, + status: 'active', + joined_at: isNewWorker ? new Date().toISOString() : state.workers[sdl_id].joined_at, + last_seen: new Date().toISOString() + }; + + return state; +} + +function computeStats(state) { + const stats = { + workers: { + allocated: 0, + available: 0, + used: 0 + }, + resources: { + cpus: { + allocated: 0, + available: 0, + used: 0 + }, + memory: { + allocated: 0, + available: 0, + used: 0 + }, + gpus: { + allocated: 0, + available: 0, + used: 0 + } + } + }; + + // Count workers and aggregate resources + for (const worker of Object.values(state.workers)) { + stats.workers.allocated++; + + if (worker.status === 'active') { + stats.workers.available++; + + stats.resources.cpus.available += worker.resources.cpus.available; + stats.resources.memory.available += worker.resources.memory.available; + stats.resources.gpus.available += worker.resources.gpus.available; + } + + stats.resources.cpus.allocated += worker.resources.cpus.allocated; + stats.resources.cpus.used += worker.resources.cpus.used; + + stats.resources.memory.allocated += worker.resources.memory.allocated; + stats.resources.memory.used += worker.resources.memory.used; + + stats.resources.gpus.allocated += worker.resources.gpus.allocated; + stats.resources.gpus.used += worker.resources.gpus.used; + } + + state.meta.stats = stats; + return state; +} + +// Worker Join Handler +function startJoinHandler() { + const sdlCfg = config.modules[module]; + const mqttCfg = config.modules.mqtt; + + if (!sdlCfg?.enabled) { + log(`${module}: disabled, not starting join handler`); + return; + } + + if (!mqttCfg?.enabled) { + log(`${module}: MQTT disabled, cannot handle joins`); + return; + } + + // Load current cluster state + let clusterState = loadClusterState(config); + + const joinReqTopic = mqttCfg.topics?.[module]?.sub?.['sdl_join-req']; + const joinAuthzTopic = mqttCfg.topics?.[module]?.pub?.['sdl_join-authz']; + + if (!joinReqTopic || !joinAuthzTopic) { + log(`${module}: join topics not configured`); + return; + } + + const mqttUrl = `mqtt://127.0.0.1:${mqttCfg.mqtt_port}`; + const client = mqtt.connect(mqttUrl); + + client.on('connect', () => { + log(`${module}: join handler connected to MQTT at ${mqttUrl}`); + + client.subscribe(joinReqTopic, { qos: 1 }, err => { + if (err) { + log(`${module}: failed to subscribe to ${joinReqTopic}: ${err}`); + } else { + log(`${module}: subscribed to ${joinReqTopic}`); + } + }); + }); + + client.on('message', (topic, message) => { + if (topic !== joinReqTopic) return; + + try { + const joinReq = JSON.parse(message.toString()); + + log(`${module}: received join request from ${joinReq.host} (${joinReq.sdl_id})`); + + // Validate version + const workerVersion = joinReq.msg?.worker?.version; + const clusterVersion = config.package.version; + + let authorized = false; + let reason = null; + + if (workerVersion === clusterVersion) { + authorized = true; + + // Add worker to cluster state + clusterState = addWorker(clusterState, joinReq.msg.worker); + clusterState = computeStats(clusterState); + saveClusterState(config, clusterState); + + log(`${module}: worker ${joinReq.host} authorized and added to cluster`); + } else { + authorized = false; + reason = 'version_mismatch'; + log(`${module}: worker ${joinReq.host} denied - version mismatch (worker: ${workerVersion}, cluster: ${clusterVersion})`); + } + + // Publish authorization response + const authzResponse = { + ts: new Date().toISOString(), + sdl_id: config.identity.sdl_id, + role: 'sdl-mgr', + host: config.identity.hostname, + type: 'join-authz', + msg: { + sdl_id: joinReq.sdl_id, + authorized: authorized + } + }; + + client.publish( + joinAuthzTopic, + JSON.stringify(authzResponse), + { qos: 1 }, + err => { + if (err) { + log(`${module}: failed to publish join authz: ${err}`); + } else { + log(`${module}: published join authz to ${joinAuthzTopic}`); + } + } + ); + + } catch (err) { + log(`${module}: failed to process join request: ${err}`); + } + }); + + client.on('error', err => { + log(`${module}: join handler MQTT error: ${err}`); + }); +} + function startSDLStatusPub() { @@ -300,4 +518,5 @@ function startUdpBeacon() { // ------------------------------- startSDLStatusPub(); startUdpBeacon(); -loadClusterState(config); \ No newline at end of file +loadClusterState(config); +startJoinHandler(); \ No newline at end of file diff --git a/sdl-wkr/app/index.js b/sdl-wkr/app/index.js index a81d329..e37d307 100644 --- a/sdl-wkr/app/index.js +++ b/sdl-wkr/app/index.js @@ -10,7 +10,7 @@ const config = load_config(); log('======================================================================================='); log(config.package.name + ': STARTING: ' + config.package.description + ' v' + config.package.version); -// log(JSON.stringify(config, null, 2)); +log(JSON.stringify(config, null, 2)); // log(JSON.stringify(config.dirs, null, 2), false); for (let m in config.modules) { diff --git a/sdl-wkr/app/modules/nwa-lib/index.js b/sdl-wkr/app/modules/nwa-lib/index.js index d4eb9a7..a0de696 100644 --- a/sdl-wkr/app/modules/nwa-lib/index.js +++ b/sdl-wkr/app/modules/nwa-lib/index.js @@ -182,6 +182,10 @@ function load_config() { }; + config.sdl = { + version: config.package.version + } + config.identity = { sdl_id: null, created_at: null, diff --git a/sdl-wkr/app/modules/sdl-wkr/index.js b/sdl-wkr/app/modules/sdl-wkr/index.js index 472596d..9b12ce2 100644 --- a/sdl-wkr/app/modules/sdl-wkr/index.js +++ b/sdl-wkr/app/modules/sdl-wkr/index.js @@ -130,12 +130,15 @@ function joinCluster(mqttInfo, cluster) { type: 'join-request', msg: { cluster_id: cluster.id, - worker: { + "sdl-wkr": { sdl_id: config.identity.sdl_id, hostname: config.identity.hostname, - version: config.package.version, - platform: os.platform(), - arch: os.arch(), + sdl_version: config.package.version, + platform: config.host.os.platform, + arch: config.host.cpu.arch, + distro: config.host.os.pretty_name, + distro_name: config.host.os.name, + distro_version: config.host.os.version, cpus: os.cpus().length, totalmem: os.totalmem() } diff --git a/sdl-wkr/conf/default-sdl-wkr-config.json b/sdl-wkr/conf/default-sdl-wkr-config.json index ef898e7..0de61b6 100644 --- a/sdl-wkr/conf/default-sdl-wkr-config.json +++ b/sdl-wkr/conf/default-sdl-wkr-config.json @@ -6,8 +6,7 @@ "modules": "APP/modules", "logs": "SDLHOME/logs", "data": "SDLHOME/data", - "proj": "DATA/projects", - "jobs": "DATA/jobs", + "sdl-wkr": "DATA/sdl-wkr", "conf": "SDLHOME/conf", "modconf": "BASE/conf/modules" },