Merge remote-tracking branch 'origin/develop' into chore.log-filter

CHANGELOG.md

@@ -2,6 +2,13 @@
 
 All notable changes to this project will be documented in this file. See [standard-version](https://github.com/conventional-changelog/standard-version) for commit guidelines.
 
+### [1.69.1](https://github.com/rudderlabs/rudder-transformer/compare/v1.69.0...v1.69.1) (2024-06-25)
+
+
+### Bug Fixes
+
+* enhancement: introduce user model for one signal ([#3499](https://github.com/rudderlabs/rudder-transformer/issues/3499)) ([1c8e950](https://github.com/rudderlabs/rudder-transformer/commit/1c8e950f3d8789b33bba69a30c9eb21c40ce3d04))
+
 ## [1.69.0](https://github.com/rudderlabs/rudder-transformer/compare/v1.68.2...v1.69.0) (2024-06-10)
 
 

package.json

@@ -1,6 +1,6 @@
 {
   "name": "rudder-transformer",
-  "version": "1.69.0",
+  "version": "1.69.1",
   "description": "",
   "homepage": "https://github.com/rudderlabs/rudder-transformer#readme",
   "bugs": {

src/legacy/router.js

@@ -539,9 +539,7 @@ if (startDestTransformer) {
           (event) => `${event.metadata.destinationId}_${event.metadata.sourceId}`,
         );
       }
-      stats.counter('user_transform_function_group_size', Object.entries(groupedEvents).length, {
-        processSessions,
-      });
+      stats.counter('user_transform_function_group_size', Object.entries(groupedEvents).length, {});
 
       let ctxStatusCode = 200;
       const transformedEvents = [];
@@ -646,16 +644,10 @@ 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,
-      });
-      stats.increment('user_transform_requests', { processSessions });
-      stats.histogram('user_transform_output_events', transformedEvents.length, {
-        processSessions,
-      });
+      stats.timing('user_transform_request_latency', startTime, {});
+      stats.timingSummary('user_transform_request_latency_summary', startTime, {});
+      stats.increment('user_transform_requests', {});
+      stats.histogram('user_transform_output_events', transformedEvents.length, {});
     });
   }
 }

src/util/customTransformer-faas.js

