HarborForge.OpenclawPlugin/server/telemetry.mjs

/**
 * HarborForge Monitor Telemetry Server
 * 
 * Runs as separate process from Gateway.
 * Collects system metrics and OpenClaw status, sends to Monitor.
 */
import { readFile, access, readdir } from 'fs/promises';
import { constants } from 'fs';
import { exec } from 'child_process';
import { promisify } from 'util';
import { platform, hostname, freemem, totalmem, uptime, loadavg } from 'os';

const execAsync = promisify(exec);

// Config from environment (set by plugin)
const openclawPath = process.env.OPENCLAW_PATH || `${process.env.HOME}/.openclaw`;
const CONFIG = {
  backendUrl: process.env.HF_MONITOR_BACKEND_URL || 'https://monitor.hangman-lab.top',
  identifier: process.env.HF_MONITOR_IDENTIFIER || hostname(),
  apiKey: process.env.HF_MONITOR_API_KEY,
  reportIntervalSec: parseInt(process.env.HF_MONITOR_REPORT_INTERVAL || '30', 10),
  httpFallbackIntervalSec: parseInt(process.env.HF_MONITOR_HTTP_FALLBACK_INTERVAL || '60', 10),
  logLevel: process.env.HF_MONITOR_LOG_LEVEL || 'info',
  openclawPath,
  pluginVersion: process.env.HF_MONITOR_PLUGIN_VERSION || 'unknown',
  cachePath: process.env.HF_MONITOR_CACHE_PATH || `${openclawPath}/telemetry_cache.json`,
  maxCacheSize: parseInt(process.env.HF_MONITOR_MAX_CACHE_SIZE || '100', 10),
};

// Logging
const log = {
  debug: (...args) => CONFIG.logLevel === 'debug' && console.log('[DEBUG]', ...args),
  info: (...args) => ['debug', 'info'].includes(CONFIG.logLevel) && console.log('[INFO]', ...args),
  warn: (...args) => console.log('[WARN]', ...args),
  error: (...args) => console.error('[ERROR]', ...args),
};

// State
let wsConnection = null;
let lastSuccessfulSend = null;
let consecutiveFailures = 0;
let isShuttingDown = false;
let cachedOpenclawVersion = null;

/**
 * Collect system metrics
 */
async function collectSystemMetrics() {
  try {
    const cpuUsage = await getCpuUsage();
    const memTotal = totalmem();
    const memFree = freemem();
    const memUsed = memTotal - memFree;
    const diskInfo = await getDiskUsage();
    const loadAvg = platform() !== 'win32' ? loadavg() : [0, 0, 0];
    
    return {
      cpu_pct: cpuUsage,
      mem_pct: Math.round((memUsed / memTotal) * 100 * 10) / 10,
      mem_used_mb: Math.round(memUsed / 1024 / 1024),
      mem_total_mb: Math.round(memTotal / 1024 / 1024),
      disk_pct: diskInfo.usedPct,
      disk_used_gb: Math.round(diskInfo.usedGB * 10) / 10,
      disk_total_gb: Math.round(diskInfo.totalGB * 10) / 10,
      swap_pct: diskInfo.swapUsedPct || 0,
      uptime_seconds: Math.floor(uptime()),
      load_avg: [
        Math.round(loadAvg[0] * 100) / 100,
        Math.round(loadAvg[1] * 100) / 100,
        Math.round(loadAvg[2] * 100) / 100,
      ],
      platform: platform(),
      hostname: hostname(),
    };
  } catch (err) {
    log.error('Failed to collect system metrics:', err.message);
    return {};
  }
}

/**
 * Get CPU usage percentage
 */
