feat: add structured logging

package.json

@@ -113,7 +113,8 @@
     "ua-parser-js": "^1.0.35",
     "unset-value": "^2.0.1",
     "uuid": "^9.0.0",
-    "valid-url": "^1.0.9"
+    "valid-url": "^1.0.9",
+    "winston": "^3.11.0"
   },
   "devDependencies": {
     "@commitlint/config-conventional": "^17.6.3",

src/controllers/bulkUpload.ts

@@ -1,6 +1,6 @@
 /* eslint-disable global-require, import/no-dynamic-require, @typescript-eslint/no-unused-vars */
 import { client as errNotificationClient } from '../util/errorNotifier';
-import logger from '../logger';
+import logger from '../logger/structured-logger';
 import { CatchErr, ContextBodySimple } from '../util/types';
 // TODO: To be refactored and redisgned
 

src/controllers/delivery.ts

@@ -12,7 +12,7 @@ import {
 import { ServiceSelector } from '../helpers/serviceSelector';
 import { DeliveryTestService } from '../services/delivertTest/deliveryTest';
 import { ControllerUtility } from './util';
-import logger from '../logger';
+import logger from '../logger/structured-logger';
 import { DestinationPostTransformationService } from '../services/destination/postTransformation';
 import tags from '../v0/util/tags';
 import { FixMe } from '../util/types';

src/controllers/destination.ts

@@ -11,7 +11,7 @@ import {
 import { ServiceSelector } from '../helpers/serviceSelector';
 import { ControllerUtility } from './util';
 import stats from '../util/stats';
-import logger from '../logger';
+import logger from '../logger/structured-logger';
 import { getIntegrationVersion } from '../util/utils';
 import tags from '../v0/util/tags';
 import { DynamicConfigParser } from '../util/dynamicConfigParser';

src/controllers/regulation.ts

@@ -1,5 +1,5 @@
 import { Context } from 'koa';
-import logger from '../logger';
+import logger from '../logger/structured-logger';
 import { UserDeletionRequest, UserDeletionResponse } from '../types';
 import { ServiceSelector } from '../helpers/serviceSelector';
 import tags from '../v0/util/tags';

src/controllers/source.ts

@@ -2,7 +2,7 @@ import { Context } from 'koa';
 import { MiscService } from '../services/misc';
 import { ServiceSelector } from '../helpers/serviceSelector';
 import { ControllerUtility } from './util';
-import logger from '../logger';
+import logger from '../logger/structured-logger';
 import { SourcePostTransformationService } from '../services/source/postTransformation';
 
 export class SourceController {

src/index.ts

@@ -2,7 +2,7 @@ import Koa from 'koa';
 import bodyParser from 'koa-bodyparser';
 import gracefulShutdown from 'http-graceful-shutdown';
 import dotenv from 'dotenv';
-import logger from './logger';
+import logger from './logger/structured-logger';
 import cluster from './util/cluster';
 import { metricsRouter } from './routes/metricsRouter';
 import { addStatMiddleware, addRequestSizeMiddleware, initPyroscope } from './middleware';

src/logger/structured-logger.ts

@@ -0,0 +1,90 @@
+import { createLogger, transports, format } from 'winston';
+import { LoggableExtraData } from '../types';
+
+const { timestamp, combine, metadata, printf, align, errors, json } = format;
+
+const metadataKey = 'extraData';
+
+const { LOG_LEVEL } = process.env;
+
+const envLevelInt = parseInt(LOG_LEVEL ?? '1', 10);
+const winstonConfigLevel = ['debug', 'info', 'warn', 'error'];
+
+const logger = createLogger({
+  level: winstonConfigLevel[envLevelInt],
+  format: combine(
+    timestamp({
+      format: 'MMM-DD-YYYY HH:mm:ss',
+    }),
+    errors({ stack: true }),
+    align(),
+    metadata({
+      key: metadataKey,
+      fillExcept: [
+        'timestamp',
+        'level',
+        'message',
+        'stack',
+        'destinationResponse',
+        'authErrorCategory',
+      ],
+    }),
+    printf((info) => {
+      let extraData = '';
+      if (info?.[metadataKey]) {
+        extraData += ` ${JSON.stringify(info[metadataKey])}`;
+      }
+      return `${info.timestamp} ${info.level}: ${info.message}${extraData}`;
+    }),
+    json(),
+  ),
+  transports: [new transports.Console()],
+});
+
+interface LeveledLogMethod {
+  (message: string, extraData?: Partial<LoggableExtraData>): void;
+  (infoObject: { message: string } & Partial<LoggableExtraData>): void;
+}
+
+interface CustLogger {
+  errorw: LeveledLogMethod;
+  infow: LeveledLogMethod;
+  debugw: LeveledLogMethod;
+  warnw: LeveledLogMethod;
+}
+
+const objectLogger: CustLogger = Object.entries({
+  debugw: 'debug',
+  infow: 'info',
+  warnw: 'warning',
+  errorw: 'error',
+}).reduce(
+  (agg, curr) => {
+    const [custLogMethod, winstonLogLevel] = curr;
+
+    const method = (
+      msg: string | ({ message: string } & Partial<LoggableExtraData>),
+      ex?: Partial<LoggableExtraData>,
+    ) => {
+      const loggableExtraData: Partial<LoggableExtraData> = {
+        ...(ex?.destinationId && { destinationId: ex.destinationId }),
+        ...(ex?.workspaceId && { workspaceId: ex.workspaceId }),
+        ...(ex?.destType && { destType: ex.destType }),
+        ...(ex?.sourceId && { sourceId: ex.sourceId }),
+        ...(ex?.module && { module: ex.module }),
+        ...(ex?.implementation && { implementation: ex.implementation }),
+      };
+      if (typeof msg === 'object') {
+        logger.log(winstonLogLevel, msg.message, loggableExtraData);
+        return;
+      }
+      logger.log(winstonLogLevel, msg, loggableExtraData);
+    };
+    return { ...agg, [custLogMethod]: method };
+  },
+  { errorw: () => {}, infow: () => {}, debugw: () => {}, warnw: () => {} },
+);
+
+const customLogger = Object.assign(logger, objectLogger);
+
+export default customLogger;

src/services/destination/cdkV2Integration.ts

@@ -22,6 +22,7 @@ import tags from '../../v0/util/tags';
 import { DestinationPostTransformationService } from './postTransformation';
 import stats from '../../util/stats';
 import { CatchErr } from '../../util/types';
+import { MiscService } from '../misc';
 
 export class CDKV2DestinationService implements DestinationService {
   public init() {}
@@ -59,14 +60,25 @@ export class CDKV2DestinationService implements DestinationService {
     // TODO: Change the promise type
     const respList: ProcessorTransformationResponse[][] = await Promise.all(
       events.map(async (event) => {
+        const metaTo = this.getTags(
+          destinationType,
+          event.metadata.destinationId,
+          event.metadata.workspaceId,
+          tags.FEATURES.PROCESSOR,
+        );
+        metaTo.metadata = event.metadata;
+        const metadataWithSvcCtx = {
+          ...requestMetadata,
+          serviceContext: MiscService.getLoggableData(metaTo.errorDetails),
+        };
         try {
           const transformedPayloads:
             | ProcessorTransformationOutput
             | ProcessorTransformationOutput[] = await processCdkV2Workflow(
             destinationType,
             event,
             tags.FEATURES.PROCESSOR,
-            requestMetadata,
+            metadataWithSvcCtx,
           );
 
           stats.increment('event_transform_success', {
@@ -85,13 +97,6 @@ export class CDKV2DestinationService implements DestinationService {
             undefined,
           );
         } catch (error: CatchErr) {
-          const metaTo = this.getTags(
-            destinationType,
-            event.metadata.destinationId,
-            event.metadata.workspaceId,
-            tags.FEATURES.PROCESSOR,
-          );
-          metaTo.metadata = event.metadata;
           const erroredResp =
             DestinationPostTransformationService.handleProcessorTransformFailureEvents(
               error,
@@ -127,13 +132,17 @@ export class CDKV2DestinationService implements DestinationService {
             tags.FEATURES.ROUTER,
           );
           metaTo.metadata = destInputArray[0].metadata;
+          const metadataWithSvcCtx = {
+            ...requestMetadata,
+            serviceContext: MiscService.getLoggableData(metaTo.errorDetails),
+          };
           try {
             const doRouterTransformationResponse: RouterTransformationResponse[] =
               await processCdkV2Workflow(
                 destinationType,
                 destInputArray,
                 tags.FEATURES.ROUTER,
-                requestMetadata,
+                metadataWithSvcCtx,
               );
             return DestinationPostTransformationService.handleRouterTransformSuccessEvents(
               doRouterTransformationResponse,

src/services/destination/nativeIntegration.ts

@@ -26,6 +26,7 @@ import networkHandlerFactory from '../../adapters/networkHandlerFactory';
 import { FetchHandler } from '../../helpers/fetchHandlers';
 import tags from '../../v0/util/tags';
 import stats from '../../util/stats';
+import { MiscService } from '../misc';
 
 export class NativeIntegrationDestinationService implements DestinationService {
   public init() {}
@@ -63,23 +64,30 @@ export class NativeIntegrationDestinationService implements DestinationService {
     const destHandler = FetchHandler.getDestHandler(destinationType, version);
     const respList: ProcessorTransformationResponse[][] = await Promise.all(
       events.map(async (event) => {
+        const metaTO = this.getTags(
+          destinationType,
+          event.metadata?.destinationId,
+          event.metadata?.workspaceId,
+          tags.FEATURES.PROCESSOR,
+        );
+        metaTO.metadata = event.metadata;
+        const metadataWithSvcCtx = {
+          ...requestMetadata,
+          serviceContext: MiscService.getLoggableData(metaTO.errorDetails),
+        };
         try {
           const transformedPayloads:
             | ProcessorTransformationOutput
-            | ProcessorTransformationOutput[] = await destHandler.process(event, requestMetadata);
+            | ProcessorTransformationOutput[] = await destHandler.process(
+            event,
+            metadataWithSvcCtx,
+          );
           return DestinationPostTransformationService.handleProcessorTransformSucessEvents(
             event,
             transformedPayloads,
             destHandler,
           );
         } catch (error: any) {
-          const metaTO = this.getTags(
-            destinationType,
-            event.metadata?.destinationId,
-            event.metadata?.workspaceId,
-            tags.FEATURES.PROCESSOR,
-          );
-          metaTO.metadata = event.metadata;
           const erroredResp =
             DestinationPostTransformationService.handleProcessorTransformFailureEvents(
               error,
@@ -113,8 +121,12 @@ export class NativeIntegrationDestinationService implements DestinationService {
           tags.FEATURES.ROUTER,
         );
         try {
+          const metadataWithSvcCtx = {
+            ...requestMetadata,
+            serviceContext: MiscService.getLoggableData(metaTO.errorDetails),
+          };
           const doRouterTransformationResponse: RouterTransformationResponse[] =
-            await destHandler.processRouterDest(cloneDeep(destInputArray), requestMetadata);
+            await destHandler.processRouterDest(cloneDeep(destInputArray), metadataWithSvcCtx);
           metaTO.metadata = destInputArray[0].metadata;
           return DestinationPostTransformationService.handleRouterTransformSuccessEvents(
             doRouterTransformationResponse,
@@ -152,20 +164,24 @@ export class NativeIntegrationDestinationService implements DestinationService {
     );
     const groupedEvents: RouterTransformationRequestData[][] = Object.values(allDestEvents);
     const response = groupedEvents.map((destEvents) => {
+      const metaTO = this.getTags(
+        destinationType,
+        destEvents[0].metadata.destinationId,
+        destEvents[0].metadata.workspaceId,
+        tags.FEATURES.BATCH,
+      );
+      metaTO.metadatas = events.map((event) => event.metadata);
+      const metadataWithSvcCtx = {
+        ...requestMetadata,
+        serviceContext: MiscService.getLoggableData(metaTO.errorDetails),
+      };
       try {
         const destBatchedRequests: RouterTransformationResponse[] = destHandler.batch(
           destEvents,
-          requestMetadata,
+          metadataWithSvcCtx,
         );
         return destBatchedRequests;
       } catch (error: any) {
-        const metaTO = this.getTags(
-          destinationType,
-          destEvents[0].metadata.destinationId,
-          destEvents[0].metadata.workspaceId,
-          tags.FEATURES.BATCH,
-        );
-        metaTO.metadatas = events.map((event) => event.metadata);
         const errResp = DestinationPostTransformationService.handleBatchTransformFailureEvents(
           error,
           metaTO,
@@ -262,10 +278,12 @@ export class NativeIntegrationDestinationService implements DestinationService {
             error: `${destType}: Doesn't support deletion of users`,
           } as UserDeletionResponse;
         }
+        const metaTO = this.getTags(destType, 'unknown', 'unknown', tags.FEATURES.USER_DELETION);
         try {
           const result: UserDeletionResponse = await destUserDeletionHandler.processDeleteUsers({
             ...request,
             rudderDestInfo,
+            serviceContext: MiscService.getLoggableData(metaTO.errorDetails),
           });
           stats.timing('regulation_worker_requests_dest_latency', startTime, {
             feature: tags.FEATURES.USER_DELETION,
@@ -274,7 +292,6 @@ export class NativeIntegrationDestinationService implements DestinationService {
           });
           return result;
         } catch (error: any) {
-          const metaTO = this.getTags(destType, 'unknown', 'unknown', tags.FEATURES.USER_DELETION);
           return DestinationPostTransformationService.handleUserDeletionFailureEvents(
             error,
             metaTO,