statsd-realtime-stats-backend.js

'use strict';

const WebSocket = require('ws');

const express = require('express');
const http = require('http');

let debug = false;

function arrayShiftRight(array) {
  for (let idx = array.length - 1; idx > 0; idx--) {
    array[idx] = array[idx - 1];
  }
  array[0] = 0;
}

const RealtimeBackend = {
  init(startup_time, statsdConfig, events) {
    const config = statsdConfig.realtime || {};
    const {
      backendPort,
      historyLength,
      allowOrigin,
    } = config;

    if (!backendPort || !historyLength) {
      console.error('Realtime configuration missing backendPort or historyLength');
      return false;
    }


    const app = express();
    const server = http.createServer(app);
    const wss = new WebSocket.Server({
      server,
    });

    server.listen(backendPort, function listening() {
      console.log('Listening on %d', server.address().port);
    });

    let history = {
      lastTimestamp: null,
      counters: {},
      gauges: {},
    };

    app.get('/history', function (req, res) {
      if (allowOrigin) {
        res.header('Access-Control-Allow-Origin', allowOrigin);
      }
      res.send(history)
    })

    wss.on('connection', function connection(ws) {
      ws.send(JSON.stringify(Object.assign({
        event: 'history',
      }, history)));
    });

    events.on('flush', (timestamp, stats) => {
      const counters = Object.assign({}, stats.counters, {
        random: Math.floor(Math.random() * 10),
      });

      const gauges = {};
      for (let key of Object.keys(stats.gauges)) {
        if (stats.gauges[key] !== history.gauges[key]) {
          gauges[key] = stats.gauges[key];
          history.gauges[key] = stats.gauges[key];
        }
      }

      // @WARN: Shift counters right once
      // We don't deal with the problems of time here. Just assume this is
      // called accurately every `flushInterval` [even though it's not true.]
      for (let key of Object.keys(history.counters)) {
        arrayShiftRight(history.counters[key]);
      }

      // Add new value to the history
      for (let key of Object.keys(counters)) {
        if (!history.counters.hasOwnProperty(key)) {
          history.counters[key] = new Array(historyLength).fill(0);
        }
        history.counters[key][0] = counters[key];
      }

      const data = JSON.stringify({
        event: 'update',
        timestamp,
        counters,
        gauges,
      });

      wss.clients.forEach(function each(client) {
        if (client.readyState === WebSocket.OPEN) {
          client.send(data);
        }
      });

      history.lastTimestamp = timestamp;
    });

    return true;
  },
};

module.exports = RealtimeBackend;