async function getCpuUsage() {
  try {
    if (platform() === 'linux') {
      const { stdout } = await execAsync("top -bn1 | grep 'Cpu(s)' | awk '{print $2}' | cut -d'%' -f1");
      const usage = parseFloat(stdout.trim());
      return isNaN(usage) ? 0 : Math.round(usage * 10) / 10;
    } else if (platform() === 'darwin') {
      const { stdout } = await execAsync("top -l 1 | grep 'CPU usage' | awk '{print $3}' | cut -d'%' -f1");
      const usage = parseFloat(stdout.trim());
      return isNaN(usage) ? 0 : Math.round(usage * 10) / 10;
    }
  } catch {
    try {
      const stat = await readFile('/proc/stat', 'utf8');
      const cpuLine = stat.split('\n')[0];
      const parts = cpuLine.split(/\s+/).slice(1).map(Number);
      const idle = parts[3];
      const total = parts.reduce((a, b) => a + b, 0);
      const usage = ((total - idle) / total) * 100;
      return Math.round(usage * 10) / 10;
    } catch {
      return 0;
    }
  }
  return 0;
}

/**
 * Get disk usage
 */
async function getDiskUsage() {
  try {
    if (platform() === 'linux' || platform() === 'darwin') {
      const { stdout } = await execAsync("df -h / | tail -1 | awk '{print $2,$3,$5}'");
      const [total, used, pct] = stdout.trim().split(/\s+/);
      return {
        totalGB: parseSizeToGB(total),
        usedGB: parseSizeToGB(used),
        usedPct: parseInt(pct.replace('%', ''), 10),
      };
    }
  } catch (err) {
    log.debug('Failed to get disk usage:', err.message);
  }
  return { totalGB: 0, usedGB: 0, usedPct: 0 };
}

function parseSizeToGB(size) {
  const num = parseFloat(size);
  if (size.includes('T')) return num * 1024;
  if (size.includes('G')) return num;
  if (size.includes('M')) return num / 1024;
  if (size.includes('K')) return num / 1024 / 1024;
  return num;
}

async function resolveOpenclawVersion() {
  if (cachedOpenclawVersion) return cachedOpenclawVersion;

  try {
    const { stdout } = await execAsync('openclaw --version');
    const version = stdout.trim();
    cachedOpenclawVersion = version || 'unknown';
    return cachedOpenclawVersion;
  } catch (err) {
    log.debug('Failed to resolve OpenClaw version:', err.message);
    cachedOpenclawVersion = 'unknown';
    return cachedOpenclawVersion;
  }
}

/**
 * Collect OpenClaw status
 */
async function collectOpenclawStatus() {
  try {
    const [agents, openclawVersion] = await Promise.all([
      getOpenclawAgents(),
      resolveOpenclawVersion(),
    ]);
    return {
      openclawVersion,
      pluginVersion: CONFIG.pluginVersion,
      agent_count: agents.length,
      agents: agents.map(a => ({
        id: a.id,
        name: a.name,
        status: a.status,
      })),
    };
  } catch (err) {
    log.debug('Failed to collect OpenClaw status:', err.message);
    return {
      openclawVersion: await resolveOpenclawVersion(),
      pluginVersion: CONFIG.pluginVersion,
      agent_count: 0,
      agents: [],
    };
  }
}

/**
 * Get list of OpenClaw agents from local state
 */
function extractJsonPrefix(text) {
  const trimmed = text.trim();
  if (!trimmed) return null;

  const startsWith = trimmed[0];
  if (startsWith !== '[' && startsWith !== '{') return null;

  let depth = 0;
  let inString = false;
  let escape = false;
  for (let i = 0; i < trimmed.length; i += 1) {
    const ch = trimmed[i];
    if (escape) {
      escape = false;
      continue;
    }
    if (ch === '\\') {
      escape = true;
      continue;
    }
    if (ch === '"') {
      inString = !inString;
      continue;
    }
    if (inString) continue;
    if (ch === '[' || ch === '{') depth += 1;
    if (ch === ']' || ch === '}') depth -= 1;
    if (depth === 0) {
      return trimmed.slice(0, i + 1);
    }
  }
  return null;
}

