flush logs + refactoring

Author: Artem

redisinsight/api/src/modules/profiler/emitters/client.logs-emitter.ts

@@ -2,7 +2,7 @@ import { Socket } from 'socket.io';
 import { ILogsEmitter } from 'src/modules/profiler/interfaces/logs-emitter.interface';
 import { ProfilerServerEvents } from 'src/modules/profiler/constants';
 
-class ClientLogsEmitter implements ILogsEmitter {
+export class ClientLogsEmitter implements ILogsEmitter {
   private readonly client: Socket;
 
   public readonly id: string;
@@ -16,9 +16,9 @@ class ClientLogsEmitter implements ILogsEmitter {
     return this.client.emit(ProfilerServerEvents.Data, items);
   }
 
-  public addClientObserver() {}
+  public addProfilerClient() {}
 
-  public removeClientObserver() {}
-}
+  public removeProfilerClient() {}
 
-export default ClientLogsEmitter;
+  public flushLogs() {}
+}

redisinsight/api/src/modules/profiler/emitters/file.logs-emitter.ts

@@ -14,7 +14,7 @@ class FileLogsEmitter implements ILogsEmitter {
   /**
    * Write batch of logs to a file
    */
-  public async emit(items: any[]) {
+  async emit(items: any[]) {
     try {
       if (!this.logFile.getWriteStream()) {
         return;
@@ -31,12 +31,16 @@ class FileLogsEmitter implements ILogsEmitter {
     }
   }
 
-  public addClientObserver(id: string) {
-    return this.logFile.addClientObserver(id);
+  async addProfilerClient(id: string) {
+    return this.logFile.addProfilerClient(id);
   }
 
-  public removeClientObserver(id: string) {
-    return this.logFile.removeClientObserver(id);
+  async removeProfilerClient(id: string) {
+    return this.logFile.removeProfilerClient(id);
+  }
+
+  async flushLogs() {
+    return this.logFile.destroy();
   }
 }
 export default FileLogsEmitter;

redisinsight/api/src/modules/profiler/interfaces/logs-emitter.interface.ts

@@ -1,6 +1,7 @@
 export interface ILogsEmitter {
   id: string;
   emit: (items: any[]) => void;
-  addClientObserver: (id: string) => void;
-  removeClientObserver: (id: string) => void;
+  addProfilerClient: (id: string) => void;
+  removeProfilerClient: (id: string) => void;
+  flushLogs: () => void;
 }

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

@@ -72,12 +72,12 @@ export class LogFile {
     this.alias = alias;
   }
 
-  public addClientObserver(id: string) {
+  addProfilerClient(id: string) {
     this.clientObservers.set(id, id);
     this.idleSince = 0;
   }
 
-  public removeClientObserver(id: string) {
+  removeProfilerClient(id: string) {
     this.clientObservers.delete(id);
 
     if (!this.clientObservers.size) {
@@ -92,10 +92,11 @@ export class LogFile {
   }
 
   /**
-   * Remove file after finish
+   * Remove file and delete write stream after finish
    */
   async destroy() {
     try {
+      this.writeStream?.close();
       this.writeStream = null;
       await fs.unlink(this.filePath);
     } catch (e) {

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

@@ -56,10 +56,14 @@ export class ProfilerClient {
 
   public addLogsEmitter(emitter: ILogsEmitter) {
     this.logsEmitters.set(emitter.id, emitter);
-    emitter.addClientObserver(this.id);
+    emitter.addProfilerClient(this.id);
+  }
+
+  async flushLogs() {
+    this.logsEmitters.forEach((emitter) => emitter.flushLogs());
   }
 
   public destroy() {
-    this.logsEmitters.forEach((emitter) => emitter.removeClientObserver(this.id));
+    this.logsEmitters.forEach((emitter) => emitter.removeProfilerClient(this.id));
   }
 }

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

@@ -9,7 +9,7 @@ import ERROR_MESSAGES from 'src/constants/error-messages';
 export class RedisObserver {
   private readonly redis: IORedis.Redis | IORedis.Cluster;
 
-  private clientMonitorObservers: Map<string, ProfilerClient> = new Map();
+  private profilerClients: Map<string, ProfilerClient> = new Map();
 
   private shardsObservers: IShardObserver[] = [];
 
@@ -20,65 +20,82 @@ export class RedisObserver {
     this.status = RedisObserverStatus.Wait;
   }
 
-  public async subscribe(client: ProfilerClient) {
+  /**
+   * Create "monitor" clients for each shard if not exists
+   * Subscribe profiler client to each each shard
+   * Ignore when profiler client with such id already exists
+   * @param profilerClient
+   */
+  public async subscribe(profilerClient: ProfilerClient) {
     if (this.status !== RedisObserverStatus.Ready) {
       await this.connect();
     }
-    if (this.clientMonitorObservers.has(client.id)) {
+
+    if (this.profilerClients.has(profilerClient.id)) {
       return;
     }
 
     this.shardsObservers.forEach((observer) => {
       observer.on('monitor', (time, args, source, database) => {
-        client.handleOnData({
+        profilerClient.handleOnData({
           time, args, database, source, shardOptions: observer.options,
         });
       });
       observer.on('end', () => {
-        client.handleOnDisconnect();
+        profilerClient.handleOnDisconnect();
         this.clear();
       });
     });
-    this.clientMonitorObservers.set(client.id, client);
+    this.profilerClients.set(profilerClient.id, profilerClient);
   }
 
   public unsubscribe(id: string) {
-    this.clientMonitorObservers.delete(id);
-    if (this.clientMonitorObservers.size === 0) {
+    this.profilerClients.delete(id);
+    if (this.profilerClients.size === 0) {
       this.clear();
     }
   }
 
   public disconnect(id: string) {
-    const userClient = this.clientMonitorObservers.get(id);
+    const userClient = this.profilerClients.get(id);
     if (userClient) {
       userClient.destroy();
     }
-    this.clientMonitorObservers.delete(id);
-    if (this.clientMonitorObservers.size === 0) {
+    this.profilerClients.delete(id);
+    if (this.profilerClients.size === 0) {
       this.clear();
     }
   }
 
   public clear() {
-    this.clientMonitorObservers.clear();
-    this.shardsObservers.forEach((observer) => observer.disconnect());
+    this.profilerClients.clear();
+    this.shardsObservers.forEach((observer) => {
+      observer.removeAllListeners('end');
+      observer.disconnect();
+    });
     this.shardsObservers = [];
     this.status = RedisObserverStatus.End;
   }
 
-  public getSize(): number {
-    return this.clientMonitorObservers.size;
+  /**
+   * Return number of profilerClients for current Redis Observer instance
+   */
+  public getProfilerClientsSize(): number {
+    return this.profilerClients.size;
   }
 
+  /**
+   * Create shard observer for each Redis shard to receive "monitor" data
+   * @private
+   */
   private async connect(): Promise<void> {
     try {
       if (this.redis instanceof IORedis.Cluster) {
         this.shardsObservers = await Promise.all(
-          this.redis.nodes('all').filter((node) => node.status === 'ready').map(this.createShardObserver),
+          this.redis.nodes('all').filter((node) => node.status === 'ready').map(RedisObserver.createShardObserver),
         );
       } else {
-        this.shardsObservers = [await this.createShardObserver(this.redis)];
+        this.shardsObservers = [await RedisObserver.createShardObserver(this.redis)];
       }
       this.status = RedisObserverStatus.Ready;
     } catch (error) {
@@ -92,12 +109,20 @@ export class RedisObserver {
     }
   }
 
-  private async createShardObserver(redis: IORedis.Redis): Promise<IShardObserver> {
-    // HACK: ioredis impropriety throw error a user has no permissions to run the 'monitor' command
+  /**
+   * Create and return shard observer using IORedis common client
+   * @param redis
+   */
+  static async createShardObserver(redis: IORedis.Redis): Promise<IShardObserver> {
     await RedisObserver.isMonitorAvailable(redis);
     return await redis.monitor() as IShardObserver;
   }
 
+  /**
+   * HACK: ioredis do not handle error when a user has no permissions to run the 'monitor' command
+   * Here we try to send "monitor" command directly to throw error (like NOPERM) if any
+   * @param redis
+   */
   static async isMonitorAvailable(redis: IORedis.Redis): Promise<boolean> {
     // @ts-ignore
     const duplicate = redis.duplicate({

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

@@ -30,6 +30,7 @@ export class ProfilerGateway implements OnGatewayConnection, OnGatewayDisconnect
       await this.service.addListenerForInstance(ProfilerGateway.getInstanceId(client), client, settings);
       return { status: 'ok' };
     } catch (error) {
+      this.logger.error('Unable to add listener', error);
       throw new WsException(error);
     }
   }
@@ -40,16 +41,18 @@ export class ProfilerGateway implements OnGatewayConnection, OnGatewayDisconnect
       await this.service.removeListenerFromInstance(ProfilerGateway.getInstanceId(client), client.id);
       return { status: 'ok' };
     } catch (error) {
+      this.logger.error('Unable to pause monitor', error);
       throw new WsException(error);
     }
   }
 
   @SubscribeMessage(ProfilerClientEvents.FlushLogs)
   async flushLogs(client: Socket): Promise<any> {
     try {
-      await this.service.flushLogs(ProfilerGateway.getInstanceId(client), client.id);
+      await this.service.flushLogs(client.id);
       return { status: 'ok' };
     } catch (error) {
+      this.logger.error('Unable to flush logs', error);
       throw new WsException(error);
     }
   }

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

@@ -2,15 +2,19 @@ import { Module } from '@nestjs/common';
 import { SharedModule } from 'src/modules/shared/shared.module';
 import { LogFileProvider } from 'src/modules/profiler/providers/log-file.provider';
 import { ProfilerController } from 'src/modules/profiler/profiler.controller';
+import { RedisObserverProvider } from 'src/modules/profiler/providers/redis-observer.provider';
+import { ProfilerClientProvider } from 'src/modules/profiler/providers/profiler-client.provider';
 import { ProfilerGateway } from './profiler.gateway';
 import { ProfilerService } from './profiler.service';
 
 @Module({
   imports: [SharedModule],
   providers: [
+    RedisObserverProvider,
+    ProfilerClientProvider,
+    LogFileProvider,
     ProfilerGateway,
     ProfilerService,
-    LogFileProvider,
   ],
   controllers: [ProfilerController],
 })