agent_monitoring.js

'use strict';

const { opendir, mkdir, readdir, rmdir, rm, readFile, writeFile, stat } = require('node:fs/promises');
const { sep: PATH_SEP, normalize: normalizePath } = require('node:path');
const { hostname } = require('os');

const {
  nop,
  createReadyPromise,
  aquireCacheWriteLock,
  aquireCacheReadLock,
  CounterMap,
  convertObjectToPrometheusLabels,
  startHeartbeat,
} = require('./utils')({
  CACHE_HEARTBEAT_INTERVAL: 30_000,
  CACHE_HEARTBEAT_OFFSET: 500,
});

const kDataSymbol = Symbol('AgentMonitoringData');
const kCacheSymbol = Symbol('AgentMonitoringCache');

const CONNECT_TYPE_MESHAGENT = 1;

let MeshServer_NotifyUserOfDeviceStateChange;

let meshserver;
let webserver;
let monitoring;
let cacheDir;
let heartbeatFilePath;
let prometheusJobName;

const [ready, resolveReady] = createReadyPromise();
const eventTarget = new EventTarget();
const stableAgents = new Map();
const startupTime = Date.now();

async function collectMetricFromCache(write, name, type, help, dirname, caches, clearCache) {
  let counters, countersZero;
  if (name === 'up') {
    counters = new CounterMap();
    countersZero = new CounterMap();
  }
  write(
    '# TYPE ' + name + ' ' + type + '\n' +
    (help ? '# HELP ' + name + ' ' + help + '\n' : '')
  );
  for (const [encodedMeshName, _agentName, agentDir, labels] of caches) {
    const metricDir = agentDir + dirname + PATH_SEP;
    const meshName = decodeURIComponent(encodedMeshName);
    let cacheContent;
    if (!(cacheContent = await readdir(metricDir).catch(nop))) {
      continue;
    }
    cacheContent = cacheContent
      .map((fileName) => ([fileName, fileName.split('#')]))
      .map((data) => (data[1][0] = +data[1][0], data))
      .sort((a, b) => (a[1][0] - b[1][0]));
    for (const [_, [timestamp, value]] of cacheContent) {
      if (name === 'up' && timestamp != startupTime) {
        counters.inc(meshName);
        if (+value === 0) {
          countersZero.inc(meshName);
        }
      }
      const line = name + labels + ' ' + value + ' ' + timestamp + '\n';
      write(line);
    }
    if (clearCache) {
      await Promise.all(cacheContent.map(([name]) => (rm(metricDir + name))));
      await rmdir(metricDir);
    }
  }
  return [counters, countersZero];
}