async function getOpenclawAgents() {
  try {
    try {
      const { stdout } = await execAsync('openclaw agents list --json 2>/dev/null');
      const jsonPrefix = extractJsonPrefix(stdout);
      if (jsonPrefix) {
        const agents = JSON.parse(jsonPrefix);
        if (Array.isArray(agents)) {
          return agents.map((agent) => ({
            id: agent.id,
            name: agent.name || agent.id,
            status: agent.isDefault ? 'default' : 'configured',
          }));
        }
      }
    } catch (err) {
      log.debug('Failed to get agents from `openclaw agents list --json`:', err.message);
    }

    const agentConfigPath = `${CONFIG.openclawPath}/agents.json`;
    try {
      await access(agentConfigPath, constants.R_OK);
      const data = JSON.parse(await readFile(agentConfigPath, 'utf8'));
      if (Array.isArray(data.agents) && data.agents.length > 0) {
        return data.agents;
      }
    } catch {
      // fall through to directory-based discovery
    }

    const agentsDir = `${CONFIG.openclawPath}/agents`;
    await access(agentsDir, constants.R_OK);
    const entries = await readdir(agentsDir, { withFileTypes: true });
    return entries
      .filter((entry) => entry.isDirectory())
      .filter((entry) => entry.name !== 'main')
      .map((entry) => ({
        id: entry.name,
        name: entry.name,
        status: 'configured',
      }));
  } catch {
    return [];
  }
}

/**
 * Build telemetry payload
 */
async function buildPayload() {
  const system = await collectSystemMetrics();
  const openclaw = await collectOpenclawStatus();
  
  return {
    identifier: CONFIG.identifier,
    timestamp: new Date().toISOString(),
    ...system,
    openclaw_version: openclaw.openclawVersion,
    plugin_version: openclaw.pluginVersion,
    agents: openclaw.agents,
  };
}

/**
 * Send telemetry via HTTP
 */
async function sendHttpHeartbeat() {
  try {
    const payload = await buildPayload();
    log.debug('Sending HTTP heartbeat...');
    
    const headers = {
      'Content-Type': 'application/json',
      'X-Server-Identifier': CONFIG.identifier,
    };
    
    if (CONFIG.apiKey) {
      headers['X-API-Key'] = CONFIG.apiKey;
    }
    
    const response = await fetch(`${CONFIG.backendUrl}/monitor/server/heartbeat-v2`, {
      method: 'POST',
      headers,
      body: JSON.stringify(payload),
    });
    
    if (response.ok) {
      log.debug('HTTP heartbeat sent successfully');
      lastSuccessfulSend = Date.now();
      consecutiveFailures = 0;
      return true;
    } else {
      throw new Error(`HTTP ${response.status}: ${await response.text()}`);
    }
  } catch (err) {
    log.error('HTTP heartbeat failed:', err.message);
    consecutiveFailures++;
    return false;
  }
}

/**
 * Main reporting loop
 */
async function reportingLoop() {
  while (!isShuttingDown) {
    try {
      const success = await sendHttpHeartbeat();
      
      let interval = CONFIG.reportIntervalSec * 1000;
      if (!success) {
        const backoff = Math.min(consecutiveFailures * 10000, 300000);
        interval = Math.max(interval, backoff);
        log.info(`Retry in ${interval}ms (backoff)`);
      }
      
      await new Promise(resolve => setTimeout(resolve, interval));
    } catch (err) {
      log.error('Reporting loop error:', err.message);
      await new Promise(resolve => setTimeout(resolve, 30000));
    }
  }
}

/**
 * Graceful shutdown
 */
function shutdown() {
  log.info('Shutting down telemetry server...');
  isShuttingDown = true;
  
  if (wsConnection) {
    wsConnection.close();
  }
  
  sendHttpHeartbeat().finally(() => {
    process.exit(0);
  });
}

// Handle signals
process.on('SIGTERM', shutdown);
process.on('SIGINT', shutdown);

// Start
log.info('HarborForge Monitor Telemetry Server starting...');
log.info('Config:', {
  identifier: CONFIG.identifier,
  backendUrl: CONFIG.backendUrl,
  reportIntervalSec: CONFIG.reportIntervalSec,
  hasApiKey: !!CONFIG.apiKey,
  pluginVersion: CONFIG.pluginVersion,
});

if (!CONFIG.apiKey) {
  log.warn('Missing HF_MONITOR_API_KEY environment variable - API authentication will fail');
}

reportingLoop();