Merge branch 'develop' into feat.amplitudeSkipTraits

package.json

@@ -108,6 +108,7 @@
     "rudder-transformer-cdk": "^1.4.11",
     "set-value": "^4.1.0",
     "sha256": "^0.2.0",
+    "sqlstring": "^2.3.3",
     "stacktrace-parser": "^0.1.10",
     "statsd-client": "^0.4.7",
     "truncate-utf8-bytes": "^1.0.2",

src/controllers/userTransform.ts

@@ -15,9 +15,10 @@ export class UserTransformController {
       '(User transform - router:/customTransform ):: Request to transformer',
       JSON.stringify(ctx.request.body),
     );
+    const requestSize = Number(ctx.request.get('content-length'));
     const events = ctx.request.body as ProcessorTransformationRequest[];
     const processedRespone: UserTransformationServiceResponse =
-      await UserTransformService.transformRoutine(events, ctx.state.features);
+      await UserTransformService.transformRoutine(events, ctx.state.features, requestSize);
     ctx.body = processedRespone.transformedEvents;
     ControllerUtility.postProcess(ctx, processedRespone.retryStatus);
     logger.debug(

src/services/userTransform.ts

@@ -14,7 +14,7 @@ import {
   RetryRequestError,
   extractStackTraceUptoLastSubstringMatch,
 } from '../util/utils';
-import { getMetadata, isNonFuncObject } from '../v0/util';
+import { getMetadata, getTransformationMetadata, isNonFuncObject } from '../v0/util';
 import { SUPPORTED_FUNC_NAMES } from '../util/ivmFactory';
 import logger from '../logger';
 import stats from '../util/stats';
@@ -28,6 +28,7 @@ export class UserTransformService {
   public static async transformRoutine(
     events: ProcessorTransformationRequest[],
     features: FeatureFlags = {},
+    requestSize = 0,
   ): Promise<UserTransformationServiceResponse> {
     let retryStatus = 200;
     const groupedEvents: NonNullable<unknown> = groupBy(
@@ -162,16 +163,19 @@ export class UserTransformService {
             ),
           );
           stats.counter('user_transform_errors', eventsToProcess.length, {
-            transformationId: eventsToProcess[0]?.metadata?.transformationId,
-            workspaceId: eventsToProcess[0]?.metadata?.workspaceId,
             status,
             ...metaTags,
+            ...getTransformationMetadata(eventsToProcess[0]?.metadata),
           });
         } finally {
           stats.timing('user_transform_request_latency', userFuncStartTime, {
-            workspaceId: eventsToProcess[0]?.metadata?.workspaceId,
-            transformationId: eventsToProcess[0]?.metadata?.transformationId,
             ...metaTags,
+            ...getTransformationMetadata(eventsToProcess[0]?.metadata),
+          });
+
+          stats.histogram('user_transform_batch_size', requestSize, {
+            ...metaTags,
+            ...getTransformationMetadata(eventsToProcess[0]?.metadata),
           });
         }
 

src/util/prometheus.js

@@ -593,6 +593,10 @@ class Prometheus {
         name: 'tp_batch_size',
         help: 'Size of batch of events for tracking plan validation',
         type: 'histogram',
+        buckets: [
+          1024, 102400, 524288, 1048576, 10485760, 20971520, 52428800, 104857600, 209715200,
+          524288000,
+        ],
         labelNames: [
           'sourceType',
           'destinationType',
@@ -670,6 +674,22 @@ class Prometheus {
           'k8_namespace',
         ],
       },
+      {
+        name: 'user_transform_batch_size',
+        help: 'user_transform_batch_size',
+        type: 'histogram',
+        labelNames: [
+          'workspaceId',
+          'transformationId',
+          'sourceType',
+          'destinationType',
+          'k8_namespace',
+        ],
+        buckets: [
+          1024, 102400, 524288, 1048576, 10485760, 20971520, 52428800, 104857600, 209715200,
+          524288000,
+        ], // 1KB, 100KB, 0.5MB, 1MB, 10MB, 20MB, 50MB, 100MB, 200MB, 500MB
+      },
       {
         name: 'source_transform_request_latency',
         help: 'source_transform_request_latency',

src/v0/destinations/google_adwords_enhanced_conversions/networkHandler.js

@@ -1,6 +1,7 @@
 const { get, set } = require('lodash');
 const sha256 = require('sha256');
 const { NetworkError, NetworkInstrumentationError } = require('@rudderstack/integrations-lib');
+const SqlString = require('sqlstring');
 const { prepareProxyRequest, handleHttpRequest } = require('../../../adapters/network');
 const { isHttpStatusSuccess, getAuthErrCategoryFromStCode } = require('../../util/index');
 const { CONVERSION_ACTION_ID_CACHE_TTL } = require('./config');
@@ -29,8 +30,12 @@ const ERROR_MSG_PATH = 'response[0].error.message';
 const getConversionActionId = async (method, headers, params) => {
   const conversionActionIdKey = sha256(params.event + params.customerId).toString();
   return conversionActionIdCache.get(conversionActionIdKey, async () => {
+    const queryString = SqlString.format(
+      'SELECT conversion_action.id FROM conversion_action WHERE conversion_action.name = ?',
+      [params.event],
+    );
     const data = {
-      query: `SELECT conversion_action.id FROM conversion_action WHERE conversion_action.name = '${params.event}'`,
+      query: queryString,
     };
     const requestBody = {
       url: `${BASE_ENDPOINT}/${params.customerId}/googleAds:searchStream`,
@@ -117,7 +122,7 @@ const responseHandler = (responseParams) => {
     // Ref - https://github.com/googleapis/googleapis/blob/master/google/rpc/code.proto
     if (partialFailureError && partialFailureError.code !== 0) {
       throw new NetworkError(
-        `[Google Ads Offline Conversions]:: partialFailureError - ${JSON.stringify(
+        `[Google Adwords Enhanced Conversions]:: partialFailureError - ${JSON.stringify(
           partialFailureError,
         )}`,
         400,

src/v0/destinations/google_adwords_offline_conversions/utils.js

@@ -1,4 +1,5 @@
 const sha256 = require('sha256');
+const SqlString = require('sqlstring');
 const { get, set, cloneDeep } = require('lodash');
 const {
   AbortedError,
@@ -53,8 +54,12 @@ const validateDestinationConfig = ({ Config }) => {
 const getConversionActionId = async (headers, params) => {
   const conversionActionIdKey = sha256(params.event + params.customerId).toString();
   return conversionActionIdCache.get(conversionActionIdKey, async () => {
+    const queryString = SqlString.format(
+      'SELECT conversion_action.id FROM conversion_action WHERE conversion_action.name = ?',
+      [params.event],
+    );
     const data = {
-      query: `SELECT conversion_action.id FROM conversion_action WHERE conversion_action.name = '${params.event}'`,
+      query: queryString,
     };
     const endpoint = SEARCH_STREAM.replace(':customerId', params.customerId);
     const requestOptions = {

src/v0/destinations/pardot/networkHandler.js

@@ -46,6 +46,19 @@ const getStatus = (code) => {
 const pardotRespHandler = (destResponse, stageMsg) => {
   const { status, response } = destResponse;
   const respAttributes = response['@attributes'];
+
+  // to handle errors like service unavilable, wrong url, no response
+  if (!respAttributes) {
+    throw new NetworkError(
+      `${JSON.stringify(response)} ${stageMsg}`,
+      status,
+      {
+        [tags.TAG_NAMES.ERROR_TYPE]: getDynamicErrorType(status),
+      },
+      response,
+    );
+  }
+
   const { stat, err_code: errorCode } = respAttributes;
 
   if (isHttpStatusSuccess(status) && stat !== 'fail') {

src/v1/destinations/algolia/networkHandler.js

@@ -1,7 +1,7 @@
 /* eslint-disable no-restricted-syntax */
 const { TransformerProxyError } = require('../../../v0/util/errorTypes');
 const { prepareProxyRequest, proxyRequest } = require('../../../adapters/network');
-const { isHttpStatusSuccess, getAuthErrCategoryFromStCode } = require('../../../v0/util/index');
+const { isHttpStatusSuccess } = require('../../../v0/util/index');
 
 const {
   processAxiosResponse,
@@ -13,12 +13,6 @@ const responseHandler = (responseParams) => {
   const { destinationResponse, rudderJobMetadata } = responseParams;
   const message = `[ALGOLIA Response V1 Handler] - Request Processed Successfully`;
   const responseWithIndividualEvents = [];
-  // response:
-  // {status: 200, message: 'OK'}
-  // {response:'[ENOTFOUND] :: DNS lookup failed', status: 400}
-  // destinationResponse = {
-  //   response: {"status": 422, "message": "EventType must be one of \"click\", \"conversion\" or \"view\""}, status: 422
-  // }
   const { response, status } = destinationResponse;
 
   if (isHttpStatusSuccess(status)) {
@@ -41,35 +35,38 @@ const responseHandler = (responseParams) => {
 
   // in case of non 2xx status sending 500 for every event, populate response and update dontBatch to true
   const errorMessage = response?.error?.message || response?.message || 'unknown error format';
-  let serverStatus = 400;
   for (const metadata of rudderJobMetadata) {
-    // handling case if dontBatch is true, and again we got invalid from destination
-    if (metadata.dontBatch && status === 422) {
-      responseWithIndividualEvents.push({
-        statusCode: 400,
-        metadata,
-        error: errorMessage,
-      });
-    } else {
-      serverStatus = 500;
-      metadata.dontBatch = true;
-      responseWithIndividualEvents.push({
-        statusCode: 500,
-        metadata,
-        error: errorMessage,
-      });
-    }
+    metadata.dontBatch = true;
+    responseWithIndividualEvents.push({
+      statusCode: 500,
+      metadata,
+      error: errorMessage,
+    });
+  }
+
+  // At least one event in the batch is invalid.
+  if (status === 422) {
+    // sending back 500 for retry
+    throw new TransformerProxyError(
+      `ALGOLIA: Error transformer proxy v1 during ALGOLIA response transformation`,
+      500,
+      {
+        [tags.TAG_NAMES.ERROR_TYPE]: getDynamicErrorType(500),
+      },
+      destinationResponse,
+      '',
+      responseWithIndividualEvents,
+    );
   }
 
-  // sending back 500 for retry
   throw new TransformerProxyError(
     `ALGOLIA: Error transformer proxy v1 during ALGOLIA response transformation`,
-    serverStatus,
+    status,
     {
       [tags.TAG_NAMES.ERROR_TYPE]: getDynamicErrorType(status),
     },
     destinationResponse,
-    getAuthErrCategoryFromStCode(status),
+    '',
     responseWithIndividualEvents,
   );
 };