321 lines
10 KiB
JavaScript
321 lines
10 KiB
JavaScript
'use strict';
|
|
|
|
exports = module.exports = {
|
|
get,
|
|
add,
|
|
update,
|
|
setCompleted,
|
|
setCompletedByType,
|
|
listByTypePaged,
|
|
|
|
getLogs,
|
|
|
|
startTask,
|
|
stopTask,
|
|
stopAllTasks,
|
|
|
|
removePrivateFields,
|
|
|
|
_del: del,
|
|
|
|
// task types. if you add a task here, fill up the function table in taskworker and dashboard client.js
|
|
TASK_APP: 'app',
|
|
TASK_BACKUP: 'backup',
|
|
TASK_UPDATE: 'update',
|
|
TASK_CHECK_CERTS: 'checkCerts',
|
|
TASK_SETUP_DNS_AND_CERT: 'setupDnsAndCert',
|
|
TASK_CLEAN_BACKUPS: 'cleanBackups',
|
|
TASK_SYNC_EXTERNAL_LDAP: 'syncExternalLdap',
|
|
TASK_CHANGE_MAIL_LOCATION: 'changeMailLocation',
|
|
TASK_SYNC_DNS_RECORDS: 'syncDnsRecords',
|
|
|
|
// error codes
|
|
ESTOPPED: 'stopped',
|
|
ECRASHED: 'crashed',
|
|
ETIMEOUT: 'timeout',
|
|
|
|
// testing
|
|
_TASK_IDENTITY: '_identity',
|
|
_TASK_CRASH: '_crash',
|
|
_TASK_ERROR: '_error',
|
|
_TASK_SLEEP: '_sleep'
|
|
};
|
|
|
|
const assert = require('assert'),
|
|
BoxError = require('./boxerror.js'),
|
|
database = require('./database.js'),
|
|
debug = require('debug')('box:tasks'),
|
|
path = require('path'),
|
|
paths = require('./paths.js'),
|
|
safe = require('safetydance'),
|
|
shell = require('./shell.js'),
|
|
spawn = require('child_process').spawn,
|
|
split = require('split'),
|
|
_ = require('underscore');
|
|
|
|
let gTasks = {}; // indexed by task id
|
|
|
|
const START_TASK_CMD = path.join(__dirname, 'scripts/starttask.sh');
|
|
const STOP_TASK_CMD = path.join(__dirname, 'scripts/stoptask.sh');
|
|
|
|
const TASKS_FIELDS = [ 'id', 'type', 'argsJson', 'percent', 'message', 'errorJson', 'creationTime', 'resultJson', 'ts' ];
|
|
|
|
function postProcess(task) {
|
|
assert.strictEqual(typeof task, 'object');
|
|
|
|
assert(task.argsJson === null || typeof task.argsJson === 'string');
|
|
task.args = safe.JSON.parse(task.argsJson) || [];
|
|
delete task.argsJson;
|
|
|
|
task.id = String(task.id);
|
|
|
|
task.result = JSON.parse(task.resultJson);
|
|
delete task.resultJson;
|
|
|
|
task.error = safe.JSON.parse(task.errorJson);
|
|
delete task.errorJson;
|
|
|
|
return task;
|
|
}
|
|
|
|
function updateStatus(result) {
|
|
assert.strictEqual(typeof result, 'object');
|
|
|
|
result.active = !!gTasks[result.id];
|
|
|
|
// we rely on 'percent' to determine success. maybe this can become a db field
|
|
result.success = result.percent === 100 && !result.error;
|
|
|
|
// we rely on 'percent' to determine pending. maybe this can become a db field
|
|
result.pending = result.percent === 1;
|
|
|
|
// the error in db will be empty if we didn't get a chance to handle task exit
|
|
if (!result.active && result.percent !== 100 && !result.error) {
|
|
result.error = { message: 'Cloudron crashed/stopped', code: exports.ECRASHED };
|
|
}
|
|
|
|
return result;
|
|
}
|
|
|
|
async function get(id) {
|
|
assert.strictEqual(typeof id, 'string');
|
|
|
|
const result = await database.query(`SELECT ${TASKS_FIELDS} FROM tasks WHERE id = ?`, [ id ]);
|
|
if (result.length === 0) return null;
|
|
|
|
return updateStatus(postProcess(result[0]));
|
|
}
|
|
|
|
async function update(id, task) {
|
|
assert.strictEqual(typeof id, 'string');
|
|
assert.strictEqual(typeof task, 'object');
|
|
|
|
debug(`${id}: ${JSON.stringify(task)}`);
|
|
|
|
let args = [ ];
|
|
let fields = [ ];
|
|
for (let k in task) {
|
|
if (k === 'result' || k === 'error') {
|
|
fields.push(`${k}Json = ?`);
|
|
args.push(JSON.stringify(task[k]));
|
|
} else {
|
|
fields.push(k + ' = ?');
|
|
args.push(task[k]);
|
|
}
|
|
}
|
|
args.push(id);
|
|
|
|
const result = await database.query('UPDATE tasks SET ' + fields.join(', ') + ' WHERE id = ?', args);
|
|
if (result.affectedRows !== 1) throw new BoxError(BoxError.NOT_FOUND, 'Task not found');
|
|
}
|
|
|
|
async function setCompleted(id, task) {
|
|
assert.strictEqual(typeof id, 'string');
|
|
assert.strictEqual(typeof task, 'object');
|
|
|
|
debug(`setCompleted - ${id}: ${JSON.stringify(task)}`);
|
|
|
|
await update(id, _.extend({ percent: 100 }, task));
|
|
}
|
|
|
|
async function setCompletedByType(type, task) {
|
|
assert.strictEqual(typeof type, 'string');
|
|
assert.strictEqual(typeof task, 'object');
|
|
|
|
const results = await listByTypePaged(type, 1, 1);
|
|
if (results.length !== 1) throw new BoxError(BoxError.NOT_FOUND, 'No such task');
|
|
|
|
await setCompleted(results[0].id, task);
|
|
}
|
|
|
|
async function add(type, args) {
|
|
assert.strictEqual(typeof type, 'string');
|
|
assert(Array.isArray(args));
|
|
|
|
const result = await database.query('INSERT INTO tasks (type, argsJson, percent, message) VALUES (?, ?, ?, ?)', [ type, JSON.stringify(args), 0, 'Queued' ]);
|
|
return String(result.insertId);
|
|
}
|
|
|
|
function startTask(id, options, callback) {
|
|
assert.strictEqual(typeof id, 'string');
|
|
assert.strictEqual(typeof options, 'object');
|
|
assert.strictEqual(typeof callback, 'function');
|
|
|
|
const logFile = options.logFile || `${paths.TASKS_LOG_DIR}/${id}.log`;
|
|
debug(`startTask - starting task ${id} with options ${JSON.stringify(options)}. logs at ${logFile}`);
|
|
|
|
let killTimerId = null, timedOut = false;
|
|
|
|
gTasks[id] = shell.sudo('startTask', [ START_TASK_CMD, id, logFile, options.nice || 0, options.memoryLimit || 400 ], { preserveEnv: true }, async function (sudoError) {
|
|
if (!gTasks[id]) return; // ignore task exit since we are shutting down. see stopAllTasks
|
|
|
|
const code = sudoError ? sudoError.code : 0;
|
|
const signal = sudoError ? sudoError.signal : 0;
|
|
|
|
debug(`startTask: ${id} completed with code ${code} and signal ${signal}`);
|
|
|
|
if (options.timeout) clearTimeout(killTimerId);
|
|
|
|
const [getError, task] = await safe(get(id));
|
|
|
|
let taskError;
|
|
if (!getError && task.percent !== 100) { // taskworker crashed or was killed by us
|
|
if (code === 0) {
|
|
taskError = {
|
|
message: `Task ${id} ${timedOut ? 'timed out' : 'stopped'}` ,
|
|
code: timedOut ? exports.ETIMEOUT : exports.ESTOPPED
|
|
};
|
|
} else { // task crashed
|
|
taskError = {
|
|
message: signal === 9 ? `Task ${id} crashed as it ran out of memory` : `Task ${id} crashed with code ${code} and signal ${signal}`,
|
|
code: exports.ECRASHED
|
|
};
|
|
}
|
|
|
|
// note that despite the update() here, we should handle the case where the box code was restarted and never got taskworker exit
|
|
await safe(setCompleted(id, { error: taskError }));
|
|
} else if (!getError && task.error) {
|
|
taskError = task.error;
|
|
} else if (!task) { // db got cleared in tests
|
|
taskError = new BoxError(BoxError.NOT_FOUND, `No such task ${id}`);
|
|
}
|
|
|
|
delete gTasks[id];
|
|
|
|
callback(taskError, task ? task.result : null);
|
|
|
|
debug(`startTask: ${id} done`);
|
|
});
|
|
|
|
if (options.timeout) {
|
|
killTimerId = setTimeout(async function () {
|
|
debug(`startTask: task ${id} took too long. killing`);
|
|
timedOut = true;
|
|
const [error] = await safe(stopTask(id));
|
|
if (error) debug(`startTask: error stopping task: ${error.message}`);
|
|
}, options.timeout);
|
|
}
|
|
}
|
|
|
|
async function stopTask(id) {
|
|
assert.strictEqual(typeof id, 'string');
|
|
|
|
if (!gTasks[id]) throw new BoxError(BoxError.BAD_STATE, 'task is not active');
|
|
|
|
debug(`stopTask: stopping task ${id}`);
|
|
|
|
await shell.promises.sudo('stopTask', [ STOP_TASK_CMD, id, ], {});
|
|
}
|
|
|
|
async function stopAllTasks() {
|
|
debug('stopAllTasks: stopping all tasks');
|
|
|
|
gTasks = {}; // this signals startTask() to not set completion status as "crashed"
|
|
const [error] = await safe(shell.promises.sudo('stopTask', [ STOP_TASK_CMD, 'all' ], { cwd: paths.baseDir() }));
|
|
if (error) debug(`stopAllTasks: error stopping stasks: ${error.message}`);
|
|
}
|
|
|
|
async function listByTypePaged(type, page, perPage) {
|
|
assert(typeof type === 'string' || type === null);
|
|
assert.strictEqual(typeof page, 'number');
|
|
assert.strictEqual(typeof perPage, 'number');
|
|
|
|
let data = [];
|
|
let query = `SELECT ${TASKS_FIELDS} FROM tasks`;
|
|
|
|
if (type) {
|
|
query += ' WHERE TYPE=?';
|
|
data.push(type);
|
|
}
|
|
|
|
query += ' ORDER BY creationTime DESC, id DESC LIMIT ?,?'; // put latest task first
|
|
|
|
data.push((page-1)*perPage);
|
|
data.push(perPage);
|
|
|
|
const results = await database.query(query, data);
|
|
results.forEach(postProcess);
|
|
results.forEach(updateStatus);
|
|
return results;
|
|
}
|
|
|
|
function getLogs(taskId, options) {
|
|
assert.strictEqual(typeof taskId, 'string');
|
|
assert(options && typeof options === 'object');
|
|
|
|
assert.strictEqual(typeof options.lines, 'number');
|
|
assert.strictEqual(typeof options.format, 'string');
|
|
assert.strictEqual(typeof options.follow, 'boolean');
|
|
|
|
debug(`Getting logs for ${taskId}`);
|
|
|
|
const lines = options.lines === -1 ? '+1' : options.lines,
|
|
format = options.format || 'json',
|
|
follow = options.follow;
|
|
|
|
const cmd = '/usr/bin/tail';
|
|
let args = [ '--lines=' + lines ];
|
|
|
|
if (follow) args.push('--follow', '--retry', '--quiet'); // same as -F. to make it work if file doesn't exist, --quiet to not output file headers, which are no logs
|
|
args.push(`${paths.TASKS_LOG_DIR}/${taskId}.log`);
|
|
|
|
const cp = spawn(cmd, args);
|
|
|
|
const transformStream = split(function mapper(line) {
|
|
if (format !== 'json') return line + '\n';
|
|
|
|
const data = line.split(' '); // logs are <ISOtimestamp> <msg>
|
|
let timestamp = (new Date(data[0])).getTime();
|
|
if (isNaN(timestamp)) timestamp = 0;
|
|
const message = line.slice(data[0].length+1);
|
|
|
|
// ignore faulty empty logs
|
|
if (!timestamp && !message) return;
|
|
|
|
return JSON.stringify({
|
|
realtimeTimestamp: timestamp * 1000,
|
|
message: message,
|
|
source: taskId
|
|
}) + '\n';
|
|
});
|
|
|
|
transformStream.close = cp.kill.bind(cp, 'SIGKILL'); // closing stream kills the child process
|
|
|
|
cp.stdout.pipe(transformStream);
|
|
|
|
return transformStream;
|
|
}
|
|
|
|
// removes all fields that are strictly private and should never be returned by API calls
|
|
function removePrivateFields(task) {
|
|
var result = _.pick(task, 'id', 'type', 'percent', 'message', 'error', 'active', 'pending', 'creationTime', 'result', 'ts', 'success');
|
|
return result;
|
|
}
|
|
|
|
async function del(id) {
|
|
assert.strictEqual(typeof id, 'string');
|
|
|
|
const result = await database.query('DELETE FROM tasks WHERE id = ?', [ id ]);
|
|
if (result.affectedRows !== 1) throw new BoxError(BoxError.NOT_FOUND, 'Task not found');
|
|
}
|