Merge pull request #210 from telefonicaid/retryBucketsLUA

Retry buckets using LUA

Author: Aitor Magán García

lib/configBase.js

@@ -71,6 +71,15 @@ exports.queue = {redisHost:'localhost', redisPort:6379};
  */
 exports.expireTime = 60 * 60;
 
+/*
+ Retry buckets
+ When a petition fails it will be inserted on a bucket (every time on one different: from the first one to the last one)
+ Petitions inserted on a bucket will be retried sometime in the future.
+ This property defines when petitions should be retried.
+ The number of elements in this array will establish the number of buckets
+ */
+exports.retryBuckets = [1, 10, 60];
+
 
 /*
  HIGH AVAILABILITY CONFIGURATION
@@ -106,15 +115,6 @@ exports.longFailingTimeout = 10000;
  */
 exports.minQuorum = 2;
 
-/*
- Retry buckets
- When a petition fails it will be inserted on a bucket (every time on one different: from the first one to the last one)
- Petitions inserted on a bucket will be retried sometime in the future.
- This property defines when petitions should be retried.
- The number of elements in this array will establish the number of buckets
- */
-exports.retryBuckets = [1, 10, 60];
-
 
 /* LISTENER AND CONSUMER CONFIGURATION
  ======================================

lib/listener.js

@@ -73,6 +73,7 @@ function assignRequest(request, data, callback) {
   function processTask(err, routeObj) {
     if (!err) {
 
+      routeObj.task.service = routeObj.service;
       store.put(routeObj.service, routeObj.task, function onWrittenReq(err) {
         var st;
         if (err) {

lib/retryBuckets.js

@@ -28,17 +28,32 @@ logger.prefix = path.basename(module.filename, '.js');
 
 var RETRY_BUCKETS = config.retryBuckets;
 var BUCKET_PREFIX = 'bucket';
-var FIRST_CHARCODE = "A".charCodeAt(0); //Needed to auto-create bucket names (bucketA, bucketB,...)
+var FIRST_CHARCODE = 'A'.charCodeAt(0); //Needed to auto-create bucket names (bucketA, bucketB,...)
 
 var stopped;
 
-var lrangeAndDelete = "\
+var getTasksAndRepsuh = "\
 local bucketName = KEYS[1]\n\
+local serviceQueue = KEYS[2]\n\
 local tasks = redis.call('lrange', bucketName, 0, -1)\n\
-redis.call('del', bucketName)\n\
+local size = table.getn(tasks)\n\
+if size > 0 then\n\
+  redis.call('lpush', serviceQueue, unpack(tasks))\n\
+  redis.call('del', bucketName)\n\
+end\n\
 return tasks\n\
 ";
 
+/**
+ * Get bucket name
+ * @param index Position on buckets array
+ * @param service Service name
+ * @returns {string} The name of the bucket
+ */
+function getBucketName(index, service) {
+  return BUCKET_PREFIX + String.fromCharCode(FIRST_CHARCODE + index) + ':' + service;
+}
+
 /**
  *
  * @param task The task to be pushed on the bucket
@@ -62,7 +77,7 @@ exports.insertOnBucket = function (task, error, maxRetry, callback) {
 
   } else {
 
-    var bucketName = BUCKET_PREFIX + String.fromCharCode(FIRST_CHARCODE + retried);
+    var bucketName = getBucketName(retried, task.service);
     var bucketTime = RETRY_BUCKETS[retried];
     var db = dbCluster.getDb(bucketName);
 
@@ -98,105 +113,62 @@ exports.insertOnBucket = function (task, error, maxRetry, callback) {
 
 exports.initBucketTimers = function() {
 
-  function repushBucketTasks(bucketName, cb) {
+  function repushBucketTasks(bucketName, serviceQueue, cb) {
 
     var db = dbCluster.getDb(bucketName);
 
     //Get all tasks from the bucket
-    db.eval(lrangeAndDelete, 1, bucketName,  function (errEval, data) {
-
+    db.eval(getTasksAndRepsuh, 2, bucketName, serviceQueue, function (errEval, data) {
       if (!errEval && data.length > 0) {
-
-        //Insert on the queue again (using the standard method: as listener)
-        async.map(data,
-
-            //Iterate function
-            function(task, callback) {
-              router.route(JSON.parse(task), callback);
-            },
-
-            //Function to be called when all items of the array has been processed
-            function (err, results) {
-
-              if (err) {
-
-                logger.error('Error routing tasks to be respushed', { op: 'REPUSH BUCKET TASKS', error: err });
-                cb();
-
-              } else {
-
-                //Get tasks for each service
-                var putArguments = {};
-                for (var i = 0; i < results.length; i++) {
-                  var service = results[i].service;
-                  var task = results[i].task;
-                  if (!putArguments[service]) {
-                    putArguments[service] = [];
-                  }
-                  putArguments[service].push(task);
-                }
-
-                //For each service, repush all the tasks at the same time
-                //In order to improve efficiency, only up to 100 tasks will be repushed at the same time
-                var putFunctions = [];
-                for (service in putArguments) {
-                  if (putArguments.hasOwnProperty(service)) {
-                    var concurrentPushes = MG.CONCURRENT_PUSHES;
-                    for (i=0; i < putArguments[service].length; i += concurrentPushes){
-                      putFunctions.push(store.put.bind(this, service,
-                          putArguments[service].slice(i, i + concurrentPushes)));
-                    }
-                  }
-                }
-
-                async.series(putFunctions, cb);
-              }
-            }
-        );
-
+        logger.info('Tasks repushed', { op: 'REPUSH BUCKET TASKS', tasks: data });
+        cb();
       } else if (errEval) {
-
         logger.error('Error getting bucket elements', { op: 'RESPUSH BUCKET TASKS', error: errEval });
         cb();
-
       } else if (data.length === 0) {
-
         logger.debug('Empty bucket', { op: 'REPUSH BUCKET TASKS'});
         cb();
-
       }
     });
   }
 
+  //Stop condition
   stopped = false;
 
-  RETRY_BUCKETS.forEach(function(bucket, index) {
+  var services = router.getQueues();
 
-    var bucketName = BUCKET_PREFIX + String.fromCharCode(FIRST_CHARCODE + index);
-    var intervalTime = bucket * 2 * 1000;
+  for (var service in services) {
+    if (services.hasOwnProperty(service)) {
 
-    async.whilst(
+      RETRY_BUCKETS.forEach(function(bucketTimeout, index) {
 
-        //Stop condition
-        function() {
-          return stopped === false;
-        },
+        var serviceName = services[service];
+        var bucketName = getBucketName(index, serviceName);
+        var intervalTime = bucketTimeout * 2 * 1000;
 
-        //Function to be executed forever
-        function (callback) {
-          repushBucketTasks(bucketName, function() {
-            setTimeout(callback, intervalTime);
-          })
-        },
+        async.whilst(
 
-        //Function to be called on error
-        function(err) {
-          logger.error('Error processing respush', { op: 'BUCKET TIMER', error: err });
-        }
-    );
+            //Stop condition
+            function() {
+              return stopped === false;
+            },
 
-  });
+            //Function to be executed forever
+            function (callback) {
+              repushBucketTasks(bucketName, serviceName, function() {
+                setTimeout(callback, intervalTime);
+              })
+            },
 
+            //Function to be called on error
+            function(err) {
+              logger.error('Error processing respush', { op: 'BUCKET TIMER', error: err });
+            }
+        );
+
+      });
+    }
+  }
 };
 
 exports.stopTimers = function() {