Merge tag '1.42.3' into cdb

1.42.3 - 2016-11-07
-------------------

Announcements:
 * Raise payload limit for batch-queries to 16kb.

Author: Daniel García Aubert

NEWS.md

@@ -1,3 +1,31 @@
+1.42.3 - 2016-11-07
+-------------------
+
+Announcements:
+ * Raise payload limit for batch-queries to 16kb.
+
+
+1.42.2 - 2016-11-07
+-------------------
+
+Bug fixes:
+ * Improve error handling while registering jobs to be tracked.
+
+
+1.42.1 - 2016-11-03
+-------------------
+
+Bug fixes:
+ * Avoid to use SCAN command to find work-in-progress queues.
+
+
+1.42.0 - 2016-11-02
+-------------------
+
+Announcements:
+ * Adds endpoint to check running batch queries
+
+
 1.41.0 - 2016-10-21
 -------------------
 

app/controllers/job_controller.js

@@ -8,7 +8,7 @@ var authenticatedMiddleware = require('../middlewares/authenticated-request');
 var handleException = require('../utils/error_handler');
 
 var ONE_KILOBYTE_IN_BYTES = 1024;
-var MAX_LIMIT_QUERY_SIZE_IN_KB = 8;
+var MAX_LIMIT_QUERY_SIZE_IN_KB = 16;
 var MAX_LIMIT_QUERY_SIZE_IN_BYTES = MAX_LIMIT_QUERY_SIZE_IN_KB * ONE_KILOBYTE_IN_BYTES;
 
 function getMaxSizeErrorMessage(sql) {
@@ -48,6 +48,11 @@ JobController.prototype.route = function (app) {
         bodyPayloadSizeMiddleware, userMiddleware, authenticatedMiddleware(this.userDatabaseService),
         this.createJob.bind(this)
     );
+    app.get(
+        global.settings.base_url + '/sql/job/wip',
+        userMiddleware, authenticatedMiddleware(this.userDatabaseService),
+        this.listWorkInProgressJobs.bind(this)
+    );
     app.get(
         global.settings.base_url + '/sql/job/:job_id',
         userMiddleware, authenticatedMiddleware(this.userDatabaseService),
@@ -82,6 +87,36 @@ JobController.prototype.createJob = function (req, res) {
     this.jobService.create(data, jobResponse(req, res, this.statsdClient, 'create', 201));
 };
 
+JobController.prototype.listWorkInProgressJobs = function (req, res) {
+    var self = this;
+
+    this.jobService.listWorkInProgressJobs(function (err, list) {
+        if (err) {
+            self.statsdClient.increment('sqlapi.job.error');
+            return handleException(err, res);
+        }
+
+        res.header('X-Served-By-DB-Host', req.context.userDatabase.host);
+
+        req.profiler.done('list');
+        req.profiler.end();
+        req.profiler.sendStats();
+
+        res.header('X-SQLAPI-Profiler', req.profiler.toJSONString());
+        self.statsdClient.increment('sqlapi.job.success');
+
+        if (process.env.NODE_ENV !== 'test') {
+            console.info(JSON.stringify({
+                type: 'sql_api_batch_job',
+                username: req.context.user,
+                action: 'list'
+            }));
+        }
+
+        res.status(200).send(list);
+    });
+};
+
 function jobResponse(req, res, statsdClient, action, status) {
     return function handler(err, job) {
         status = status || 200;

batch/batch.js

@@ -17,7 +17,7 @@ function Batch(name, jobSubscriber, jobQueue, jobRunner, jobService, jobPublishe
     this.jobService = jobService;
     this.jobPublisher = jobPublisher;
     this.logger = logger;
-    this.hostScheduler = new HostScheduler(name, { run: this.processJob.bind(this) }, redisPool);
+    this.hostScheduler = new HostScheduler(this.name, { run: this.processJob.bind(this) }, redisPool);
 
     // map: user => jobId. Will be used for draining jobs.
     this.workInProgressJobs = {};
@@ -63,10 +63,7 @@ Batch.prototype.processJob = function (user, callback) {
             return callback(null, EMPTY_QUEUE);
         }
 
-        self.setWorkInProgressJob(user, jobId);
-        self.jobRunner.run(jobId, function (err, job) {
-            self.clearWorkInProgressJob(user);
-
+        self._processWorkInProgressJob(user, jobId, function (err, job) {
             if (err) {
                 debug(err);
                 if (err.name === 'JobNotRunnable') {
@@ -87,6 +84,26 @@ Batch.prototype.processJob = function (user, callback) {
     });
 };
 
+Batch.prototype._processWorkInProgressJob = function (user, jobId, callback) {
+    var self = this;
+
+    self.setWorkInProgressJob(user, jobId, function (errSet) {
+        if (errSet) {
+            debug(new Error('Could not add job to work-in-progress list. Reason: ' + errSet.message));
+        }
+
+        self.jobRunner.run(jobId, function (err, job) {
+            self.clearWorkInProgressJob(user, jobId, function (errClear) {
+                if (errClear) {
+                    debug(new Error('Could not clear job from work-in-progress list. Reason: ' + errClear.message));
+                }
+
+                return callback(err, job);
+            });
+        });
+    });
+};
+
 Batch.prototype.drain = function (callback) {
     var self = this;
     var workingUsers = this.getWorkInProgressUsers();
@@ -138,16 +155,18 @@ Batch.prototype.stop = function (callback) {
 
 /* Work in progress jobs */
 
-Batch.prototype.setWorkInProgressJob = function(user, jobId) {
+Batch.prototype.setWorkInProgressJob = function(user, jobId, callback) {
     this.workInProgressJobs[user] = jobId;
+    this.jobService.addWorkInProgressJob(user, jobId, callback);
 };
 
 Batch.prototype.getWorkInProgressJob = function(user) {
     return this.workInProgressJobs[user];
 };
 
-Batch.prototype.clearWorkInProgressJob = function(user) {
+Batch.prototype.clearWorkInProgressJob = function(user, jobId, callback) {
     delete this.workInProgressJobs[user];
+    this.jobService.clearWorkInProgressJob(user, jobId, callback);
 };
 
 Batch.prototype.getWorkInProgressUsers = function() {

batch/job_backend.js

@@ -3,12 +3,15 @@
 var REDIS_PREFIX = 'batch:jobs:';
 var REDIS_DB = 5;
 var JobStatus = require('./job_status');
+var queue = require('queue-async');
+var debug = require('./util/debug')('job-backend');
 
 function JobBackend(metadataBackend, jobQueue) {
     this.metadataBackend = metadataBackend;
     this.jobQueue = jobQueue;
     this.maxNumberOfQueuedJobs = global.settings.batch_max_queued_jobs || 64;
     this.inSecondsJobTTLAfterFinished = global.settings.finished_jobs_ttl_in_seconds || 2 * 3600; // 2 hours
+    this.hostname = global.settings.api_hostname || 'batch';
 }
 
 function toRedisParams(job) {
@@ -164,6 +167,102 @@ JobBackend.prototype.save = function (job, callback) {
     });
 };
 
+var WORK_IN_PROGRESS_JOB = {
+    DB: 5,
+    PREFIX_USER: 'batch:wip:user:',
+    USER_INDEX_KEY: 'batch:wip:users'
+};
+
+JobBackend.prototype.addWorkInProgressJob = function (user, jobId, callback) {
+    var userWIPKey = WORK_IN_PROGRESS_JOB.PREFIX_USER + user;
+    debug('add job %s to user %s (%s)', jobId, user, userWIPKey);
+    this.metadataBackend.redisMultiCmd(WORK_IN_PROGRESS_JOB.DB, [
+        ['SADD', WORK_IN_PROGRESS_JOB.USER_INDEX_KEY, user],
+        ['RPUSH', userWIPKey, jobId]
+    ], callback);
+};
+
+JobBackend.prototype.clearWorkInProgressJob = function (user, jobId, callback) {
+    var self = this;
+    var DB = WORK_IN_PROGRESS_JOB.DB;
+    var userWIPKey = WORK_IN_PROGRESS_JOB.PREFIX_USER + user;
+
+    var params = [userWIPKey, 0, jobId];
+    self.metadataBackend.redisCmd(DB, 'LREM', params, function (err) {
+        if (err) {
+            return callback(err);
+        }
+
+        params = [userWIPKey, 0, -1];
+        self.metadataBackend.redisCmd(DB, 'LRANGE', params, function (err, workInProgressJobs) {
+            if (err) {
+                return callback(err);
+            }
+
+            debug('user %s has work in progress jobs %j', user, workInProgressJobs);
+
+            if (workInProgressJobs.length < 0) {
+                return callback();
+            }
+
+            debug('delete user %s from index', user);
+
+            params = [WORK_IN_PROGRESS_JOB.USER_INDEX_KEY, user];
+            self.metadataBackend.redisCmd(DB, 'SREM', params, function (err) {
+                if (err) {
+                    return callback(err);
+                }
+
+                return callback();
+            });
+        });
+    });
+};
+
+JobBackend.prototype.listWorkInProgressJobByUser = function (user, callback) {
+    var userWIPKey = WORK_IN_PROGRESS_JOB.PREFIX_USER + user;
+    var params = [userWIPKey, 0, -1];
+    this.metadataBackend.redisCmd(WORK_IN_PROGRESS_JOB.DB, 'LRANGE', params, callback);
+};
+
+JobBackend.prototype.listWorkInProgressJobs = function (callback) {
+    var self = this;
+    var DB = WORK_IN_PROGRESS_JOB.DB;
+
+    var params = [WORK_IN_PROGRESS_JOB.USER_INDEX_KEY];
+    this.metadataBackend.redisCmd(DB, 'SMEMBERS', params, function (err, workInProgressUsers) {
+        if (err) {
+            return callback(err);
+        }
+
+        if (workInProgressUsers < 1) {
+            return callback(null, {});
+        }
+
+        debug('found %j work in progress users', workInProgressUsers);
+
+        var usersQueue = queue(4);
+
+        workInProgressUsers.forEach(function (user) {
+            usersQueue.defer(self.listWorkInProgressJobByUser.bind(self), user);
+        });
+
+        usersQueue.awaitAll(function (err, userWorkInProgressJobs) {
+            if (err) {
+                return callback(err);
+            }
+
+            var workInProgressJobs = workInProgressUsers.reduce(function (users, user, index) {
+                users[user] = userWorkInProgressJobs[index];
+                debug('found %j work in progress jobs for user %s', userWorkInProgressJobs[index], user);
+                return users;
+            }, {});
+
+            callback(null, workInProgressJobs);
+        });
+    });
+};
+
 JobBackend.prototype.setTTL = function (job, callback) {
     var self = this;
     var redisKey = REDIS_PREFIX + job.job_id;

batch/job_service.js

@@ -122,3 +122,15 @@ JobService.prototype.drain = function (job_id, callback) {
         });
     });
 };
+
+JobService.prototype.addWorkInProgressJob = function (user, jobId, callback) {
+    this.jobBackend.addWorkInProgressJob(user, jobId, callback);
+};
+
+JobService.prototype.clearWorkInProgressJob = function (user, jobId, callback) {
+    this.jobBackend.clearWorkInProgressJob(user, jobId, callback);
+};
+
+JobService.prototype.listWorkInProgressJobs = function (callback) {
+    this.jobBackend.listWorkInProgressJobs(callback);
+};

doc/batch_queries.md

@@ -2,7 +2,7 @@
 
 A Batch Query enables you to request queries with long-running CPU processing times. Typically, these kind of requests raise timeout errors when using the SQL API. In order to avoid timeouts, you can use Batch Queries to [create](#create-a-job), [read](#read-a-job) and [cancel](#cancel-a-job) queries. You can also run a [chained batch query](#chaining-batch-queries) to chain several SQL queries into one job. A Batch Query schedules the incoming jobs and allows you to request the job status for each query.
 
-_Batch Queries are not intended to be used for large query payloads that contain over 8192 characters (8kb). For instance, if you are inserting a large number of rows into your table, you still need to use the [Import API](https://carto.com/docs/carto-engine/import-api/) or [SQL API](https://carto.com/docs/carto-engine/sql-api/) for this type of data management. Batch Queries are specific to queries and CPU usage._
+_Batch Queries are not intended to be used for large query payloads that contain over 16384 characters (16kb). For instance, if you are inserting a large number of rows into your table, you still need to use the [Import API](https://carto.com/docs/carto-engine/import-api/) or [SQL API](https://carto.com/docs/carto-engine/sql-api/) for this type of data management. Batch Queries are specific to queries and CPU usage._
 
 **Note:** In order to use Batch Queries, you **must** be [authenticated](https://carto.com/docs/carto-engine/sql-api/authentication/) using API keys.
 
@@ -486,7 +486,6 @@ For best practices, follow these recommended usage notes when using Batch Querie
 
 - Batch Queries are not intended for large query payloads (e.g: inserting thousands of rows), use the [Import API](https://carto.com/docs/carto-engine/import-api/) for this type of data management.
 
-- There is a limit of 8kb per job. The following error message appears if your job exceeds this size:
-
-  `Your payload is too large. Max size allowed is 8192 (8kb)`
+- There is a limit of 16kb per job. The following error message appears if your job exceeds this size:
 
+  `Your payload is too large. Max size allowed is 16384 (16kb)`