Added instrumentation for admin dashboard interface

client/src/index.js

@@ -74,6 +74,7 @@ function Horizon({
   horizon._horizonPath = path
   horizon.authEndpoint = authEndpoint
   horizon.hasAuthToken = tokenStorage.hasAuthToken.bind(tokenStorage)
+  horizon.send = sendRequest;
 
   return horizon
 

server/package.json

@@ -21,13 +21,14 @@
     "node": ">=4.0.0"
   },
   "dependencies": {
-    "joi": "^8.0.4",
     "@horizon/client": "",
     "cookie": "^0.2.3",
+    "joi": "^8.0.4",
     "jsonwebtoken": "^5.5.4",
     "oauth": "^0.9.14",
     "pem": "^1.8.1",
     "rethinkdb": "^2.1.1",
+    "uuid": "^2.0.1",
     "winston": "^2.1.0",
     "ws": "^1.0.1"
   },

server/src/admin.js

@@ -0,0 +1,74 @@
+'use strict';
+
+const r = require('rethinkdb');
+
+const admin_db = r.db('horizon_internal');
+const client_table = admin_db.table('stats_clients');
+const request_table = admin_db.table('stats_requests');
+
+class Admin {
+  constructor(server) {
+    this.server = server;
+  }
+
+  clear_tables() {
+    this._run_query(
+      r.expr(["stats_clients", "stats_requests"])
+       .forEach(admin_db.table(r.row).delete()));
+  }
+
+  add_request(req, client) {
+    if (req.raw.type.startsWith("admin:")) return;
+
+    this._run_query(
+      request_table.insert({
+        id: [client.id, req.id],
+        request: req.id, raw: req.raw,
+        time: r.now(), cursors: 0,
+        client: {id: client.id, ip: client.get_address()}
+      }));
+  }
+
+  add_cursor(client, req) {
+    if (req.raw.type.startsWith("admin:")) return;
+
+    this._run_query(
+      request_table
+       .get([client.id, req.id])
+       .update({cursors: r.row('cursors').add(1)}));
+  }
+
+  remove_cursor(client, req) {
+    if (req.raw.type.startsWith("admin:")) return;
+
+    this._run_query(
+      request_table
+       .get([client.id, req.id])
+       .update({cursors: r.row('cursors').sub(1)}));
+  }
+
+  add_client(client) {
+    this._run_query(
+      client_table.insert({
+         connected: true, id: client.id,
+         time: r.now(), ip: client.get_address()}));
+  }
+
+  remove_client(client) {
+    this._run_query(
+      request_table
+       .between([client.id, r.minval], [client.id, r.maxval])
+       .update({cursors: 0}));
+
+    this._run_query(
+      client_table
+       .get(client.id)
+       .update({connected: false, disconnected: r.now()}));
+  }
+
+  _run_query(query) {
+    query.run(this.server._reql_conn.connection());
+  }
+}
+
+module.exports = { Admin };

server/src/client.js

@@ -7,6 +7,7 @@ const schemas = require('./schema/horizon_protocol');
 const Joi = require('joi');
 const r = require('rethinkdb');
 const websocket = require('ws');
+const uuid = require("uuid");
 
 class Request {
   constructor(client, raw_request) {
@@ -32,10 +33,12 @@ class Request {
           cursor.constructor.name === 'OrderByLimitFeed',
           'Endpoint provided a non-cursor as a cursor.');
     this.client.cursors.set(this.id, cursor);
+    this.client.parent._admin.add_cursor(this.client, this);
   }
 
   remove_cursor() {
     this.client.cursors.delete(this.id);
+    this.client.parent._admin.remove_cursor(this.client, this);
   }
 
   _run_reql() {
@@ -106,6 +109,11 @@ class Client {
     this.socket = socket;
     this.parent = parent_server;
     this.cursors = new Map();
+    this.id = uuid.v4();
+
+    // Socket is already open when client is instantiated
+    // So we should probably just invoke this here?
+    this.handle_open();
 
     this.socket.on('open', () => this.handle_open());
     this.socket.on('close', (code, msg) => this.handle_close(code, msg));
@@ -114,14 +122,20 @@ class Client {
       this.error_wrap_socket(() => this.handle_handshake(data)));
   }
 
+  get_address() {
+    return this.socket.upgradeReq.connection.remoteAddress;
+  }
+
   handle_open() {
     logger.debug('Client connection established.');
     this.parent._clients.add(this); // TODO: this is a race condition - the client could miss a reql_connection_lost call
+    this.parent._admin.add_client(this);
   }
 
   handle_close() {
     logger.debug('Client connection terminated.');
     this.parent._clients.delete(this);
+    this.parent._admin.remove_client(this);
     this.cursors.forEach((cursor) => cursor.close());
   }
 
@@ -230,7 +244,8 @@ class Client {
       }
 
       // Kick off the request - it will handle errors and send the response
-      new Request(this, request);
+      let req = new Request(this, request);
+      this.parent._admin.add_request(req, this);
     }
   }
 
@@ -239,6 +254,8 @@ class Client {
     if (this.cursors.delete(raw_request.request_id)) {
       cursor.close();
     }
+
+    this.parent._admin.remove_cursor(this, raw_request);
   }
 
   send_response(request_id, data) {

server/src/endpoint/admin.js

@@ -0,0 +1,33 @@
+'use strict';
+
+const r = require('rethinkdb');
+
+const admin_db = r.db('horizon_internal');
+const client_table = admin_db.table('stats_clients');
+const request_table = admin_db.table('stats_requests');
+
+const change_settings = {
+  include_initial: true,
+  include_states: true,
+  include_types: true
+};
+
+module.exports = {
+  clients(req, metadata) {
+    return client_table
+            .getAll(true, {index: 'connected'})
+            .changes(change_settings);
+  },
+
+  requests(req, metadata) {
+    return (req.options.live ?
+           request_table.getAll(true, {index: "cursors"}) :
+           request_table.orderBy({index: r.desc("time")}).limit(100))
+           .changes(change_settings);
+  },
+
+  collections(req, metadata) {
+    return admin_db.table("collections")
+                   .changes(change_settings);
+  }
+}

server/src/metadata.js

@@ -148,6 +148,11 @@ class Metadata {
     this._auto_create_index = auto_create_index;
     this._ready = false;
 
+    let createIndex = (db, table, index, expr) =>
+      r.branch(
+        r.db(db).table(table).indexList().contains(index), r.expr(true),
+        r.db(db).table(table).indexCreate(index, expr));
+
     let query =
       r.db('horizon_internal')
        .table('collections')
@@ -159,9 +164,15 @@ class Metadata {
       query = r.expr([ 'horizon', 'horizon_internal' ])
        .forEach((db) => r.branch(r.dbList().contains(db), [], r.dbCreate(db)))
        .do(() =>
-         r.expr([ 'collections', 'users_auth', 'users' ])
+         r.expr([ 'collections', 'users_auth', 'users', 'stats_clients', 'stats_requests' ])
           .forEach((table) => r.branch(r.db('horizon_internal').tableList().contains(table),
                                        [], r.db('horizon_internal').tableCreate(table)))
+          .do(() => r.expr([
+            createIndex('horizon_internal', 'stats_clients', 'connected'),
+            createIndex('horizon_internal', 'stats_requests', 'time'),
+            createIndex('horizon_internal', 'stats_requests', 'cursors',
+                        row => row('cursors').gt(0))
+          ]))
           .do(() => query));
     }
 

server/src/server.js

@@ -3,6 +3,7 @@
 const Auth = require('./auth').Auth;
 const check = require('./error').check;
 const Client = require('./client').Client;
+const Admin = require('./admin').Admin;
 const ReqlConnection = require('./reql_connection').ReqlConnection;
 const logger = require('./logger');
 const horizon_protocol = require('./schema/horizon_protocol');
@@ -12,6 +13,8 @@ const options_schema = require('./schema/server_options').server;
 // library. Minified, Rx included etc.
 const horizon_client_path = require.resolve('@horizon/client/dist/horizon');
 
+const admin_endpoints = require("./endpoint/admin.js");
+
 const endpoints = {
   insert: require('./endpoint/insert'),
   query: require('./endpoint/query'),
@@ -76,11 +79,20 @@ class Server {
                                          opts.auto_create_table,
                                          opts.auto_create_index,
                                          this._clients);
+
     this._auth = new Auth(this, opts.auth);
+    this._admin = new Admin(this);
+    this._reql_conn.ready().then(() => this._admin.clear_tables());
+
     for (let key of Object.keys(endpoints)) {
       this.add_request_handler(key, endpoints[key].make_reql, endpoints[key].handle_response);
     }
 
+    for (let key of Object.keys(admin_endpoints))
+      this.add_request_handler(`admin:${key}`,
+                               admin_endpoints[key],
+                               endpoints.subscribe.handle_response);
+
     const verify_client = (info, cb) => {
       // Reject connections if we aren't synced with the database
       if (!this._reql_conn.is_ready()) {