🔊 back: use sub loggers for platform and message queue logging

tdrive/backend/node/src/core/platform/framework/api/component.ts

@@ -3,7 +3,7 @@ import { TdriveService } from "./service";
 import { TdriveServiceProvider } from "./service-provider";
 import { ServiceDefinition } from "./service-definition";
 import { TdriveServiceState } from "./service-state";
-import { logger } from "../logger";
+import { platformLogger } from "../logger";
 
 export class TdriveComponent {
   instance: TdriveService<TdriveServiceProvider>;
@@ -38,7 +38,7 @@ export class TdriveComponent {
     recursionDepth?: number,
   ): Promise<void> {
     if (recursionDepth > 10) {
-      logger.error("Maximum recursion depth exceeded (will exit process)");
+      platformLogger.error("Maximum recursion depth exceeded (will exit process)");
       process.exit(1);
     }
 
@@ -50,10 +50,10 @@ export class TdriveComponent {
       for (const component of this.components) {
         await component.switchToState(state, (recursionDepth || 0) + 1);
       }
-      logger.info(`Children of ${this.name} are all in ${state} state`);
+      platformLogger.info(`Children of ${this.name} are all in ${state} state`);
-      logger.info(this.getStateTree());
+      platformLogger.info(this.getStateTree());
     } else {
-      logger.info(`${this.name} does not have children`);
+      platformLogger.info(`${this.name} does not have children`);
     }
 
     // eslint-disable-next-line @typescript-eslint/no-explicit-any

tdrive/backend/node/src/core/platform/framework/api/service.ts

@@ -6,7 +6,7 @@ import { TdriveServiceConfiguration } from "./service-configuration";
 import { TdriveContext } from "./context";
 import { TdriveServiceOptions } from "./service-options";
 import { CONSUMES_METADATA, PREFIX_METADATA } from "./constants";
-import { getLogger, logger } from "../logger";
+import { getLogger, platformLogger } from "../logger";
 import { TdriveLogger } from "..";
 
 const pendingServices: any = {};
@@ -39,23 +39,23 @@ export abstract class TdriveService<T extends TdriveServiceProvider>
 
   async init(): Promise<this> {
     if (this.state.value !== TdriveServiceState.Ready) {
-      logger.info("Service %s is already initialized", this.name);
+      platformLogger.info("Service %s is already initialized", this.name);
       return this;
     }
 
     try {
-      logger.info("Initializing service %s", this.name);
+      platformLogger.info("Initializing service %s", this.name);
       pendingServices[this.name] = true;
       this.state.next(TdriveServiceState.Initializing);
       await this.doInit();
       this.state.next(TdriveServiceState.Initialized);
-      logger.info("Service %s is initialized", this.name);
+      platformLogger.info("Service %s is initialized", this.name);
       delete pendingServices[this.name];
-      logger.info("Pending services: %s", JSON.stringify(Object.keys(pendingServices)));
+      platformLogger.info("Pending services: %s", JSON.stringify(Object.keys(pendingServices)));
       return this;
     } catch (err) {
-      logger.error("Error while initializing service %s", this.name);
+      platformLogger.error("Error while initializing service %s", this.name);
-      logger.error(err);
+      platformLogger.error(err);
       this.state.error(new Error(`Error while initializing service ${this.name}`));
 
       throw err;
@@ -75,21 +75,21 @@ export abstract class TdriveService<T extends TdriveServiceProvider>
       this.state.value === TdriveServiceState.Starting ||
       this.state.value === TdriveServiceState.Started
     ) {
-      logger.info("Service %s is already started", this.name);
+      platformLogger.info("Service %s is already started", this.name);
       return this;
     }
 
     try {
-      logger.info("Starting service %s", this.name);
+      platformLogger.info("Starting service %s", this.name);
       this.state.next(TdriveServiceState.Starting);
       await this.doStart();
       this.state.next(TdriveServiceState.Started);
-      logger.info("Service %s is started", this.name);
+      platformLogger.info("Service %s is started", this.name);
 
       return this;
     } catch (err) {
-      logger.error("Error while starting service %s", this.name, err);
+      platformLogger.error("Error while starting service %s", this.name, err);
-      logger.error(err);
+      platformLogger.error(err);
       this.state.error(new Error(`Error while starting service ${this.name}`));
 
       throw err;
@@ -101,26 +101,26 @@ export abstract class TdriveService<T extends TdriveServiceProvider>
       this.state.value === TdriveServiceState.Stopping ||
       this.state.value === TdriveServiceState.Stopped
     ) {
-      logger.info("Service %s is already stopped", this.name);
+      platformLogger.info("Service %s is already stopped", this.name);
       return this;
     }
 
     if (this.state.value !== TdriveServiceState.Started) {
-      logger.info("Service %s can not be stopped until started", this.name);
+      platformLogger.info("Service %s can not be stopped until started", this.name);
       return this;
     }
 
     try {
-      logger.info("Stopping service %s", this.name);
+      platformLogger.info("Stopping service %s", this.name);
       this.state.next(TdriveServiceState.Stopping);
       await this.doStop();
       this.state.next(TdriveServiceState.Stopped);
-      logger.info("Service %s is stopped", this.name);
+      platformLogger.info("Service %s is stopped", this.name);
 
       return this;
     } catch (err) {
-      logger.error("Error while stopping service %s", this.name, err);
+      platformLogger.error("Error while stopping service %s", this.name, err);
-      logger.error(err);
+      platformLogger.error(err);
       this.state.error(new Error(`Error while stopping service ${this.name}`));
 
       throw err;

tdrive/backend/node/src/core/platform/framework/logger.ts

@@ -23,3 +23,7 @@ export const logger = pino({
 
 export const getLogger = (name?: string): TdriveLogger =>
   logger.child({ name: `tdrive${name ? "." + name : ""}` });
+
+export const platformLogger = getLogger("platform");
+
+export const messageQueueLogger = getLogger("message-queue");

tdrive/backend/node/src/core/platform/framework/utils/component-utils.ts

@@ -1,6 +1,6 @@
 /* eslint-disable @typescript-eslint/no-explicit-any, @typescript-eslint/explicit-module-boundary-types */
 import {
-  logger,
+  platformLogger,
   TdriveComponent,
   TdriveContext,
   TdriveServiceFactory,
@@ -28,7 +28,7 @@ export async function buildDependenciesTree(
             `The component dependency ${dependencyName} has not been found for component ${name}`,
           );
         } else {
-          logger.warn(
+          platformLogger.warn(
             `(warning) The component dependency ${dependencyName} has not been found for component ${name} it will be imported asynchronously`,
           );
 
@@ -99,10 +99,10 @@ export async function switchComponentsToState(
   const states = [];
 
   for (const [name, component] of components) {
-    logger.info(`Asking for ${state} on ${name} dependencies`);
+    platformLogger.info(`Asking for ${state} on ${name} dependencies`);
     states.push(component.getServiceInstance().state);
     await component.switchToState(state);
   }
 
-  logger.info(`All components are now in ${state} state`);
+  platformLogger.info(`All components are now in ${state} state`);
 }

tdrive/backend/node/src/core/platform/framework/utils/loader.ts

@@ -1,4 +1,4 @@
-import { logger } from "../logger";
+import { platformLogger } from "../logger";
 import fs from "fs";
 
 export class Loader {
@@ -14,7 +14,7 @@ export class Loader {
           try {
             return await import(modulePath);
           } catch (err) {
-            logger.debug(
+            platformLogger.debug(
               { err },
               `${modulePath} can not be loaded (file was found but we were unable to import the module)`,
             );
@@ -28,7 +28,9 @@ export class Loader {
     if (!classes || !classes.length) {
       modulesPaths.map(modulePath => {
         if (fs.existsSync(modulePath)) {
-          logger.debug(`${modulePath} content was: [${fs.readdirSync(modulePath).join(", ")}]`);
+          platformLogger.debug(
+            `${modulePath} content was: [${fs.readdirSync(modulePath).join(", ")}]`,
+          );
         }
       });
       throw new Error(
@@ -38,7 +40,7 @@ export class Loader {
       );
     }
 
-    logger.debug(`Loaded ${componentName}`);
+    platformLogger.debug(`Loaded ${componentName}`);
 
     return classes[0].default;
   }

tdrive/backend/node/src/core/platform/services/message-queue/api.ts

@@ -1,6 +1,6 @@
 import { Subject } from "rxjs";
 import { v4 as uuidv4 } from "uuid";
-import { Initializable, logger, TdriveServiceProvider } from "../../framework";
+import { Initializable, messageQueueLogger, TdriveServiceProvider } from "../../framework";
 import { Processor } from "./processor";
 import { ExecutionContext } from "../../framework/api/crud-service";
 
@@ -187,7 +187,7 @@ export class MessageQueueServiceProcessor<In, Out>
     try {
       await this.subscribe(this.messageQueue);
     } catch (err) {
-      logger.warn(
+      messageQueueLogger.warn(
         { err },
         `MessageQueueServiceProcessor.handler.${this.handler.name} -  Not able to start handler`,
       );
@@ -202,7 +202,7 @@ export class MessageQueueServiceProcessor<In, Out>
   }
 
   async process(message: IncomingMessageQueueMessage<In>): Promise<Out> {
-    logger.info(
+    messageQueueLogger.info(
       `MessageQueueServiceProcessor.handler.${this.handler.name}:${message.id} - Processing message`,
     );
     return this.handler.process(message.data);
@@ -212,7 +212,7 @@ export class MessageQueueServiceProcessor<In, Out>
     //TODO this is where we do not receive the call
 
     if (this.handler.topics && this.handler.topics.in) {
-      logger.info(
+      messageQueueLogger.info(
         `MessageQueueServiceProcessor.handler.${this.handler.name} - Subscribing to topic ${this.handler?.topics?.in} with options %o`,
         this.handler.options,
       );
@@ -233,7 +233,7 @@ export class MessageQueueServiceProcessor<In, Out>
       const isValid = this.handler.validate(message.data);
 
       if (!isValid) {
-        logger.error(
+        messageQueueLogger.error(
           `MessageQueueServiceProcessor.handler.${this.handler.name}:${message.id} - Message is invalid`,
         );
 
@@ -246,7 +246,7 @@ export class MessageQueueServiceProcessor<In, Out>
     try {
       const result = await this.process(message);
       if (this.handler?.options?.ack && message?.ack) {
-        logger.debug(
+        messageQueueLogger.debug(
           `MessageQueueServiceProcessor.handler.${this.handler.name}:${message.id} - Acknowledging message %o`,
           message,
         );
@@ -266,13 +266,13 @@ export class MessageQueueServiceProcessor<In, Out>
 
   private async sendResult(message: IncomingMessageQueueMessage<In>, result: Out): Promise<void> {
     if (!this.handler.topics.out) {
-      logger.info(
+      messageQueueLogger.info(
         `MessageQueueServiceProcessor.handler.${this.handler.name}:${message.id} - Message processing result is skipped`,
       );
       return;
     }
 
-    logger.info(
+    messageQueueLogger.info(
       `MessageQueueServiceProcessor.handler.${this.handler.name}:${message.id} - Sending processing result to ${this.handler.topics.out}`,
     );
 
@@ -284,7 +284,7 @@ export class MessageQueueServiceProcessor<In, Out>
 
   // eslint-disable-next-line @typescript-eslint/no-explicit-any
   private async handleError(message: IncomingMessageQueueMessage<In>, err: any) {
-    logger.error(
+    messageQueueLogger.error(
       { err },
       `MessageQueueServiceProcessor.handler.${this.handler.name}:${message.id} - Error while processing message`,
     );

tdrive/backend/node/src/core/platform/services/message-queue/processor.ts

@@ -1,4 +1,4 @@
-import { logger } from "../../framework";
+import { messageQueueLogger } from "../../framework";
 import { MessageQueueHandler, MessageQueueServiceAPI, MessageQueueServiceProcessor } from "./api";
 
 const LOG_PREFIX = "service.message-queue.Processor";
@@ -20,9 +20,9 @@ export class Processor {
     this.started = true;
     await Promise.all(
       Array.from(this.registry.processors.keys()).map(async name => {
-        logger.info(`${LOG_PREFIX} - Starting notification processor ${name}`);
+        messageQueueLogger.info(`${LOG_PREFIX} - Starting notification processor ${name}`);
         await this.registry.processors.get(name)?.init();
-        logger.info(`${LOG_PREFIX} - notification processor ${name} is started`);
+        messageQueueLogger.info(`${LOG_PREFIX} - notification processor ${name} is started`);
       }),
     );
   }
@@ -41,7 +41,7 @@ export class Processor {
       throw new Error(`${LOG_PREFIX} - Can not add null handler`);
     }
 
-    logger.info(`${LOG_PREFIX} - Adding message-queue handler ${handler.name}`);
+    messageQueueLogger.info(`${LOG_PREFIX} - Adding message-queue handler ${handler.name}`);
     this.registry.register(handler);
 
     if (this.started) {
@@ -50,17 +50,17 @@ export class Processor {
   }
 
   async startHandler(name: string): Promise<void> {
-    logger.info(`${LOG_PREFIX} - Starting message-queue handler ${name}`);
+    messageQueueLogger.info(`${LOG_PREFIX} - Starting message-queue handler ${name}`);
     await this.registry.processors.get(name)?.init();
   }
 
   stopHandler(name: string): void {
-    logger.info(`${LOG_PREFIX} - Stopping message-queue handler ${name}`);
+    messageQueueLogger.info(`${LOG_PREFIX} - Stopping message-queue handler ${name}`);
     this.registry.processors.get(name)?.stop();
   }
 
   removeHandler(name: string): void {
-    logger.info(`${LOG_PREFIX} - Removing message-queue handler ${name}`);
+    messageQueueLogger.info(`${LOG_PREFIX} - Removing message-queue handler ${name}`);
     this.stopHandler(name);
     this.registry.processors.delete(name);
   }

tdrive/backend/node/src/core/platform/services/message-queue/proxy.ts

@@ -1,5 +1,5 @@
 import { isEqual } from "lodash";
-import { logger } from "../../framework";
+import { messageQueueLogger } from "../../framework";
 import {
   MessageQueueClient,
   MessageQueueListener,
@@ -40,7 +40,7 @@ export default class MessageQueueProxyService implements MessageQueueProxy {
   constructor(private client?: MessageQueueClient) {}
 
   async setClient(client: MessageQueueClient): Promise<void> {
-    logger.info(`${LOG_PREFIX} Setting new message-queue client`);
+    messageQueueLogger.info(`${LOG_PREFIX} Setting new message-queue client`);
     // TODO: The client can be removed or replaced while we are looping here
     // We may wait until things are done, or discard some...
     if (!client) {
@@ -48,7 +48,7 @@ export default class MessageQueueProxyService implements MessageQueueProxy {
     }
 
     if (this.client) {
-      logger.info(`${LOG_PREFIX} MessageQueue client already set. Overriding`);
+      messageQueueLogger.info(`${LOG_PREFIX} MessageQueue client already set. Overriding`);
     }
 
     this.client = client;
@@ -59,11 +59,11 @@ export default class MessageQueueProxyService implements MessageQueueProxy {
         const listeners = entries[1];
 
         return Array.from(listeners).map(async listener => {
-          logger.debug(`${LOG_PREFIX} Subscribing to topic ${topic} from cache`);
+          messageQueueLogger.debug(`${LOG_PREFIX} Subscribing to topic ${topic} from cache`);
           try {
             await this.subscribe(topic, listener.listener, listener.options);
           } catch (err) {
-            logger.warn(
+            messageQueueLogger.warn(
               { err },
               `${LOG_PREFIX} Error while subscribing with cached subscription to topic ${topic}`,
             );
@@ -75,10 +75,12 @@ export default class MessageQueueProxyService implements MessageQueueProxy {
     await Promise.all(
       this.publicationsBuffer.map(async publication => {
         try {
-          logger.debug(`${LOG_PREFIX} Publishing to topic ${publication.topic} from cache`);
+          messageQueueLogger.debug(
+            `${LOG_PREFIX} Publishing to topic ${publication.topic} from cache`,
+          );
           await this.publish(publication.topic, publication.message);
         } catch (err) {
-          logger.warn(
+          messageQueueLogger.warn(
             { err },
             `${LOG_PREFIX} Error while publishing cached data on topic ${publication.topic}`,
           );
@@ -101,7 +103,7 @@ export default class MessageQueueProxyService implements MessageQueueProxy {
     try {
       await this.client?.close?.();
     } catch (err) {
-      logger.debug({ err }, `${LOG_PREFIX} Error on closing the message-queue layer`);
+      messageQueueLogger.debug({ err }, `${LOG_PREFIX} Error on closing the message-queue layer`);
     }
   }
 
@@ -139,7 +141,9 @@ export default class MessageQueueProxyService implements MessageQueueProxy {
     }
 
     const subscriptions = this.subscriptionsCache.get(topic);
-    logger.debug(`${LOG_PREFIX} Subscriptions for topic ${topic}: ${subscriptions.size}`);
+    messageQueueLogger.debug(
+      `${LOG_PREFIX} Subscriptions for topic ${topic}: ${subscriptions.size}`,
+    );
 
     const values = [...subscriptions];
     const cachedListener = values.find(
@@ -147,16 +151,16 @@ export default class MessageQueueProxyService implements MessageQueueProxy {
     );
 
     if (!cachedListener) {
-      logger.debug(`${LOG_PREFIX} Caching subscription to ${topic} topic: Yes`);
+      messageQueueLogger.debug(`${LOG_PREFIX} Caching subscription to ${topic} topic: Yes`);
       this.subscriptionsCache.get(topic).add({ listener, options });
     } else {
-      logger.debug(`${LOG_PREFIX} Caching subscription to ${topic} topic: No`);
+      messageQueueLogger.debug(`${LOG_PREFIX} Caching subscription to ${topic} topic: No`);
     }
   }
 
   // eslint-disable-next-line @typescript-eslint/no-explicit-any
   private addPublishCache(topic: string, message: any): void {
-    logger.debug(`${LOG_PREFIX} Caching publication to ${topic} topic`);
+    messageQueueLogger.debug(`${LOG_PREFIX} Caching publication to ${topic} topic`);
     this.publicationsBuffer.push({ topic, message });
   }
 
@@ -166,7 +170,10 @@ export default class MessageQueueProxyService implements MessageQueueProxy {
     listener: MessageQueueListener<any>,
     options?: MessageQueueSubscriptionOptions,
   ): Promise<void> {
-    logger.debug(`${LOG_PREFIX} Trying to subscribe to ${topic} topic with options %o`, options);
+    messageQueueLogger.debug(
+      `${LOG_PREFIX} Trying to subscribe to ${topic} topic with options %o`,
+      options,
+    );
     return this.client?.subscribe(topic, listener, options);
   }
 }