Add support for sets, counting unique events

Sets are backed using a set data-structure, discarding duplicate
values being inserted. This allows backend to retrieve the number
of unique events that happened since the last flush.

Sets are all emptied after each flush.

README.md

@@ -61,6 +61,13 @@ StatsD now also supports gauges, arbitrary values, which can be recorded.
 
     gaugor:333|g
 
+Sets
+----
+StatsD supports counting unique occurences of events between flushes,
+using a Set to store all occuring events.
+
+    uniques:765|s
+
 All metrics can also be batch send in a single UDP packet, separated by a
 newline character.
 

lib/set.js

@@ -0,0 +1,30 @@
+var Set = function() {
+  this.store = {};
+}
+
+Set.prototype = {
+  has: function(value) {
+    if (value) {
+      return this.store.hasOwnProperty(value);
+    } else {
+      return false;
+    }
+  },
+  insert: function(value) {
+    if (value) {
+      this.store[value] = value;
+    }
+  },
+  clear: function() {
+    this.store = {};
+  },
+  values: function() {
+    var values = [];
+    for (value in this.store) {
+      values.push(value);
+    }
+    return values;
+  }
+}
+
+exports.Set = Set;

stats.js

@@ -5,6 +5,7 @@ var dgram  = require('dgram')
   , fs     = require('fs')
   , events = require('events')
   , logger = require('./lib/logger')
+  , set = require('./lib/set')
 
 // initialize data structures with defaults for statsd stats
 var keyCounter = {};
@@ -16,6 +17,8 @@ var timers = {
   "statsd.packet_process_time": []
 };
 var gauges = {};
+var sets = {
+};
 var pctThreshold = null;
 var debugInt, flushInterval, keyFlushInt, server, mgmtServer;
 var startup_time = Math.round(new Date().getTime() / 1000);
@@ -44,6 +47,7 @@ function flushMetrics() {
     counters: counters,
     gauges: gauges,
     timers: timers,
+    sets: sets,
     pctThreshold: pctThreshold
   }
 
@@ -58,6 +62,11 @@ function flushMetrics() {
     for (key in metrics.timers) {
       metrics.timers[key] = [];
     }
+
+    // Clear the sets
+    for (key in metrics.sets) {
+      metrics.sets[key] = new set.Set();
+    }
   });
 
   // Flush metrics to each backend.
@@ -139,6 +148,11 @@ config.configFile(process.argv[2], function (config, oldConfig) {
             timers[key].push(Number(fields[0] || 0));
           } else if (fields[1].trim() == "g") {
             gauges[key] = Number(fields[0] || 0);
+          } else if (fields[1].trim() == "s") {
+            if (! sets[key]) {
+              sets[key] = new set.Set();
+            }
+            sets[key].insert(fields[0] || '0');
           } else {
             if (fields[2] && fields[2].match(/^@([\d\.]+)/)) {
               sampleRate = Number(fields[2].match(/^@([\d\.]+)/)[1]);

test/set_tests.js

@@ -0,0 +1,41 @@
+var set = require('../lib/set')
+
+module.exports = {
+  has_returns_expected_values: function(test) {
+    test.expect(2);
+    var s = new set.Set();
+    s.insert('a');
+    test.ok(s.has('a'));
+    test.ok(!s.has('b'));
+    test.done();
+  },
+  clear_empties_the_set: function(test) {
+    test.expect(3);
+    var s = new set.Set();
+    s.insert('a');
+    test.equal(1, s.values().length);
+    s.clear();
+    test.equal(0, s.values().length);
+    test.equal([], s.values().length);
+    test.done();
+  },
+  values_returns_values: function(test) {
+    test.expect(3);
+    var s = new set.Set();
+    s.insert('a');
+    s.insert('b');
+    test.equal(2, s.values().length);
+    test.ok(s.values().indexOf('a') != -1);
+    test.ok(s.values().indexOf('b') != -1);
+    test.done();
+  },
+  values_are_unique: function(test) {
+    test.expect(1);
+    var s = new set.Set();
+    s.insert('a');
+    s.insert('a');
+    s.insert('b');
+    test.equal(2, s.values().length);
+    test.done();
+  }
+}