eduvia/workavia-drive
11
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

🔊 back: use sub loggers for platform and message queue logging (#761)

Browse Source 
Move platform and message queue logging to their own loggers for better
filtering in dev
This commit is contained in:
Anton Shepilov
2024-12-19 14:46:52 +01:00
committed by GitHub
parent ffa0902244 b59fc33cc9
commit 946fabbadb
8 changed files with 73 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tdrive/backend/node/src/core/platform/framework/api/component.ts
+5 -5
View File
@@ -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`);
logger.info(this.getStateTree());
platformLogger.info(`Children of ${this.name} are all in ${state} state`);
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
+18 -18
View File
@@ -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);
logger.error(err);
platformLogger.error("Error while initializing service %s", this.name);
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);
logger.error(err);
platformLogger.error("Error while starting service %s", this.name, 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);
logger.error(err);
platformLogger.error("Error while stopping service %s", this.name, 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
+4
View File
@@ -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
+4 -4
View File
@@ -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
+6 -4
View File
@@ -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
+9 -9
View File
@@ -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
+7 -7
View File
@@ -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
+20 -13
View File
@@ -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);
}
}