Merge tag '1.41.0' into cdb

Version 1.41.0

2016-10-21

Announcements:
 * Stop migrating old queues by default.

Bug fixes:
 * Fix some scenarios where batch queries got stuck waiting for available slots.

Author: rochoa

NEWS.md

@@ -1,3 +1,43 @@
+1.41.0 - 2016-10-21
+-------------------
+
+Announcements:
+ * Stop migrating old queues by default.
+
+Bug fixes:
+ * Fix some scenarios where batch queries got stuck waiting for available slots.
+
+
+1.40.0 - 2016-10-20
+-------------------
+
+New features:
+ * Batch queries are handled per db host.
+   - There is an scheduler controlling how many queries and in what order they are run.
+     - Priority is based on: number of queries already ran, and oldest user in queue.
+ * Batch queries capacity: allow to configure how many jobs to run per db host.
+
+
+1.39.1 - 2016-10-17
+-------------------
+
+Enhancements:
+ * Log creation and waiting time for fallback jobs' queries.
+
+
+1.39.0 - 2016-10-17
+-------------------
+
+Enhancements:
+ * Use just one Redis pool across the whole application.
+
+New features:
+ * Batch queries use per user-queues.
+ * Batch queries queues can limit the number of queued jobs per user.
+   - Default is 64 jobs.
+   - Configuration key `batch_max_queued_jobs` allows to modify the limit.
+
+
 1.38.2 - 2016-10-13
 -------------------
 

app/server.js

@@ -23,6 +23,7 @@ var _ = require('underscore');
 var LRU = require('lru-cache');
 
 var RedisPool = require('redis-mpool');
+var cartodbRedis = require('cartodb-redis');
 var UserDatabaseService = require('./services/user_database_service');
 var JobPublisher = require('../batch/pubsub/job-publisher');
 var JobQueue = require('../batch/job_queue');
