Merge pull request #117 from jsnoble/resilent_logging

improved logging to elasticsearch error handling and schema definition

README.md

@@ -38,7 +38,6 @@ name | name of application| String | optional, defaults to terafoundation
 ops_directory | 'path/to/directory', to look for more custom connectors. Usually this is where you place your custom code not part of core, unless you want to leave your code in place. | String | optional
 descriptors | Object listing all the different modules that a child process could run, used when child process will have different behaviors | Object | optional, default to creation of children running the worker module passed in
 cluster_name | name of application| String or Function | optional, defaults to terafoundation
-logging_connection | If storing logs to elasticsearch, use this parameter to specify which endpoint it will use in the connectors configuration| String or Function | optional
 script | javascript execution of code | Function | optional
 config_schema | system schema for the top level service| Object or Function | optional
 plugin_schema |system schema for plugins associated with the top level service |Object or Function | optional
@@ -71,9 +70,8 @@ var foundation = require('terafoundation')({
     config_schema: config_schema,
     schema_formats: schema_formats,
     ops_directory: ops_directory,
-    cluster_name: cluster_name,
-    logging_connection: logging_connection
-});
+    cluster_name: cluster_name
+  });
 ```
 You may pass in multiple different type of child process that behave differently. To use them you must specify them on the descriptors. After this you may create them using `context.foundation.startWorkers` api to create that specific worker type. The descriptor is primarily needed for child worker restarts
 
@@ -203,4 +201,5 @@ logging | a list to where logs will be written to, settings available are ['cons
 log_level | this determines what level of logs are shown, options: trace, debug, info, warn , error, fatal | String or Array | optional, defaults to info, if a string is set then all log destinations in logging will use this, you may also customize log levels for each destination: [{console: 'debug'}, {file: 'warn}, {elasticsearch: 'info'}]
 log_buffer_limit | the number of logs stored in the ringbuffer on the logger before sent, logging must have elasticsearch set as a value for this to take effect | Number | optional, defaults to 30
 log_buffer_interval | interval (number in milliseconds) that the log buffer will send up its logs, this is used if logging is set to use elasticsearch | Number | optional, defaults to 60000 ms
+log_connection | logging connection endpoint if logging is saved to elasticsearch, this is only in use if logs are being saved to a DB | String | optional, uses the 'default' endpoint of elasticsearch connectors
 connectors | An object containing database client information for you service | Object | required

index.js

@@ -11,7 +11,6 @@ module.exports = function module(config) {
     const api = require('./lib/api');
 
     const name = config.name ? config.name : 'terafoundation';
-    let loggingConnection = 'default';
 
     const { argv } = require('yargs')
         .usage('Usage: $0 [options]')
@@ -36,6 +35,7 @@ module.exports = function module(config) {
     let logger;
 
     const sysconfig = validateConfigs(cluster, config, configFile);
+
     // set by initAPI
 
     function errorHandler(err) {
@@ -100,10 +100,6 @@ module.exports = function module(config) {
             context.cluster_name = config.cluster_name(context.sysconfig);
         }
 
-        if (typeof config.loggingConnection === 'function') {
-            loggingConnection = config.loggingConnection(context.sysconfig);
-        }
-
         // Initialize the API
         api(context);
 
@@ -113,7 +109,7 @@ module.exports = function module(config) {
 
         // FIXME: this should probably be refactored to actually create the
         // logger as it stands this function is very confusing
-        loggerClient(context, context.logger, loggingConnection);
+        loggerClient(context);
 
         if (config.script) {
             config.script(context);

lib/api/makeLogger.js

@@ -3,8 +3,8 @@
 const bunyan = require('bunyan');
 const fs = require('fs');
 const _ = require('lodash');
-const { RingBuffer } = require('../logger_utils');
 const Promise = require('bluebird');
+const { RingBuffer } = require('../logger_utils');
 
 function getLogLevel(level) {
     // Set the same level for all logging types.
@@ -21,7 +21,6 @@ function getLogLevel(level) {
 
 module.exports = function module(context) {
     const loggingConfig = context.sysconfig.terafoundation;
-
     const logLevel = getLogLevel(loggingConfig.log_level);
 
     // This is the root logger. Module specific loggers will be created
@@ -84,9 +83,11 @@ module.exports = function module(context) {
         }
 
         if (_.includes(loggingConfig.logging, 'elasticsearch')) {
-            const limit = loggingConfig.log_buffer_limit;
-            const delay = loggingConfig.log_buffer_interval;
-            const timeseriesFormat = loggingConfig.log_index_rollover_frequency;
+            const {
+                log_buffer_interval: delay,
+                log_buffer_limit: limit,
+                log_index_rollover_frequency: timeseriesFormat
+            } = loggingConfig;
             const name = context.cluster_name;
 
             const level = logLevel.elasticsearch ? logLevel.elasticsearch : 'info';

lib/logger_utils.js

@@ -4,6 +4,7 @@ const util = require('util');
 const moment = require('moment');
 const bunyan = require('bunyan');
 const Promise = require('bluebird');
+const esApi = require('@terascope/elasticsearch-api');
 
 const levelsObj = {
     10: 'trace',
@@ -33,11 +34,11 @@ function RingBuffer(index, limit, delay, client, timeseriesFormat) {
     // set a timer to flush logs
     this.interval = setInterval(() => {
         self.sendBulk();
-    }, delay);
+    }, 1000);
 }
 util.inherits(RingBuffer, bunyan.RingBuffer);
 
-RingBuffer.prototype.write = function (record) {
+RingBuffer.prototype.write = function write(record) {
     if (!this.writable) {
         throw (new Error('RingBuffer has been ended already'));
     }
@@ -50,9 +51,9 @@ RingBuffer.prototype.write = function (record) {
     return (true);
 };
 
-RingBuffer.prototype.sendBulk = function () {
+RingBuffer.prototype.sendBulk = async function sendBulk() {
     const {
-        records, client, index, timeseriesFormat
+        records, client, index, timeseriesFormat, logger
     } = this;
 
     const esData = records.reduce((prev, record) => {
@@ -70,25 +71,21 @@ RingBuffer.prototype.sendBulk = function () {
     this.records = [];
 
     if (esData.length > 0) {
-        return client.bulk({ body: esData })
-            .then((res) => {
-                if (res.errors) {
-                    console.log('whats this error', JSON.stringify(res));
-                    res.items.forEach((item) => {
-                        console.log(item.create);
-                    });
-                }
-            })
-            .catch((err) => {
-                const errMsg = err.toJSON ? err.toJSON : err.stack;
-                return Promise.reject(errMsg);
-            });
+        try {
+            const results = await client.bulkSend(esData, logger);
+            return results;
+        } catch (err) {
+            logger.error(`failed to send logs to index: ${index}`, err);
+            // non retrialbe error, so we return. We do not return an error
+            // as that could have other side effects in teraslice
+            return true;
+        }
     }
 
-    return Promise.resolve(true);
+    return true;
 };
 
-RingBuffer.prototype.flush = function () {
+RingBuffer.prototype.flush = function flush() {
     const { interval } = this;
     if (this.records.length > 0) {
         clearInterval(interval);
@@ -99,25 +96,32 @@ RingBuffer.prototype.flush = function () {
     return Promise.resolve(true);
 };
 
-RingBuffer.prototype.setBufferClient = function (client) {
+RingBuffer.prototype.setBufferClient = function setBufferClient(client, logger) {
     this.client = client;
+    this.logger = logger;
 };
 
-function loggerClient(context, logger, loggingConnection) {
+function loggerClient(context) {
+    const {
+        logger,
+        sysconfig: { terafoundation: { log_connection: endpoint } }
+    } = context;
+
     const esClient = logger.streams.filter((stream) => {
         if (stream.stream instanceof RingBuffer) {
             return stream;
         }
+        return false;
     });
 
     if (esClient.length > 0) {
         const { client } = context.foundation.getConnection({
             type: 'elasticsearch',
-            endpoint: loggingConnection,
+            endpoint,
             cached: true
         });
-
-        esClient[0].stream.setBufferClient(client);
+        const api = esApi(client, logger);
+        esClient[0].stream.setBufferClient(api, logger);
     }
 }
 

package.json

@@ -1,6 +1,6 @@
 {
   "name": "terafoundation",
-  "version": "0.3.0",
+  "version": "0.4.0",
   "scripts": {
     "lint": "eslint *.js lib",
     "lint:fix": "eslint --fix *.js lib",
@@ -28,6 +28,7 @@
   },
   "license": "Apache-2.0",
   "dependencies": {
+    "@terascope/elasticsearch-api": "^1.1.1",
     "agentkeepalive": "^3.4.1",
     "bluebird": "^3.5.2",
     "bunyan": "^1.8.12",

system_schema.js

@@ -94,6 +94,11 @@ module.exports = {
         default: 'monthly',
         format: ['daily', 'monthly', 'yearly']
     },
+    log_connection: {
+        doc: 'logging connection endpoint if logging is saved to elasticsearch',
+        default: 'default',
+        format: 'optional_String'
+    },
     workers: {
         doc: 'Number of workers per server',
         default: workerCount

yarn.lock

@@ -96,6 +96,21 @@
     lodash "^4.17.5"
     to-fast-properties "^2.0.0"
 
+"@terascope/elasticsearch-api@^1.1.1":
+  version "1.1.1"
+  resolved "https://registry.yarnpkg.com/@terascope/elasticsearch-api/-/elasticsearch-api-1.1.1.tgz#87de959d443289c77cbd0a2131de3b71b6fc4b8a"
+  dependencies:
+    "@terascope/error-parser" "^1.0.0"
+    bluebird "^3.5.0"
+    lodash "^4.17.4"
+    uuid "^3.0.1"
+
+"@terascope/error-parser@^1.0.0":
+  version "1.0.0"
+  resolved "https://registry.yarnpkg.com/@terascope/error-parser/-/error-parser-1.0.0.tgz#0853b6d10845babda7a7ed596632f866d2fde86d"
+  dependencies:
+    lodash "^4.17.4"
+
 acorn-jsx@^4.1.1:
   version "4.1.1"
   resolved "https://registry.yarnpkg.com/acorn-jsx/-/acorn-jsx-4.1.1.tgz#e8e41e48ea2fe0c896740610ab6a4ffd8add225e"
@@ -244,7 +259,7 @@ [email protected]:
   version "3.5.0"
   resolved "https://registry.yarnpkg.com/bluebird/-/bluebird-3.5.0.tgz#791420d7f551eea2897453a8a77653f96606d67c"
 
-bluebird@^3.5.2:
+bluebird@^3.5.0, bluebird@^3.5.2:
   version "3.5.2"
   resolved "https://registry.yarnpkg.com/bluebird/-/bluebird-3.5.2.tgz#1be0908e054a751754549c270489c1505d4ab15a"
 
@@ -2361,14 +2376,14 @@ util-deprecate@~1.0.1:
   version "1.0.2"
   resolved "https://registry.yarnpkg.com/util-deprecate/-/util-deprecate-1.0.2.tgz#450d4dc9fa70de732762fbd2d4a28981419a0ccf"
 
+uuid@^3.0.1, uuid@^3.3.2:
+  version "3.3.2"
+  resolved "https://registry.yarnpkg.com/uuid/-/uuid-3.3.2.tgz#1b4af4955eb3077c501c23872fc6513811587131"
+
 uuid@^3.1.0:
   version "3.2.1"
   resolved "https://registry.yarnpkg.com/uuid/-/uuid-3.2.1.tgz#12c528bb9d58d0b9265d9a2f6f0fe8be17ff1f14"
 
-uuid@^3.3.2:
-  version "3.3.2"
-  resolved "https://registry.yarnpkg.com/uuid/-/uuid-3.3.2.tgz#1b4af4955eb3077c501c23872fc6513811587131"
-
 validate-npm-package-license@^3.0.1:
   version "3.0.3"
   resolved "https://registry.yarnpkg.com/validate-npm-package-license/-/validate-npm-package-license-3.0.3.tgz#81643bcbef1bdfecd4623793dc4648948ba98338"