Merge pull request #96 from mcollina/mosquitto-stats-align

Mosquitto stats align

.travis.yml

@@ -6,4 +6,6 @@ services:
   - redis-server
   - mongodb
 script:
+  - npm run coverage
+after_success:
   - npm run publish-coverage

lib/stats.js

@@ -26,24 +26,47 @@ OTHER DEALINGS IN THE SOFTWARE.
 "use strict";
 
 var moment = require("moment");
+var movingAverage = require("moving-average");
 var version = "mosca " + require("../package").version;
 
 /**
  * Create a new load object.
  *
  * @api private
  */
-function Load() {
-  this.publishedMessages = 0;
-  this.connectedClients = 0;
+function Load(minutes) {
+  this.maPublishedMessages = movingAverage(minutes * 60 * 1000);
+  this.maPublishedMessages.push(Date.now(), 0);
+  this.maConnectedClients = movingAverage(minutes * 60 * 1000);
+  this.maConnectedClients.push(Date.now(), 0);
 }
 
+Object.defineProperties(Load.prototype, {
+  "publishedMessages": {
+    get: function () {
+      var value = this.maPublishedMessages.movingAverage();
+      value = Math.round(value * 100) / 100;
+      return value;
+    }
+  },
+  "connectedClients": {
+    get: function () {
+      var value = this.maConnectedClients.movingAverage();
+      value = Math.round(value * 100) / 100;
+      return value;
+    }
+  }
+});
+
 /**
  * A Stats object is used to keep track of the state of a mosca.Server
  * and can be wired() there.
  *
  * It provides the following stats:
  *  - connectedClients: the number of connected clients at this point in time
+ *  - publishedMessages: the number of publish messages received since the the start
+ *
+ * It also track the load at 1min, 5min, and 15min of the same events.
  *
  * @api public
  */
@@ -54,15 +77,13 @@ function Stats() {
 
   this.connectedClients = 0;
   this.publishedMessages = 0;
+  this.lastIntervalPublishedMessages = 0;
   this.started = new Date();
 
-  this.current = {
-    m15: new Load(),
-    m1: new Load()
-  };
   this.load = {
-    m15: new Load(),
-    m1: new Load()
+    m15: new Load(15),
+    m5: new Load(5),
+    m1: new Load(1)
   };
 }
 
@@ -75,8 +96,6 @@ function Stats() {
 function clientConnected() {
   /*jshint validthis:true */
   this.stats.connectedClients++;
-  this.stats.current.m1.connectedClients++;
-  this.stats.current.m15.connectedClients++;
 }
 
 /**
@@ -99,8 +118,7 @@ function clientDisconnected() {
 function published() {
   /*jshint validthis:true */
   this.stats.publishedMessages++;
-  this.stats.current.m1.publishedMessages++;
-  this.stats.current.m15.publishedMessages++;
+  this.stats.lastIntervalPublishedMessages++;
 }
 
 /**
@@ -137,27 +155,33 @@ Stats.prototype.wire = function wire(server) {
   var timer = setInterval(function() {
     var stats = server.stats;
     var mem = process.memoryUsage();
-    stats.load.m1 = stats.current.m1;
-    stats.current.m1 = new Load();
 
-    if (++count % 15 === 0) {
-      stats.load.m15 = stats.current.m15;
-      stats.current.m15 = new Load();
-      count = 0;
-    }
+    var date = new Date();
+
+    stats.load.m1.maConnectedClients.push(date, stats.connectedClients);
+    stats.load.m5.maConnectedClients.push(date, stats.connectedClients);
+    stats.load.m15.maConnectedClients.push(date, stats.connectedClients);
+
+    stats.load.m1.maPublishedMessages.push(date, stats.lastIntervalPublishedMessages);
+    stats.load.m5.maPublishedMessages.push(date, stats.lastIntervalPublishedMessages);
+    stats.load.m15.maPublishedMessages.push(date, stats.lastIntervalPublishedMessages);
+    stats.lastIntervalPublishedMessages = 0;
 
     doPublish("version", version);
+    doPublish("started_at", server.stats.started.toISOString());
     doPublish("uptime", mom.from(Date.now(), true));
-    doPublish("connectedClients", stats.connectedClients);
-    doPublish("publishedMessages", stats.publishedMessages);
-    doPublish("load/15m/connectedClients", stats.load.m15.connectedClients);
-    doPublish("load/15m/publishedMessages", stats.load.m15.publishedMessages);
-    doPublish("load/1m/connectedClients", stats.load.m1.connectedClients);
-    doPublish("load/1m/publishedMessages", stats.load.m1.publishedMessages);
+    doPublish("clients/connected", stats.connectedClients);
+    doPublish("publish/received", stats.publishedMessages);
+    doPublish("load/connections/15min", stats.load.m15.connectedClients);
+    doPublish("load/publish/received/15min", stats.load.m15.publishedMessages);
+    doPublish("load/connections/5min", stats.load.m5.connectedClients);
+    doPublish("load/publish/received/5min", stats.load.m5.publishedMessages);
+    doPublish("load/connections/1min", stats.load.m1.connectedClients);
+    doPublish("load/publish/received/1min", stats.load.m1.publishedMessages);
     doPublish("memory/rss", mem.rss);
-    doPublish("memory/heapUsed", mem.heapUsed);
-    doPublish("memory/heapTotal", mem.heapTotal);
-  }, 60 * 1000);
+    doPublish("memory/heap/current", mem.heapUsed);
+    doPublish("memory/heap/maximum", mem.heapTotal);
+  }, 10 * 1000);
 
   events.forEach(function(event) {
     server.on(event.name, event);

package.json

@@ -10,7 +10,7 @@
     "test": "mocha --recursive --bail --reporter spec test",
     "ci": "mocha --recursive --bail --watch test",
     "coverage": "rm -rf coverage; istanbul cover _mocha -- --recursive --reporter spec --bail",
-    "publish-coverage": "npm run coverage && (cat coverage/lcov.info | coveralls)",
+    "publish-coverage": "(cat coverage/lcov.info | coveralls)",
     "jshint-lib": "jshint lib/*.js",
     "jshint-test": "jshint test/*.js",
     "start": "./bin/mosca -v | bunyan"
@@ -84,7 +84,8 @@
     "json-buffer": "~2.0.7",
     "brfs": "0.0.8",
     "pre-commit": "0.0.4",
-    "moment": "~2.5.1"
+    "moment": "~2.5.1",
+    "moving-average": "0.0.4"
   },
   "optionalDependencies": {
     "leveldown": "~0.10.0",

test/abstract_server.js

@@ -1705,7 +1705,7 @@ module.exports = function(moscaSettings, createConnection) {
 
     it("should publish data each minute", function(done) {
       buildAndConnect(done, function(client1) {
-        var topic = "$SYS/" + instance.id + "/connectedClients";
+        var topic = "$SYS/" + instance.id + "/clients/connected";
         instance.ascoltatore.subscribe(topic, function(topic, value) {
           expect(value).to.eql("1");
           client1.disconnect();