async function collectMetrics(req, res) {
  const now = Date.now();
  const clearCache = req.headers['user-agent'].startsWith('Prometheus/') && req.query['clear-cache'] === 'yes';
  const lastScrapeTime = parseInt((await stat(cacheDir + 'last_scrape').catch(() => ({}))).mtimeMs);
  const timeElapsedSinceLastScrape = now - lastScrapeTime;
  const releaseCacheReadLock = await aquireCacheReadLock(kCacheSymbol);
  const encodedMeshNames = (await readdir(cacheDir, { withFileTypes: true }))
    .filter((dirent) => (dirent.isDirectory()))
    .map((dirent) => (dirent.name));
  const caches = [];
  const quietMeshes = [];
  for (const encodedMeshName of encodedMeshNames) {
    const meshDir = cacheDir + encodedMeshName + PATH_SEP;
    const dirents = (await readdir(meshDir, { withFileTypes: true }))
      .filter((dirent) => (dirent.isDirectory()));
    if (dirents.length === 0) {
      quietMeshes.push([decodeURIComponent(encodedMeshName), encodedMeshName]);
      continue
    }
    for (const dirent of dirents) {
      caches.push([encodedMeshName, dirent.name])
    }
  }
  for (const cache of caches) {
    const caheDirWithPathSep = cacheDir + cache[0] + PATH_SEP + cache[1] + PATH_SEP;
    const labels = await readFile(caheDirWithPathSep + 'labels', { encoding: 'utf-8' });
    cache.push(caheDirWithPathSep);
    cache.push(labels);
  }
  const write = (text) => (/*console.log(text),*/ res.write(text));
  const [stateChangeCounters, stateChangeDownCounters] = await collectMetricFromCache(write, 'up', 'untyped', false, 'up', caches, clearCache);
  if (stateChangeCounters.size) {
    write('###\n');
  }
  for (const [_, { [kDataSymbol]: { labels } }] of stableAgents) {
    write('up' + labels + ' 1 ' + now + '\n');
  }
  write('\n');
  await collectMetricFromCache(write, 'meshcentral_agent_warmup_seconds', 'gauge', 'Time until agent became stable', 'warmup', caches, clearCache);
  write('\n');
  await collectMetricFromCache(write, 'meshcentral_agent_stable_seconds', 'gauge', 'Time agent was up and stable', 'stable', caches, clearCache);
  write('\n');
  write([
    '# TYPE meshcentral_agent_state_changes_per_scrape gauge',
    '# HELP meshcentral_agent_state_changes_per_scrape MeshAgent state changes per scrape interval',
    ''
  ].join('\n'));
  const tray = [
    '# TYPE meshcentral_agent_state_changes_per_second gauge',
    '# HELP meshcentral_agent_state_changes_per_second MeshAgent state changes per second',
  ];
  function writeStateChangeMetrics(meshName, upCounter, downCounter) {
    write([
      'meshcentral_agent_state_changes_per_scrape{to_state="down",mesh_name="' + meshName + '",job="' + prometheusJobName + '"} ' + downCounter + ' ' + lastScrapeTime,
      'meshcentral_agent_state_changes_per_scrape{to_state="up",mesh_name="' + meshName + '",job="' + prometheusJobName + '"} ' + upCounter + ' ' + lastScrapeTime,
      '',
    ].join('\n'));
    tray.push(
      'meshcentral_agent_state_changes_per_second{to_state="down",mesh_name="' + meshName + '",job="' + prometheusJobName + '"} ' + (downCounter * 1000 / timeElapsedSinceLastScrape) + ' ' + lastScrapeTime,
      'meshcentral_agent_state_changes_per_second{to_state="up",mesh_name="' + meshName + '",job="' + prometheusJobName + '"} ' + (upCounter * 1000 / timeElapsedSinceLastScrape) + ' ' + lastScrapeTime,
    );
  }
  for (const [meshName, downCounter] of stateChangeDownCounters) {
    const upCounter = (stateChangeCounters.get(meshName) - downCounter);
    writeStateChangeMetrics(meshName, upCounter, downCounter);
    stateChangeCounters.delete(meshName)
  }
  for (const [meshName, upCounter] of stateChangeCounters) {
    writeStateChangeMetrics(meshName, upCounter, 0);
  }
  for (const [meshName, encodedMeshName] of quietMeshes) {
    writeStateChangeMetrics(meshName, 0, 0);
    if (clearCache) {
      await rmdir(cacheDir + encodedMeshName).catch(nop);
    }
  }
  write('\n');
  if (tray.length > 2) {
    tray.push('');
  }
  write(tray.join('\n'));
  write('\n');
  if (clearCache) {
    for (const [_1, _2, agentDir] of caches) {
      await rm(agentDir + 'labels');
      await rmdir(agentDir);
    }
    await writeFile(cacheDir + 'last_scrape', '');
  }
  releaseCacheReadLock();
}

