feat: add reconnect and process old files

tamasdomokos · tamasdomokos · commit c810e8416018 · 2022-11-16T14:47:59.000+02:00
diff --git a/config/default.yaml b/config/default.yaml
@@ -15,6 +15,8 @@ server:
 
 features:
     disableFeatExtraction: false
+    reconnectTimout: 60000
+    requenceNumberSendingInterval: 60000
 
 amplitude:
     key:
diff --git a/src/app.js b/src/app.js
@@ -2,10 +2,12 @@ const JSONStream = require('JSONStream');
 const assert = require('assert').strict;
 const config = require('config');
 const fs = require('fs');
+const { stat } = require('fs/promises');
 const http = require('http');
 const https = require('https');
 const path = require('path');
 const { pipeline } = require('stream');
+const url = require('url');
 const WebSocket = require('ws');
 
 const { name: appName, version: appVersion } = require('../package');
@@ -17,6 +19,7 @@ const DemuxSink = require('./demux');
 const logger = require('./logging');
 const PromCollector = require('./metrics/PromCollector');
 const { saveEntryAssureUnique } = require('./store/dynamo');
+const fileStore = require('./store/file');
 const initS3Store = require('./store/s3.js');
 const { getStatsFormat } = require('./utils/stats-detection');
 const { asyncDeleteFile, getEnvName, getIdealWorkerCount, RequestType, ResponseType } = require('./utils/utils');
@@ -26,6 +29,7 @@ let amplitude;
 let store;
 let featPublisher;
 let tempPath;