@@ -91,7 +91,7 @@ async function setOpenFaasUserTransform(
   trMetadata = {},
 ) {
   const tags = {
-    transformerVersionId: userTransformation.versionId,
+    transformationId: userTransformation.id,
     identifier: 'openfaas',
     testMode,
   };

src/util/customTransformer-v1.js

@@ -65,7 +65,6 @@ async function userTransformHandlerV1(
   const isolatevmFactory = await getFactory(
     userTransformation.code,
     libraryVersionIds,
-    userTransformation.versionId,
     userTransformation.id,
     userTransformation.workspaceId,
     credentialsMap,

src/util/customTransformer.js

@@ -16,9 +16,11 @@ async function runUserTransform(
   code,
   secrets,
   eventsMetadata,
-  versionId,
+  transformationId,
+  workspaceId,
   testMode = false,
 ) {
+  const trTags = { identifier: 'v0', transformationId, workspaceId };
   // TODO: Decide on the right value for memory limit
   const isolate = new ivm.Isolate({ memoryLimit: ISOLATE_VM_MEMORY });
   const context = await isolate.createContext();
@@ -36,9 +38,9 @@ async function runUserTransform(
     new ivm.Reference(async (resolve, ...args) => {
       try {
         const fetchStartTime = new Date();
-        const res = await fetchWithDnsWrapper(versionId, ...args);
+        const res = await fetchWithDnsWrapper(trTags, ...args);
         const data = await res.json();
-        stats.timing('fetch_call_duration', fetchStartTime, { versionId });
+        stats.timing('fetch_call_duration', fetchStartTime, trTags);
         resolve.applyIgnored(undefined, [new ivm.ExternalCopy(data).copyInto()]);
       } catch (error) {
         resolve.applyIgnored(undefined, [new ivm.ExternalCopy('ERROR').copyInto()]);
@@ -51,7 +53,7 @@ async function runUserTransform(
     new ivm.Reference(async (resolve, reject, ...args) => {
       try {
         const fetchStartTime = new Date();
-        const res = await fetchWithDnsWrapper(versionId, ...args);
+        const res = await fetchWithDnsWrapper(trTags, ...args);
         const headersContent = {};
         res.headers.forEach((value, header) => {
           headersContent[header] = value;
@@ -67,7 +69,7 @@ async function runUserTransform(
           data.body = JSON.parse(data.body);
         } catch (e) {}
 
-        stats.timing('fetchV2_call_duration', fetchStartTime, { versionId });
+        stats.timing('fetchV2_call_duration', fetchStartTime, trTags);
         resolve.applyIgnored(undefined, [new ivm.ExternalCopy(data).copyInto()]);
       } catch (error) {
         const err = JSON.parse(JSON.stringify(error, Object.getOwnPropertyNames(error)));
@@ -93,7 +95,7 @@ async function runUserTransform(
           throw new Error(`request to fetch geolocation failed with status code: ${res.status}`);
         }
         const geoData = await res.json();
-        stats.timing('geo_call_duration', geoStartTime, { versionId });
+        stats.timing('geo_call_duration', geoStartTime, trTags);
         resolve.applyIgnored(undefined, [new ivm.ExternalCopy(geoData).copyInto()]);
       } catch (error) {
         const err = JSON.parse(JSON.stringify(error, Object.getOwnPropertyNames(error)));
@@ -251,10 +253,9 @@ async function runUserTransform(
     isolate.dispose();
 
     const tags = {
-      identifier: 'v0',
       errored: transformationError ? true : false,
-      ...(events.length && events[0].metadata ? getMetadata(events[0].metadata) : {}),
-      ...(events.length && events[0].metadata ? getTransformationMetadata(events[0].metadata) : {}),
+      ...(Object.keys(eventsMetadata).length ? getMetadata(Object.values(eventsMetadata)[0]) : {}),
+      ...trTags,
     };
 
     stats.counter('user_transform_function_input_events', events.length, tags);
@@ -316,7 +317,8 @@ async function userTransformHandler(
           res.code,
           res.secrets || {},
           eventsMetadata,
-          versionId,
+          res.id,
+          res.workspaceId,
           testMode,
         );
 

src/util/customTransforrmationsStore-v1.js

@@ -19,79 +19,54 @@ const getRudderLibrariesUrl = `${CONFIG_BACKEND_URL}/rudderstackTransformationLi
 async function getTransformationCodeV1(versionId) {
   const transformation = transformationCache[versionId];
   if (transformation) return transformation;
-  const tags = {
-    versionId,
-    version: 1,
-  };
   try {
     const url = `${getTransformationURL}?versionId=${versionId}`;
-    const startTime = new Date();
     const response = await fetchWithProxy(url);
 
     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;
     return myJson;
   } catch (error) {
-    logger.error(error);
-    stats.increment('get_transformation_code', { success: 'false', ...tags });
+    logger.error(
+      `Error fetching transformation V1 code for versionId: ${versionId}`,
+      error.message,
+    );
     throw error;
   }
 }
 
 async function getLibraryCodeV1(versionId) {
   const library = libraryCache[versionId];
   if (library) return library;
-  const tags = {
-    libraryVersionId: versionId,
-    version: 1,
-  };
   try {
     const url = `${getLibrariesUrl}?versionId=${versionId}`;
-    const startTime = new Date();
     const response = await fetchWithProxy(url);
 
     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;
     return myJson;
   } catch (error) {
-    logger.error(error);
-    stats.increment('get_libraries_code', { success: 'false', ...tags });
+    logger.error(`Error fetching library code for versionId: ${versionId}`, error.message);
     throw error;
   }
 }
 
 async function getRudderLibByImportName(importName) {
   const rudderLibrary = rudderLibraryCache[importName];
   if (rudderLibrary) return rudderLibrary;
-  const tags = {
-    libraryVersionId: importName,
-    version: 1,
-    type: 'rudderlibrary',
-  };
   try {
     const [name, version] = importName.split('/').slice(-2);
     const url = `${getRudderLibrariesUrl}/${name}?version=${version}`;
-    const startTime = new Date();
     const response = await fetchWithProxy(url);
 
     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;
     return myJson;
   } catch (error) {
-    logger.error(error);
-    stats.increment('get_libraries_code', { success: 'false', ...tags });
+    logger.error(`Error fetching rudder library code for importName: ${importName}`, error.message);
     throw error;
   }
 }

src/util/customTransforrmationsStore.js

@@ -2,7 +2,6 @@ const NodeCache = require('node-cache');
 const { fetchWithProxy } = require('./fetch');
 const logger = require('../logger');
 const { responseStatusHandler } = require('./utils');
-const stats = require('./stats');
 
 const myCache = new NodeCache({ stdTTL: 60 * 60 * 24 * 1 });
 
@@ -18,19 +17,14 @@ async function getTransformationCode(versionId) {
   if (transformation) return transformation;
   try {
     const url = `${getTransformationURL}?versionId=${versionId}`;
-    const startTime = new Date();
     const response = await fetchWithProxy(url);
 
     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);
     return myJson;
   } catch (error) {
-    logger.error(error);
-    stats.increment('get_transformation_code', { versionId, success: 'false' });
+    logger.error(`Error fetching transformation code for versionId: ${versionId}`, error.message);
     throw error;
   }
 }

src/util/ivmFactory.js

@@ -33,13 +33,13 @@ async function loadModule(isolateInternal, contextInternal, moduleName, moduleCo
 async function createIvm(
   code,
   libraryVersionIds,
-  versionId,
   transformationId,
   workspaceId,
   credentials,
   secrets,
   testMode,
 ) {
+  const trTags = { identifier: 'V1', transformationId, workspaceId };
   const createIvmStartTime = new Date();
   const logs = [];
   const libraries = await Promise.all(
@@ -187,9 +187,9 @@ async function createIvm(
     new ivm.Reference(async (resolve, ...args) => {
       try {
         const fetchStartTime = new Date();
-        const res = await fetchWithDnsWrapper(versionId, ...args);
+        const res = await fetchWithDnsWrapper(trTags, ...args);
         const data = await res.json();
-        stats.timing('fetch_call_duration', fetchStartTime, { versionId });
+        stats.timing('fetch_call_duration', fetchStartTime, trTags);
         resolve.applyIgnored(undefined, [new ivm.ExternalCopy(data).copyInto()]);
       } catch (error) {
         resolve.applyIgnored(undefined, [new ivm.ExternalCopy('ERROR').copyInto()]);
@@ -202,7 +202,7 @@ async function createIvm(
     new ivm.Reference(async (resolve, reject, ...args) => {
       try {
         const fetchStartTime = new Date();
-        const res = await fetchWithDnsWrapper(versionId, ...args);
+        const res = await fetchWithDnsWrapper(trTags, ...args);
         const headersContent = {};
         res.headers.forEach((value, header) => {
           headersContent[header] = value;
@@ -218,7 +218,7 @@ async function createIvm(
           data.body = JSON.parse(data.body);
         } catch (e) {}
 
-        stats.timing('fetchV2_call_duration', fetchStartTime, { versionId });
+        stats.timing('fetchV2_call_duration', fetchStartTime, trTags);
         resolve.applyIgnored(undefined, [new ivm.ExternalCopy(data).copyInto()]);
       } catch (error) {
         const err = JSON.parse(JSON.stringify(error, Object.getOwnPropertyNames(error)));
@@ -243,7 +243,7 @@ async function createIvm(
           throw new Error(`request to fetch geolocation failed with status code: ${res.status}`);
         }
         const geoData = await res.json();
-        stats.timing('geo_call_duration', geoStartTime, { versionId });
+        stats.timing('geo_call_duration', geoStartTime, trTags);
         resolve.applyIgnored(undefined, [new ivm.ExternalCopy(geoData).copyInto()]);
       } catch (error) {
         const err = JSON.parse(JSON.stringify(error, Object.getOwnPropertyNames(error)));
@@ -257,7 +257,7 @@ async function createIvm(
       logger.error(
         `Error fetching credentials map for transformationID: ${transformationId} and workspaceId: ${workspaceId}`,
       );
-      stats.increment('credential_error_total', { transformationId, workspaceId });
+      stats.increment('credential_error_total', trTags);
       return undefined;
     }
     if (key === null || key === undefined) {
@@ -416,7 +416,7 @@ async function createIvm(
     reference: true,
   });
   const fName = availableFuncNames[0];
-  stats.timing('createivm_duration', createIvmStartTime);
+  stats.timing('createivm_duration', createIvmStartTime, trTags);
   // TODO : check if we can resolve this
   // eslint-disable-next-line no-async-promise-executor
 
@@ -446,7 +446,6 @@ async function getFactory(
   libraryVersionIds,
   transformationId,
   workspaceId,
-  versionId,
   credentials,
   secrets,
   testMode,
@@ -456,7 +455,6 @@ async function getFactory(
       return createIvm(
         code,
         libraryVersionIds,
-        versionId,
         transformationId,
         workspaceId,
         credentials,