Repository: brightcove/hot-shots
Pull request #123 (2019-09-26 07:45:36)

Add udsGracefulErrorHandling
Tried fixing the feedback from #116 

To be totally correct, it probably should've been a PR against NinjaBanjo/hot-shots, but I figured this might move a bit quicker (otherwise: feel free to cherry-pick my changes into #116)


Patch:
diff --git a/README.md b/README.md
index 50b7e43..d3c7073 100644
--- a/README.md
+++ b/README.md
@@ -47,6 +47,8 @@ Parameters (specified as one object passed into hot-shots):
 * `useDefaultRoute`: Use the default interface on a Linux system. Useful when running in containers
 * `protocol`: Use `tcp` option for TCP protocol, or `uds` for the Unix Domain Socket protocol. Defaults to UDP otherwise
 * `path`: Used only when the protocol is `uds`. Defaults to `/var/run/datadog/dsd.socket`.
+* `udsGracefulErrorHandling`: Used only when the protocol is `uds`. Boolean indicating whether to handle socket errors gracefully. Defaults to true.
+* `udsGracefulRestartRateLimit`: Used only when the protocol is `uds`. Time (ms) between re-creating the socket. Defaults to `1000`.
 
 ### StatsD methods
 All StatsD methods other than `event`, `close`, and `check` have the same API:
diff --git a/lib/statsd.js b/lib/statsd.js
index 0663d0b..3a98a44 100644
--- a/lib/statsd.js
+++ b/lib/statsd.js
@@ -13,6 +13,7 @@ const UDS = 'uds';
 // UDP constant not needed/used, defaults to udp
 
 const UDS_PATH_DEFAULT = '/var/run/datadog/dsd.socket';
