feat: additional prom stats / relay protocol fix (#121)

* client type prom metrics / relay protocol fix

* fix lint

* fix user agent log

Author: andrei-gavrilescu

package-lock.json

@@ -1,14 +1,14 @@
 {
   "name": "rtcstats-server",
-  "version": "2.24.3",
+  "version": "2.24.5",
   "description": "The rtcstats-server represents the server side component of the rtcstats ecosystem, the client side being https://github.com/jitsi/rtcstats which collects and sends WebRTC related statistics.",
   "main": "websocket.js",
   "private": true,
   "scripts": {
     "lint:fix": "eslint --fix ./src/",
     "lint": "eslint ./src/",
     "integration": "node ./src/test/client.js",
-    "test": "jest",
+    "test": "NODE_ENV=production jest",
     "test:fix": "jest ./src/test/jest/extract.test.js -- --fix",
     "start": "NODE_ENV=production node ./src/app.js",
     "watch:dev": "nodemon --exec 'NODE_ENV=debug node ./src/app.js'",

package.json

@@ -18,7 +18,7 @@ const logger = require('./logging');
 const PromCollector = require('./metrics/PromCollector');
 const S3Manager = require('./store/S3Manager');
 const { saveEntryAssureUnique } = require('./store/dynamo');
-const { getStatsFormat } = require('./utils/stats-detection');
+const { ClientManager } = require('./utils/ClientManager');
 const { asyncDeleteFile,
     getEnvName,
     getIdealWorkerCount,
@@ -207,7 +207,6 @@ function setupFeaturesPublisher() {
 function setupWorkDirectory() {
     try {
         // Temporary path for stats dumps must be configured.
-        tempPath = config.server.tempPath;
         assert(tempPath);
 
         if (fs.existsSync(tempPath)) {
@@ -275,42 +274,37 @@ function wsConnectionHandler(client, upgradeReq) {
     try {
         PromCollector.connected.inc();
 
-        // the url the client is coming from
-        const referer = upgradeReq.headers.origin + upgradeReq.url;
-        const ua = upgradeReq.headers['user-agent'];
-
-        // During feature extraction we need information about the browser in order to decide which algorithms use.
-        const connectionInfo = {
-            path: upgradeReq.url,
-            origin: upgradeReq.headers.origin,
-            url: referer,
-            userAgent: ua,
-            clientProtocol: client.protocol
-        };
+        const clientManager = new ClientManager(client, upgradeReq);
+        const clientDetails = clientManager.getDetails();
+        const {
+            userAgent,
+            clientProtocol,
+            url,
+            statsFormat,
+            clientType
+        } = clientDetails;
 
         logger.info(
-        '[App] New app connected: ua: %s, protocol: %s, referer: %s',
-        ua,
-        client.protocol,
-        referer
+            '[App] New app connected: user-agent: %s, protocol: %s, url: %s',
+            userAgent,
+            clientProtocol,
+            url
         );
 
         // The if statement is used to maintain compatibility with the reconnect functionality on the client
         // it should be removed once the server also supports this functionality.
         // TODO: Remove once reconnect is added to server
-        if (isSessionOngoing(referer, tempPath) || isSessionReconnect(referer)) {
-            logger.warn(`[APP] Reconnect not supported, closing connection for ${referer}`);
+        if (isSessionOngoing(url, tempPath) || isSessionReconnect(url)) {
+            logger.warn(`[APP] Reconnect not supported, closing connection for ${url}`);
 
             client.close(3001);
 
             return;
         }
 
-        connectionInfo.statsFormat = getStatsFormat(connectionInfo);
-
         const demuxSinkOptions = {
-            connectionInfo,
-            dumpFolder: './temp',
+            clientDetails,
+            dumpFolder: tempPath,
             log: logger
         };
 
@@ -324,6 +318,7 @@ function wsConnectionHandler(client, upgradeReq) {
             const dumpData = {
                 app: meta.applicationName || 'Undefined',
                 clientId: id,
+                clientType,
                 conferenceId: meta.confName,
                 conferenceUrl: meta.confID,
                 dumpPath: meta.dumpPath,
@@ -335,30 +330,28 @@ function wsConnectionHandler(client, upgradeReq) {
                 ampSessionId: meta.sessionId,
                 ampUserId: meta.userId,
                 ampDeviceId: meta.deviceId,
-                statsFormat: connectionInfo.statsFormat,
+                statsFormat,
                 isBreakoutRoom: meta.isBreakoutRoom,
                 breakoutRoomId: meta.roomId,
                 parentStatsSessionId: meta.parentStatsSessionId,
                 ...tenantInfo
             };
 
+            PromCollector.collectClientDumpSizeMetrics(dumpData);
+
             const obfuscatedDumpData = obfuscatePII(dumpData);
 
             logger.info('[App] Processing dump id %s, metadata %o', id, obfuscatedDumpData);
 
             // Don't process dumps generated by JVB & Jigasi, there should be a more formal process to
-            if (config.features.disableFeatExtraction
-                || connectionInfo.clientProtocol?.includes('JVB')
-                || connectionInfo.clientProtocol?.includes('JIGASI')
-                || connectionInfo.clientProtocol?.includes('JICOFO')
-                || connectionInfo.userAgent?.includes('react-native')) {
-                persistDumpData(dumpData);
-            } else {
-            // Add the clientId in the worker pool so it can process the associated dump file.
+            if (clientManager.supportsFeatureExtraction()) {
+                // Add the clientId in the worker pool so it can process the associated dump file.
                 workerPool.addTask({
                     type: RequestType.PROCESS,
                     body: dumpData
                 });
+            } else {
+                persistDumpData(dumpData);
             }
         });
 
@@ -372,12 +365,12 @@ function wsConnectionHandler(client, upgradeReq) {
                 // the whole pipeline does as well,
                 PromCollector.sessionErrorCount.inc();
 
-                logger.error('[App] Connection pipeline: %o;  error: %o', connectionInfo, err);
+                logger.error('[App] Connection pipeline: %o;  error: %o', clientDetails, err);
             }
         });
 
         connectionPipeline.on('finish', () => {
-            logger.info('[App] Connection pipeline successfully finished %o', connectionInfo);
+            logger.info('[App] Connection pipeline successfully finished %o', clientDetails);
 
             // We need to explicity close the ws, you might notice that we don't do the same in case of an error
             // that's because in that case the error will propagate up the pipeline chain and the ws stream will also
@@ -404,7 +397,7 @@ function wsConnectionHandler(client, upgradeReq) {
             PromCollector.connected.dec();
         });
     } catch (error) {
-        logger.error('[App] Error while handling ws connection: %j', error);
+        logger.error('[App] Error while handling ws connection: %o', error);
     }
 }
 
@@ -526,6 +519,8 @@ function setupSecretManager() {
 async function startRtcstatsServer() {
     logger.info('[App] Initializing: %s; version: %s; env: %s ...', appName, appVersion, getEnvName());
 
+    tempPath = config.server.tempPath;
+
     setupSecretManager();
     await setupWebhookSender();
     setupWorkDirectory();

src/app.js

@@ -23,16 +23,16 @@ class DemuxSink extends Writable {
      * C'tor
      *
      * @param {string} dumpFolder - Path to where sink files will be temporarily stored.
-     * @param {Object} connectionInfo - Object containing information about the ws connection which stream the data.
+     * @param {Object} clientDetails - Object containing information about the ws connection which stream the data.
      * @param {Object} log - Log object.
      * @param {boolean} persistDump - Flag used for generating a complete dump of the data coming to the stream.
      * Required when creating mock tests.
      */
-    constructor({ dumpFolder, connectionInfo, log, persistDump = false }) {
+    constructor({ dumpFolder, clientDetails, log, persistDump = false }) {
         super({ objectMode: true });
 
         this.dumpFolder = dumpFolder;
-        this.connectionInfo = connectionInfo;
+        this.clientDetails = clientDetails;
         this.log = log;
         this.timeoutId = -1;
         this.sinkMap = new Map();
@@ -179,7 +179,7 @@ class DemuxSink extends Writable {
             }
         }
 
-        this.log.info('[Demux] open-sink id: %s; path %s; connection: %o', id, filePath, this.connectionInfo);
+        this.log.info('[Demux] open-sink id: %s; path %s; connection: %o', id, filePath, this.clientDetails);
 
         const sink = fs.createWriteStream(idealPath, { fd });
 
@@ -204,7 +204,7 @@ class DemuxSink extends Writable {
         // by visualizer tools for identifying the originating client (browser, jvb or other).
         this._sinkWrite(
             sink,
-            JSON.stringify([ 'connectionInfo', null, JSON.stringify(this.connectionInfo), Date.now() ]));
+            JSON.stringify([ 'connectionInfo', null, JSON.stringify(this.clientDetails), Date.now() ]));
 
         return sinkData;
     }

src/demux.js

@@ -1,8 +1,10 @@
 // Initialize prometheus metrics.
+const fsPromises = require('fs').promises;
 const getFolderSize = require('get-folder-size');
 const prom = require('prom-client');
 
 const logger = require('../logging');
+const { ClientType } = require('../utils/ClientManager');
 
 const PromCollector = {
     connected: new prom.Gauge({
@@ -169,6 +171,59 @@ const PromCollector = {
         help: 'number of open websocket connections that failed with an error'
     }),
 
+    rtcstatsDumpSizeBytes: new prom.Summary({
+        name: 'rtcstats_dump_size_bytes',
+        help: 'Size of processed dumps sent by rtcstats clients'
+    }),
+    jvbDumpSizeBytes: new prom.Summary({
+        name: 'jvb_dump_size_bytes',
+        help: 'Size of processed dumps sent by jvb clients'
+    }),
+    jicofoDumpSizeBytes: new prom.Summary({
+        name: 'jicofo_dump_size_bytes',
+        help: 'Size of processed dumps sent by jicofo clients'
+    }),
+    jibriDumpSizeBytes: new prom.Summary({
+        name: 'jibri_dump_size_bytes',
+        help: 'Size of processed dumps sent by jibri clients'
+    }),
+    jigasiDumpSizeBytes: new prom.Summary({
+        name: 'jigasi_dump_size_bytes',
+        help: 'Size of processed dumps sent by jigasi clients'
+    }),
+    unknownDumpSizeBytes: new prom.Summary({
+        name: 'unknown_dump_size_bytes',
+        help: 'Size of processed dumps sent by unknown clients'
+    }),
+
+    collectClientDumpSizeMetrics: async dumpData => {
+        const { dumpPath, clientType } = dumpData;
+
+        const dumpStats = await fsPromises.stat(dumpPath);
+        const dumpSize = dumpStats.size;
+
+        switch (clientType) {
+        case ClientType.RTCSTATS:
+            PromCollector.rtcstatsDumpSizeBytes.observe(dumpSize);
+            break;
+        case ClientType.JVB:
+            PromCollector.jvbDumpSizeBytes.observe(dumpSize);
+            break;
+        case ClientType.JICOFO:
+            PromCollector.jicofoDumpSizeBytes.observe(dumpSize);
+            break;
+        case ClientType.JIBRI:
+            PromCollector.jibriDumpSizeBytes.observe(dumpSize);
+            break;
+        case ClientType.JIGASI:
+            PromCollector.jigasiDumpSizeBytes.observe(dumpSize);
+            break;
+        default:
+            PromCollector.unknownDumpSizeBytes.observe(dumpSize);
+            break;
+        }
+    },
+
     metrics: () => prom.register.metrics(),
 
     collectDefaultMetrics: () => prom.collectDefaultMetrics(),

src/metrics/PromCollector.js

@@ -27,8 +27,8 @@ const BrowserUASamples = Object.freeze({
 });
 
 const ProtocolV = Object.freeze({
-    LEGACY: '3_LEGACY',
-    STANDARD: '3_STANDARD'
+    LEGACY: '3.1_LEGACY',
+    STANDARD: '3.1_STANDARD'
 });
 
 /**