metrics: add stream api for system info

dashboard/src/components/CpuUsage.vue

@@ -38,33 +38,25 @@ const period = ref(6);
 const busy = ref(true);
 
 let gGraph = null;
-
-let gLiveDataPoints = {};
+let gMetricStream = null;
 
 async function liveRefresh() {
-  // stop and clear
-  if (period.value !== 0) return gLiveDataPoints = {};
+  gMetricStream = await systemModel.getMetricStream();
+  gMetricStream.onerror = (error) => console.log('event stream error:', error);
+  gMetricStream.onmessage = (message) => {
+    const data = JSON.parse(message.data);
+    if (data.cpu[0]) return; // value can be null if no previous value
+    gGraph.data.labels.push(moment(data.cpu[1]*1000).format('hh:mm'));
+    gGraph.data.datasets[0].data.push(data.cpu[0]);
 
-  const [error, result] = await systemModel.getMetrics({ fromSecs: 60, intervalSecs: 300 });
-  if (error) return console.error(error);
+    // Limit the number of data points to keep the chart readable
+    if (gGraph.data.datasets[0].data.length > 40) {
+      gGraph.data.labels.shift();
+      gGraph.data.datasets[0].data.shift();
+    }
 
-  for (const v of result.cpu) {
-    if (gLiveDataPoints[v[1]]) continue;
-
-    gLiveDataPoints[v[1]] = v[0];
-    gGraph.data.labels.push(moment(v[1]*1000).format('hh:mm'));
-    gGraph.data.datasets[0].data.push(v[0]);
-  }
-
-  // Limit the number of data points to keep the chart readable
-  if (gGraph.data.datasets[0].data.length > 40) {
-    gGraph.data.labels.shift();
-    gGraph.data.datasets[0].data.shift();
-  }
-
-  gGraph.update();
-
-  setTimeout(liveRefresh, 2000);
+    gGraph.update();
+  };
 }
 
 async function refresh() {
@@ -79,6 +71,10 @@ async function refresh() {
 
   const data = result.cpu.map(v => v[0]); // already scaled to cpu*100
 
+  if (gMetricStream) {
+    gMetricStream.close();
+    gMetricStream = null;
+  }
   if (gGraph) gGraph.destroy();
   gGraph = new Chart(graph.value, {
     type: 'line',

dashboard/src/models/SystemModel.js

@@ -94,6 +94,9 @@ function create() {
       if (error || result.status !== 200) return [error || result];
       return [null, result.body];
     },
+    async getMetricStream() {
+      return new EventSource(`${API_ORIGIN}/api/v1/system/metricstream?access_token=${accessToken}`);
+    }
   };
 }
 

src/apps.js

@@ -150,8 +150,7 @@ exports = module.exports = {
     _clear: clear
 };
 
-const appstore = require('./appstore.js'),
-    appTaskManager = require('./apptaskmanager.js'),
+const appTaskManager = require('./apptaskmanager.js'),
     archives = require('./archives.js'),
     assert = require('assert'),
     backups = require('./backups.js'),

src/metrics.js

@@ -2,6 +2,8 @@
 
 exports = module.exports = {
     getSystem,
+    getSystemStream,
+
     getContainers,
 
     sendToGraphite
@@ -16,6 +18,7 @@ const apps = require('./apps.js'),
     execSync = require('child_process').execSync,
     net = require('net'),
     os = require('os'),
+    { Readable } = require('stream'),
     safe = require('safetydance'),
     services = require('./services.js'),
     superagent = require('./superagent.js');
@@ -226,7 +229,7 @@ async function getContainers(name, options) {
     };
 }
 
-async function getSystemStats(options) {
+async function readSystemFromGraphite(options) {
     assert.strictEqual(typeof options, 'object');
 
     const { fromSecs, intervalSecs, noNullPoints } = options;
@@ -269,7 +272,7 @@ async function getSystemStats(options) {
 async function getSystem(options) {
     assert.strictEqual(typeof options, 'object');
 
-    const systemStats = await getSystemStats(options);
+    const systemStats = await readSystemFromGraphite(options);
 
     const appStats = {};
     for (const app of await apps.list()) {
@@ -289,3 +292,32 @@ async function getSystem(options) {
         cpuCount: os.cpus().length
     };
 }
+
+async function getSystemStream() {
+    const INTERVAL_SECS = 2;
+    let intervalId = null, oldCpuMetrics = null;
+
+    const metricsStream = new Readable({
+        read(/*size*/) { /* ignored, we push via interval */ },
+        destroy(error, callback) {
+            clearInterval(intervalId);
+            callback(error);
+        }
+    });
+
+    intervalId = setInterval(async () => {
+        const memoryMetrics = await getMemoryMetrics();
+        const cpuMetrics = await getCpuMetrics();
+
+        const cpuPercent = oldCpuMetrics ? (cpuMetrics.userMsecs + cpuMetrics.sysMsecs - oldCpuMetrics.userMsecs - oldCpuMetrics.sysMsecs) * 0.1 / INTERVAL_SECS : null;
+        oldCpuMetrics = cpuMetrics;
+
+        const now = Date.now();
+        metricsStream.push(JSON.stringify({
+            cpu: [ cpuPercent, now ],
+            memory: [ memoryMetrics.used, now ]
+        }));
+    }, INTERVAL_SECS*1000);
+
+    return metricsStream;
+}

src/routes/system.js

@@ -9,6 +9,7 @@ exports = module.exports = {
     getLogs,
     getLogStream,
     getMetrics,
+    getMetricStream,
     getBlockDevices,
     getCpus,
 };
@@ -130,6 +131,30 @@ async function getMetrics(req, res, next) {
     next(new HttpSuccess(200, result));
 }
 
+async function getMetricStream(req, res, next) {
+    if (req.headers.accept !== 'text/event-stream') return next(new HttpError(400, 'This API call requires EventStream'));
+
+    const [error, metricStream] = await safe(metrics.getSystemStream());
+    if (error) return next(BoxError.toHttpError(error));
+
+    res.writeHead(200, {
+        'Content-Type': 'text/event-stream',
+        'Cache-Control': 'no-cache',
+        'Connection': 'keep-alive',
+        'X-Accel-Buffering': 'no', // disable nginx buffering
+        'Access-Control-Allow-Origin': '*'
+    });
+    res.write('retry: 3000\n');
+    res.on('close', () => metricStream.destroy());
+    metricStream.on('data', function (data) {
+        const obj = JSON.parse(data);
+        const sse = `data: ${JSON.stringify(obj)}\n\n`;
+        res.write(sse);
+    });
+    metricStream.on('end', res.end.bind(res));
+    metricStream.on('error', res.end.bind(res, null));
+}
+
 async function getBlockDevices(req, res, next) {
     const [error, devices] = await safe(system.getBlockDevices());
     if (error) return next(new HttpError(500, error));

src/server.js

@@ -114,6 +114,7 @@ async function initializeExpressSync() {
     router.get ('/api/v1/system/info',                           token, authorizeAdmin, routes.system.getInfo); // vendor, product name etc
     router.post('/api/v1/system/reboot',                   json, token, authorizeAdmin, routes.system.reboot);
     router.get ('/api/v1/system/metrics',                        token, authorizeAdmin, routes.system.getMetrics);
+    router.get ('/api/v1/system/metricstream',                   token, authorizeAdmin, routes.system.getMetricStream);
     router.get ('/api/v1/system/disk_usage',                     token, authorizeAdmin, routes.system.getDiskUsage);
     router.post('/api/v1/system/disk_usage',                     token, authorizeAdmin, routes.system.updateDiskUsage);
     router.get ('/api/v1/system/block_devices',                  token, authorizeAdmin, routes.system.getBlockDevices);