+const UDS_DEFAULT_GRACEFUL_RESTART_LIMIT = 1000;
 
 /**
  * The Client for StatsD.  The main entry-point for hot-shots.  Note adding new parameters
@@ -120,6 +121,7 @@ const Client = function (host, port, prefix, suffix, globalize, cacheDns, mock,
   this.bufferFlushInterval = options.bufferFlushInterval || 1000;
   this.bufferHolder = options.isChild ? options.bufferHolder : { buffer: '' };
   this.errorHandler = options.errorHandler;
+  this.udsGracefulErrorHandling = 'udsGracefulErrorHandling' in options ? options.udsGracefulErrorHandling : true;
 
   // If we're mocking the client, create a buffer to record the outgoing calls.
   if (this.mock) {
@@ -153,6 +155,47 @@ const Client = function (host, port, prefix, suffix, globalize, cacheDns, mock,
     global.statsd = this;
   }
 
+  // only for uds (options.protocol uds)
+  // enabled with the extra flag options.udsGracefulErrorHandling
+  // will gracefully (attempt) to re-open the socket with a small delay
+  // options.udsGracefulRestartRateLimit is the minimum time (ms) between creating sockets
+  // does not support options.isChild (how to re-create a socket you didn't create?)
+  if (!options.isChild && options.protocol === 'uds' && options.udsGracefulErrorHandling) {
+    const socketCreateLimit = options.udsGracefulRestartRateLimit || UDS_DEFAULT_GRACEFUL_RESTART_LIMIT; // only recreate once per second
+    const lastSocketCreateTime = Date.now();
+    this.socket.on('error', (err) => {
+      const code = err.code;
+      switch (code) {
+        case 107:
+        case 111: {
+          if (Date.now() - lastSocketCreateTime >= socketCreateLimit) {
+            // recreate the socket, but only once per 30 seconds
+            if (this.errorHandler) {
+              this.socket.removeListener('error', this.errorHandler);
+            }
+            this.socket.close();
+            this.socket = createSocket(this, {
+              host: this.host,
+              path: options.path,
+              port: this.port,
+              protocol: this.protocol
+            });
+
+            if (this.errorHandler) {
+              this.socket.on('error', this.errorHandler);
+            } else {
+              this.socket.on('error', error => console.error(`hot-shots UDS error: ${error}`));
+            }
+          }
+          break;
+        }
+        default: {
+          break;
+        }
+      }
+    });
+  }
+
   if (options.useDefaultRoute) {
     const defaultRoute = helpers.getDefaultRoute();
     if (defaultRoute) {


Test patch:
diff --git a/test/errorHandling.js b/test/errorHandling.js
index 7b7839d..caa109d 100644
--- a/test/errorHandling.js
+++ b/test/errorHandling.js
@@ -180,6 +180,124 @@ describe('#errorHandling', () => {
 
         statsd.send('test title');
       });
+
+      if (serverType === 'uds' && clientType === 'client') {
+        it('should re-create the socket on 111 error for type uds', (done) => {
+          const code = 111;
+          const realDateNow = Date.now;
+          Date.now = () => '4857394578';
+          // emit an error, like a socket would
+          // 111 is connection refused
+          server = createServer('uds_broken', address => {
+            const client = statsd = createHotShotsClient({
+              host: address.address,
+              port: address.port,
+              protocol: 'uds',
+              udsGracefulErrorHandling: true,
+              errorHandler(error) {
+                assert.ok(error);
+                assert.equal(error.code, code);
+              }
+            }, 'client');
+            const initialSocket = client.socket;
+            setTimeout(() => {
+              initialSocket.emit('error', { code });
+              assert.ok(Object.is(initialSocket, client.socket));
+              // it should not create the socket if it breaks too quickly
+              // change time and make another error
+              Date.now = () => 4857394578 + 1000; // 1 second later
+              initialSocket.emit('error', { code });
+              setTimeout(() => {
+                // make sure the socket was re-created
+                assert.notEqual(initialSocket, client.socket);
+                // put things back
+                Date.now = realDateNow;
+                done();
+              }, 5);
+            }, 5);
+          });
+        });
+
+        it('should re-create the socket on 107 error for type uds', (done) => {
+          const code = 107;
+          const realDateNow = Date.now;
+          Date.now = () => '4857394578';
+          // emit an error, like a socket would
+          // 111 is connection refused
+          server = createServer('uds_broken', address => {
+            const client = statsd = createHotShotsClient({
+              host: address.address,
+              port: address.port,
+              protocol: 'uds',
+              udsGracefulErrorHandling: true,
+              errorHandler(error) {
+                assert.ok(error);
+                assert.equal(error.code, code);
+              }
+            }, 'client');
+            const initialSocket = client.socket;
+            setTimeout(() => {
+              initialSocket.emit('error', { code });
+              assert.ok(Object.is(initialSocket, client.socket));
+              // it should not create the socket if it breaks too quickly
+              // change time and make another error
+              Date.now = () => 4857394578 + 1000; // 1 second later
+              initialSocket.emit('error', { code });
+              setTimeout(() => {
+                // make sure the socket was re-created
+                assert.notEqual(initialSocket, client.socket);
+                // put things back
+                Date.now = realDateNow;
+                done();
+              }, 5);
+            }, 5);
+          });
+        });
+
+        it('should re-create the socket on error for type uds with the configurable limit', (done) => {
+          const code = 111;
+          const limit = 4000;
+          const realDateNow = Date.now;
+          Date.now = () => '4857394578';
+          // emit an error, like a socket would
+          // 111 is connection refused
+          server = createServer('uds_broken', address => {
+            const client = statsd = createHotShotsClient({
+              host: address.address,
+              port: address.port,
+              protocol: 'uds',
+              udsGracefulErrorHandling: true,
+              udsGracefulRestartRateLimit: limit,
+              errorHandler(error) {
+                assert.ok(error);
+                assert.equal(error.code, code);
+              }
+            }, 'client');
+            const initialSocket = client.socket;
+            setTimeout(() => {
+              initialSocket.emit('error', { code });
+              assert.ok(Object.is(initialSocket, client.socket));
+              // it should not create the socket if it breaks too quickly
+              // change time and make another error
+              Date.now = () => 4857394578 + 1000; // 1 second later
+              initialSocket.emit('error', { code });
+              setTimeout(() => {
+                // make sure the socket was NOT re-created
+                assert.equal(initialSocket, client.socket);
+                Date.now = () => 4857394578 + limit; // 1 second later
+                initialSocket.emit('error', { code });
+                setTimeout(() => {
+                  // make sure the socket was re-created
+                  assert.notEqual(initialSocket, client.socket);
+                  // put things back
+                  Date.now = realDateNow;
+                  done();
+                }, 5);
+              }, 5);
+            }, 5);
+          });
+        });
+      }
     });
   });
 });
diff --git a/test/helpers/helpers.js b/test/helpers/helpers.js
index b578d2f..21ba156 100644
--- a/test/helpers/helpers.js
+++ b/test/helpers/helpers.js
@@ -22,6 +22,7 @@ const TCP = 'tcp';
 const UDP = 'udp';
 const UDS = 'uds';
 const TCP_BROKEN = 'tcp_broken';
+const UDS_BROKEN = 'uds_broken';
 
 // tcp puts a newline at the end but udp/uds do not
 const TCP_METRIC_END = '\n';
@@ -137,6 +138,26 @@ function createServer(serverType, onListening) {
     });
     server.bind(UDS_TEST_PATH);
   }
+  else if (serverType === UDS_BROKEN) {
+    // we always have to manually unlink the test socket
+    if (fs.existsSync(UDS_TEST_PATH)) { // eslint-disable-line no-sync
+      fs.unlinkSync(UDS_TEST_PATH); // eslint-disable-line no-sync
+    }
+
+    server = unixDgram.createSocket('unix_dgram', buf => {
+      const metrics = buf.toString();
+      server.emit('metrics', metrics);
+      server.close();
+    });
+    server.on('listening', () => {
+      onListening('127.0.0.1');
+    });
+    server.on('error', (err) => {
+      console.log('uds connection failed', err);
+      onListening('127.0.0.1');
+    });
+    server.bind(UDS_TEST_PATH);
+  }
   else if (serverType === TCP) {
     server = net.createServer(socket => {
       socket.setEncoding('ascii');