function _hook_deviceStateChange(meshid, nodeid, connectTime, connectType, powerState, serverid, stateSet, extraInfo) {
  if ((connectType & CONNECT_TYPE_MESHAGENT) !== 0) {
    if (stateSet) {
      webserver.wsagents[nodeid][kDataSymbol] = { connectTime };
    } else {
      const meshagent = stableAgents.get(nodeid);
      if (meshagent) {
        meshagent[kDataSymbol].disconnectTime = connectTime;
        hook_agentDisconnect(meshagent);
      }
    }
  }
  MeshServer_NotifyUserOfDeviceStateChange.call(meshid, nodeid, connectTime, connectType, powerState, serverid, stateSet, extraInfo);
}

function hook_agentCoreIsStable(meshagent) {
  const data = meshagent[kDataSymbol];
  const now = data.stableTime = Date.now();
  data.warmupDuration = now - data.connectTime;
  const nodeKey = meshagent.dbNodeKey;
  const encodedNodeKey = encodeURIComponent(nodeKey);
  const meshKey = meshagent.dbMeshKey;
  const meshName = webserver.meshes[meshKey].name;
  const encodedMeshName = encodeURIComponent(meshName);
  const agentName = meshagent.agentName || meshagent.name || meshagent.agentInfo?.computerName || '';
  const agentDir = cacheDir + encodedMeshName + PATH_SEP+ encodedNodeKey + PATH_SEP;
  const upDir = agentDir + 'up' + PATH_SEP;
  const warmupDir = agentDir + 'warmup' + PATH_SEP;
  const labels = data.labels = convertObjectToPrometheusLabels({
    instance: 'meshagent' + meshKey.slice(meshKey.lastIndexOf('/')) + nodeKey.slice(nodeKey.lastIndexOf('/')),
    agent_name: agentName || '',
    mesh_name: meshName || '',
    job: prometheusJobName,
  });
  stableAgents.set(nodeKey, meshagent);
  (async function () {
    const releaseCacheWriteLock = await aquireCacheWriteLock(kCacheSymbol);
    await mkdir(agentDir, { recursive: true });
    await Promise.all([
      // Don't fail if exists.
      mkdir(upDir).catch(nop),
      mkdir(warmupDir).catch(nop),
    ]);
    await Promise.all([
      writeFile(agentDir + 'labels', labels),
      // Make sure the agent was down at startup. If the agent came up before
      // the last scrape there is no "down" event on unclean shutdown.
      writeFile(upDir + startupTime + '#0', ''),
      writeFile(upDir + now + '#1', ''),
      writeFile(warmupDir + now + '#' + (data.warmupDuration / 1000), ''),
    ]);
    releaseCacheWriteLock();
    eventTarget.dispatchEvent(new CustomEvent('agenCoreIsStable', { detail: { meshagent } }));
  })();
}

function hook_agentDisconnect(meshagent) {
  const data = meshagent[kDataSymbol];
  const now = data.disconnectTime;
  data.stableDuration = now - data.stableTime;
  const nodeKey = meshagent.dbNodeKey;
  const encodedNodeKey = encodeURIComponent(nodeKey);
  const meshKey = meshagent.dbMeshKey;
  const meshName = webserver.meshes[meshKey].name;
  const encodedMeshName = encodeURIComponent(meshName);
  const agentDir = cacheDir + encodedMeshName + PATH_SEP + encodedNodeKey + PATH_SEP;
  const upDir = agentDir + 'up' + PATH_SEP;
  const stableDir = agentDir + 'stable' + PATH_SEP;
  const labels = data.labels;
  stableAgents.delete(nodeKey);
  (async function () {
    const releaseCacheWriteLock = await aquireCacheWriteLock(kCacheSymbol);
    await mkdir(agentDir, { recursive: true });
    await Promise.all([
      // Don't fail if exists.
      mkdir(upDir).catch(nop),
      mkdir(stableDir).catch(nop),
    ]);
    await Promise.all([
      writeFile(agentDir + 'labels', labels),
      writeFile(upDir + now + '#0', ''),
      writeFile(stableDir + now + '#' + (data.stableDuration / 1000), ''),
    ]);
    releaseCacheWriteLock();
    eventTarget.dispatchEvent(new CustomEvent('disconnect', { detail: { meshagent } }));
  })();
}

