Replace signale with pino (#346)

Author: cbovis

config/default.js

@@ -9,6 +9,13 @@ module.exports = {
   maxPollingInterval: parseInt(process.env.MAX_POLLING_INTERVAL, 10),
   minConfirmations: 12,
   minPollingInterval: parseInt(process.env.MIN_POLLING_INTERVAL, 10),
+  pino: {
+    elasticsearch: {
+      batchSize: 200,
+      index: 'logs_event_extractor',
+      url: process.env.PINO_ELASTIC_SEARCH_URL || null,
+    },
+  },
   v1: {
     genesisBlock: 4145578,
   },

package.json

@@ -2,6 +2,7 @@
   "name": "@0x-event-extractor/app",
   "main": "src/index.js",
   "scripts": {
+    "dev": "nodemon pino . | pino-pretty -t",
     "lint": "eslint .",
     "start": "node src",
     "test": "jest",
@@ -22,11 +23,13 @@
     "husky": "3.0.8",
     "jest": "25.1.0",
     "lint-staged": "9.4.2",
+    "nodemon": "2.0.2",
+    "pino-pretty": "3.6.1",
     "prettier": "1.19.1"
   },
   "engines": {
     "node": "10.17.0",
-    "yarn": "1.21.1"
+    "yarn": "1.22.4"
   },
   "private": true,
   "workspaces": [

packages/core/package.json

@@ -12,7 +12,8 @@
     "delay": "^4.3.0",
     "lodash": "^4.17.15",
     "mongoose": "^5.7.3",
-    "promise-poller": "^1.9.1",
-    "signale": "^1.4.0"
+    "pino": "6.0.0",
+    "pino-elasticsearch": "4.4.0",
+    "promise-poller": "^1.9.1"
   }
 }

packages/core/src/index.js

@@ -7,14 +7,18 @@ const db = require('./util/db');
 const errorLogger = require('./util/error-logger');
 const getJobs = require('./jobs');
 const jobRunner = require('./util/job-runner');
+const logging = require('./util/logging');
+const model = require('./model');
 const web3 = require('../../shared/src/web3');
 
 const configure = async initialConfig => {
   config.init(initialConfig);
+  logging.init(config.get('pino'));
   errorLogger.configure({
     bugsnagToken: config.get('bugsnag.token'),
   });
   await db.connect(config.get('database.connectionString'));
+  await model.init();
   web3.configure({ endpoint: config.get('web3.endpoint') });
   extractorV1.configure({ networkId: config.get('web3.networkId') });
   extractorV2.configure({ networkId: config.get('web3.networkId') });

packages/core/src/jobs/extract-events.js

@@ -1,45 +1,45 @@
 const extractorV1 = require('@0x-event-extractor/extractor-v1');
 const extractorV2 = require('@0x-event-extractor/extractor-v2');
 const extractorV3 = require('@0x-event-extractor/extractor-v3');
-const signale = require('signale');
 
+const { getLogger } = require('../util/logging');
 const BlockRange = require('../model/block-range');
 const Event = require('../model/event');
 const getCurrentBlock = require('../ethereum/get-current-block');
 const getNextBlockRange = require('../events/get-next-block-range');
-const withTimer = require('../util/with-timer');
 const withTransaction = require('../util/with-transaction');
 
 const extractEventsForProtocol = async (protocolVersion, extractorConfig) => {
   // Scope all logging for the job to the specified protocol version
-  const logger = signale.scope(`extract events v${protocolVersion}`);
+  const logger = getLogger(`extract events v${protocolVersion}`);
 
   // Determine which blocks we should fetch log entries from
   const { currentBlock, fetchLogEntries, getEventData } = extractorConfig;
+
+  logger.info(`current block is ${currentBlock}`);
+
   const nextBlockRange = await getNextBlockRange({
     currentBlock,
     protocolVersion,
   });
 
-  logger.info(`current block is ${currentBlock}`);
-
   if (nextBlockRange === null) {
     logger.info('no more blocks to process');
     return;
   }
 
   const { fromBlock, toBlock } = nextBlockRange;
 
-  logger.info(`next block range is ${fromBlock} to ${toBlock}`);
+  logger.info(`fetching events from block range ${fromBlock} to ${toBlock}`);
 
-  const logEntries = await withTimer(logger, `fetch entries`, () =>
-    fetchLogEntries(fromBlock, toBlock),
-  );
+  const logEntries = await fetchLogEntries(fromBlock, toBlock);
 
   if (logEntries.length === 0) {
-    logger.info(`no entries were found`);
+    logger.info(`no events found in block range ${fromBlock} to ${toBlock}`);
   } else {
-    logger.info(`${logEntries.length} entries were found`);
+    logger.info(
+      `${logEntries.length} events found in block range ${fromBlock} to ${toBlock}`,
+    );
   }
 
   // Persistence operations are wrapped in a transaction to ensure consistency
@@ -55,34 +55,34 @@ const extractEventsForProtocol = async (protocolVersion, extractorConfig) => {
         type: logEntry.event,
       }));
 
-      await withTimer(logger, `persist ${events.length} events`, async () => {
-        await Event.insertMany(events, { session });
-      });
+      await Event.insertMany(events, { session });
     }
 
     // Log details of the queried block range so that we know where
     // to start from in the next iteration.
