Feature/sre 484 test out aggregating metrics in a worker thread with reset

src/legacy/router.js

@@ -646,9 +646,6 @@ if (startDestTransformer) {
       ctx.status = ctxStatusCode;
       ctx.set('apiVersion', API_VERSION);
 
-      stats.timing('user_transform_request_latency', startTime, {
-        processSessions,
-      });
       stats.timingSummary('user_transform_request_latency_summary', startTime, {
         processSessions,
       });

src/middleware.js

@@ -1,12 +1,24 @@
 const Pyroscope = require('@pyroscope/nodejs');
 const { getDestTypeFromContext } = require('@rudderstack/integrations-lib');
+const cluster = require('cluster');
 const stats = require('./util/stats');
 
 function initPyroscope() {
   Pyroscope.init({
-    appName: 'rudder-transformer',
+    appName: 'ut-rudder-transformer',
+    serverAddress: 'https://sdk-pyroscope.rudderstack.com',
+    tags: {
+      workerId: cluster.worker ? cluster.worker.id : 0,
+      isMaster: cluster.isMaster,
+    },
   });
+  Pyroscope.start();
   Pyroscope.startHeapCollecting();
+  process.on('SIGINT', () => {
+    Pyroscope.stop();
+    Pyroscope.stopHeapCollecting();
+  });
+
 }
 
 function getCPUProfile(seconds) {

src/routes/metricsRouter.js

@@ -16,6 +16,16 @@ if (enableStats) {
       ctx.body = error.message;
     }
   });
+
+  metricsRouter.get('/resetMetrics', async (ctx) => {
+    try {
+      await stats.resetMetricsController(ctx);
+    } catch (error) {
+      logger.error(error);
+      ctx.status = 400;
+      ctx.body = error.message;
+    }
+  });
 }
 
 module.exports = { metricsRouter };

src/services/userTransform.ts

