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

Adds start and end time for batch queries with fallback

Browse Source
This commit is contained in:
Raul Ochoa 2016-06-29 13:56:45 +02:00
parent 30ee62e70d
commit 1d8f5539a7
4 changed files with 254 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
NEWS.md
View File

@ -1,6 +1,9 @@
1.30.2 - 2016-mm-dd
-------------------
New features:
* Adds start and end time for batch queries with fallback.
1.30.1 - 2016-06-23
-------------------

6
batch/models/query/query.js
View File

@ -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;
}

69
test/acceptance/job.fallback.test.js
View File

@ -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
View 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);
});
});
});