cartodb/CartoDB-SQL-API
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Added uknown status to jobs when drain fails

Browse Source
This commit is contained in:
Daniel García Aubert 2016-01-25 20:07:41 +01:00
parent a960bd18f1
commit da16f32789
3 changed files with 50 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
batch/batch.js
View File

@ -80,6 +80,10 @@ Batch.prototype._drainJob = function (host, callback) {
var queue = self.jobQueuePool.getQueue(host);
this.jobCanceller.drain(job_id, function (err) {
if (err && err.name === 'CancelNotAllowedError') {
return callback();
}
if (err) {
return callback(err);
}
@ -109,6 +113,8 @@ Batch.prototype._consumeJobs = function (host, queue, callback) {
self.jobQueuePool.setCurrentJobId(host, job_id);
self.jobRunner.run(job_id, function (err, job) {
self.jobQueuePool.removeCurrentJobId(host);
if (err && err.name === 'InvalidJobStatus') {
console.log(err.message);
return callback();
@ -124,8 +130,6 @@ Batch.prototype._consumeJobs = function (host, queue, callback) {
console.log('Job %s %s in %s', job_id, job.status, host);
}
self.jobQueuePool.removeCurrentJobId(host);
self.emit('job:' + job.status, job_id);
callback();

35
batch/job_backend.js
View File

@ -313,5 +313,40 @@ JobBackend.prototype.setCancelled = function (job, callback) {
});
};
JobBackend.prototype.setUnknown = function (job_id, callback) {
var self = this;
this.get(job_id, function (err, job) {
if (err) {
return callback(err);
}
var now = new Date().toISOString();
var redisKey = self.redisPrefix + job.job_id;
var redisParams = [
redisKey,
'status', 'unknown',
'updated_at', now
];
self.metadataBackend.redisCmd(self.db, 'HMSET', redisParams , function (err) {
if (err) {
return callback(err);
}
self.metadataBackend.redisCmd(self.db, 'EXPIRE', [ redisKey, JOBS_TTL_IN_SECONDS ], function (err) {
if (err) {
return callback(err);
}
job.status = 'unknown';
job.updated_at = now;
callback(null, job);
});
});
});
};
module.exports = JobBackend;

11
batch/job_canceller.js
View File

@ -21,7 +21,9 @@ JobCanceller.prototype.cancel = function (job_id, callback) {
}
if (job.status !== 'running') {
return callback(new Error('Job is ' + job.status + ', cancel is not allowed'));
var cancelNotAllowedError = new Error('Job is ' + job.status + ', cancel is not allowed');
cancelNotAllowedError.name = 'CancelNotAllowedError';
return callback(cancelNotAllowedError);
}
self.userDatabaseMetadataService.getUserMetadata(job.user, function (err, userDatabaseMetadata) {
@ -44,10 +46,15 @@ JobCanceller.prototype.drain = function (job_id, callback) {
var self = this;
this.cancel(job_id, function (err, job) {
if (err) {
if (err && err.name === 'CancelNotAllowedError') {
return callback(err);
}
if (err) {
console.error('There was an error while draining job %s, %s ', job_id, err);
return self.jobBackend.setUnknown(job_id, callback);
}
self.jobBackend.setPending(job, callback);
});