async function startupCache() {
  await mkdir(cacheDir, { recursive: true });
  heartbeatFilePath = cacheDir + 'heartbeat';
  const lastHeartbeat = parseInt((await stat(heartbeatFilePath).catch(() => ({}))).mtimeMs);
  // create missing "down" events
  for await (const dirent of await opendir(cacheDir)) {
    if (!dirent.isDirectory()) {
      continue;
    }
    const upDir = cacheDir + dirent.name + PATH_SEP + 'up';
    const upDirContent = await readdir(upDir).catch(nop);
    if (!upDirContent) {
      continue;
    }
    const [lastTimestamp, lastValue] = (upDirContent)
      .map((name) => (name.split('#').map((val) => (+val))))
      .sort((a, b) => (a[0] - b[0]))
      .pop();
    if (lastValue === 1) {
      const stableDir = cacheDir + dirent.name + PATH_SEP + 'stable';
      if (lastHeartbeat > lastTimestamp) {
        // last "up" event was before last heartbeat
        await Promise.all([
          writeFile(upDir + PATH_SEP + lastHeartbeat + '#0', ''),
          writeFile(stableDir + PATH_SEP + lastHeartbeat + '#' + (lastHeartbeat - lastTimestamp), ''),
        ]);
      } else {
        // last "up" event was after last heartbeat
        const lastDown = lastTimestamp + 1000;
        await Promise.all([
          writeFile(upDir + PATH_SEP + lastDown + '#0', ''),
          writeFile(stableDir + PATH_SEP + lastDown + '#1', ''),
        ]);
      }
    }
  }

  await Promise.all([
    rm(heartbeatFilePath).catch(nop),
    writeFile(cacheDir + 'last_scrape', ''),
  ]);

  startHeartbeat(heartbeatFilePath);
}

function startup() {
  let config = Object.assign({
    cachedirpath: '.cache',
    prometheusjobname: 'meshcentral@' + hostname(),
  }, meshserver.config.settings.plugins.pluginsettings?.agent_monitoring);
  webserver = meshserver.webserver;
  monitoring = meshserver.monitoring;
  MeshServer_NotifyUserOfDeviceStateChange = meshserver.NotifyUserOfDeviceStateChange;
  meshserver.NotifyUserOfDeviceStateChange = _hook_deviceStateChange;
  prometheusJobName = config.prometheusjobname;
  if (process.platform === 'win32') {
    throw new Error('SUPPORT FOR WIN32 IS NOT TESTED YET!');
    const tmp = config.cachedirpath;
    // either UNC or absolute on current drive or absolute with drive letter
    if (tmp[0] === '\\' || tmp.slice(1, 3) === ':\\' || tmp.slice(1, 3) === ':/') {
      cacheDir = '';
    } else {
      cacheDir = __dirname + PATH_SEP;
    }
  } else {
    cacheDir = config.cachedirpath[0] === '/' ? '' : __dirname + PATH_SEP;
  }
  cacheDir = normalizePath( cacheDir + config.cachedirpath );
  if (!cacheDir.endsWith(PATH_SEP)) {
    cacheDir += PATH_SEP;
  }
  (async function () {
    await startupCache();
    monitoring.collectors.push(collectMetrics);
    resolveReady();
  })();
}

module.exports = {
  agent_monitoring: function (pluginHandler) {
    meshserver = pluginHandler.parent;
    return {
      hook_agentCoreIsStable: hook_agentCoreIsStable,
      addEventListener: eventTarget.addEventListener.bind(eventTarget),
      server_startup: startup,
    };
  },
  ready,
  kDataSymbol,
  nop,
  createReadyPromise,
  aquireCacheReadLock,
  aquireCacheWriteLock,
  CounterMap,
};