Steward fix and IncludeHeader option (#33)

Author: uaArsen

.eslintrc.js

@@ -8,7 +8,7 @@ const NEVER = 'never';
 
 module.exports = {
     'parserOptions': {
-        ecmaVersion: 2017
+        ecmaVersion: 2019
     },
     'env': {
         es6: true,

CHANGELOG.md

@@ -1,3 +1,7 @@
+## 1.1.5 (July 15, 2019)
+
+* Add retries mechanism to all request, refactor component to use axios library
+
 ## 1.1.4 (June 27, 2019)
 
 * Added emitAll feature for CSV Write action

README.md

@@ -13,8 +13,12 @@ attachment. It can also write a CSV file from the incoming events.
 
 ## Environment variables
 
-Component does not has any required environment variables, but we suggest to use `EIO_REQUIRED_RAM_MB` - recommended value of allocated memory is `512` MB
-
+1. `EIO_REQUIRED_RAM_MB` - recommended value of allocated memory is `512` MB
+2. `REQUEST_TIMEOUT` - HTTP request timeout in milliseconds, default value 10000
+3. `REQUEST_RETRY_DELAY` - delay between retry attempts in milliseconds, default value 7000
+4. `REQUEST_MAX_RETRY` - number of HTTP request retry attempts, default value 7
+5. `REQUEST_MAX_CONTENT_LENGTH`  - max size of http request in bytes, default value: 104857600
+6. `TIMEOUT_BETWEEN_EVENTS` - number of milliseconds write action wait before creating separate attachments, default value: 10000
 
 ## Credentials
 
@@ -56,6 +60,8 @@ a `JSON` object. To configure this action the following fields can be used:
 
 ### Write CSV attachment
 
+* `Include Header` - this select configures output behavior of the component. If option is `Yes` or no value chosen than header of csv file will be written to attachment, this is default behavior. If value `No` selected than csv header will be omitted from attachment.
+
 This action will combine multiple incoming events into a CSV file until there is a gap
 of more than 10 seconds between events. Afterwards, the CSV file will be closed
 and attached to the outgoing message.
@@ -95,6 +101,6 @@ able to handle file attachments.
 
 1. You may get `Component run out of memory and terminated.` error during run-time, that means that component needs more memory, please add
  `EIO_REQUIRED_RAM_MB` environment variable with an appropriate value (e.g. value `512` means that 512 MB will be allocated) for the component in this case.
-2. Maximal possible size for an attachment is 10 MB.
-3. Attachments mechanism does not work with [Local Agent Installation](https://support.elastic.io/support/solutions/articles/14000076461-announcing-the-local-agent-)
-
+2. You may get `Error: write after end` error, as a current workaround try increase value of environment variable: `TIMEOUT_BETWEEN_EVENTS`. 
+3. Maximal possible size for an attachment is 10 MB.
+4. Attachments mechanism does not work with [Local Agent Installation](https://support.elastic.io/support/solutions/articles/14000076461-announcing-the-local-agent-)

component.json

@@ -48,6 +48,17 @@
       "main": "./lib/actions/write.js",
       "title": "Write CSV attachment",
       "fields": {
+        "includeHeaders": {
+          "label" : "Include Headers",
+          "required": false,
+          "viewClass" : "SelectView",
+          "description" : "Default Yes",
+          "model": {
+            "Yes" : "Yes",
+            "No" : "No"
+          },
+          "prompt": "Include headers? Default Yes."
+        },
         "writer": {
           "viewClass": "CSVWriteView"
         }

lib/actions/write.js

@@ -1,26 +1,34 @@
 'use strict';
 
-const request = require('request-promise');
+const axios = require('axios');
 const co = require('co');
 const csv = require('csv');
 const _ = require('lodash');
 const messages = require('elasticio-node').messages;
 const client = require('elasticio-rest-node')();
 
-const TIMEOUT_BETWEEN_EVENTS = 10000;
+const util = require('../util/util');
+
+const REQUEST_TIMEOUT = process.env.REQUEST_TIMEOUT || 10000; // 10s
+const REQUEST_MAX_RETRY = process.env.REQUEST_MAX_RETRY || 7;
+const REQUEST_RETRY_DELAY = process.env.REQUEST_RETRY_DELAY || 7000; // 7s
+const REQUEST_MAX_CONTENT_LENGTH = process.env.REQUEST_MAX_CONTENT_LENGTH || 104857600; // 100MB
+const TIMEOUT_BETWEEN_EVENTS = process.env.TIMEOUT_BETWEEN_EVENTS || 10000; // 10s;
 
 let stringifier;
 let signedUrl;
 let timeout;
 let rowCount = 0;
-
-exports.init = function init(cfg) {
+let ax;
+let putUrl;
+exports.init = async function init(cfg) {
     return co(function* gen() {
 
         const delimiter = cfg.writer.separator || ',';
+        const header = cfg.includeHeaders !== 'No';
         console.log('Using delimiter: \'%s\'', delimiter);
         const options = {
-            header: true,
+            header,
             delimiter
         };
 
@@ -33,13 +41,11 @@ exports.init = function init(cfg) {
         }
 
         stringifier = csv.stringify(options);
-
         signedUrl = yield client.resources.storage.createSignedUrl();
-
-        let putUrl = signedUrl.put_url;
+        putUrl = signedUrl.put_url;
         console.log('CSV file to be uploaded file to uri=%s', putUrl);
-
-        stringifier.pipe(request.put(putUrl));
+        ax = axios.create();
+        util.addRetryCountInterceptorToAxios(ax);
     });
 };
 
@@ -51,13 +57,20 @@ exports.process = async function ProcessAction(msg, cfg) {
         clearTimeout(timeout);
     }
 
+
     timeout = setTimeout(() => {
         console.log('Closing the stream due to inactivity');
         co(function* gen() {
             const finalRowCount = rowCount;
             console.log('The resulting CSV file contains %s rows', finalRowCount);
+            ax.put(putUrl, stringifier ,{
+                method: 'PUT',
+                timeout: REQUEST_TIMEOUT,
+                retry: REQUEST_MAX_RETRY,
+                delay: REQUEST_RETRY_DELAY,
+                maxContentLength: REQUEST_MAX_CONTENT_LENGTH
+            });
             stringifier.end();
-
             const messageToEmit = messages.newMessageWithBody({
                 rowCount: finalRowCount
             });
@@ -68,14 +81,11 @@ exports.process = async function ProcessAction(msg, cfg) {
             };
             signedUrl = null;
             rowCount = 0;
-
             console.log('Emitting message %j', messageToEmit);
             yield self.emit('data', messageToEmit);
-
             yield exports.init(cfg);
         });
     }, TIMEOUT_BETWEEN_EVENTS);
-
     let row = msg.body.writer;
     console.log(`Incoming data: ${JSON.stringify(row)}`);
     if (cfg.writer.columns) {

lib/read.js

@@ -1,14 +1,20 @@
 'use strict';
 
+const util = require('./util/util');
+
 const _ = require('underscore');
 const csv = require('csv');
 const messages = require('elasticio-node').messages;
 const moment = require('moment');
 const debug = require('debug')('csv');
-const request = require('request');
+const axios = require('axios');
 const { Writable } = require('stream');
 
 const formatters = {};
+const REQUEST_TIMEOUT = process.env.REQUEST_TIMEOUT || 10000; //ms
+const REQUEST_MAX_RETRY = process.env.REQUEST_MAX_RETRY || 7;
+const REQUEST_RETRY_DELAY = process.env.REQUEST_RETRY_DELAY || 7000; // ms
+
 
 formatters.date = function formatDate(value, col) {
 
@@ -123,16 +129,23 @@ async function ProcessRead(msg, cfg) {
     const writer = new CsvWriter();
 
     debug('Sending GET request to url=%s', csvURL);
-    request.get(csvURL)
-        .on('response', async function onResponse(response) {
-            debug('Have got response status=%s headers=%j', response.statusCode, response.headers);
-            if (response.statusCode !== 200) {
-                await that.emit('error', 'Unexpected response code code=' + response.statusCode);
-                ended = true;
-                throw Error('Unexpected response code code=' + response.statusCode);
-            }
-        })
-        .pipe(parser).pipe(writer);
+    const ax = axios.create();
+    util.addRetryCountInterceptorToAxios(ax);
+    const response = await ax({
+        method: 'get',
+        url: csvURL,
+        responseType: 'stream',
+        timeout: REQUEST_TIMEOUT,
+        retry: REQUEST_MAX_RETRY,
+        delay: REQUEST_RETRY_DELAY
+    });
+    debug('Have got response status=%s headers=%j', response.status, response.headers);
+    if (response.status !== 200) {
+        await that.emit('error', 'Unexpected response code code=' + response.status);
+        ended = true;
+        throw Error('Unexpected response code code=' + response.status);
+    }
+    response.data.pipe(parser).pipe(writer);
 
     // Need to wait for processing all messages
     while (!ended) {

lib/util/util.js

@@ -0,0 +1,12 @@
+exports.addRetryCountInterceptorToAxios = (ax) => {
+    ax.interceptors.response.use(undefined, (err) => { //  Retry count interceptor for axios
+        const { config } = err;
+        if (!config || !config.retry || !config.delay) {return Promise.reject(err);}
+        config.currentRetryCount = config.currentRetryCount || 0;
+        if (config.currentRetryCount >= config.retry) {
+            return Promise.reject(err);
+        }
+        config.currentRetryCount += 1;
+        return new Promise(resolve => setTimeout(() => resolve(ax(config)), config.delay));
+    });
+};