@@ -53,14 +54,15 @@ function App() {
 
     var app = express();
 
-    var redisConfig = {
+    var redisPool = new RedisPool({
+        name: 'sql-api',
         host: global.settings.redis_host,
         port: global.settings.redis_port,
         max: global.settings.redisPool,
         idleTimeoutMillis: global.settings.redisIdleTimeoutMillis,
         reapIntervalMillis: global.settings.redisReapIntervalMillis
-    };
-    var metadataBackend = require('cartodb-redis')(redisConfig);
+    });
+    var metadataBackend = cartodbRedis({ pool: redisPool });
 
 
     // Set default configuration
@@ -181,8 +183,7 @@ function App() {
 
     var userDatabaseService = new UserDatabaseService(metadataBackend);
 
-    var redisPoolPublisher = new RedisPool(_.extend(redisConfig, { name: 'job-publisher'}));
-    var jobPublisher = new JobPublisher(redisPoolPublisher);
+    var jobPublisher = new JobPublisher(redisPool);
     var jobQueue = new JobQueue(metadataBackend, jobPublisher);
     var jobBackend = new JobBackend(metadataBackend, jobQueue);
     var userDatabaseMetadataService = new UserDatabaseMetadataService(metadataBackend);
@@ -212,7 +213,7 @@ function App() {
     if (global.settings.environment !== 'test' && isBatchProcess) {
         var batchName = global.settings.api_hostname || 'batch';
         app.batch = batchFactory(
-            metadataBackend, redisConfig, batchName, statsd_client, global.settings.batch_log_filename
+            metadataBackend, redisPool, batchName, statsd_client, global.settings.batch_log_filename
         );
         app.batch.start();
     }

batch/README.md

@@ -1,9 +1,40 @@
 # Batch Queries
 
-This document describes the currently supported query types, and what they are missing in terms of features.
+This document describes features from Batch Queries, it also details some internals that might be useful for maintainers
+and developers.
+
+
+## Redis data structures
+
+### Jobs definition
+
+Redis Hash: `batch:jobs:{UUID}`.
+
+Redis DB: 5.
+
+It stores the job definition, the user, and some metadata like the final status, the failure reason, and so.
+
+### Job queues
+
+Redis List: `batch:queue:{username}`.
+
+Redis DB: 5.
+
+It stores a pending list of jobs per user. It points to a job definition with the `{UUID}`.
+
+### Job notifications
+
+Redis Pub/Sub channel: `batch:users`.
+
+Redis DB: 0.
+
+In order to notify new jobs, it uses a Pub/Sub channel were the username for the queued job is published.
+
 
 ## Job types
 
+Format for the currently supported query types, and what they are missing in terms of features.
+
 ### Simple
 
 ```json

batch/batch.js

@@ -3,11 +3,12 @@
 var util = require('util');
 var EventEmitter = require('events').EventEmitter;
 var debug = require('./util/debug')('batch');
-var forever = require('./util/forever');
 var queue = require('queue-async');
-var Locker = require('./leader/locker');
+var HostScheduler = require('./scheduler/host-scheduler');
 
-function Batch(name, jobSubscriber, jobQueue, jobRunner, jobService, jobPublisher, redisConfig, logger) {
+var EMPTY_QUEUE = true;
+
+function Batch(name, jobSubscriber, jobQueue, jobRunner, jobService, jobPublisher, redisPool, logger) {
     EventEmitter.call(this);
     this.name = name || 'batch';
     this.jobSubscriber = jobSubscriber;
@@ -16,10 +17,10 @@ function Batch(name, jobSubscriber, jobQueue, jobRunner, jobService, jobPublishe
     this.jobService = jobService;
     this.jobPublisher = jobPublisher;
     this.logger = logger;
-    this.locker = Locker.create('redis-distlock', { redisConfig: redisConfig });
+    this.hostScheduler = new HostScheduler(name, { run: this.processJob.bind(this) }, redisPool);
 
-    // map: host => jobId
-    this.workingQueues = {};
+    // map: user => jobId. Will be used for draining jobs.
+    this.workInProgressJobs = {};
 }
 util.inherits(Batch, EventEmitter);
 
@@ -29,33 +30,16 @@ Batch.prototype.start = function () {
     var self = this;
 
     this.jobSubscriber.subscribe(
-        function onJobHandler(host) {
-            if (self.isProcessingHost(host)) {
-                return debug('%s is already processing host=%s', self.name, host);
-            }
-
-            // do forever, it does not throw a stack overflow
-            forever(
-                function (next) {
-                    self.locker.lock(host, function(err) {
-                        // we didn't get the lock for the host
-                        if (err) {
-                            debug('Could not lock host=%s from %s. Reason: %s', host, self.name, err.message);
-                            return next(err);
-                        }
-                        debug('Locked host=%s from %s', host, self.name);
-                        self.processNextJob(host, next);
-                    });
-                },
-                function (err) {
-                    if (err) {
-                        debug(err.name === 'EmptyQueue' ? err.message : err);
-                    }
-
-                    self.finishedProcessingHost(host);
-                    self.locker.unlock(host, debug);
+        function onJobHandler(user, host) {
+            debug('[%s] onJobHandler(%s, %s)', self.name, user, host);
+            self.hostScheduler.add(host, user, function(err) {
+                if (err) {
+                    return debug(
+                        'Could not schedule host=%s user=%s from %s. Reason: %s',
+                        host, self.name, user, err.message
+                    );
                 }
-            );
+            });
         },
         function onJobSubscriberReady(err) {
             if (err) {
@@ -67,50 +51,49 @@ Batch.prototype.start = function () {
     );
 };
 
-Batch.prototype.processNextJob = function (host, callback) {
+Batch.prototype.processJob = function (user, callback) {
     var self = this;
-    self.jobQueue.dequeue(host, function (err, jobId) {
+    self.jobQueue.dequeue(user, function (err, jobId) {
         if (err) {
-            return callback(err);
+            return callback(new Error('Could not get job from "' + user + '". Reason: ' + err.message), !EMPTY_QUEUE);
         }
 
         if (!jobId) {
-            var emptyQueueError = new Error('Queue ' + host + ' is empty');
-            emptyQueueError.name = 'EmptyQueue';
-            return callback(emptyQueueError);
+            debug('Queue empty user=%s', user);
+            return callback(null, EMPTY_QUEUE);
         }
 
-        self.setProcessingJobId(host, jobId);
-
+        self.setWorkInProgressJob(user, jobId);
         self.jobRunner.run(jobId, function (err, job) {
-            self.setProcessingJobId(host, null);
+            self.clearWorkInProgressJob(user);
 
             if (err) {
                 debug(err);
                 if (err.name === 'JobNotRunnable') {
-                    return callback();
+                    return callback(null, !EMPTY_QUEUE);
                 }
-                return callback(err);
+                return callback(err, !EMPTY_QUEUE);
             }
 
-            debug('Job[%s] status=%s in host=%s (failed_reason=%s)', jobId, job.data.status, host, job.failed_reason);
+            debug(
+                '[%s] Job=%s status=%s user=%s (failed_reason=%s)',
+                self.name, jobId, job.data.status, user, job.failed_reason
+            );
 
             self.logger.log(job);
 
-            self.emit('job:' + job.data.status, jobId);
-
-            callback();
+            return callback(null, !EMPTY_QUEUE);
         });
     });
 };
 
 Batch.prototype.drain = function (callback) {
     var self = this;
-    var workingHosts = this.getWorkingHosts();
-    var batchQueues = queue(workingHosts.length);
+    var workingUsers = this.getWorkInProgressUsers();
+    var batchQueues = queue(workingUsers.length);
 
-    workingHosts.forEach(function (host) {
-        batchQueues.defer(self._drainJob.bind(self), host);
+    workingUsers.forEach(function (user) {
+        batchQueues.defer(self._drainJob.bind(self), user);
     });
 
     batchQueues.awaitAll(function (err) {
@@ -124,9 +107,9 @@ Batch.prototype.drain = function (callback) {
     });
 };
 
-Batch.prototype._drainJob = function (host, callback) {
+Batch.prototype._drainJob = function (user, callback) {
     var self = this;
-    var job_id = this.getProcessingJobId(host);
+    var job_id = this.getWorkInProgressJob(user);
 
     if (!job_id) {
         return process.nextTick(function () {
@@ -143,7 +126,7 @@ Batch.prototype._drainJob = function (host, callback) {
             return callback(err);
         }
 
-        self.jobQueue.enqueueFirst(job_id, host, callback);
+        self.jobQueue.enqueueFirst(user, job_id, callback);
     });
 };
 
@@ -152,22 +135,21 @@ Batch.prototype.stop = function (callback) {
     this.jobSubscriber.unsubscribe(callback);
 };
 
-Batch.prototype.isProcessingHost = function(host) {
-    return this.workingQueues.hasOwnProperty(host);
-};
 
-Batch.prototype.getWorkingHosts = function() {
-    return Object.keys(this.workingQueues);
+/* Work in progress jobs */
+
+Batch.prototype.setWorkInProgressJob = function(user, jobId) {
+    this.workInProgressJobs[user] = jobId;
 };
 
-Batch.prototype.setProcessingJobId = function(host, jobId) {
-    this.workingQueues[host] = jobId;
+Batch.prototype.getWorkInProgressJob = function(user) {
+    return this.workInProgressJobs[user];
 };
 
-Batch.prototype.getProcessingJobId = function(host) {
-    return this.workingQueues[host];
+Batch.prototype.clearWorkInProgressJob = function(user) {
+    delete this.workInProgressJobs[user];
 };
 
-Batch.prototype.finishedProcessingHost = function(host) {
-    delete this.workingQueues[host];
+Batch.prototype.getWorkInProgressUsers = function() {
+    return Object.keys(this.workInProgressJobs);
 };

batch/index.js

@@ -1,7 +1,5 @@
 'use strict';
 
-var RedisPool = require('redis-mpool');
-var _ = require('underscore');
 var JobRunner = require('./job_runner');
 var QueryRunner = require('./query_runner');
 var JobCanceller = require('./job_canceller');
@@ -14,18 +12,18 @@ var JobService = require('./job_service');
 var BatchLogger = require('./batch-logger');
 var Batch = require('./batch');
 
-module.exports = function batchFactory (metadataBackend, redisConfig, name, statsdClient, loggerPath) {
-    var pubSubRedisPool = new RedisPool(_.extend({ name: 'batch-pubsub'}, redisConfig));
-    var jobSubscriber = new JobSubscriber(pubSubRedisPool);
-    var jobPublisher = new JobPublisher(pubSubRedisPool);
+module.exports = function batchFactory (metadataBackend, redisPool, name, statsdClient, loggerPath) {
+    var userDatabaseMetadataService = new UserDatabaseMetadataService(metadataBackend);
+
+    var jobSubscriber = new JobSubscriber(redisPool, userDatabaseMetadataService);
+    var jobPublisher = new JobPublisher(redisPool);
 
     var jobQueue =  new JobQueue(metadataBackend, jobPublisher);
     var jobBackend = new JobBackend(metadataBackend, jobQueue);
-    var userDatabaseMetadataService = new UserDatabaseMetadataService(metadataBackend);
     var queryRunner = new QueryRunner(userDatabaseMetadataService);
     var jobCanceller = new JobCanceller(userDatabaseMetadataService);
     var jobService = new JobService(jobBackend, jobCanceller);
-    var jobRunner = new JobRunner(jobService, jobQueue, queryRunner, statsdClient);
+    var jobRunner = new JobRunner(jobService, jobQueue, queryRunner, metadataBackend, statsdClient);
     var logger = new BatchLogger(loggerPath);
 
     return new Batch(
@@ -35,7 +33,7 @@ module.exports = function batchFactory (metadataBackend, redisConfig, name, stat
         jobRunner,
         jobService,
         jobPublisher,
-        redisConfig,
+        redisPool,
         logger
     );
 };