fix: marketo bulk upload bugs and refactor (#2414)

* fix marketo bulk upload * fix: marketo bulk upload poll * fix: review comments addressed * fix: marketo fixes * fix: marketo fixes * fix: marketo fixes * fix: marketo bugs * fix: edit marketo error handling * fix: adding comments and refs * fix: prometheus tags and labels * fix: adding default status code for poll * fix: fixing resp parsing * fix: update prometheus.js * fix: marketo bulk upload refactor (#2505) * fix: initial commit * fix: initial commit * fix: small edit * fix: small edit * fix: small edit * fix: edit part 2 * fix: bug fix * fix: small edit * fix: refactor * fix: small fix * fix: file upload related review comments * fix: adding function doc strings * fix: adding error message in getImportId * fix: deduplication of code * fix: bug fixes * fix: review address * fix: handling warning jobs * fix: adding test cases * fix: adding doc string * fix: making the job status response server competent * fix: review comments addressed * chore: clean up * fix: review comments addressed * fix: review comments addressed * fix: bug * fix: bug * fix: refactor, sonar issue solve and formatting * fix: the data type check for fetch job status * fix: comment delete * fix: comment delete * fix: issue solve * fix: review comments address part 1 * fix: review comments address part 2 * fix: review comments address part 3 * fix: update status code for 4XX scenarios * fix: the poll activity stat * fix: the poll status return and code clean up * fix: update error handling * fix: test cases fix --------- Co-authored-by: shrouti1507 <[email protected]> Co-authored-by: shrouti1507 <[email protected]> Co-authored-by: Chandra shekar Varkala <[email protected]>
rudderlabs · Sep 4, 2023 · 9e3ace1 · 9e3ace1
1 parent 360e09a
commit 9e3ace1
Show file tree

Hide file tree

Showing 11 changed files with 1,015 additions and 677 deletions.
diff --git a/src/controllers/bulkUpload.ts b/src/controllers/bulkUpload.ts
@@ -44,7 +44,7 @@ export const fileUpload = async (ctx) => {
     response = await destFileUploadHandler.processFileData(ctx.request.body);
   } catch (error: any) {
     response = {
-      statusCode: error.response ? error.response.status : 400,
+      statusCode: error?.response?.status || error?.status || 400,
       error: error.message || 'Error occurred while processing payload.',
       metadata: error.response ? error.response.metadata : null,
     };
@@ -89,7 +89,7 @@ export const pollStatus = async (ctx) => {
     response = await destFileUploadHandler.processPolling(ctx.request.body);
   } catch (error: any) {
     response = {
-      statusCode: error.response?.status || 400,
+      statusCode: error.response?.status || error?.status || 400,
       error: error.message || 'Error occurred while processing payload.',
     };
     errNotificationClient.notify(error, 'Poll Status', {

diff --git a/src/util/prometheus.js b/src/util/prometheus.js
@@ -417,7 +417,7 @@ class Prometheus {
         name: 'marketo_bulk_upload_polling',
         help: 'marketo_bulk_upload_polling',
         type: 'counter',
-        labelNames: ['status', 'state'],
+        labelNames: ['status', 'state', 'requestTime'],
       },
       {
         name: 'marketo_fetch_token',
@@ -776,6 +776,18 @@ class Prometheus {
         type: 'counter',
         labelNames: ['http_status', 'destination_id'],
       },
+      {
+        name: 'marketo_bulk_upload_upload_file_succJobs',
+        help: 'marketo_bulk_upload_upload_file_succJobs',
+        type: 'counter',
+        labelNames: [],
+      },
+      {
+        name: 'marketo_bulk_upload_upload_file_unsuccJobs',
+        help: 'marketo_bulk_upload_upload_file_unsuccJobs',
+        type: 'counter',
+        labelNames: [],
+      },
       {
         name: 'braze_lookup_time',
         help: 'braze look-up time',
@@ -861,6 +873,43 @@ class Prometheus {
         labelNames: ['processSessions'],
         buckets: [10, 20, 30, 40, 50, 60, 70, 80, 90, 100, 150, 200],
       },
+      {
+        name: 'marketo_bulk_upload_create_header_time',
+        help: 'marketo_bulk_upload_create_header_time',
+        type: 'histogram',
+        labelNames: [],
+      },
+      {
+        name: 'marketo_bulk_upload_fetch_job_time',
+        help: 'marketo_bulk_upload_fetch_job_time',
+        type: 'histogram',
+        labelNames: [],
+      },
+      {
+        name: 'marketo_bulk_upload_fetch_job_create_response_time',
+        help: 'marketo_bulk_upload_fetch_job_create_response_time',
+        type: 'histogram',
+        labelNames: [],
+      },
+      {
+        name: 'marketo_bulk_upload_create_file_time',
+        help: 'marketo_bulk_upload_create_file_time',
+        type: 'histogram',
+        labelNames: [],
+      },
+      {
+        name: 'marketo_bulk_upload_upload_file_time',
+        help: 'marketo_bulk_upload_upload_file_time',
+        type: 'histogram',
+        labelNames: [],
+      },
+      {
+        name: 'marketo_bulk_upload_create_csvloop_time',
+        help: 'marketo_bulk_upload_create_csvloop_time',
+        type: 'histogram',
+        labelNames: [],
+      },
+
     ];
 
     metrics.forEach((metric) => {

diff --git a/src/v0/destinations/marketo_bulk_upload/config.js b/src/v0/destinations/marketo_bulk_upload/config.js
@@ -0,0 +1,36 @@
+const ABORTABLE_CODES = ['601', '603', '605', '609', '610'];
+const RETRYABLE_CODES = ['713', '602', '604', '611'];
+const THROTTLED_CODES = ['502', '606', '607', '608', '615'];
+
+const MARKETO_FILE_SIZE = 10485760;
+const MARKETO_FILE_PATH = `${__dirname}/uploadFile/marketo_bulkupload.csv`;
+
+const FETCH_ACCESS_TOKEN = 'marketo_bulk_upload_access_token_fetching';
+
+const POLL_ACTIVITY = 'marketo_bulk_upload_polling';
+const POLL_STATUS_ERR_MSG = 'Could not poll status';
+
+const UPLOAD_FILE = 'marketo_bulk_upload_upload_file';
+const FILE_UPLOAD_ERR_MSG = 'Could not upload file';
+
+const JOB_STATUS_ACTIVITY = 'marketo_bulk_upload_get_job_status';
+const FETCH_FAILURE_JOB_STATUS_ERR_MSG = 'Could not fetch failure job status';
+const FETCH_WARNING_JOB_STATUS_ERR_MSG = 'Could not fetch warning job status';
+const ACCESS_TOKEN_FETCH_ERR_MSG = 'Error during fetching access token';
+
+module.exports = {
+  ABORTABLE_CODES,
+  RETRYABLE_CODES,
+  THROTTLED_CODES,
+  MARKETO_FILE_SIZE,
+  POLL_ACTIVITY,
+  UPLOAD_FILE,
+  JOB_STATUS_ACTIVITY,
+  MARKETO_FILE_PATH,
+  FETCH_ACCESS_TOKEN,
+  POLL_STATUS_ERR_MSG,
+  FILE_UPLOAD_ERR_MSG,
+  FETCH_FAILURE_JOB_STATUS_ERR_MSG,
+  FETCH_WARNING_JOB_STATUS_ERR_MSG,
+  ACCESS_TOKEN_FETCH_ERR_MSG,
+};