+const sessionIdTimeouts = {};
 
 /**
  * Store the dump to the configured store. The dump file might be stored under a different
@@ -180,6 +184,28 @@ function setupFeaturesPublisher() {
  * Initialize the directory where temporary dump files will be stored.
  */
 function setupWorkDirectory() {
+    try {
+        // Temporary path for stats dumps must be configured.
+        tempPath = config.server.tempPath;
+        assert(tempPath);
+
+        if (!fs.existsSync(tempPath)) {
+            logger.debug(`[App] Creating working dir ${tempPath}`);
+            fs.mkdirSync(tempPath);
+        }
+    } catch (e) {
+        logger.error(`[App] Error while accessing working dir ${tempPath} - ${e}`);
+
+        // The app is probably in an inconsistent state at this point, throw and stop process.
+        throw e;
+    }
+}
+
+/**
+ * Remove old files from the temp folder.
+ */
+function processOldFiles() {
+    logger.info('[App] Waiting for connections to reconnect.');
     try {
         // Temporary path for stats dumps must be configured.
         tempPath = config.server.tempPath;
@@ -188,15 +214,37 @@ function setupWorkDirectory() {
         if (fs.existsSync(tempPath)) {
             fs.readdirSync(tempPath).forEach(fname => {
                 try {
-                    logger.debug(`[App] Removing file ${`${tempPath}/${fname}`}`);
-                    fs.unlinkSync(`${tempPath}/${fname}`);
+                    const filePath = `${tempPath}/${fname}`;
+
+                    logger.debug(`[App] Trying to process file ${filePath}`);
+                    fs.stat(filePath, (err, stats) => {
+                        if (err) {
+                            throw err;
+                        }
+
+                        if (Math.abs(Date.now() - stats.mtime.getTime()) > config.features.reconnectTimout) {
+                            logger.debug(`[App] Start processing the file ${`${filePath}`}`);
+                            const response = fileStore?.getObjectsByKeys(
+                                filePath, [ 'connectionInfo', 'identity' ]);
+
+                            response.then(
+                                obj => {
+                                    logger.info(`[App] Meta and connectionInfo response: ${response}`);
+                                    const meta = obj?.connectionInfo;
+                                    const connectionInfo = obj?.identity;
+
+                                    setTimeout(processData,
+                                        config.features.reconnectTimout, fname, meta, connectionInfo);
+                                })
+                                .catch(() => {
+                                    logger.info('[App] New connection. File doesn\'t exist.');
+                                });
+                        }
+                    });
                 } catch (e) {
                     logger.error(`[App] Error while unlinking file ${fname} - ${e}`);
                 }
             });
-        } else {
-            logger.debug(`[App] Creating working dir ${tempPath}`);
-            fs.mkdirSync(tempPath);
         }
     } catch (e) {
         logger.error(`[App] Error while accessing working dir ${tempPath} - ${e}`);
@@ -248,10 +296,16 @@ function setupMetricsServer() {
  */
 function wsConnectionHandler(client, upgradeReq) {
     PromCollector.connected.inc();
+    logger.info('[App] Websocket connection handler');
 
     // the url the client is coming from
     const referer = upgradeReq.headers.origin + upgradeReq.url;
     const ua = upgradeReq.headers['user-agent'];
+    const queryObject = url.parse(referer, true).query;
+    const statsSessionId = queryObject?.statsSessionId;
+
+    clearConnectionTimeout(statsSessionId);
+    sendLastSequenceNumber(client, statsSessionId);
 
     // During feature extraction we need information about the browser in order to decide which algorithms use.
     const connectionInfo = {
@@ -273,39 +327,11 @@ function wsConnectionHandler(client, upgradeReq) {
     const demuxSink = new DemuxSink(demuxSinkOptions);
 
     demuxSink.on('close-sink', ({ id, meta }) => {
-        logger.info('[App] Queue for processing id %s', id);
-
-        // Metadata associated with a dump can get large so just select the necessary fields.
-        const dumpData = {
-            app: meta.applicationName || 'Undefined',
-            clientId: id,
-            conferenceId: meta.confName,
-            conferenceUrl: meta.confID,
-            dumpPath: meta.dumpPath,
-            endDate: Date.now(),
-            endpointId: meta.endpointId,
-            startDate: meta.startDate,
-            sessionId: meta.meetingUniqueId,
-            userId: meta.displayName,
-            ampSessionId: meta.sessionId,
-            ampUserId: meta.userId,
-            ampDeviceId: meta.deviceId,
-            statsFormat: connectionInfo.statsFormat,
-            isBreakoutRoom: meta.isBreakoutRoom,
-            breakoutRoomId: meta.roomId,
-            parentStatsSessionId: meta.parentStatsSessionId
-        };
-
-        // Don't process dumps generated by JVB, there should be a more formal process to
-        if (config.features.disableFeatExtraction || connectionInfo.clientProtocol?.includes('JVB')) {
-            persistDumpData(dumpData);
-        } else {
-            // Add the clientId in the worker pool so it can process the associated dump file.
-            workerPool.addTask({
-                type: RequestType.PROCESS,
-                body: dumpData
-            });
-        }
+        logger.info('[App] Websocket disconnected waiting for processing the data %s', id);
+
+        const timemoutId = setTimeout(processData, config.features.reconnectTimout, id, meta, connectionInfo);
+
+        sessionIdTimeouts[id] = timemoutId;
     });
 
     const connectionPipeline = pipeline(
@@ -348,6 +374,61 @@ function wsConnectionHandler(client, upgradeReq) {
     });
 }
 
+/**
+ * Clear the connection timeout if the user is reconnected/
+ *
+ * @param {*} id
+ */
+function clearConnectionTimeout(id) {
+    const timeoutId = sessionIdTimeouts[id];
+
+    if (timeoutId) {
+        logger.info('[App] Clear timeout for connectionId: %s', id);
+        clearTimeout(timeoutId);
+    }
+}
+
+/**
+ *
+ * @param {*} meta
+ * @param {*} connectionInfo
+ */
+function processData(id, meta, connectionInfo) {
+    logger.info('[App] Queue for processing id %s', id);
+
+    // Metadata associated with a dump can get large so just select the necessary fields.
+    const dumpData = {
+        app: meta.applicationName || 'Undefined',
+        clientId: id,
+        conferenceId: meta.confName,
+        conferenceUrl: meta.confID,
+        dumpPath: meta.dumpPath,
+        endDate: Date.now(),
+        endpointId: meta.endpointId,
+        startDate: meta.startDate,
+        sessionId: meta.meetingUniqueId,
+        userId: meta.displayName,
+        ampSessionId: meta.sessionId,
+        ampUserId: meta.userId,
+        ampDeviceId: meta.deviceId,
+        statsFormat: connectionInfo.statsFormat,
+        isBreakoutRoom: meta.isBreakoutRoom,
+        breakoutRoomId: meta.roomId,
+        parentStatsSessionId: meta.parentStatsSessionId
+    };
+
+    // Don't process dumps generated by JVB, there should be a more formal process to
+    if (config.features.disableFeatExtraction || connectionInfo.clientProtocol?.includes('JVB')) {
+        persistDumpData(dumpData);
+    } else {
+        // Add the clientId in the worker pool so it can process the associated dump file.
+        workerPool.addTask({
+            type: RequestType.PROCESS,
+            body: dumpData
+        });
+    }
+}
+
 /**
  *
  * @param {*} wsServer
@@ -358,6 +439,33 @@ function setupWebSocketsServer(wsServer) {
     wss.on('connection', wsConnectionHandler);
 }
 
+/**
+ * @param {*} client
+ * @param {*} id
+ */
+function sendLastSequenceNumber(client, id) {
+    const dumpPath = `${tempPath}/${id}`;
+    let sequenceNumber = 0;
+
+    fileStore?.getLastLine(dumpPath, 1)
+        .then(
+            lastLine => {
+                const jsonData = JSON.parse(lastLine);
+
+                if (Array.isArray(jsonData) && jsonData[4] !== undefined) {
+                    sequenceNumber = jsonData[4];
+                }
+            })
+        .catch(() => {
+            logger.info('[App] New connection. File doesn\'t exist.');
+        })
+        .finally(() => {
+            client.send(`{"sn":${sequenceNumber}}`);
+
+            setTimeout(sendLastSequenceNumber, config.features.requenceNumberSendingInterval, client, id);
+        });
+}
+
 /**
  * Handler used for basic availability checks.
  *
@@ -439,6 +547,7 @@ function startRtcstatsServer() {
     logger.info('[App] Initializing: %s; version: %s; env: %s ...', appName, appVersion, getEnvName());
 
     setupWorkDirectory();
+    processOldFiles();
     setupDumpStorage();
     setupFeaturesPublisher();
     setupAmplitudeConnector();
diff --git a/src/demux.js b/src/demux.js
@@ -151,12 +151,11 @@ class DemuxSink extends Writable {
     async _sinkCreate(id) {
         PromCollector.sessionCount.inc();
 
-        let resolvedId = id;
-        let i = 0;
+        const resolvedId = id;
         let fd;
 
         const idealPath = path.resolve(cwd, this.dumpFolder, id);
-        let filePath = idealPath;
+        const filePath = idealPath;
 
         // If a client reconnects the same client id will be provided thus cases can occur where the previous dump
         // with the same id is still present on the disk, in order to avoid conflicts and states where multiple
@@ -168,15 +167,7 @@ class DemuxSink extends Writable {
         // if the entry already exists because some other instance uploaded first, the same incremental approach needs
         // to be taken.
         while (!fd) {
-            try {
-                fd = await fsOpen(filePath, 'wx');
-            } catch (err) {
-                if (err.code !== 'EEXIST') {
-                    throw err;
-                }
-                resolvedId = `${id}_${++i}`;
-                filePath = path.resolve(cwd, this.dumpFolder, resolvedId);
-            }
+            fd = await fsOpen(filePath, 'w');
         }
 
         this.log.info('[Demux] open-sink id: %s; path %s; connection: %o', id, filePath, this.connectionInfo);
diff --git a/src/store/file.js b/src/store/file.js
@@ -0,0 +1,54 @@
+const fs = require('fs');
+const readline = require('readline');
+const Stream = require('stream');
+
+const logger = require('../logging');
+
+exports.getLastLine = (fileName, minLength) => {
+    const inStream = fs.createReadStream(fileName);
+    const outStream = new Stream();
+
+    return new Promise((resolve, reject) => {
+        const rl = readline.createInterface(inStream, outStream);
+
+        let lastLine = '';
+
+        rl.on('line', line => {
+            if (line.length >= minLength) {
+                lastLine = line;
+            }
+        });
+
+        rl.on('error', reject);
+
+        rl.on('close', () => {
+            resolve(lastLine);
+        });
+    });
+};
+
+exports.getObjectsByKeys = (fileName, keys) => {
+    const inStream = fs.createReadStream(fileName);
+    const outStream = new Stream();
+    const response = {};
+
+    return new Promise((resolve, reject) => {
+        const rl = readline.createInterface(inStream, outStream);
+
+        rl.on('line', line => {
+            if (line !== '') {
+                const jsonLine = JSON.parse(line);
+
+                if (keys.indexOf(jsonLine[0]) >= 0 && jsonLine[2] !== null) {
+                    response[jsonLine[0]] = jsonLine[2];
+                }
+            }
+        });
+
+        rl.on('close', () => {
+            resolve(response);
+        });
+
+        rl.on('error', reject);
+    });
+};