@@ -169,16 +169,6 @@ export class UserTransformService {
             ...getTransformationMetadata(eventsToProcess[0]?.metadata),
           });
         } finally {
-          stats.timing('user_transform_request_latency', userFuncStartTime, {
-            ...metaTags,
-            ...getTransformationMetadata(eventsToProcess[0]?.metadata),
-          });
-
-          stats.timing('user_transform_batch_size', requestSize, {
-            ...metaTags,
-            ...getTransformationMetadata(eventsToProcess[0]?.metadata),
-          });
-
           stats.timingSummary('user_transform_request_latency_summary', userFuncStartTime, {
             ...metaTags,
             ...getTransformationMetadata(eventsToProcess[0]?.metadata),

src/util/customTransformer-v1.js

@@ -92,7 +92,6 @@ async function userTransformHandlerV1(
       ...(events.length && events[0].metadata ? getTransformationMetadata(events[0].metadata) : {}),
     };
     stats.counter('user_transform_function_input_events', events.length, tags);
-    stats.timing('user_transform_function_latency', invokeTime, tags);
     stats.timingSummary('user_transform_function_latency_summary', invokeTime, tags);
   }
 

src/util/customTransformer.js

@@ -259,7 +259,6 @@ async function runUserTransform(
     };
 
     stats.counter('user_transform_function_input_events', events.length, tags);
-    stats.timing('user_transform_function_latency', invokeTime, tags);
     stats.timingSummary('user_transform_function_latency_summary', invokeTime, tags);
   }
 

src/util/customTransforrmationsStore-v1.js

@@ -30,7 +30,6 @@ async function getTransformationCodeV1(versionId) {
 
     responseStatusHandler(response.status, 'Transformation', versionId, url);
     stats.increment('get_transformation_code', { success: 'true', ...tags });
-    stats.timing('get_transformation_code_time', startTime, tags);
     stats.timingSummary('get_transformation_code_time_summary', startTime, tags);
     const myJson = await response.json();
     transformationCache[versionId] = myJson;
@@ -56,7 +55,6 @@ async function getLibraryCodeV1(versionId) {
 
     responseStatusHandler(response.status, 'Transformation Library', versionId, url);
     stats.increment('get_libraries_code', { success: 'true', ...tags });
-    stats.timing('get_libraries_code_time', startTime, tags);
     stats.timingSummary('get_libraries_code_time_summary', startTime, tags);
     const myJson = await response.json();
     libraryCache[versionId] = myJson;
@@ -84,7 +82,6 @@ async function getRudderLibByImportName(importName) {
 
     responseStatusHandler(response.status, 'Rudder Library', importName, url);
     stats.increment('get_libraries_code', { success: 'true', ...tags });
-    stats.timing('get_libraries_code_time', startTime, tags);
     stats.timingSummary('get_libraries_code_time_summary', startTime, tags);
     const myJson = await response.json();
     rudderLibraryCache[importName] = myJson;

src/util/customTransforrmationsStore.js

@@ -23,7 +23,6 @@ async function getTransformationCode(versionId) {
 
     responseStatusHandler(response.status, 'Transformation', versionId, url);
     stats.increment('get_transformation_code', { versionId, success: 'true' });
-    stats.timing('get_transformation_code_time', startTime, { versionId });
     stats.timingSummary('get_transformation_code_time_summary', startTime, { versionId });
     const myJson = await response.json();
     myCache.set(versionId, myJson);

src/util/metricsAggregator.js

@@ -6,6 +6,7 @@
 const GET_METRICS_RES = 'rudder-transformer:getMetricsRes';
 const AGGREGATE_METRICS_REQ = 'rudder-transformer:aggregateMetricsReq';
 const AGGREGATE_METRICS_RES = 'rudder-transformer:aggregateMetricsRes';
+const RESET_METRICS_REQUEST = 'rudder-transformer:resetMetricsReq';
 
 class MetricsAggregator {
   constructor(prometheusInstance) {
@@ -18,28 +19,62 @@
     this.registerCallbacks();
   }
 
+  timing(name, start, tags = {}) {
+    try {
+      let metric = this.prometheusInstance.prometheusRegistry.getSingleMetric(name);
+      if (!metric) {
+        logger.warn(
+          `Prometheus: Timing metric ${name} not found in the registry. Creating a new one`,
+        );
+        metric = this.prometheusInstance.newHistogramStat(name, name, Object.keys(tags));
+      }
+      metric.observe(tags, (new Date() - start) / 1000);
+    } catch (e) {
+      logger.error(`Prometheus: Timing metric ${name} failed with error ${e}`);
+    }
+  }
+
   registerCallbacks() {
     if (cluster.isPrimary) {
       // register callback for master process
       cluster.on('message', async (worker, message) => {
+        const startTime = Date.now();
         if (message.type === GET_METRICS_RES) {
-          logger.debug(`[MetricsAggregator] Master received metrics from worker ${worker.id}`);
+          logger.info(
+            `${new Date().toISOString()} [MetricsAggregator] Master received metrics from worker ${worker.id}`,
+          );
+          this.timing('getMetricsReq', message.startTime, { workerId: worker.id });
           await this.handleMetricsResponse(message);
         }
+        logger.info(
+          `[MetricsAggregator] Master processed ${message.type} in ${new Date() - startTime} ms`,
+        );
       });
       return;
     }
     // register callback for worker process
     cluster.worker.on('message', async (message) => {
+      const startTime = new Date();
       if (message.type === GET_METRICS_REQ) {
-        logger.debug(`[MetricsAggregator] Worker ${cluster.worker.id} received metrics request`);
+        logger.info(
+          `${new Date().toISOString()} [MetricsAggregator] Worker ${cluster.worker.id} received metrics request`,
+        );
         try {
           const metrics = await this.prometheusInstance.prometheusRegistry.getMetricsAsJSON();
-          cluster.worker.send({ type: GET_METRICS_RES, metrics });
+          cluster.worker.send({ type: GET_METRICS_RES, metrics, startTime: message.startTime });
         } catch (error) {
           cluster.worker.send({ type: GET_METRICS_RES, error: error.message });
         }
+      } else if (message.type === RESET_METRICS_REQUEST) {
+        logger.info(
+          `[MetricsAggregator] Worker ${cluster.worker.id} received reset metrics request `,
+        );
+        this.prometheusInstance.prometheusRegistry.resetMetrics();
+        logger.info(`[MetricsAggregator] Worker ${cluster.worker.id} reset metrics successfully`);
       }
+      logger.info(
+        `[MetricsAggregator] Worker ${cluster.worker.id} processed ${message.type} in ${new Date() - startTime} ms`,
+      );
     });
   }
 
@@ -51,7 +86,12 @@
       );
 
       this.workerThread.on('message', (message) => {
+        const startTime = new Date();
         if ((message.type = AGGREGATE_METRICS_RES)) {
+          logger.info(
+            `${new Date().toISOString()} [MetricsAggregator] Master received aggregated metrics from worker thread`,
+          );
+          this.timing('aggregateMetricsReq', message.startTime);
           if (message.error) {
             this.rejectFunc(new Error(message.error));
             this.resetAggregator();
@@ -60,6 +100,9 @@
           this.resolveFunc(message.metrics);
           this.resetAggregator();
         }
+        logger.info(
+          `[MetricsAggregator] Master processed ${message.type} in ${new Date() - startTime} ms`,
+        );
       });
     }
   }
@@ -88,16 +131,22 @@
     return new Promise((resolve, reject) => {
       this.resolveFunc = resolve;
       this.rejectFunc = reject;
+      const startTime = Date.now();
       for (const id in cluster.workers) {
         this.pendingMetricRequests++;
-        logger.debug(`[MetricsAggregator] Requesting metrics from worker ${id}`);
-        cluster.workers[id].send({ type: GET_METRICS_REQ });
+        logger.info(`[MetricsAggregator] Requesting metrics from worker ${id}`);
+        cluster.workers[id].send({ type: GET_METRICS_REQ, startTime });
       }
     });
   }
 
   async aggregateMetricsInWorkerThread() {
-    this.workerThread.postMessage({ type: AGGREGATE_METRICS_REQ, metrics: this.metricsBuffer });
+    this.metricsBuffer.push(await this.prometheusInstance.prometheusRegistry.getMetricsAsJSON());
+    this.workerThread.postMessage({
+      type: AGGREGATE_METRICS_REQ,
+      metrics: this.metricsBuffer,
+      startTime: Date.now(),
+    });
   }
 
   async handleMetricsResponse(message) {
@@ -109,7 +158,11 @@
     this.metricsBuffer.push(message.metrics);
     this.pendingMetricRequests--;
     if (this.pendingMetricRequests === 0) {
-      this.aggregateMetricsInWorkerThread();
+      this.timing('getMetricsAll', message.startTime);
+      logger.info(
+        `${new Date().toISOString()} [MetricsAggregator] All metrics received, sending metrics to worker thread`,
+      );
+      await this.aggregateMetricsInWorkerThread();
     }
   }
 
@@ -118,6 +171,13 @@
       `[MetricsAggregator] Worker thread terminated with exit code ${await this.workerThread.terminate()}`,
     );
   }
+
+  resetMetrics() {
+    for (const id in cluster.workers) {
+      logger.info(`[MetricsAggregator] Resetting metrics for worker ${id}`);
+      cluster.workers[id].send({ type: RESET_METRICS_REQUEST });
+    }
+  }
 }
 
 module.exports = { MetricsAggregator, AGGREGATE_METRICS_REQ, AGGREGATE_METRICS_RES };

src/util/openfaas/index.js

@@ -379,7 +379,6 @@ const executeFaasFunction = async (
     };
 
     stats.counter('user_transform_function_input_events', events.length, tags);
-    stats.timing('user_transform_function_latency', startTime, tags);
     stats.timingSummary('user_transform_function_latency_summary', startTime, tags);
   }
 };