elasticio/elasticio#6359 https.globalAgent enable keepAlive (#194)

* elasticio/elasticio#6359 https.globalAgent enable keepAlive

* elasticio/elasticio#6359 test keep alive is set

Author: PaulAnnekov

CHANGELOG.md

@@ -1,3 +1,7 @@
+## 2.6.29 (July 14, 2022)
+
+* Enabled keep-alive for global HTTPS agent ([#6359](https://github.com/elasticio/elasticio/issues/6359))
+
 ## 2.6.28 (June 21, 2022)
 
 * Fix: "sailor-nodejs ignores errors from maester during lightweight message upload" [#6233](https://github.com/elasticio/elasticio/issues/6233))

mocha_spec/integration_component/component.json

@@ -16,6 +16,10 @@
       "main": "./triggers/trigger_with_no_hooks.js",
       "title": "No hooks at all. The only process function."
     },
+    "trigger_with_keep_alive": {
+      "main": "./triggers/trigger_with_keep_alive.js",
+      "title": "Returns info whether keep alive agent was used during HTTP request"
+    },
     "fails_to_init": {
       "main": "./triggers/fails_to_init.js",
       "title": "Fails to init"

mocha_spec/integration_component/triggers/trigger_with_keep_alive.js

@@ -0,0 +1,20 @@
+const rp = require('request-promise-native');
+
+exports.process = async function process(msg) {
+    const options = {
+        uri: `${msg.body.isHttps ? 'https' : 'http'}://api.acme.com/customers`,
+        json: true,
+        resolveWithFullResponse: true
+    };
+
+    const response = await rp.get(options);
+    this.emit('data', {
+        id: 'f45be600-f770-11e6-b42d-b187bfbf19fd',
+        body: {
+            originalMsg: msg,
+            customers: response.body,
+            keepAlive: response.request.agent.keepAlive
+        }
+    });
+    this.emit('end');
+};

mocha_spec/run.spec.js

@@ -164,81 +164,6 @@ describe('Integration Test', () => {
                     });
                 });
 
-                it('should run trigger successfully', async () => {
-                    helpers.mockApiTaskStepResponse(env);
-
-                    nock('https://api.acme.com')
-                        .post('/subscribe')
-                        .reply(200, {
-                            id: 'subscription_12345'
-                        })
-                        .get('/customers')
-                        .reply(200, customers);
-
-                    await amqpHelper.publishMessage(inputMessage, {
-                        parentMessageId,
-                        threadId
-                    });
-                    runner.run(settings.readFrom(env), ipc);
-                    const { message, queueName } = await new Promise(resolve => amqpHelper.on(
-                        'data',
-                        (message, queueName) => resolve({ message, queueName })
-                    ));
-
-                    const { properties, content } = message;
-                    const { body } = encryptor.decryptMessageContent(content, encoding);
-                    expect(queueName).to.eql(amqpHelper.nextStepQueue);
-
-                    expect(properties.headers.messageId).to.be.a('string');
-                    delete properties.headers.start;
-                    delete properties.headers.end;
-                    delete properties.headers.cid;
-                    delete properties.headers.messageId;
-
-                    expect(properties.headers).to.deep.equal({
-                        execId: env.ELASTICIO_EXEC_ID,
-                        taskId: env.ELASTICIO_FLOW_ID,
-                        workspaceId: env.ELASTICIO_WORKSPACE_ID,
-                        containerId: env.ELASTICIO_CONTAINER_ID,
-                        userId: env.ELASTICIO_USER_ID,
-                        stepId: env.ELASTICIO_STEP_ID,
-                        compId: env.ELASTICIO_COMP_ID,
-                        function: env.ELASTICIO_FUNCTION,
-                        threadId,
-                        parentMessageId,
-                        protocolVersion: protocolVersion
-                    });
-
-                    delete properties.headers;
-
-                    expect(properties).to.deep.equal({
-                        contentType: 'application/json',
-                        contentEncoding: 'utf8',
-                        deliveryMode: undefined,
-                        priority: undefined,
-                        correlationId: undefined,
-                        replyTo: undefined,
-                        expiration: undefined,
-                        messageId: undefined,
-                        timestamp: undefined,
-                        type: undefined,
-                        userId: undefined,
-                        appId: undefined,
-                        clusterId: undefined
-                    });
-
-                    expect(body).to.deep.equal({
-                        originalMsg: inputMessage,
-                        customers: customers,
-                        subscription: {
-                            id: 'subscription_12345',
-                            cfg: {
-                                apiKey: 'secret'
-                            }
-                        }
-                    });
-                });
-
                 it('should run trigger successfully for input protocolVersion 2', async () => {
                     helpers.mockApiTaskStepResponse(env);
 
@@ -1250,6 +1175,70 @@ describe('Integration Test', () => {
             });
         }
 
+        it('should make all HTTP requests with keep alive agent', async () => {
+            env.ELASTICIO_STEP_ID = 'step_2';
+            env.ELASTICIO_FLOW_ID = '5559edd38968ec0736000003';
+            env.ELASTICIO_FUNCTION = 'trigger_with_keep_alive';
+
+            helpers.mockApiTaskStepResponse(env);
+
+            nock('http://api.acme.com')
+                .get('/customers')
+                .reply(200, customers);
+
+            await amqpHelper.publishMessage(inputMessage, {
+                parentMessageId,
+                threadId
+            });
+            runner.run(settings.readFrom(env), ipc);
+            const { message } = await new Promise(resolve => amqpHelper.on(
+                'data',
+                (message, queueName) => resolve({ message, queueName })
+            ));
+
+            const { content } = message;
+            const { body } = encryptor.decryptMessageContent(content, 'base64');
+
+            expect(body).to.deep.equal({
+                originalMsg: inputMessage,
+                customers,
+                keepAlive: true
+            });
+        });
+
+        it('should make all HTTPS requests with keep alive agent', async () => {
+            env.ELASTICIO_STEP_ID = 'step_2';
+            env.ELASTICIO_FLOW_ID = '5559edd38968ec0736000003';
+            env.ELASTICIO_FUNCTION = 'trigger_with_keep_alive';
+
+            inputMessage.body.isHttps = true;
+
+            helpers.mockApiTaskStepResponse(env);
+
+            nock('https://api.acme.com')
+                .get('/customers')
+                .reply(200, customers);
+
+            await amqpHelper.publishMessage(inputMessage, {
+                parentMessageId,
+                threadId
+            });
+            runner.run(settings.readFrom(env), ipc);
+            const { message } = await new Promise(resolve => amqpHelper.on(
+                'data',
+                (message, queueName) => resolve({ message, queueName })
+            ));
+
+            const { content } = message;
+            const { body } = encryptor.decryptMessageContent(content, 'base64');
+
+            expect(body).to.deep.equal({
+                originalMsg: inputMessage,
+                customers,
+                keepAlive: true
+            });
+        });
+
         it('should fail if queue deleted', async () => {
             helpers.mockApiTaskStepResponse(env);
 

package.json

@@ -1,7 +1,7 @@
 {
   "name": "elasticio-sailor-nodejs",
   "description": "The official elastic.io library for bootstrapping and executing for Node.js connectors",
-  "version": "2.6.28",
+  "version": "2.6.29",
   "main": "run.js",
   "scripts": {
     "audit": "better-npm-audit audit --level high --production",

run.js

@@ -7,16 +7,21 @@ const settings = require('./lib/settings.js');
 const { IPC } = require('./lib/ipc.js');
 const Q = require('q');
 const http = require('http');
+const https = require('https');
 
 let sailor;
 let sailorInit;
 let disconnectRequired;
 
-// miserable try to workaround issue described in https://github.com/elasticio/elasticio/issues/4874
-const { Agent } = http;
-http.globalAgent = new Agent({
-    keepAlive: true
-});
+function prepareSandbox() {
+    // enable keep alive by default to handle issues like https://github.com/elasticio/elasticio/issues/4874
+    http.globalAgent = new http.Agent({
+        keepAlive: true
+    });
+    https.globalAgent = new https.Agent({
+        keepAlive: true
+    });
+}
 
 async function putOutToSea(settings, ipc) {
     ipc.send('init:started');
@@ -88,6 +93,7 @@ async function gracefulShutdown() {
 }
 
 async function run(settings, ipc) {
+    prepareSandbox();
     try {
         await putOutToSea(settings, ipc);
         logger.info('Fully initialized and waiting for messages');