Repository: brightcove/hot-shots
Pull request #93 (2019-01-01 19:33:39)

Fix up how close calls work
This fixes long-requested https://github.com/brightcove/hot-shots/issues/41 and adds to tests for the different permutations of this case.

Also as I dug into this, I realized that the client.socket.on('error') pattern mentioned in the docs isn't the best choice, and it's better to suggest using the errorHandler so it can catch more cases.  I've updated the docs for this while not breaking backwards compatibility.




Patch:
diff --git a/.eslintrc.js b/.eslintrc.js
index 00a7f10..91d2351 100644
--- a/.eslintrc.js
+++ b/.eslintrc.js
@@ -88,7 +88,7 @@ module.exports = {
             "error",
             "unix"
         ],
-        "lines-around-comment": "error",
+        "lines-around-comment": "off",
         "lines-around-directive": "error",
         "lines-between-class-members": [
             "error",
@@ -104,7 +104,7 @@ module.exports = {
         "max-statements": "off",
         "max-statements-per-line": "off",
         "multiline-comment-style": [
-            "error",
+            "off",
             "separate-lines"
         ],
         "multiline-ternary": "off",
@@ -204,7 +204,7 @@ module.exports = {
         "no-useless-return": "error",
         "no-var": "error",
         "no-void": "error",
-        "no-warning-comments": "error",
+        "no-warning-comments": "off",
         "no-whitespace-before-property": "error",
         "no-with": "error",
         "nonblock-statement-body-position": "error",
diff --git a/README.md b/README.md
index add42a7..3fd71be 100644
--- a/README.md
+++ b/README.md
@@ -84,14 +84,7 @@ The check method has the following API:
 
 ```javascript
   var StatsD = require('hot-shots'),
-      client = new StatsD({ port: 8020, globalTags: { env:
-  process.env.NODE_ENV });
-
-  // Catch socket errors so they don't go unhandled, as explained
-  // in the Errors section below
-  client.socket.on('error', function(error) {
-    console.error("Error in socket: ", error);
-  });
+      client = new StatsD({ port: 8020, globalTags: { env: process.env.NODE_ENV, errorHandler: errorHandler });
 
   // Timing: sends a timing command with the specified milliseconds
   client.timing('response_time', 42);
@@ -200,9 +193,13 @@ Some of the functionality mentioned above is specific to DogStatsD or Telegraf.
 
 As usual, callbacks will have an error as their first parameter.  You can have an error in both the message and close callbacks.
 
-If the optional callback is not given, an error is thrown in some cases and a console.log message is used in others.  An error will only be thrown when there is a missing callback if it is some potential configuration issue to be fixed.
+If the optional callback is not given, an error is thrown in some
+cases and a console.log message is used in others.  An error will only
+be explitly thrown when there is a missing callback or if it is some potential configuration issue to be fixed.
 
-In the event that there is a socket error, `hot-shots` will allow this error to bubble up unless an `errorHandler` is specified.  If you would like to catch the errors, either specify an `errorHandler` in your root client or just attach a listener to the socket property on the instance.
+If you would like ensure all errors are caught, specify an `errorHandler` in your root
+client. This will catch errors in socket setup, sending of messages,
+and closing of the socket.  If you specify an errorHandler and a callback, the callback will take precedence.
 
 ```javascript
 // Using errorHandler
@@ -213,13 +210,6 @@ var client = new StatsD({
 })
 ```
 
-```javascript
-// Attaching an error handler to client's socket
-client.socket.on('error', function(error) {
-  console.error("Error in socket: ", error);
-});
-```
-
 ## Submitting changes
 
 Thanks for considering making any updates to this project!  Here are the steps to take in your fork:
diff --git a/lib/statsd.js b/lib/statsd.js
index f8a8c8b..9382fe1 100644
--- a/lib/statsd.js
+++ b/lib/statsd.js
@@ -120,6 +120,7 @@ const Client = function (host, port, prefix, suffix, globalize, cacheDns, mock,
     }
   }
 
+  this.messagesInFlight = 0;
   this.CHECKS = {
     OK: 0,
     WARNING: 1,
@@ -207,7 +208,6 @@ Client.prototype.sendAll = function (stat, value, type, sampleRate, tags, callba
  */
 Client.prototype.sendStat = function (stat, value, type, sampleRate, tags, callback) {
   let message = `${this.prefix + stat + this.suffix}:${value}|${type}`;
-
   sampleRate = sampleRate || this.sampleRate;
   if (sampleRate && sampleRate < 1) {
     if (Math.random() < sampleRate) {
@@ -310,8 +310,7 @@ Client.prototype.flushQueue = function (callback) {
  * @param callback {Function=} Callback when message is done being delivered. Optional.
  */
 Client.prototype.sendMessage = function (message, callback) {
-  // Guard against 'RangeError: Offset into buffer too large' in node 0.10
-  // https://github.com/nodejs/node-v0.x-archive/issues/7884
+  // don't waste the time if we aren't sending anything
   if (message === '') {
     if (callback) {
       callback(null);
@@ -323,22 +322,33 @@ Client.prototype.sendMessage = function (message, callback) {
     message += '\n';
   }
 
+  const handleCallback = (err) => {
+    this.messagesInFlight--;
+    const errFormatted = err ? new Error(`Error sending hot-shots message: ${err}`) : null;
+    if (errFormatted) {
+      errFormatted.code = err.code;
+    }
+    if (callback) {
+      callback(errFormatted);
+    } else if (this.errorHandler) {
+      this.errorHandler(errFormatted);
+    } else if (errFormatted) {
+      console.log(String(errFormatted));
+      // emit error ourselves on the socket for backwards compatibility
+      this.socket.emit('error', errFormatted);
+    }
+  };
+
   const buf = Buffer.from(message);
   try {
+    this.messagesInFlight++;
     if (this.protocol === 'tcp') {
-      this.socket.write(buf, 'ascii', callback);
+      this.socket.write(buf, 'ascii', handleCallback);
     } else {
-      this.socket.send(buf, 0, buf.length, this.port, this.host, callback);
+      this.socket.send(buf, 0, buf.length, this.port, this.host, handleCallback);
     }
   } catch (err) {
-    const errMessage = `Error sending hot-shots message: ${err}`;
-    if (callback) {
-      callback(new Error(errMessage));
-    } else if (this.errorHandler) {
-      this.errorHandler(new Error(errMessage));
-    } else {
-      console.log(errMessage);
-    }
+    handleCallback(err);
   }
 };
 
@@ -353,12 +363,44 @@ Client.prototype.onBufferFlushInterval = function () {
  * Close the underlying socket and stop listening for data on it.
  */
 Client.prototype.close = function (callback) {
+  // stop trying to flush the queue on an interval
   if (this.intervalHandle) {
     clearInterval(this.intervalHandle);
   }
 
-  this.flushQueue();
+  // flush the queue one last time, if needed
+  this.flushQueue((err) => {
+    if (err) {
+      if (callback) {
+        return callback(err);
+      }
+      else {
+        return console.log(err);
+      }
+    }
+
+    // FIXME: we have entered callback hell, and this whole file is in need of an async rework
+
+    // wait until there are no more messages in flight before really closing the socket
+    let intervalAttempts = 0;
+    const waitForMessages = setInterval(() => {
+      intervalAttempts++;
+      if (intervalAttempts > 10) {
+        console.log('hot-shots could not clear out messages in flight but closing anyways');
+        this.messagesInFlight = 0;
+      }
+      if (this.messagesInFlight <= 0) {
+        clearInterval(waitForMessages);
+        this._close(callback);
+      }
+    }, 50);
+  });
+};
 
+/**
+ * Really close the socket and handle any errors related to it
+ */
+Client.prototype._close = function (callback) {
   // error function to use in callback and catch below
   let handledError = false;
   const handleErr = (err) => {


Test patch:
diff --git a/test/close.js b/test/close.js
index c2972ec..b5696dc 100644
--- a/test/close.js
+++ b/test/close.js
@@ -1,3 +1,4 @@
+const assert = require('assert');
 const helpers = require('./helpers/helpers.js');
 
 const testTypes = helpers.testTypes;
@@ -8,7 +9,7 @@ describe('#close', () => {
   let server;
   let statsd;
 
-  testTypes().forEach(([description, serverType, clientType]) => {
+  testTypes().forEach(([description, serverType, clientType, metricsEnd]) => {
     describe(description, () => {
       it('should call callback after close call', done => {
         server = createServer(serverType, address => {
@@ -24,6 +25,87 @@ describe('#close', () => {
         });
       });
 
+      it('should send metrics before close call', done => {
+        let metricSeen = false;
+        server = createServer(serverType, address => {
+          statsd = createHotShotsClient({
+            host: address.address,
+            port: address.port,
+            protocol: serverType
+          }, clientType);
+          statsd.set('test', 42);
+          statsd.close(() => {
+            // give the metric a bit of time to get handled by the server
+            const serverClose = setInterval(() => {
+              server.close();
+              clearInterval(serverClose);
+              assert.ok(metricSeen, 'Metric was not seen as expected');
+              done();
+            }, 100);
+          });
+        });
+        server.on('metrics', metrics => {
+          assert.equal(metrics, `test:42|s${metricsEnd}`);
+          metricSeen = true;
+        });
+      });
+
+      it('should send metric before close call when buffering enabled', done => {
+        let metricSeen = false;
+        server = createServer(serverType, address => {
+          statsd = createHotShotsClient({
+            host: address.address,
+            port: address.port,
+            protocol: serverType,
+            maxBufferSize: 1
+          }, clientType);
+          statsd.set('test', 42);
+          statsd.close(() => {
+            // give the metric a bit of time to get handled by the server
+            const serverClose = setInterval(() => {
+              server.close();
+              clearInterval(serverClose);
+              assert.ok(metricSeen, 'Metric was not seen as expected');
+              done();
+            }, 100);
+          });
+        });
+        server.on('metrics', metrics => {
+          // this uses '\n' instead of metricsEnd because that's how things are set up when
+          // maxBufferSize is in use
+          assert.equal(metrics, 'test:42|s\n');
+          metricSeen = true;
+        });
+      });
+
+      it('should send metric before close call when buffered', done => {
+        let metricSeen = false;
+        server = createServer(serverType, address => {
+          statsd = createHotShotsClient({
+            host: address.address,
+            port: address.port,
+            protocol: serverType,
+            maxBufferSize: 5000
+          }, clientType);
+          statsd.set('test', 42);
+          statsd.close(() => {
+            // give the metric a bit of time to get handled by the server
+            const serverClose = setInterval(() => {
+              server.close();
+              clearInterval(serverClose);
+              assert.ok(metricSeen, 'Metric was not seen as expected');
+              done();
+            }, 100);
+          });
+        });
+        server.on('metrics', metrics => {
+          // this uses '\n' instead of metricsEnd because that's how things are set up when
+          // maxBufferSize is in use
+          assert.equal(metrics, 'test:42|s\n');
+          metricSeen = true;
+        });
+      });
+
       it('should use errorHandler on close issue', done => {
         server = createServer(serverType, address => {
           statsd = createHotShotsClient({
diff --git a/test/errorHandling.js b/test/errorHandling.js
index a1f0b52..02bf9fd 100644
--- a/test/errorHandling.js
+++ b/test/errorHandling.js
@@ -92,7 +92,7 @@ describe('#errorHandling', () => {
         statsd.socket.on('error', error => {
           assert.ok(error);
           assert.equal(error.code, 'ENOTFOUND');
-          // skip closing, because the unresolvable host hangs
+          statsd.close();
           done();
         });