Merge pull request #205 from internxt/feat/add-better-observability

feat: add requestId and clientId to logs

Author: apsantiso

lib/core/buckets/usecase.ts

@@ -1,6 +1,7 @@
 import axios, { AxiosError } from 'axios';
 import { v4 } from 'uuid';
 import lodash from 'lodash';
+import log from '../../logger';
 
 import { Bucket } from './Bucket';
 import { Frame } from '../frames/Frame';
@@ -240,7 +241,7 @@ export class BucketsUsecase {
     });
 
     if (!bucketEntry) {
-      console.warn(`[getFileInfo] Bucket entry not found in DB - bucketId: ${bucketId}, fileId: ${fileId}`);
+      log.warn(`[getFileInfo] Bucket entry not found in DB - bucketId: ${bucketId}, fileId: ${fileId}`);
       throw new BucketEntryNotFoundError(fileId);
     }
 
@@ -253,7 +254,7 @@ export class BucketsUsecase {
         if (axios.isAxiosError(error)) {
           const axiosError = error as AxiosError;
           if (axiosError.response?.status === 404) {
-            console.warn(`[getFileInfo] Download links returned 404 - bucketEntryId: ${bucketEntry.id}, fileId: ${fileId}, url: ${axiosError.config?.url}`);
+            log.warn(`[getFileInfo] Download links returned 404 - bucketEntryId: ${bucketEntry.id}, fileId: ${fileId}, url: ${axiosError.config?.url}`);
             throw new BucketEntryNotFoundError(fileId);
           }
         }
@@ -597,7 +598,7 @@ export class BucketsUsecase {
       .deleteManyByUuids(uploads.map((u) => u.uuid))
       .catch((err) => {
       // TODO: Move to EventBus
-        console.log(
+        log.info(
           'completeUpload/uploads-deletion: Failed due to %s. %s',
           err.message,
           err.stack
@@ -680,21 +681,23 @@ export class BucketsUsecase {
       } else {
         await this.validateObjectInStorage(contact, uuid, data_size).catch((error) => {
           if (error instanceof UploadSizeDoesNotMatchError) {
-            console.error(`[finishUpload][SizeDoesNotMatchError] Error size match ${JSON.stringify({ uuid, expectedSize: data_size, contactId: contact.id, message: error.message })}`);
+            log.error(`[finishUpload][SizeDoesNotMatchError] ${JSON.stringify({ uuid, expectedSize: data_size, contactId: contact.id, message: error.message, isMultipartUpload })}`);
+            return;
           }
 
           if (error instanceof UploadNotFoundInStorageError) {
-            console.error(`[finishUpload][UploadNotFoundInStorageError] Error getting bucket meta ${JSON.stringify({ uuid, contactId: contact.id, error: error.message, stack: error.stack })}`);
+            log.error(`[finishUpload][UploadNotFoundInStorageError] ${JSON.stringify({ uuid, contactId: contact.id, error: error.message, stack: error.stack, isMultipartUpload, size: data_size })}`);
+            return;
           }
 
-          console.error(`[finishUpload][unexpectedError] Error getting bucket meta ${JSON.stringify({ uuid, contactId: contact.id, error: error.message, stack: error.stack })}`);
+          log.error(`[finishUpload][unexpectedError] Error getting bucket meta ${JSON.stringify({ uuid, contactId: contact.id, error: error.message, stack: error.stack })}`);
         });
         contactsThatStoreTheShard.push(contact);
       }
     }
 
     if (contactsThatStoreTheShard.length === 0) {
-      console.log('createShardAndMirrors | object %s NOT FOUND in contacts', uuid, contracts.map(c => c.nodeID));
+      log.error('createShardAndMirrors | object %s NOT FOUND in contacts', uuid, contracts.map(c => c.nodeID));
       throw new Error('Shard not uploaded');
     }
 

lib/engine.js

@@ -19,6 +19,8 @@ const MongoDBStorageAdapter = require('storj-mongodb-adapter');
 const NetworkMessageQueue = require('./server/queues/networkQueue');
 const { bindNewRoutes } = require('./server/http');
 const { json } = require('express');
+const { store: requestContextStore } = require('./requestContext');
+const { requestLogger, unexpectedErrorLogger } = require('./server/middleware/requestLogger');
 const { Notifications } = require('./server/notifications');
 const bullQueueModule = require('./core/queue/bullQueue');
 
@@ -224,49 +226,34 @@ Engine.prototype._configureApp = function () {
 
   function bindRoute(route) {
     let verb = route.shift().toLowerCase();
-    app[verb].apply(app, route);
+    const path = route.shift();
+    const handler = route.pop();
+
+    // Re-establish AsyncLocalStorage context for the
+    // final handler. Auth middleware (storj-service-middleware)
+    // calls next() from a MongoDB callback, which breaks
+    // the async context chain.
+    const wrappedHandler = (req, res, next) => {
+      const ctx = res.locals.requestContext;
+      if (ctx) {
+        requestContextStore.run(ctx, () => handler(req, res, next));
+      } else {
+        handler(req, res, next);
+      }
+    };
+    app[verb].apply(app, [path, ...route, wrappedHandler]);
   }
 
   self._keepPendingResponsesClean();
 
-  const getDurationInMilliseconds = (start) => {
-    const NS_PER_SEC = 1e9;
-    const NS_TO_MS = 1e6;
-    const diff = process.hrtime(start);
-
-    return (diff[0] * NS_PER_SEC + diff[1]) / NS_TO_MS;
-  };
-
-  app.use((req, res, next) => {
-    log.info(`${req.method} ${req.originalUrl} [STARTED]`);
-    const start = process.hrtime();
-
-    res.on('finish', () => {
-      const durationInMilliseconds = getDurationInMilliseconds(start);
-      log.info(
-        `${req.method} ${
-          req.originalUrl
-        } [FINISHED] ${durationInMilliseconds.toLocaleString()} ms`
-      );
-    });
-
-    res.on('close', () => {
-      const durationInMilliseconds = getDurationInMilliseconds(start);
-      log.info(
-        `${req.method} ${
-          req.originalUrl
-        } [CLOSED] ${durationInMilliseconds.toLocaleString()} ms`
-      );
-    });
-
-    next();
-  });
+  app.use(requestLogger);
   app.use(middleware.querystring);
   app.use(this._trackResponseStatus.bind(this));
   app.use(crossorigin(corsOptions));
   app.use(helmet());
   app.get('/', this._handleRootGET.bind(this));
   routers.forEach(bindRoute);
+  app.use(unexpectedErrorLogger);
   app.use(middleware.errorhandler({ logger: log }));
 
   app.use(json());

lib/logger.js

@@ -6,6 +6,7 @@
 
 const config = require('./config')(process.env.NODE_ENV);
 const CONSTANTS = require('./constants');
+const { getContext } = require('./requestContext');
 
 const Winston = require('winston');
 
@@ -48,12 +49,18 @@ module.exports = (() => {
       Winston.format.timestamp({ format: 'YYYY-MM-DD HH:mm:ss' }),
       Winston.format.splat(),
       Winston.format.printf(info => {
-        return `${info.timestamp} ${info.level}: ${info.message}`;
+        const { requestId, clientId } = getContext();
+
+        return `${info.timestamp} ${info.level} [${requestId || '-'}] [${clientId || '-'}]: ${info.message}`;
       })
     ) :
     Winston.format.combine(
       Winston.format.splat(),
-      Winston.format.printf(info => info.message)
+      Winston.format.printf(info => {
+        const { requestId, clientId } = getContext();
+
+        return `[${requestId || '-'}] [${clientId || '-'}] ${info.message}`;
+      })
     );
 
   return Winston.createLogger(

lib/requestContext.ts

@@ -0,0 +1,14 @@
+import { AsyncLocalStorage } from 'async_hooks';
+
+interface RequestContext {
+  requestId: string;
+  clientId: string | undefined;
+}
+
+const store = new AsyncLocalStorage<RequestContext>();
+
+function getContext(): Partial<RequestContext> {
+  return store.getStore() ?? {};
+}
+
+export { store, getContext };

lib/server/middleware/requestLogger.ts

@@ -0,0 +1,47 @@
+import { Request, Response, NextFunction } from 'express';
+import { v4 as uuidv4 } from 'uuid';
+import { store as requestContextStore } from '../../requestContext';
+import log from '../../logger';
+
+const getDurationInMilliseconds = (start: [number, number]): number => {
+  const NS_PER_SEC = 1e9;
+  const NS_TO_MS = 1e6;
+  const diff = process.hrtime(start);
+
+  return (diff[0] * NS_PER_SEC + diff[1]) / NS_TO_MS;
+};
+
+export function requestLogger(req: Request, res: Response, next: NextFunction): void {
+  const ctx = {
+    requestId: uuidv4(),
+    clientId: req.headers['internxt-client'] as string | undefined,
+  };
+
+  requestContextStore.run(ctx, () => {
+    res.locals.requestContext = ctx;
+    res.setHeader('X-Request-Id', ctx.requestId);
+    log.info(`${req.method} ${req.originalUrl} [STARTED]`);
+    const start = process.hrtime();
+
+    res.on('finish', () => {
+      requestContextStore.run(ctx, () => {
+        const durationInMilliseconds = getDurationInMilliseconds(start);
+        log.info(`${req.method} ${req.originalUrl} [FINISHED] ${res.statusCode} ${durationInMilliseconds.toLocaleString()} ms`);
+      });
+    });
+
+    res.on('close', () => {
+      requestContextStore.run(ctx, () => {
+        const durationInMilliseconds = getDurationInMilliseconds(start);
+        log.info(`${req.method} ${req.originalUrl} [CLOSED] ${res.statusCode} ${durationInMilliseconds.toLocaleString()} ms`);
+      });
+    });
+
+    next();
+  });
+}
+
+export function unexpectedErrorLogger(err: Error, req: Request, res: Response, next: NextFunction): void {
+  log.error(`${req.method} ${req.originalUrl} [UNEXPECTED_ERROR] ${err.message}`, err.stack);
+  next(err);
+}