Merge branch 'master' into batch-onerror-template
Conflicts: NEWS.md
This commit is contained in:
commit
85fa9d3c2b
1
NEWS.md
1
NEWS.md
@ -2,6 +2,7 @@
|
||||
-------------------
|
||||
|
||||
New features:
|
||||
* Adds start and end time for batch queries with fallback.
|
||||
* Add `<%= error_message %>` template support for onerror fallback queries.
|
||||
|
||||
|
||||
|
@ -32,6 +32,12 @@ Query.prototype.setStatus = function (status, job, errorMesssage) {
|
||||
|
||||
if (isValid) {
|
||||
job.query.query[this.index].status = status;
|
||||
if (status === jobStatus.RUNNING) {
|
||||
job.query.query[this.index].started_at = new Date().toISOString();
|
||||
}
|
||||
if (this.isFinalStatus(status)) {
|
||||
job.query.query[this.index].ended_at = new Date().toISOString();
|
||||
}
|
||||
if (status === jobStatus.FAILED && errorMesssage) {
|
||||
job.query.query[this.index].failed_reason = errorMesssage;
|
||||
}
|
||||
|
@ -15,6 +15,25 @@ var jobStatus = require('../../batch/job_status');
|
||||
|
||||
describe('Batch API fallback job', function () {
|
||||
|
||||
function validateExpectedResponse(actual, expected) {
|
||||
actual.query.forEach(function(actualQuery, index) {
|
||||
var expectedQuery = expected.query[index];
|
||||
assert.ok(expectedQuery);
|
||||
Object.keys(expectedQuery).forEach(function(expectedKey) {
|
||||
assert.equal(actualQuery[expectedKey], expectedQuery[expectedKey]);
|
||||
});
|
||||
var propsToCheckDate = ['started_at', 'ended_at'];
|
||||
propsToCheckDate.forEach(function(propToCheckDate) {
|
||||
if (actualQuery.hasOwnProperty(propToCheckDate)) {
|
||||
assert.ok(new Date(actualQuery[propToCheckDate]));
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
assert.equal(actual.onsuccess, expected.onsuccess);
|
||||
assert.equal(actual.onerror, expected.onerror);
|
||||
}
|
||||
|
||||
var batch = batchFactory(metadataBackend);
|
||||
|
||||
before(function () {
|
||||
@ -85,7 +104,7 @@ describe('Batch API fallback job', function () {
|
||||
var job = JSON.parse(res.body);
|
||||
if (job.status === jobStatus.DONE) {
|
||||
clearInterval(interval);
|
||||
assert.deepEqual(job.query, expectedQuery);
|
||||
validateExpectedResponse(job.query, expectedQuery);
|
||||
done();
|
||||
} else if (job.status === jobStatus.FAILED || job.status === jobStatus.CANCELLED) {
|
||||
clearInterval(interval);
|
||||
@ -152,7 +171,7 @@ describe('Batch API fallback job', function () {
|
||||
var job = JSON.parse(res.body);
|
||||
if (job.status === jobStatus.DONE) {
|
||||
clearInterval(interval);
|
||||
assert.deepEqual(job.query, expectedQuery);
|
||||
validateExpectedResponse(job.query, expectedQuery);
|
||||
done();
|
||||
} else if (job.status === jobStatus.FAILED || job.status === jobStatus.CANCELLED) {
|
||||
clearInterval(interval);
|
||||
@ -220,7 +239,7 @@ describe('Batch API fallback job', function () {
|
||||
var job = JSON.parse(res.body);
|
||||
if (job.status === jobStatus.FAILED) {
|
||||
clearInterval(interval);
|
||||
assert.deepEqual(job.query, expectedQuery);
|
||||
validateExpectedResponse(job.query, expectedQuery);
|
||||
done();
|
||||
} else if (job.status === jobStatus.DONE || job.status === jobStatus.CANCELLED) {
|
||||
clearInterval(interval);
|
||||
@ -289,7 +308,7 @@ describe('Batch API fallback job', function () {
|
||||
var job = JSON.parse(res.body);
|
||||
if (job.status === jobStatus.FAILED) {
|
||||
clearInterval(interval);
|
||||
assert.deepEqual(job.query, expectedQuery);
|
||||
validateExpectedResponse(job.query, expectedQuery);
|
||||
done();
|
||||
} else if (job.status === jobStatus.DONE || job.status === jobStatus.CANCELLED) {
|
||||
clearInterval(interval);
|
||||
@ -357,7 +376,7 @@ describe('Batch API fallback job', function () {
|
||||
var job = JSON.parse(res.body);
|
||||
if (job.status === jobStatus.DONE && job.fallback_status === jobStatus.DONE) {
|
||||
clearInterval(interval);
|
||||
assert.deepEqual(job.query, expectedQuery);
|
||||
validateExpectedResponse(job.query, expectedQuery);
|
||||
done();
|
||||
} else if (job.status === jobStatus.FAILED || job.status === jobStatus.CANCELLED) {
|
||||
clearInterval(interval);
|
||||
@ -425,7 +444,7 @@ describe('Batch API fallback job', function () {
|
||||
var job = JSON.parse(res.body);
|
||||
if (job.status === jobStatus.FAILED && job.fallback_status === jobStatus.SKIPPED) {
|
||||
clearInterval(interval);
|
||||
assert.deepEqual(job.query, expectedQuery);
|
||||
validateExpectedResponse(job.query, expectedQuery);
|
||||
done();
|
||||
} else if (job.status === jobStatus.FAILED || job.status === jobStatus.CANCELLED) {
|
||||
clearInterval(interval);
|
||||
@ -494,7 +513,7 @@ describe('Batch API fallback job', function () {
|
||||
var job = JSON.parse(res.body);
|
||||
if (job.status === jobStatus.FAILED && job.fallback_status === jobStatus.DONE) {
|
||||
clearInterval(interval);
|
||||
assert.deepEqual(job.query, expectedQuery);
|
||||
validateExpectedResponse(job.query, expectedQuery);
|
||||
done();
|
||||
} else if (job.status === jobStatus.DONE || job.status === jobStatus.CANCELLED) {
|
||||
clearInterval(interval);
|
||||
@ -561,7 +580,7 @@ describe('Batch API fallback job', function () {
|
||||
var job = JSON.parse(res.body);
|
||||
if (job.status === jobStatus.DONE && job.fallback_status === jobStatus.SKIPPED) {
|
||||
clearInterval(interval);
|
||||
assert.deepEqual(job.query, expectedQuery);
|
||||
validateExpectedResponse(job.query, expectedQuery);
|
||||
done();
|
||||
} else if (job.status === jobStatus.FAILED || job.status === jobStatus.CANCELLED) {
|
||||
clearInterval(interval);
|
||||
@ -632,7 +651,7 @@ describe('Batch API fallback job', function () {
|
||||
var job = JSON.parse(res.body);
|
||||
if (job.status === jobStatus.DONE && job.fallback_status === jobStatus.DONE) {
|
||||
clearInterval(interval);
|
||||
assert.deepEqual(job.query, expectedQuery);
|
||||
validateExpectedResponse(job.query, expectedQuery);
|
||||
done();
|
||||
} else if (job.status === jobStatus.FAILED || job.status === jobStatus.CANCELLED) {
|
||||
clearInterval(interval);
|
||||
@ -708,7 +727,7 @@ describe('Batch API fallback job', function () {
|
||||
var job = JSON.parse(res.body);
|
||||
if (job.status === jobStatus.DONE) {
|
||||
clearInterval(interval);
|
||||
assert.deepEqual(job.query, expectedQuery);
|
||||
validateExpectedResponse(job.query, expectedQuery);
|
||||
done();
|
||||
} else if (job.status === jobStatus.FAILED || job.status === jobStatus.CANCELLED) {
|
||||
clearInterval(interval);
|
||||
@ -785,7 +804,7 @@ describe('Batch API fallback job', function () {
|
||||
var job = JSON.parse(res.body);
|
||||
if (job.status === jobStatus.FAILED) {
|
||||
clearInterval(interval);
|
||||
assert.deepEqual(job.query, expectedQuery);
|
||||
validateExpectedResponse(job.query, expectedQuery);
|
||||
done();
|
||||
} else if (job.status === jobStatus.DONE || job.status === jobStatus.CANCELLED) {
|
||||
clearInterval(interval);
|
||||
@ -863,7 +882,7 @@ describe('Batch API fallback job', function () {
|
||||
var job = JSON.parse(res.body);
|
||||
if (job.status === jobStatus.FAILED) {
|
||||
clearInterval(interval);
|
||||
assert.deepEqual(job.query, expectedQuery);
|
||||
validateExpectedResponse(job.query, expectedQuery);
|
||||
done();
|
||||
} else if (job.status === jobStatus.DONE || job.status === jobStatus.CANCELLED) {
|
||||
clearInterval(interval);
|
||||
@ -939,7 +958,7 @@ describe('Batch API fallback job', function () {
|
||||
var job = JSON.parse(res.body);
|
||||
if (job.status === jobStatus.DONE) {
|
||||
clearInterval(interval);
|
||||
assert.deepEqual(job.query, expectedQuery);
|
||||
validateExpectedResponse(job.query, expectedQuery);
|
||||
done();
|
||||
} else if (job.status === jobStatus.FAILED || job.status === jobStatus.CANCELLED) {
|
||||
clearInterval(interval);
|
||||
@ -1008,7 +1027,7 @@ describe('Batch API fallback job', function () {
|
||||
var job = JSON.parse(res.body);
|
||||
if (job.status === jobStatus.FAILED) {
|
||||
clearInterval(interval);
|
||||
assert.deepEqual(job.query, expectedQuery);
|
||||
validateExpectedResponse(job.query, expectedQuery);
|
||||
done();
|
||||
} else if (job.status === jobStatus.DONE || job.status === jobStatus.CANCELLED) {
|
||||
clearInterval(interval);
|
||||
@ -1077,7 +1096,7 @@ describe('Batch API fallback job', function () {
|
||||
var job = JSON.parse(res.body);
|
||||
if (job.status === jobStatus.FAILED && job.fallback_status === jobStatus.SKIPPED) {
|
||||
clearInterval(interval);
|
||||
assert.deepEqual(job.query, expectedQuery);
|
||||
validateExpectedResponse(job.query, expectedQuery);
|
||||
done();
|
||||
} else if (job.status === jobStatus.DONE || job.status === jobStatus.CANCELLED) {
|
||||
clearInterval(interval);
|
||||
@ -1154,7 +1173,7 @@ describe('Batch API fallback job', function () {
|
||||
var job = JSON.parse(res.body);
|
||||
if (job.status === jobStatus.DONE) {
|
||||
clearInterval(interval);
|
||||
assert.deepEqual(job.query, expectedQuery);
|
||||
validateExpectedResponse(job.query, expectedQuery);
|
||||
done();
|
||||
} else if (job.status === jobStatus.FAILED || job.status === jobStatus.CANCELLED) {
|
||||
clearInterval(interval);
|
||||
@ -1231,7 +1250,7 @@ describe('Batch API fallback job', function () {
|
||||
var job = JSON.parse(res.body);
|
||||
if (job.status === jobStatus.DONE) {
|
||||
clearInterval(interval);
|
||||
assert.deepEqual(job.query, expectedQuery);
|
||||
validateExpectedResponse(job.query, expectedQuery);
|
||||
done();
|
||||
} else if (job.status === jobStatus.FAILED || job.status === jobStatus.CANCELLED) {
|
||||
clearInterval(interval);
|
||||
@ -1309,7 +1328,7 @@ describe('Batch API fallback job', function () {
|
||||
var job = JSON.parse(res.body);
|
||||
if (job.status === jobStatus.DONE && job.fallback_status === jobStatus.DONE) {
|
||||
clearInterval(interval);
|
||||
assert.deepEqual(job.query, expectedQuery);
|
||||
validateExpectedResponse(job.query, expectedQuery);
|
||||
done();
|
||||
} else if (job.status === jobStatus.FAILED || job.status === jobStatus.CANCELLED) {
|
||||
clearInterval(interval);
|
||||
@ -1389,7 +1408,7 @@ describe('Batch API fallback job', function () {
|
||||
var job = JSON.parse(res.body);
|
||||
if (job.status === jobStatus.DONE && job.fallback_status === jobStatus.DONE) {
|
||||
clearInterval(interval);
|
||||
assert.deepEqual(job.query, expectedQuery);
|
||||
validateExpectedResponse(job.query, expectedQuery);
|
||||
done();
|
||||
} else if (job.status === jobStatus.FAILED || job.status === jobStatus.CANCELLED) {
|
||||
clearInterval(interval);
|
||||
@ -1460,7 +1479,7 @@ describe('Batch API fallback job', function () {
|
||||
var job = JSON.parse(res.body);
|
||||
if (job.status === jobStatus.RUNNING && job.fallback_status === jobStatus.PENDING) {
|
||||
clearInterval(interval);
|
||||
assert.deepEqual(job.query, expectedQuery);
|
||||
validateExpectedResponse(job.query, expectedQuery);
|
||||
done();
|
||||
} else if (job.status === jobStatus.DONE ||
|
||||
job.status === jobStatus.FAILED ||
|
||||
@ -1498,7 +1517,7 @@ describe('Batch API fallback job', function () {
|
||||
}
|
||||
var job = JSON.parse(res.body);
|
||||
if (job.status === jobStatus.CANCELLED && job.fallback_status === jobStatus.SKIPPED) {
|
||||
assert.deepEqual(job.query, expectedQuery);
|
||||
validateExpectedResponse(job.query, expectedQuery);
|
||||
done();
|
||||
} else if (job.status === jobStatus.DONE || job.status === jobStatus.FAILED) {
|
||||
done(new Error('Job ' + job.job_id + ' is ' + job.status + ', expected to be cancelled'));
|
||||
@ -1567,7 +1586,7 @@ describe('Batch API fallback job', function () {
|
||||
if (job.query.query[0].status === jobStatus.DONE &&
|
||||
job.query.query[0].fallback_status === jobStatus.RUNNING) {
|
||||
clearInterval(interval);
|
||||
assert.deepEqual(job.query, expectedQuery);
|
||||
validateExpectedResponse(job.query, expectedQuery);
|
||||
done();
|
||||
} else if (job.query.query[0].status === jobStatus.DONE ||
|
||||
job.query.query[0].status === jobStatus.FAILED ||
|
||||
@ -1607,7 +1626,7 @@ describe('Batch API fallback job', function () {
|
||||
}
|
||||
var job = JSON.parse(res.body);
|
||||
if (job.status === jobStatus.CANCELLED && job.fallback_status === jobStatus.SKIPPED) {
|
||||
assert.deepEqual(job.query, expectedQuery);
|
||||
validateExpectedResponse(job.query, expectedQuery);
|
||||
done();
|
||||
} else if (job.status === jobStatus.DONE || job.status === jobStatus.FAILED) {
|
||||
done(new Error('Job ' + job.job_id + ' is ' + job.status + ', expected to be cancelled'));
|
||||
@ -1687,7 +1706,7 @@ describe('Batch API fallback job', function () {
|
||||
var job = JSON.parse(res.body);
|
||||
if (job.status === jobStatus.FAILED && job.fallback_status === jobStatus.DONE) {
|
||||
clearInterval(interval);
|
||||
assert.deepEqual(job.query, expectedQuery);
|
||||
validateExpectedResponse(job.query, expectedQuery);
|
||||
done();
|
||||
} else if (job.status === jobStatus.DONE || job.status === jobStatus.CANCELLED) {
|
||||
clearInterval(interval);
|
||||
@ -1767,7 +1786,7 @@ describe('Batch API fallback job', function () {
|
||||
var job = JSON.parse(res.body);
|
||||
if (job.status === jobStatus.FAILED && job.fallback_status === jobStatus.FAILED) {
|
||||
clearInterval(interval);
|
||||
assert.deepEqual(job.query, expectedQuery);
|
||||
validateExpectedResponse(job.query, expectedQuery);
|
||||
done();
|
||||
} else if (job.status === jobStatus.DONE || job.status === jobStatus.CANCELLED) {
|
||||
clearInterval(interval);
|
||||
|
201
test/acceptance/job.timing.test.js
Normal file
201
test/acceptance/job.timing.test.js
Normal file
@ -0,0 +1,201 @@
|
||||
require('../helper');
|
||||
|
||||
var assert = require('../support/assert');
|
||||
var app = require(global.settings.app_root + '/app/app')();
|
||||
var querystring = require('qs');
|
||||
var metadataBackend = require('cartodb-redis')({
|
||||
host: global.settings.redis_host,
|
||||
port: global.settings.redis_port,
|
||||
max: global.settings.redisPool,
|
||||
idleTimeoutMillis: global.settings.redisIdleTimeoutMillis,
|
||||
reapIntervalMillis: global.settings.redisReapIntervalMillis
|
||||
});
|
||||
var batchFactory = require('../../batch');
|
||||
var jobStatus = require('../../batch/job_status');
|
||||
|
||||
describe('Batch API query timing', function () {
|
||||
|
||||
function createJob(jobDefinition, callback) {
|
||||
assert.response(app, {
|
||||
url: '/api/v2/sql/job?api_key=1234',
|
||||
headers: {
|
||||
'Content-Type': 'application/x-www-form-urlencoded',
|
||||
host: 'vizzuality.cartodb.com'
|
||||
},
|
||||
method: 'POST',
|
||||
data: querystring.stringify(jobDefinition)
|
||||
}, {
|
||||
status: 201
|
||||
}, function (res, err) {
|
||||
if (err) {
|
||||
return callback(err);
|
||||
}
|
||||
return callback(null, JSON.parse(res.body));
|
||||
});
|
||||
}
|
||||
|
||||
function getJobStatus(jobId, callback) {
|
||||
assert.response(app, {
|
||||
url: '/api/v2/sql/job/' + jobId + '?api_key=1234&',
|
||||
headers: {
|
||||
host: 'vizzuality.cartodb.com'
|
||||
},
|
||||
method: 'GET'
|
||||
}, {
|
||||
status: 200
|
||||
}, function (res, err) {
|
||||
if (err) {
|
||||
return callback(err);
|
||||
}
|
||||
return callback(null, JSON.parse(res.body));
|
||||
});
|
||||
}
|
||||
|
||||
function validateExpectedResponse(actual, expected) {
|
||||
actual.query.forEach(function(actualQuery, index) {
|
||||
var expectedQuery = expected.query[index];
|
||||
assert.ok(expectedQuery);
|
||||
Object.keys(expectedQuery).forEach(function(expectedKey) {
|
||||
assert.equal(actualQuery[expectedKey], expectedQuery[expectedKey]);
|
||||
});
|
||||
var propsToCheckDate = ['started_at', 'ended_at'];
|
||||
propsToCheckDate.forEach(function(propToCheckDate) {
|
||||
if (actualQuery.hasOwnProperty(propToCheckDate)) {
|
||||
assert.ok(new Date(actualQuery[propToCheckDate]));
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
assert.equal(actual.onsuccess, expected.onsuccess);
|
||||
assert.equal(actual.onerror, expected.onerror);
|
||||
}
|
||||
|
||||
var batch = batchFactory(metadataBackend);
|
||||
|
||||
before(function () {
|
||||
batch.start();
|
||||
});
|
||||
|
||||
after(function (done) {
|
||||
batch.stop();
|
||||
batch.drain(function () {
|
||||
metadataBackend.redisCmd(5, 'DEL', [ 'batch:queues:localhost' ], done);
|
||||
});
|
||||
});
|
||||
|
||||
describe('should report start and end time for each query with fallback queries', function () {
|
||||
var jobResponse;
|
||||
before(function(done) {
|
||||
createJob({
|
||||
"query": {
|
||||
"query": [
|
||||
{
|
||||
"query": "SELECT * FROM untitle_table_4 limit 1",
|
||||
"onerror": "SELECT * FROM untitle_table_4 limit 2"
|
||||
},
|
||||
{
|
||||
"query": "SELECT * FROM untitle_table_4 limit 3",
|
||||
"onerror": "SELECT * FROM untitle_table_4 limit 4"
|
||||
}
|
||||
],
|
||||
"onerror": "SELECT * FROM untitle_table_4 limit 5"
|
||||
}
|
||||
}, function(err, job) {
|
||||
jobResponse = job;
|
||||
return done(err);
|
||||
});
|
||||
});
|
||||
|
||||
it('should expose started_at and ended_at for all queries with fallback mechanism', function (done) {
|
||||
var expectedQuery = {
|
||||
query: [{
|
||||
query: 'SELECT * FROM untitle_table_4 limit 1',
|
||||
onerror: 'SELECT * FROM untitle_table_4 limit 2',
|
||||
status: 'done',
|
||||
fallback_status: 'skipped'
|
||||
}, {
|
||||
query: 'SELECT * FROM untitle_table_4 limit 3',
|
||||
onerror: 'SELECT * FROM untitle_table_4 limit 4',
|
||||
status: 'done',
|
||||
fallback_status: 'skipped'
|
||||
}],
|
||||
onerror: 'SELECT * FROM untitle_table_4 limit 5'
|
||||
};
|
||||
|
||||
var interval = setInterval(function () {
|
||||
getJobStatus(jobResponse.job_id, function(err, job) {
|
||||
if (job.status === jobStatus.DONE) {
|
||||
clearInterval(interval);
|
||||
validateExpectedResponse(job.query, expectedQuery);
|
||||
job.query.query.forEach(function(actualQuery) {
|
||||
assert.ok(actualQuery.started_at);
|
||||
assert.ok(actualQuery.ended_at);
|
||||
});
|
||||
done();
|
||||
} else if (job.status === jobStatus.FAILED || job.status === jobStatus.CANCELLED) {
|
||||
clearInterval(interval);
|
||||
done(new Error('Job ' + job.job_id + ' is ' + job.status + ', expected to be "done"'));
|
||||
}
|
||||
});
|
||||
}, 50);
|
||||
});
|
||||
});
|
||||
|
||||
describe('should report start and end time for each query also for failing queries', function () {
|
||||
var jobResponse;
|
||||
before(function(done) {
|
||||
createJob({
|
||||
"query": {
|
||||
"query": [
|
||||
{
|
||||
"query": "SELECT * FROM untitle_table_4 limit 1",
|
||||
"onerror": "SELECT * FROM untitle_table_4 limit 2"
|
||||
},
|
||||
{
|
||||
"query": "SELECT * FROM untitle_table_4 limit 3 failed",
|
||||
"onerror": "SELECT * FROM untitle_table_4 limit 4"
|
||||
}
|
||||
],
|
||||
"onerror": "SELECT * FROM untitle_table_4 limit 5"
|
||||
}
|
||||
}, function(err, job) {
|
||||
jobResponse = job;
|
||||
return done(err);
|
||||
});
|
||||
});
|
||||
|
||||
it('should expose started_at and ended_at for all queries with fallback mechanism (failed)', function (done) {
|
||||
var expectedQuery = {
|
||||
query: [{
|
||||
query: 'SELECT * FROM untitle_table_4 limit 1',
|
||||
onerror: 'SELECT * FROM untitle_table_4 limit 2',
|
||||
status: 'done',
|
||||
fallback_status: 'skipped'
|
||||
}, {
|
||||
query: 'SELECT * FROM untitle_table_4 limit 3 failed',
|
||||
onerror: 'SELECT * FROM untitle_table_4 limit 4',
|
||||
status: 'failed',
|
||||
fallback_status: 'done'
|
||||
}],
|
||||
onerror: 'SELECT * FROM untitle_table_4 limit 5'
|
||||
};
|
||||
|
||||
var interval = setInterval(function () {
|
||||
getJobStatus(jobResponse.job_id, function(err, job) {
|
||||
if (job.status === jobStatus.FAILED) {
|
||||
clearInterval(interval);
|
||||
validateExpectedResponse(job.query, expectedQuery);
|
||||
job.query.query.forEach(function(actualQuery) {
|
||||
assert.ok(actualQuery.started_at);
|
||||
assert.ok(actualQuery.ended_at);
|
||||
});
|
||||
done();
|
||||
} else if (job.status === jobStatus.DONE || job.status === jobStatus.CANCELLED) {
|
||||
clearInterval(interval);
|
||||
done(new Error('Job ' + job.job_id + ' is ' + job.status + ', expected to be "failed"'));
|
||||
}
|
||||
});
|
||||
}, 50);
|
||||
});
|
||||
});
|
||||
});
|
Loading…
Reference in New Issue
Block a user