diff --git a/lib/cartodb/api/query_tables_api.js b/lib/cartodb/api/query_tables_api.js new file mode 100644 index 00000000..9782d91a --- /dev/null +++ b/lib/cartodb/api/query_tables_api.js @@ -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 + }); + }); +}; diff --git a/lib/cartodb/cache/cache_api.js b/lib/cartodb/cache/cache_api.js new file mode 100644 index 00000000..915cc929 --- /dev/null +++ b/lib/cartodb/cache/cache_api.js @@ -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); + }); +} \ No newline at end of file diff --git a/lib/cartodb/server_options.js b/lib/cartodb/server_options.js index cb97e38e..cd8098ad 100644 --- a/lib/cartodb/server_options.js +++ b/lib/cartodb/server_options.js @@ -1,10 +1,10 @@ var _ = require('underscore') , Step = require('step') , cartoData = require('cartodb-redis')(global.environment.redis) - , Cache = require('./cache_validator') + , 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,44 +313,37 @@ 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 getLastupdated(err, tableNames) { - if (req.profiler) req.profiler.done('affectedTables'); - if ( err ) throw err; - var cacheChannel = me.buildCacheChannel(dbName,tableNames); - // store for caching from me.afterLayergroupCreate - me.channelCache[cacheKey] = cacheChannel; - if (req.res && req.method == 'GET') { - var res = req.res; - if ( req.query && req.query.cache_policy == 'persist' ) { - res.header('Cache-Control', 'public,max-age=31536000'); // 1 year - } else { - var ttl = global.environment.varnish.ttl || 86400; - res.header('Cache-Control', 'public,max-age='+ttl+',must-revalidate'); + function getAffectedTablesAndLastUpdatedTime() { + queryTablesApi.getAffectedTablesAndLastUpdatedTime(usr, key, sql, this); + }, + function handleAffectedTablesAndLastUpdatedTime(err, result) { + if (req.profiler) req.profiler.done('queryTablesAndLastUpdated'); + if ( err ) throw err; + 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' ) { + res.header('Cache-Control', 'public,max-age=31536000'); // 1 year + } else { + var ttl = global.environment.varnish.ttl || 86400; + res.header('Cache-Control', 'public,max-age='+ttl+',must-revalidate'); + } + res.header('Last-Modified', (new Date()).toUTCString()); + res.header('X-Cache-Channel', cacheChannel); } - res.header('Last-Modified', (new Date()).toUTCString()); - res.header('X-Cache-Channel', cacheChannel); + + // 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) { + done(err); } - // 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(); - return null; - }, - function finish(err) { - done(err); - } ); }; diff --git a/lib/cartodb/sql/sql_api.js b/lib/cartodb/sql/sql_api.js new file mode 100644 index 00000000..44c831e2 --- /dev/null +++ b/lib/cartodb/sql/sql_api.js @@ -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); + }); +};