Repository: awslabs/aws-embedded-metrics-node
Pull request #7 (2019-11-26 12:08:49)

fix: unhandled TCP error in AgentSink constructor and TcpClient sendM…
…essage functions

*Issue #, if available:*

*Description of changes:*

Handles unhandled rejections for Tcp Client.
Improves error messages by passing original error object to `reject` functions.

By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.


jaredcnance: Build is failing due to `npm audit` ... if you can run `npm audit fix` it should let the build pass. I will look into whether or not we can make these artifacts public and if not we will probably move to TravisCI or another CI build system.

Patch:
diff --git a/src/sinks/AgentSink.ts b/src/sinks/AgentSink.ts
index 81359ab..17ef032 100644
--- a/src/sinks/AgentSink.ts
+++ b/src/sinks/AgentSink.ts
@@ -102,6 +102,8 @@ export class AgentSink implements ISink {
 
   private getSocketClient(endpoint: IEndpoint): ISocketClient {
     LOG('Getting socket client for connection.', endpoint);
-    return endpoint.protocol === TCP ? new TcpClient(endpoint).connect() : new UdpClient(endpoint);
+    const client = endpoint.protocol === TCP ? new TcpClient(endpoint) : new UdpClient(endpoint);
+    client.warmup();
+    return client;
   }
 }
diff --git a/src/sinks/connections/ISocketClient.ts b/src/sinks/connections/ISocketClient.ts
index 25c5fdc..1d7cd17 100644
--- a/src/sinks/connections/ISocketClient.ts
+++ b/src/sinks/connections/ISocketClient.ts
@@ -14,5 +14,6 @@
  */
 
 export interface ISocketClient {
+  warmup(): Promise<void>;
   sendMessage(message: Buffer): Promise<void>;
 }
diff --git a/src/sinks/connections/TcpClient.ts b/src/sinks/connections/TcpClient.ts
index 2a39f07..511caa2 100644
--- a/src/sinks/connections/TcpClient.ts
+++ b/src/sinks/connections/TcpClient.ts
@@ -33,24 +33,30 @@ export class TcpClient implements ISocketClient {
       .on('data', data => LOG('TcpClient received data.', data));
   }
 
-  public connect(): TcpClient {
-    // create the connection as soon as we can, but don't block
-    // we can block on flush if we have to.
-    this.establishConnection();
-    return this;
+  public async warmup() {
+    try {
+      await this.establishConnection();
+    } catch (err) {
+      LOG('Failed to connect', err)
+    }
   }
 
   public async sendMessage(message: Buffer): Promise<void> {
     await this.waitForOpenConnection();
     await new Promise((resolve, reject) => {
-      const wasFlushedToKernel = this.socket.write(message, (err?: Error) => {
+      const onSendError = (err: Error) => {
+        LOG('Failed to write', err);
+        LOG('Socket', this.socket);
+        reject(err);
+      }
+      const wasFlushedToKernel = this.socket
+      .once('error', onSendError)
+      .write(message, (err?: Error) => {
         if (!err) {
           LOG('Write succeeded');
           resolve();
         } else {
-          LOG('Failed to write', err);
-          LOG('Socket', this.socket);
-          reject();
+          onSendError(err)
         }
       });
 
@@ -79,10 +85,10 @@ export class TcpClient implements ISocketClient {
           LOG('TcpClient connected.', this.endpoint);
           resolve();
         })
-        .once('error', () => {
+        .once('error', (e) => {
           this.disconnect('error');
           this.socket.removeListener('connection', resolve);
-          reject();
+          reject(e);
         });
     });
   }
diff --git a/src/sinks/connections/UdpClient.ts b/src/sinks/connections/UdpClient.ts
index 8c1d944..a7a82bd 100644
--- a/src/sinks/connections/UdpClient.ts
+++ b/src/sinks/connections/UdpClient.ts
@@ -25,7 +25,12 @@ export class UdpClient implements ISocketClient {
     this.endpoint = endpoint;
   }
 
-  public sendMessage(message: Buffer): Promise<void> {
+  // No warm up for UDP
+  public warmup (): Promise<void> {
+    return Promise.resolve()
+  }
+
+  public async sendMessage(message: Buffer): Promise<void> {
     const client = dgram.createSocket('udp4');
     client.send(message, this.endpoint.port, this.endpoint.host, (error: any) => {
       if (error) {


Test patch:
diff --git a/src/sinks/__tests__/AgentSink.test.ts b/src/sinks/__tests__/AgentSink.test.ts
index 4177de4..62fd816 100644
--- a/src/sinks/__tests__/AgentSink.test.ts
+++ b/src/sinks/__tests__/AgentSink.test.ts
@@ -1,5 +1,6 @@
 import * as faker from 'faker';
 import Configuration from '../../config/Configuration';
+import { MetricsContext } from '../../logger/MetricsContext';
 import { AgentSink } from '../AgentSink';
 
 test('default endpoint is tcp', () => {
@@ -34,3 +35,17 @@ test('can parse udp endpoints', () => {
   // @ts-ignore
   expect(sink.endpoint.port).toBe(1000);
 });
+
+test('handles tcp connection error', async () => {
+  // arrange
+  const noProcessPort = faker.random.number({min: 1000, max: 9999})
+  Configuration.agentEndpoint = `tcp://127.0.0.1:${noProcessPort}`;
+  const context = MetricsContext.empty()
+  const logGroupName = faker.random.word();
+  const sink = new AgentSink(logGroupName);
+
+  // assert
+  return expect(
+    sink.accept(context)
+  ).rejects.toThrowError(/ECONNREFUSED/)
+});
diff --git a/src/sinks/__tests__/TcpClient.test.ts b/src/sinks/__tests__/TcpClient.test.ts
new file mode 100644
index 0000000..300a331
--- /dev/null
+++ b/src/sinks/__tests__/TcpClient.test.ts
@@ -0,0 +1,17 @@
+import * as faker from 'faker'
+import { TcpClient } from '../connections/TcpClient';
+
+test('handles tcp client errors', async () => {
+  // arrange
+  const noProcessPort = faker.random.number({min: 1000, max: 9999})
+  const client = new TcpClient({
+    host: '0.0.0.0',
+    port: noProcessPort,
+    protocol: 'tcp'
+  })
+
+  // assert
+  return expect(
+    client.sendMessage(Buffer.from([]))
+  ).rejects.toThrowError(/ECONNREFUSED/)
+});