290 lines
8.2 KiB
JavaScript
290 lines
8.2 KiB
JavaScript
'use strict';
|
|
|
|
var REDIS_PREFIX = 'batch:jobs:';
|
|
var REDIS_DB = 5;
|
|
var JobStatus = require('./job-status');
|
|
var queue = require('queue-async');
|
|
|
|
function JobBackend (metadataBackend, jobQueue, logger) {
|
|
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';
|
|
this.logger = logger;
|
|
}
|
|
|
|
function toRedisParams (job) {
|
|
var redisParams = [REDIS_PREFIX + job.job_id];
|
|
var obj = JSON.parse(JSON.stringify(job));
|
|
delete obj.job_id;
|
|
|
|
for (var property in obj) {
|
|
if (obj.hasOwnProperty(property)) {
|
|
redisParams.push(property);
|
|
if (property === 'query' && typeof obj[property] !== 'string') {
|
|
redisParams.push(JSON.stringify(obj[property]));
|
|
} else {
|
|
redisParams.push(obj[property]);
|
|
}
|
|
}
|
|
}
|
|
|
|
return redisParams;
|
|
}
|
|
|
|
function toObject (job_id, redisParams, redisValues) {
|
|
var obj = {};
|
|
|
|
redisParams.shift(); // job_id value
|
|
redisParams.pop(); // WARN: weird function pushed by metadataBackend
|
|
|
|
for (var i = 0; i < redisParams.length; i++) {
|
|
// TODO: this should be moved to job model
|
|
if (redisParams[i] === 'query') {
|
|
try {
|
|
obj[redisParams[i]] = JSON.parse(redisValues[i]);
|
|
} catch (e) {
|
|
obj[redisParams[i]] = redisValues[i];
|
|
}
|
|
} else if (redisValues[i]) {
|
|
obj[redisParams[i]] = redisValues[i];
|
|
}
|
|
}
|
|
|
|
obj.job_id = job_id; // adds redisKey as object property
|
|
|
|
return obj;
|
|
}
|
|
|
|
function isJobFound (redisValues) {
|
|
return !!(redisValues[0] && redisValues[1] && redisValues[2] && redisValues[3] && redisValues[4]);
|
|
}
|
|
|
|
function getNotFoundError (job_id) {
|
|
var notFoundError = new Error('Job with id ' + job_id + ' not found');
|
|
notFoundError.name = 'NotFoundError';
|
|
return notFoundError;
|
|
}
|
|
|
|
JobBackend.prototype.get = function (job_id, callback) {
|
|
if (!job_id) {
|
|
return callback(getNotFoundError(job_id));
|
|
}
|
|
|
|
var self = this;
|
|
var redisParams = [
|
|
REDIS_PREFIX + job_id,
|
|
'user',
|
|
'status',
|
|
'query',
|
|
'created_at',
|
|
'updated_at',
|
|
'host',
|
|
'failed_reason',
|
|
'fallback_status',
|
|
'host',
|
|
'port',
|
|
'pass',
|
|
'dbname',
|
|
'dbuser'
|
|
];
|
|
|
|
self.metadataBackend.redisCmd(REDIS_DB, 'HMGET', redisParams, function (err, redisValues) {
|
|
if (err) {
|
|
return callback(err);
|
|
}
|
|
|
|
if (!isJobFound(redisValues)) {
|
|
return callback(getNotFoundError(job_id));
|
|
}
|
|
|
|
var jobData = toObject(job_id, redisParams, redisValues);
|
|
|
|
callback(null, jobData);
|
|
});
|
|
};
|
|
|
|
JobBackend.prototype.create = function (job, callback) {
|
|
var self = this;
|
|
|
|
this.jobQueue.size(job.user, function (err, size) {
|
|
if (err) {
|
|
return callback(new Error('Failed to create job, could not determine user queue size'));
|
|
}
|
|
|
|
if (size >= self.maxNumberOfQueuedJobs) {
|
|
return callback(new Error(
|
|
'Failed to create job. ' +
|
|
'Max number of jobs (' + self.maxNumberOfQueuedJobs + ') queued reached'
|
|
));
|
|
}
|
|
|
|
self.get(job.job_id, function (err) {
|
|
if (err && err.name !== 'NotFoundError') {
|
|
return callback(err);
|
|
}
|
|
|
|
self.save(job, function (err, jobSaved) {
|
|
if (err) {
|
|
return callback(err);
|
|
}
|
|
|
|
self.jobQueue.enqueue(job.user, job.job_id, function (err) {
|
|
if (err) {
|
|
return callback(err);
|
|
}
|
|
|
|
return callback(null, jobSaved);
|
|
});
|
|
});
|
|
});
|
|
});
|
|
};
|
|
|
|
JobBackend.prototype.update = function (job, callback) {
|
|
var self = this;
|
|
|
|
self.get(job.job_id, function (err) {
|
|
if (err) {
|
|
return callback(err);
|
|
}
|
|
|
|
self.save(job, callback);
|
|
});
|
|
};
|
|
|
|
JobBackend.prototype.save = function (job, callback) {
|
|
var self = this;
|
|
var redisParams = toRedisParams(job);
|
|
|
|
self.metadataBackend.redisCmd(REDIS_DB, 'HMSET', redisParams, function (err) {
|
|
if (err) {
|
|
return callback(err);
|
|
}
|
|
|
|
self.setTTL(job, function (err) {
|
|
if (err) {
|
|
return callback(err);
|
|
}
|
|
|
|
self.get(job.job_id, function (err, job) {
|
|
if (err) {
|
|
return callback(err);
|
|
}
|
|
|
|
callback(null, job);
|
|
});
|
|
});
|
|
});
|
|
};
|
|
|
|
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;
|
|
this.logger.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);
|
|
}
|
|
|
|
self.logger.debug('user %s has work in progress jobs %j', user, workInProgressJobs);
|
|
|
|
if (workInProgressJobs.length < 0) {
|
|
return callback();
|
|
}
|
|
|
|
self.logger.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, {});
|
|
}
|
|
|
|
self.logger.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];
|
|
self.logger.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;
|
|
|
|
if (!JobStatus.isFinal(job.status)) {
|
|
return callback();
|
|
}
|
|
|
|
self.metadataBackend.redisCmd(REDIS_DB, 'EXPIRE', [redisKey, this.inSecondsJobTTLAfterFinished], callback);
|
|
};
|
|
|
|
module.exports = JobBackend;
|