new connection to monitor client mechanism

Author: Artem

redisinsight/api/config/default.ts

@@ -96,6 +96,7 @@ export default {
     writeKey: process.env.SEGMENT_WRITE_KEY || 'SOURCE_WRITE_KEY',
   },
   logger: {
+    logLevel: process.env.LOG_LEVEL || 'info', // log level
     stdout: process.env.STDOUT_LOGGER ? process.env.STDOUT_LOGGER === 'true' : false, // disabled by default
     files: process.env.FILES_LOGGER ? process.env.FILES_LOGGER === 'true' : true, // enabled by default
     omitSensitiveData: process.env.LOGGER_OMIT_DATA ? process.env.LOGGER_OMIT_DATA === 'true' : true,

redisinsight/api/config/development.ts

@@ -10,6 +10,7 @@ export default {
     migrationsRun: process.env.DB_MIGRATIONS ? process.env.DB_MIGRATIONS === 'true' : false,
   },
   logger: {
+    logLevel: process.env.LOG_LEVEL || 'debug',
     stdout: process.env.STDOUT_LOGGER ? process.env.STDOUT_LOGGER === 'true' : true, // enabled by default
     omitSensitiveData: process.env.LOGGER_OMIT_DATA ? process.env.LOGGER_OMIT_DATA === 'true' : false,
   },

redisinsight/api/config/logger.ts

@@ -58,6 +58,7 @@ if (LOGGER_CONFIG.files) {
 const logger: WinstonModuleOptions = {
   format: format.errors({ stack: true }),
   transports: transportsConfig,
+  level: LOGGER_CONFIG.logLevel,
 };
 
 export default logger;

redisinsight/api/src/__mocks__/monitor.ts

@@ -1,6 +1,7 @@
 import { v4 as uuidv4 } from 'uuid';
 import { IClientMonitorObserver } from 'src/modules/profiler/helpers/client-monitor-observer';
 import { IMonitorObserver, IShardObserver, MonitorObserverStatus } from 'src/modules/profiler/helpers/monitor-observer';
+import { ILogsEmitter } from 'src/modules/profiler/interfaces/logs-emitter.interface';
 
 export const mockClientMonitorObserver: IClientMonitorObserver = {
   id: uuidv4(),
@@ -34,3 +35,11 @@ export const mockRedisMonitorObserver: IShardObserver = {
   once: jest.fn(),
   disconnect: jest.fn(),
 };
+
+export const mockLogEmitter: ILogsEmitter = {
+  id: 'test',
+  emit: jest.fn(),
+  addProfilerClient: jest.fn(),
+  removeProfilerClient: jest.fn(),
+  flushLogs: jest.fn(),
+};

redisinsight/api/src/modules/profiler/constants/index.ts

@@ -10,6 +10,9 @@ export enum ProfilerServerEvents {
 }
 
 export enum RedisObserverStatus {
+  Empty = 'empty',
+  Initializing = 'initializing',
+  Connected = 'connected',
   Wait = 'wait',
   Ready = 'ready',
   End = 'end',

redisinsight/api/src/modules/profiler/models/log-file.ts

@@ -38,7 +38,7 @@ export class LogFile {
     if (!this.writeStream) {
       this.writeStream = fs.createWriteStream(this.filePath, { flags: 'a' });
     }
-
+    this.writeStream.on('error', () => {});
     return this.writeStream;
   }
 

redisinsight/api/src/modules/profiler/models/profiler.client.ts

@@ -1,12 +1,15 @@
 import { Socket } from 'socket.io';
 import { debounce } from 'lodash';
 import { WsException } from '@nestjs/websockets';
+import { Logger } from '@nestjs/common';
 import { ProfilerServerEvents } from 'src/modules/profiler/constants';
 import { ILogsEmitter } from 'src/modules/profiler/interfaces/logs-emitter.interface';
 import { IMonitorData } from 'src/modules/profiler/interfaces/monitor-data.interface';
 import ERROR_MESSAGES from 'src/constants/error-messages';
 
 export class ProfilerClient {
+  private logger = new Logger('ProfilerClient');
+
   public readonly id: string;
 
   private readonly client: Socket;
@@ -57,6 +60,7 @@ export class ProfilerClient {
   public addLogsEmitter(emitter: ILogsEmitter) {
     this.logsEmitters.set(emitter.id, emitter);
     emitter.addProfilerClient(this.id);
+    this.logCurrentState();
   }
 
   async flushLogs() {
@@ -66,4 +70,14 @@ export class ProfilerClient {
   public destroy() {
     this.logsEmitters.forEach((emitter) => emitter.removeProfilerClient(this.id));
   }
+
+  /**
+   * Logs useful information about current state for debug purposes
+   * @private
+   */
+  private logCurrentState() {
+    this.logger.debug(
+      `Emitters: ${this.logsEmitters.size}`,
+    );
+  }
 }

redisinsight/api/src/modules/profiler/models/redis.observer.ts

@@ -1,23 +1,42 @@
 import IORedis from 'ioredis';
-import { ForbiddenException, ServiceUnavailableException } from '@nestjs/common';
+import { ForbiddenException, Logger, ServiceUnavailableException } from '@nestjs/common';
 import { RedisErrorCodes } from 'src/constants';
 import { ProfilerClient } from 'src/modules/profiler/models/profiler.client';
 import { RedisObserverStatus } from 'src/modules/profiler/constants';
 import { IShardObserver } from 'src/modules/profiler/interfaces/shard-observer.interface';
 import ERROR_MESSAGES from 'src/constants/error-messages';
+import { EventEmitter2 } from '@nestjs/event-emitter';
 
-export class RedisObserver {
-  private readonly redis: IORedis.Redis | IORedis.Cluster;
+export class RedisObserver extends EventEmitter2 {
+  private logger = new Logger('RedisObserver');
+
+  private redis: IORedis.Redis | IORedis.Cluster;
 
   private profilerClients: Map<string, ProfilerClient> = new Map();
 
+  private profilerClientsListeners: Map<string, any[]> = new Map();
+
   private shardsObservers: IShardObserver[] = [];
 
   public status: RedisObserverStatus;
 
-  constructor(redis: IORedis.Redis | IORedis.Cluster) {
-    this.redis = redis;
-    this.status = RedisObserverStatus.Wait;
+  constructor() {
+    super();
+    this.status = RedisObserverStatus.Empty;
+  }
+
+  init(func: () => Promise<IORedis.Redis | IORedis.Cluster>) {
+    this.status = RedisObserverStatus.Initializing;
+
+    func()
+      .then((redis) => {
+        this.redis = redis;
+        this.status = RedisObserverStatus.Connected;
+        this.emit('connect');
+      })
+      .catch((err) => {
+        this.emit('connect_error', err);
+      });
   }
 
   /**
@@ -35,36 +54,84 @@ export class RedisObserver {
       return;
     }
 
+    if (!this.profilerClientsListeners.get(profilerClient.id)) {
+      this.profilerClientsListeners.set(profilerClient.id, []);
+    }
+
+    const profilerListeners = this.profilerClientsListeners.get(profilerClient.id);
+
     this.shardsObservers.forEach((observer) => {
-      observer.on('monitor', (time, args, source, database) => {
+      const monitorListenerFn = (time, args, source, database) => {
         profilerClient.handleOnData({
           time, args, database, source, shardOptions: observer.options,
         });
-      });
-      observer.on('end', () => {
+      };
+      const endListenerFn = () => {
         profilerClient.handleOnDisconnect();
         this.clear();
-      });
+      };
+
+      observer.on('monitor', monitorListenerFn);
+      observer.on('end', endListenerFn);
+
+      profilerListeners.push(monitorListenerFn, endListenerFn);
+      this.logger.debug(`Subscribed to shard observer. Current listeners: ${observer.listenerCount('monitor')}`);
     });
     this.profilerClients.set(profilerClient.id, profilerClient);
+
+    this.logger.debug(`Profiler Client with id:${profilerClient.id} was added`);
+    this.logCurrentState();
+  }
+
+  public removeShardsListeners(profilerClientId: string) {
+    this.shardsObservers.forEach((observer) => {
+      (this.profilerClientsListeners.get(profilerClientId) || []).forEach((listener) => {
+        observer.removeListener('monitor', listener);
+        observer.removeListener('end', listener);
+      });
+
+      this.logger.debug(
+        `Unsubscribed from from shard observer. Current listeners: ${observer.listenerCount('monitor')}`,
+      );
+    });
   }
 
   public unsubscribe(id: string) {
+    this.removeShardsListeners(id);
     this.profilerClients.delete(id);
+    this.profilerClientsListeners.delete(id);
     if (this.profilerClients.size === 0) {
       this.clear();
     }
+
+    this.logger.debug(`Profiler Client with id:${id} was unsubscribed`);
+    this.logCurrentState();
   }
 
   public disconnect(id: string) {
-    const userClient = this.profilerClients.get(id);
-    if (userClient) {
-      userClient.destroy();
+    this.removeShardsListeners(id);
+    const profilerClient = this.profilerClients.get(id);
+    if (profilerClient) {
+      profilerClient.destroy();
     }
     this.profilerClients.delete(id);
+    this.profilerClientsListeners.delete(id);
     if (this.profilerClients.size === 0) {
       this.clear();
     }
+
+    this.logger.debug(`Profiler Client with id:${id} was disconnected`);
+    this.logCurrentState();
+  }
+
+  /**
+   * Logs useful inforation about current state for debug purposes
+   * @private
+   */
+  private logCurrentState() {
+    this.logger.debug(
+      `Status: ${this.status}; Shards: ${this.shardsObservers.length}; Listeners: ${this.getProfilerClientsSize()}`,
+    );
   }
 
   public clear() {
@@ -98,6 +165,13 @@ export class RedisObserver {
       } else {
         this.shardsObservers = [await RedisObserver.createShardObserver(this.redis)];
       }
+
+      this.shardsObservers.forEach((observer) => {
+        observer.on('error', (e) => {
+          this.logger.error('Error on shard observer', e);
+        });
+      });
+
       this.status = RedisObserverStatus.Ready;
     } catch (error) {
       this.status = RedisObserverStatus.Error;
@@ -130,6 +204,7 @@ export class RedisObserver {
       ...redis.options,
       monitor: false,
       lazyLoading: false,
+      connectionName: `redisinsight-monitor-perm-check-${Math.random()}`,
     });
 
     await duplicate.send_command('monitor');

redisinsight/api/src/modules/profiler/profiler.controller.ts

@@ -25,6 +25,8 @@ export class ProfilerController {
     res.setHeader('Content-Type', 'application/octet-stream');
     res.setHeader('Content-Disposition', `attachment;filename="${filename}.txt"`);
 
-    stream.pipe(res);
+    stream
+      .on('error', () => res.status(404).send())
+      .pipe(res);
   }
 }

redisinsight/api/src/modules/profiler/providers/log-file.provider.ts

@@ -11,7 +11,7 @@ export class LogFileProvider implements OnModuleDestroy {
    * Get or create Profiler Log File to work with
    * @param id
    */
-  async getOrCreate(id: string): Promise<LogFile> {
+  getOrCreate(id: string): LogFile {
     if (!this.profilerLogFiles.has(id)) {
       this.profilerLogFiles.set(id, new LogFile(id));
     }
@@ -23,7 +23,7 @@ export class LogFileProvider implements OnModuleDestroy {
    * Get Profiler Log File or throw an error
    * @param id
    */
-  async get(id: string): Promise<LogFile> {
+  get(id: string): LogFile {
     if (!this.profilerLogFiles.has(id)) {
       throw new NotFoundException(ERROR_MESSAGES.PROFILER_LOG_FILE_NOT_FOUND);
     }