Sleep after request to give es processor time to breathe

Author: callmekatootie

config/default.js

@@ -24,7 +24,7 @@ module.exports = {
 
   TOPCODER_USERS_API: process.env.TOPCODER_USERS_API || 'https://api.topcoder-dev.com/v3/users',
 
-  PROCESS_CONCURRENCY_COUNT: process.env.PROCESS_CONCURRENCY_COUNT ? parseInt(process.env.PROCESS_CONCURRENCY_COUNT, 10) : 25,
+  PROCESS_CONCURRENCY_COUNT: process.env.PROCESS_CONCURRENCY_COUNT ? parseInt(process.env.PROCESS_CONCURRENCY_COUNT, 10) : 1,
 
   // When a user is not found, should we proceed to create the user or treat it as error
   CREATE_MISSING_USER_FLAG: process.env.CREATE_MISSING_USER_FLAG === 'true',
@@ -35,5 +35,7 @@ module.exports = {
   TOKEN_CACHE_TIME: process.env.TOKEN_CACHE_TIME,
   AUTH0_CLIENT_ID: process.env.AUTH0_CLIENT_ID,
   AUTH0_CLIENT_SECRET: process.env.AUTH0_CLIENT_SECRET,
-  AUTH0_PROXY_SERVER_URL: process.env.AUTH0_PROXY_SERVER_URL
+  AUTH0_PROXY_SERVER_URL: process.env.AUTH0_PROXY_SERVER_URL,
+
+  SLEEP_TIME: process.env.SLEEP_TIME ? parseInt(process.env.SLEEP_TIME, 10) : 1000
 }

src/common/helper.js

@@ -18,6 +18,21 @@ const topcoderM2MConfig = _.pick(config, ['AUTH0_URL', 'AUTH0_TOPCODER_AUDIENCE'
 const ubahnM2M = m2mAuth({ ...ubahnM2MConfig, AUTH0_AUDIENCE: ubahnM2MConfig.AUTH0_UBAHN_AUDIENCE })
 const topcoderM2M = m2mAuth({ ...topcoderM2MConfig, AUTH0_AUDIENCE: topcoderM2MConfig.AUTH0_TOPCODER_AUDIENCE })
 
+/**
+ * Use this function to halt execution
+ * js version of sleep()
+ * @param {Number} ms Timeout in ms
+ */
+async function sleep (ms) {
+  if (!ms) {
+    ms = config.SLEEP_TIME
+  }
+
+  logger.debug(`Sleeping for ${ms} ms`)
+
+  return new Promise(resolve => setTimeout(resolve, ms))
+}
+
 /* Function to get M2M token
  * (U-Bahn APIs only)
  * @returns {Promise}
@@ -118,6 +133,9 @@ async function getUbahnSingleRecord (path, params, isOptionRecord) {
         return status >= 200 && status < 300
       }
     })
+
+    await sleep()
+
     if (_.isArray(res.data)) {
       if (res.data.length === 1) {
         return res.data[0]
@@ -163,6 +181,9 @@ async function createUbahnRecord (path, data) {
   logger.debug(`request POST ${path} with data: ${JSON.stringify(data)}`)
   try {
     const res = await axios.post(`${config.UBAHN_API_URL}${path}`, data, { headers: { Authorization: `Bearer ${token}` } })
+
+    await sleep()
+
     return res.data
   } catch (err) {
     logger.error(err)
@@ -182,6 +203,9 @@ async function updateUBahnRecord (path, data) {
   logger.debug(`request PATCH ${path} with data: ${JSON.stringify(data)}`)
   try {
     const res = await axios.patch(`${config.UBAHN_API_URL}${path}`, data, { headers: { Authorization: `Bearer ${token}` } })
+
+    await sleep()
+
     return res.data
   } catch (err) {
     logger.error(err)
@@ -210,6 +234,9 @@ async function getUserInTopcoder (handle) {
 
   try {
     const res = await axios.get(url, { headers: { Authorization: `Bearer ${token}` }, params })
+
+    await sleep()
+
     return res.data
   } catch (err) {
     logger.error(err)
@@ -236,6 +263,9 @@ async function createUserInTopcoder (user) {
   logger.debug(`request POST ${url} with data: ${JSON.stringify(user)}`)
   try {
     const res = await axios.post(url, requestBody, { headers: { Authorization: `Bearer ${token}` } })
+
+    await sleep()
+
     return res.data
   } catch (err) {
     logger.error(err)
@@ -252,6 +282,9 @@ async function createUserInTopcoder (user) {
 async function updateProcessStatus (id, data) {
   const token = await getUbahnM2Mtoken()
   const res = await axios.patch(`${config.UBAHN_SEARCH_UI_API_URL}/uploads/${id}`, data, { headers: { Authorization: `Bearer ${token}` } })
+
+  await sleep()
+
   return res.data
 }
 

src/services/ProcessorService.js

@@ -241,6 +241,8 @@ async function processCreateRecord (record, failedRecord, organizationId) {
 async function processCreate (message) {
   const { status } = message.payload
 
+  logger.info(`Concurrency count set at ${config.PROCESS_CONCURRENCY_COUNT} with type ${typeof config.process.env.PAUSE_AFTER_REQUEST}`)
+
   if (status === 'pending') {
     try {
       const file = await helper.downloadFile(message.payload.objectKey)