-    await withTimer(logger, 'log queried block range', async () => {
-      await BlockRange.findOneAndUpdate(
-        { fromBlock, protocolVersion, toBlock },
-        {
-          $set: {
-            date: new Date(),
-            events: logEntries.length,
-            fromBlock,
-            protocolVersion,
-            toBlock,
-          },
-        },
-        {
-          upsert: true,
-          new: true,
-          runValidators: true,
-          session,
+    await BlockRange.findOneAndUpdate(
+      { fromBlock, protocolVersion, toBlock },
+      {
+        $set: {
+          date: new Date(),
+          events: logEntries.length,
+          fromBlock,
+          protocolVersion,
+          toBlock,
         },
-      );
-    });
+      },
+      {
+        upsert: true,
+        new: true,
+        runValidators: true,
+        session,
+      },
+    );
   });
+
+  if (logEntries.length > 0) {
+    logger.info(`persisted ${logEntries.length} events to database`);
+  }
 };
 
 const extractEvents = async () => {

packages/core/src/model/index.js

@@ -0,0 +1,9 @@
+const BlockRange = require('./block-range');
+const Event = require('./event');
+
+const init = async () => {
+  BlockRange.createCollection();
+  Event.createCollection();
+};
+
+module.exports = { BlockRange, Event, init };

packages/core/src/util/db.js

@@ -1,20 +1,20 @@
 const mongoose = require('mongoose');
-const signale = require('signale');
 
 const { logError } = require('./error-logger');
-
-const logger = signale.scope('mongodb');
+const { getLogger } = require('./logging');
 
 mongoose.Promise = global.Promise;
 
 module.exports = {
   connect: async connectionString => {
+    const logger = getLogger();
+
     mongoose.connection.on('connecting', () => {
       logger.info('connecting to database');
     });
 
     mongoose.connection.on('connected', () => {
-      logger.success('database connection established');
+      logger.info('database connection established');
     });
 
     mongoose.connection.on('error', err => {

packages/core/src/util/error-logger.js

@@ -1,8 +1,5 @@
 const _ = require('lodash');
 const bugsnag = require('@bugsnag/js');
-const signale = require('signale');
-
-const logger = signale.scope('application');
 
 let bugsnagClient;
 
@@ -11,7 +8,7 @@ const logError = (error, metaData) => {
     bugsnagClient.notify(error, { metaData });
   }
 
-  logger.error(error);
+  console.error(error); // eslint-disable-line no-console
 };
 
 const configure = ({ appVersion, bugsnagToken }) => {
@@ -23,8 +20,8 @@ const configure = ({ appVersion, bugsnagToken }) => {
       appVersion,
     });
   } else {
-    process.on('uncaughtException', logger.error);
-    process.on('unhandledRejection', logger.error);
+    process.on('uncaughtException', console.error); // eslint-disable-line no-console
+    process.on('unhandledRejection', console.error); // eslint-disable-line no-console
   }
 };
 

packages/core/src/util/logging.js

@@ -0,0 +1,48 @@
+const _ = require('lodash');
+const os = require('os');
+const pino = require('pino');
+const pinoElastic = require('pino-elasticsearch');
+
+let logger;
+
+const createIndexFormatter = prefix => logTime => {
+  const date = new Date(logTime);
+  const year = date.getUTCFullYear();
+  const month = _.padStart(date.getUTCMonth() + 1, 2, 0);
+  const day = _.padStart(date.getUTCDate(), 2, 0);
+
+  // Log indexes roll over every day
+  return `${prefix}_${year}_${month}_${day}`;
+};
+
+const init = config => {
+  const streamToElasticsearch =
+    config.elasticsearch.url === null || config.elasticsearch.url.length === 0
+      ? undefined
+      : pinoElastic({
+          'bulk-size': config.elasticsearch.batchSize,
+          consistency: 'one',
+          ecs: true,
+          index: createIndexFormatter(config.elasticsearch.index),
+          node: config.elasticsearch.url,
+          type: 'log',
+        });
+
+  logger = pino(
+    {
+      base: { group: 'application', pid: process.pid, hostname: os.hostname() },
+      level: 'info',
+    },
+    streamToElasticsearch,
+  );
+};
+
+const getLogger = logGroup => {
+  if (logGroup !== undefined) {
+    return logger.child({ group: logGroup });
+  }
+
+  return logger;
+};
+
+module.exports = { getLogger, init };