Moves calls to SQL API to its own entity.
Groups affected tables and last updated time for affected tables into one request.
This commit is contained in:
Raul Ochoa
parent
75088c89d3
commit
3af45e1a32
94
lib/cartodb/api/query_tables_api.js
Normal file
94
lib/cartodb/api/query_tables_api.js
Normal file
@ -0,0 +1,94 @@
|
||||
var sqlApi = require('../sql/sql_api');
|
||||
|
||||
function QueryTablesApi() {
|
||||
}
|
||||
|
||||
var affectedTableRegexCache = {
|
||||
bbox: /!bbox!/g,
|
||||
pixel_width: /!pixel_width!/g,
|
||||
pixel_height: /!pixel_height!/g
|
||||
};
|
||||
|
||||
module.exports = QueryTablesApi;
|
||||
|
||||
QueryTablesApi.prototype.getLastUpdatedTime = function (username, api_key, tableNames, callback) {
|
||||
var sql = 'SELECT EXTRACT(EPOCH FROM max(updated_at)) as max FROM CDB_TableMetadata m WHERE m.tabname = any (ARRAY['+
|
||||
tableNames.map(function(t) { return "'" + t + "'::regclass"; }).join(',') +
|
||||
'])';
|
||||
|
||||
// call sql api
|
||||
sqlApi.query(username, api_key, sql, function(err, rows){
|
||||
if (err){
|
||||
var msg = err.message ? err.message : err;
|
||||
callback(new Error('could not find last updated timestamp: ' + msg));
|
||||
return;
|
||||
}
|
||||
// when the table has not updated_at means it hasn't been changed so a default last_updated is set
|
||||
var last_updated = 0;
|
||||
if(rows.length !== 0) {
|
||||
last_updated = rows[0].max || 0;
|
||||
}
|
||||
|
||||
callback(null, last_updated*1000);
|
||||
});
|
||||
};
|
||||
|
||||
QueryTablesApi.prototype.getAffectedTablesInQuery = function (username, api_key, sql, callback) {
|
||||
// Replace mapnik tokens
|
||||
sql = sql
|
||||
.replace(affectedTableRegexCache.bbox, 'ST_MakeEnvelope(0,0,0,0)')
|
||||
.replace(affectedTableRegexCache.pixel_width, '1')
|
||||
.replace(affectedTableRegexCache.pixel_height, '1')
|
||||
;
|
||||
|
||||
// Pass to CDB_QueryTables
|
||||
sql = 'SELECT CDB_QueryTables($windshaft$' + sql + '$windshaft$)';
|
||||
|
||||
// call sql api
|
||||
sqlApi.query(username, api_key, sql, function(err, rows){
|
||||
if (err){
|
||||
var msg = err.message ? err.message : err;
|
||||
callback(new Error('could not fetch source tables: ' + msg));
|
||||
return;
|
||||
}
|
||||
var qtables = rows[0].cdb_querytables;
|
||||
var tableNames = qtables.split(/^\{(.*)\}$/)[1];
|
||||
tableNames = tableNames ? tableNames.split(',') : [];
|
||||
callback(null, tableNames);
|
||||
});
|
||||
};
|
||||
|
||||
QueryTablesApi.prototype.getAffectedTablesAndLastUpdatedTime = function (username, api_key, sql, callback) {
|
||||
sql = sql
|
||||
.replace(affectedTableRegexCache.bbox, 'ST_MakeEnvelope(0,0,0,0)')
|
||||
.replace(affectedTableRegexCache.pixel_width, '1')
|
||||
.replace(affectedTableRegexCache.pixel_height, '1')
|
||||
;
|
||||
|
||||
var query = [
|
||||
'SELECT',
|
||||
'CDB_QueryTables($windshaft$' + sql + '$windshaft$) as tablenames,',
|
||||
'EXTRACT(EPOCH FROM max(updated_at)) as max',
|
||||
'FROM CDB_TableMetadata m',
|
||||
'WHERE m.tabname = any (CDB_QueryTables($windshaft$' + sql + '$windshaft$)::regclass[])'
|
||||
].join(' ');
|
||||
|
||||
sqlApi.query(username, api_key, query, function(err, rows){
|
||||
if (err || rows.length === 0) {
|
||||
var msg = err.message ? err.message : err;
|
||||
callback(new Error('could not fetch affected tables and last updated time: ' + msg));
|
||||
return;
|
||||
}
|
||||
|
||||
var qtables = rows[0].tablenames;
|
||||
var tableNames = qtables.split(/^\{(.*)\}$/)[1];
|
||||
tableNames = tableNames ? tableNames.split(',') : [];
|
||||
|
||||
var lastUpdatedTime = rows[0].max || 0;
|
||||
|
||||
callback(null, {
|
||||
affectedTables: tableNames,
|
||||
lastUpdatedTime: lastUpdatedTime * 1000
|
||||
});
|
||||
});
|
||||
};
|
||||
159
lib/cartodb/cache/cache_api.js
vendored
Normal file
159
lib/cartodb/cache/cache_api.js
vendored
Normal file
@ -0,0 +1,159 @@
|
||||
var _ = require('underscore'),
|
||||
request = require('request');
|
||||
|
||||
function QueryTablesApi() {
|
||||
}
|
||||
|
||||
var affectedTableRegexCache = {
|
||||
bbox: /!bbox!/g,
|
||||
pixel_width: /!pixel_width!/g,
|
||||
pixel_height: /!pixel_height!/g
|
||||
};
|
||||
|
||||
module.exports = QueryTablesApi;
|
||||
|
||||
QueryTablesApi.prototype.getLastUpdatedTime = function (username, api_key, tableNames, callback) {
|
||||
var sql = 'SELECT EXTRACT(EPOCH FROM max(updated_at)) as max FROM CDB_TableMetadata m WHERE m.tabname = any (ARRAY['+
|
||||
tableNames.map(function(t) { return "'" + t + "'::regclass"; }).join(',') +
|
||||
'])';
|
||||
|
||||
// call sql api
|
||||
sqlQuery(username, api_key, sql, function(err, rows){
|
||||
if (err){
|
||||
var msg = err.message ? err.message : err;
|
||||
callback(new Error('could not find last updated timestamp: ' + msg));
|
||||
return;
|
||||
}
|
||||
// when the table has not updated_at means it hasn't been changed so a default last_updated is set
|
||||
var last_updated = 0;
|
||||
if(rows.length !== 0) {
|
||||
last_updated = rows[0].max || 0;
|
||||
}
|
||||
|
||||
callback(null, last_updated*1000);
|
||||
});
|
||||
};
|
||||
|
||||
QueryTablesApi.prototype.getAffectedTablesInQuery = function (username, api_key, sql, callback) {
|
||||
// Replace mapnik tokens
|
||||
sql = sql
|
||||
.replace(affectedTableRegexCache.bbox, 'ST_MakeEnvelope(0,0,0,0)')
|
||||
.replace(affectedTableRegexCache.pixel_width, '1')
|
||||
.replace(affectedTableRegexCache.pixel_height, '1')
|
||||
;
|
||||
|
||||
// Pass to CDB_QueryTables
|
||||
sql = 'SELECT CDB_QueryTables($windshaft$' + sql + '$windshaft$)';
|
||||
|
||||
// call sql api
|
||||
sqlQuery(username, api_key, sql, function(err, rows){
|
||||
if (err){
|
||||
var msg = err.message ? err.message : err;
|
||||
callback(new Error('could not fetch source tables: ' + msg));
|
||||
return;
|
||||
}
|
||||
var qtables = rows[0].cdb_querytables;
|
||||
var tableNames = qtables.split(/^\{(.*)\}$/)[1];
|
||||
tableNames = tableNames ? tableNames.split(',') : [];
|
||||
callback(null, tableNames);
|
||||
});
|
||||
};
|
||||
|
||||
QueryTablesApi.prototype.getAffectedTablesAndLastUpdatedTime = function (username, api_key, sql, callback) {
|
||||
sql = sql
|
||||
.replace(affectedTableRegexCache.bbox, 'ST_MakeEnvelope(0,0,0,0)')
|
||||
.replace(affectedTableRegexCache.pixel_width, '1')
|
||||
.replace(affectedTableRegexCache.pixel_height, '1')
|
||||
;
|
||||
|
||||
var query = [
|
||||
'SELECT',
|
||||
'CDB_QueryTables($windshaft$' + sql + '$windshaft$) as tablenames,',
|
||||
'EXTRACT(EPOCH FROM max(updated_at)) as max',
|
||||
'FROM CDB_TableMetadata m',
|
||||
'WHERE m.tabname = any (CDB_QueryTables($windshaft$' + sql + '$windshaft$)::regclass[])'
|
||||
].join(' ');
|
||||
|
||||
sqlQuery(username, api_key, query, function(err, rows){
|
||||
if (err || rows.length === 0) {
|
||||
var msg = err.message ? err.message : err;
|
||||
callback(new Error('could not fetch affected tables and last updated time: ' + msg));
|
||||
return;
|
||||
}
|
||||
|
||||
var qtables = rows[0].tablenames;
|
||||
var tableNames = qtables.split(/^\{(.*)\}$/)[1];
|
||||
tableNames = tableNames ? tableNames.split(',') : [];
|
||||
|
||||
var lastUpdatedTime = rows[0].max || 0;
|
||||
|
||||
callback(null, {
|
||||
affectedTables: tableNames,
|
||||
lastUpdatedTime: lastUpdatedTime * 1000
|
||||
});
|
||||
});
|
||||
};
|
||||
|
||||
function sqlQuery(username, api_key, sql, callback) {
|
||||
var api = global.environment.sqlapi;
|
||||
|
||||
// build up api string
|
||||
var sqlapihostname = username;
|
||||
if ( api.domain ) sqlapihostname += '.' + api.domain;
|
||||
|
||||
var sqlapi = api.protocol + '://';
|
||||
if ( api.host && api.host != api.domain ) sqlapi += api.host;
|
||||
else sqlapi += sqlapihostname;
|
||||
sqlapi += ':' + api.port + '/api/' + api.version + '/sql';
|
||||
|
||||
var qs = { q: sql };
|
||||
|
||||
// add api_key if given
|
||||
if (_.isString(api_key) && api_key != '') { qs.api_key = api_key; }
|
||||
|
||||
// call sql api
|
||||
//
|
||||
// NOTE: using POST to avoid size limits:
|
||||
// See http://github.com/CartoDB/Windshaft-cartodb/issues/111
|
||||
//
|
||||
// NOTE: uses "host" header to allow IP based specification
|
||||
// of sqlapi address (and avoid a DNS lookup)
|
||||
//
|
||||
// NOTE: allows for keeping up to "maxConnections" concurrent
|
||||
// sockets opened per SQL-API host.
|
||||
// See http://nodejs.org/api/http.html#http_agent_maxsockets
|
||||
//
|
||||
var maxSockets = global.environment.maxConnections || 128;
|
||||
var maxGetLen = api.max_get_sql_length || 2048;
|
||||
var maxSQLTime = api.timeout || 100; // 1/10 of a second by default
|
||||
var reqSpec = {
|
||||
url:sqlapi,
|
||||
json:true,
|
||||
headers:{host: sqlapihostname}
|
||||
// http://nodejs.org/api/http.html#http_agent_maxsockets
|
||||
,pool:{maxSockets:maxSockets}
|
||||
// timeout in milliseconds
|
||||
,timeout:maxSQLTime
|
||||
};
|
||||
if ( sql.length > maxGetLen ) {
|
||||
reqSpec.method = 'POST';
|
||||
reqSpec.body = qs;
|
||||
} else {
|
||||
reqSpec.method = 'GET';
|
||||
reqSpec.qs = qs;
|
||||
}
|
||||
request(reqSpec, function(err, res, body) {
|
||||
if (err){
|
||||
console.log('ERROR connecting to SQL API on ' + sqlapi + ': ' + err);
|
||||
callback(err);
|
||||
return;
|
||||
}
|
||||
if (res.statusCode != 200) {
|
||||
var msg = res.body.error ? res.body.error : res.body;
|
||||
callback(new Error(msg));
|
||||
console.log('unexpected response status (' + res.statusCode + ') for sql query: ' + sql + ': ' + msg);
|
||||
return;
|
||||
}
|
||||
callback(null, body.rows);
|
||||
});
|
||||
}
|
||||
@ -2,9 +2,9 @@ var _ = require('underscore')
|
||||
, Step = require('step')
|
||||
, cartoData = require('cartodb-redis')(global.environment.redis)
|
||||
, Cache = require('./cache_validator')
|
||||
, QueryTablesApi = require('./api/query_tables_api')
|
||||
, mapnik = require('mapnik')
|
||||
, crypto = require('crypto')
|
||||
, request = require('request')
|
||||
, LZMA = require('lzma/lzma_worker.js').LZMA
|
||||
;
|
||||
|
||||
@ -19,6 +19,8 @@ if ( _.isUndefined(global.environment.sqlapi.domain) ) {
|
||||
|
||||
module.exports = function(){
|
||||
|
||||
var queryTablesApi = new QueryTablesApi();
|
||||
|
||||
var rendererConfig = _.defaults(global.environment.renderer || {}, {
|
||||
cache_ttl: 60000, // milliseconds
|
||||
metatile: 4,
|
||||
@ -88,121 +90,6 @@ module.exports = function(){
|
||||
// we have no SQL after layer creation.
|
||||
me.channelCache = {};
|
||||
|
||||
// Run a query through the SQL api
|
||||
me.sqlQuery = function (username, api_key, sql, callback) {
|
||||
var api = global.environment.sqlapi;
|
||||
|
||||
// build up api string
|
||||
var sqlapihostname = username;
|
||||
if ( api.domain ) sqlapihostname += '.' + api.domain;
|
||||
|
||||
var sqlapi = api.protocol + '://';
|
||||
if ( api.host && api.host != api.domain ) sqlapi += api.host;
|
||||
else sqlapi += sqlapihostname;
|
||||
sqlapi += ':' + api.port + '/api/' + api.version + '/sql';
|
||||
|
||||
var qs = { q: sql }
|
||||
|
||||
// add api_key if given
|
||||
if (_.isString(api_key) && api_key != '') { qs.api_key = api_key; }
|
||||
|
||||
// call sql api
|
||||
//
|
||||
// NOTE: using POST to avoid size limits:
|
||||
// See http://github.com/CartoDB/Windshaft-cartodb/issues/111
|
||||
//
|
||||
// NOTE: uses "host" header to allow IP based specification
|
||||
// of sqlapi address (and avoid a DNS lookup)
|
||||
//
|
||||
// NOTE: allows for keeping up to "maxConnections" concurrent
|
||||
// sockets opened per SQL-API host.
|
||||
// See http://nodejs.org/api/http.html#http_agent_maxsockets
|
||||
//
|
||||
var maxSockets = global.environment.maxConnections || 128;
|
||||
var maxGetLen = api.max_get_sql_length || 2048;
|
||||
var maxSQLTime = api.timeout || 100; // 1/10 of a second by default
|
||||
var reqSpec = {
|
||||
url:sqlapi,
|
||||
json:true,
|
||||
headers:{host: sqlapihostname}
|
||||
// http://nodejs.org/api/http.html#http_agent_maxsockets
|
||||
,pool:{maxSockets:maxSockets}
|
||||
// timeout in milliseconds
|
||||
,timeout:maxSQLTime
|
||||
}
|
||||
if ( sql.length > maxGetLen ) {
|
||||
reqSpec.method = 'POST';
|
||||
reqSpec.body = qs;
|
||||
} else {
|
||||
reqSpec.method = 'GET';
|
||||
reqSpec.qs = qs;
|
||||
}
|
||||
request(reqSpec, function(err, res, body) {
|
||||
if (err){
|
||||
console.log('ERROR connecting to SQL API on ' + sqlapi + ': ' + err);
|
||||
callback(err);
|
||||
return;
|
||||
}
|
||||
if (res.statusCode != 200) {
|
||||
var msg = res.body.error ? res.body.error : res.body;
|
||||
callback(new Error(msg));
|
||||
console.log('unexpected response status (' + res.statusCode + ') for sql query: ' + sql + ': ' + msg);
|
||||
return;
|
||||
}
|
||||
callback(null, body.rows);
|
||||
});
|
||||
};
|
||||
|
||||
//
|
||||
// Invoke callback with number of milliseconds since
|
||||
// last update in any of the given tables
|
||||
//
|
||||
me.findLastUpdated = function (username, api_key, tableNames, callback) {
|
||||
var sql = 'SELECT EXTRACT(EPOCH FROM max(updated_at)) as max FROM CDB_TableMetadata m WHERE m.tabname = any (ARRAY['+
|
||||
tableNames.map(function(t) { return "'" + t + "'::regclass"; }).join(',') +
|
||||
'])';
|
||||
|
||||
// call sql api
|
||||
me.sqlQuery(username, api_key, sql, function(err, rows){
|
||||
if (err){
|
||||
var msg = err.message ? err.message : err;
|
||||
callback(new Error('could not find last updated timestamp: ' + msg));
|
||||
return;
|
||||
}
|
||||
// when the table has not updated_at means it hasn't been changed so a default last_updated is set
|
||||
var last_updated = 0;
|
||||
if(rows.length !== 0) {
|
||||
last_updated = rows[0].max || 0;
|
||||
}
|
||||
callback(null, last_updated*1000);
|
||||
});
|
||||
};
|
||||
|
||||
me.affectedTables = function (username, api_key, sql, callback) {
|
||||
|
||||
// Replace mapnik tokens
|
||||
sql = sql.replace(RegExp('!bbox!', 'g'), 'ST_MakeEnvelope(0,0,0,0)')
|
||||
.replace(RegExp('!pixel_width!', 'g'), '1')
|
||||
.replace(RegExp('!pixel_height!', 'g'), '1')
|
||||
;
|
||||
|
||||
// Pass to CDB_QueryTables
|
||||
sql = 'SELECT CDB_QueryTables($windshaft$' + sql + '$windshaft$)';
|
||||
|
||||
// call sql api
|
||||
me.sqlQuery(username, api_key, sql, function(err, rows){
|
||||
if (err){
|
||||
var msg = err.message ? err.message : err;
|
||||
callback(new Error('could not fetch source tables: ' + msg));
|
||||
return;
|
||||
}
|
||||
var qtables = rows[0].cdb_querytables;
|
||||
var tableNames = qtables.split(/^\{(.*)\}$/)[1];
|
||||
tableNames = tableNames ? tableNames.split(',') : [];
|
||||
callback(null, tableNames);
|
||||
});
|
||||
};
|
||||
|
||||
me.buildCacheChannel = function (dbName, tableNames){
|
||||
return dbName + ':' + tableNames.join(',');
|
||||
};
|
||||
@ -304,7 +191,7 @@ module.exports = function(){
|
||||
if ( req.profiler ) req.profiler.done('getSignerMapKey');
|
||||
key = data;
|
||||
}
|
||||
me.affectedTables(user, key, sql, this); // in addCacheChannel
|
||||
queryTablesApi.getAffectedTablesInQuery(user, key, sql, this); // in addCacheChannel
|
||||
},
|
||||
function finish(err, data) {
|
||||
next(err,data);
|
||||
@ -426,18 +313,17 @@ module.exports = function(){
|
||||
var key = req.params.map_key || req.params.api_key;
|
||||
|
||||
var cacheKey = dbName + ':' + token;
|
||||
var tabNames;
|
||||
|
||||
Step(
|
||||
function getTables() {
|
||||
me.affectedTables(usr, key, sql, this); // in afterLayergroupCreate
|
||||
function getAffectedTablesAndLastUpdatedTime() {
|
||||
queryTablesApi.getAffectedTablesAndLastUpdatedTime(usr, key, sql, this);
|
||||
},
|
||||
function getLastupdated(err, tableNames) {
|
||||
if (req.profiler) req.profiler.done('affectedTables');
|
||||
function handleAffectedTablesAndLastUpdatedTime(err, result) {
|
||||
if (req.profiler) req.profiler.done('queryTablesAndLastUpdated');
|
||||
if ( err ) throw err;
|
||||
var cacheChannel = me.buildCacheChannel(dbName,tableNames);
|
||||
// store for caching from me.afterLayergroupCreate
|
||||
var cacheChannel = me.buildCacheChannel(dbName, result.affectedTables);
|
||||
me.channelCache[cacheKey] = cacheChannel;
|
||||
|
||||
if (req.res && req.method == 'GET') {
|
||||
var res = req.res;
|
||||
if ( req.query && req.query.cache_policy == 'persist' ) {
|
||||
@ -449,16 +335,10 @@ module.exports = function(){
|
||||
res.header('Last-Modified', (new Date()).toUTCString());
|
||||
res.header('X-Cache-Channel', cacheChannel);
|
||||
}
|
||||
// find last updated
|
||||
if ( ! tableNames.length ) return 0; // skip for no affected tables
|
||||
tabNames = tableNames;
|
||||
me.findLastUpdated(usr, key, tableNames, this);
|
||||
},
|
||||
function(err, lastUpdated) {
|
||||
if ( err ) throw err;
|
||||
if (req.profiler && tabNames) req.profiler.done('findLastUpdated');
|
||||
response.layergroupid = response.layergroupid + ':' + lastUpdated; // use epoch
|
||||
response.last_updated = new Date(lastUpdated).toISOString();
|
||||
|
||||
// last update for layergroup cache buster
|
||||
response.layergroupid = response.layergroupid + ':' + result.lastUpdatedTime;
|
||||
response.last_updated = new Date(result.lastUpdatedTime).toISOString();
|
||||
return null;
|
||||
},
|
||||
function finish(err) {
|
||||
|
||||
66
lib/cartodb/sql/sql_api.js
Normal file
66
lib/cartodb/sql/sql_api.js
Normal file
@ -0,0 +1,66 @@
|
||||
var _ = require('underscore'),
|
||||
request = require('request');
|
||||
|
||||
module.exports.query = function (username, api_key, sql, callback) {
|
||||
var api = global.environment.sqlapi;
|
||||
|
||||
// build up api string
|
||||
var sqlapihostname = username;
|
||||
if ( api.domain ) sqlapihostname += '.' + api.domain;
|
||||
|
||||
var sqlapi = api.protocol + '://';
|
||||
if ( api.host && api.host != api.domain ) sqlapi += api.host;
|
||||
else sqlapi += sqlapihostname;
|
||||
sqlapi += ':' + api.port + '/api/' + api.version + '/sql';
|
||||
|
||||
var qs = { q: sql };
|
||||
|
||||
// add api_key if given
|
||||
if (_.isString(api_key) && api_key != '') { qs.api_key = api_key; }
|
||||
|
||||
// call sql api
|
||||
//
|
||||
// NOTE: using POST to avoid size limits:
|
||||
// See http://github.com/CartoDB/Windshaft-cartodb/issues/111
|
||||
//
|
||||
// NOTE: uses "host" header to allow IP based specification
|
||||
// of sqlapi address (and avoid a DNS lookup)
|
||||
//
|
||||
// NOTE: allows for keeping up to "maxConnections" concurrent
|
||||
// sockets opened per SQL-API host.
|
||||
// See http://nodejs.org/api/http.html#http_agent_maxsockets
|
||||
//
|
||||
var maxSockets = global.environment.maxConnections || 128;
|
||||
var maxGetLen = api.max_get_sql_length || 2048;
|
||||
var maxSQLTime = api.timeout || 100; // 1/10 of a second by default
|
||||
var reqSpec = {
|
||||
url:sqlapi,
|
||||
json:true,
|
||||
headers:{host: sqlapihostname}
|
||||
// http://nodejs.org/api/http.html#http_agent_maxsockets
|
||||
,pool:{maxSockets:maxSockets}
|
||||
// timeout in milliseconds
|
||||
,timeout:maxSQLTime
|
||||
};
|
||||
if ( sql.length > maxGetLen ) {
|
||||
reqSpec.method = 'POST';
|
||||
reqSpec.body = qs;
|
||||
} else {
|
||||
reqSpec.method = 'GET';
|
||||
reqSpec.qs = qs;
|
||||
}
|
||||
request(reqSpec, function(err, res, body) {
|
||||
if (err){
|
||||
console.log('ERROR connecting to SQL API on ' + sqlapi + ': ' + err);
|
||||
callback(err);
|
||||
return;
|
||||
}
|
||||
if (res.statusCode != 200) {
|
||||
var msg = res.body.error ? res.body.error : res.body;
|
||||
callback(new Error(msg));
|
||||
console.log('unexpected response status (' + res.statusCode + ') for sql query: ' + sql + ': ' + msg);
|
||||
return;
|
||||
}
|
||||
callback(null, body.rows);
|
||||
});
|
||||
};
|
||||
Loading…
Reference in New Issue
Block a user