Merge pull request #620 from CartoDB/gears

Be able to inject custom middlewares
This commit is contained in:
Daniel G. Aubert 2019-10-07 17:37:32 +02:00 committed by GitHub
commit 9a1acc6780
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
203 changed files with 975 additions and 427 deletions

View File

@ -11,7 +11,7 @@ check:
jshint: jshint:
@echo "***jshint***" @echo "***jshint***"
@./node_modules/.bin/jshint app/ batch/ test/ app.js @./node_modules/.bin/jshint lib/ test/ app.js
TEST_SUITE := $(shell find test/{unit,integration,acceptance} -name "*.js") TEST_SUITE := $(shell find test/{unit,integration,acceptance} -name "*.js")
TEST_SUITE_UNIT := $(shell find test/unit -name "*.js") TEST_SUITE_UNIT := $(shell find test/unit -name "*.js")

13
app.js
View File

@ -69,15 +69,10 @@ if (global.settings.log_filename) {
global.log4js.configure(log4jsConfig); global.log4js.configure(log4jsConfig);
global.logger = global.log4js.getLogger(); global.logger = global.log4js.getLogger();
// kick off controller
if ( ! global.settings.base_url ) {
global.settings.base_url = '/api/*';
}
const version = require("./package").version; const version = require("./package").version;
const StatsClient = require('./app/stats/client'); const StatsClient = require('./lib/stats/client');
if (global.settings.statsd) { if (global.settings.statsd) {
// Perform keyword substitution in statsd // Perform keyword substitution in statsd
if (global.settings.statsd.prefix) { if (global.settings.statsd.prefix) {
@ -86,9 +81,9 @@ if (global.settings.statsd) {
} }
const statsClient = StatsClient.getInstance(global.settings.statsd); const statsClient = StatsClient.getInstance(global.settings.statsd);
const serverFactory = require('./app/server'); const createServer = require('./lib/server');
const server = serverFactory(statsClient); const server = createServer(statsClient);
const listener = server.listen(global.settings.node_port, global.settings.node_host); const listener = server.listen(global.settings.node_port, global.settings.node_host);
listener.on('listening', function() { listener.on('listening', function() {
console.info("Using Node.js %s", process.version); console.info("Using Node.js %s", process.version);

View File

@ -1,84 +0,0 @@
'use strict';
const { Router: router } = require('express');
const UserDatabaseService = require('../services/user_database_service');
const UserLimitsService = require('../services/user_limits');
const BatchLogger = require('../../batch/batch-logger');
const JobPublisher = require('../../batch/pubsub/job-publisher');
const JobQueue = require('../../batch/job_queue');
const JobBackend = require('../../batch/job_backend');
const JobCanceller = require('../../batch/job_canceller');
const JobService = require('../../batch/job_service');
const QueryController = require('./query_controller');
const CopyController = require('./copy_controller');
const JobController = require('./job_controller');
const socketTimeout = require('../middlewares/socket-timeout');
const logger = require('../middlewares/logger');
const profiler = require('../middlewares/profiler');
const cors = require('../middlewares/cors');
const servedByHostHeader = require('../middlewares/served-by-host-header');
module.exports = class ApiRouter {
constructor ({ redisPool, metadataBackend, statsClient, dataIngestionLogger }) {
this.statsClient = statsClient;
const userLimitsServiceOptions = {
limits: {
rateLimitsEnabled: global.settings.ratelimits.rateLimitsEnabled
}
};
const userDatabaseService = new UserDatabaseService(metadataBackend);
const userLimitsService = new UserLimitsService(metadataBackend, userLimitsServiceOptions);
this.queryController = new QueryController(
metadataBackend,
userDatabaseService,
statsClient,
userLimitsService
);
this.copyController = new CopyController(
metadataBackend,
userDatabaseService,
userLimitsService,
dataIngestionLogger
);
const logger = new BatchLogger(global.settings.batch_log_filename, 'batch-queries');
const jobPublisher = new JobPublisher(redisPool);
const jobQueue = new JobQueue(metadataBackend, jobPublisher, logger);
const jobBackend = new JobBackend(metadataBackend, jobQueue, logger);
const jobCanceller = new JobCanceller();
const jobService = new JobService(jobBackend, jobCanceller, logger);
this.jobController = new JobController(
metadataBackend,
userDatabaseService,
jobService,
statsClient,
userLimitsService
);
}
route (app) {
const apiRouter = router({ mergeParams: true });
apiRouter.use(socketTimeout());
apiRouter.use(logger());
apiRouter.use(profiler({ statsClient: this.statsClient }));
apiRouter.use(cors());
apiRouter.use(servedByHostHeader());
this.queryController.route(apiRouter);
this.copyController.route(apiRouter);
this.jobController.route(apiRouter);
app.use(`${global.settings.base_url}`, apiRouter);
}
};

View File

@ -1,9 +1,36 @@
// Time in milliseconds to force GC cycle. // Time in milliseconds to force GC cycle.
// Disable by using <=0 value. // Disable by using <=0 value.
module.exports.gc_interval = 10000; module.exports.gc_interval = 10000;
// In case the base_url has a :user param the username will be the one specified in the URL, module.exports.routes = {
// otherwise it will fallback to extract the username from the host header. // Each entry corresponds with an express' router.
module.exports.base_url = '(?:/api/:version|/user/:user/api/:version)'; // You must define at least one path. However, middlewares are optional.
api: [{
// Required: path where other "routers" or "controllers" will be attached to.
paths: [
// In case the path has a :user param the username will be the one specified in the URL,
// otherwise it will fallback to extract the username from the host header.
'/api/:version',
'/user/:user/api/:version',
],
// Optional: attach middlewares at the begining of the router
// to perform custom operations.
middlewares: [
function noop () {
return function noopMiddleware (req, res, next) {
next();
}
}
],
sql: [{
// Required
paths: [
'/sql'
],
// Optional
middlewares: []
}]
}]
};
// If useProfiler is true every response will be served with an // If useProfiler is true every response will be served with an
// X-SQLAPI-Profile header containing elapsed timing for various // X-SQLAPI-Profile header containing elapsed timing for various
// steps taken for producing the response. // steps taken for producing the response.

View File

@ -1,9 +1,36 @@
// Time in milliseconds to force GC cycle. // Time in milliseconds to force GC cycle.
// Disable by using <=0 value. // Disable by using <=0 value.
module.exports.gc_interval = 10000; module.exports.gc_interval = 10000;
// In case the base_url has a :user param the username will be the one specified in the URL, module.exports.routes = {
// otherwise it will fallback to extract the username from the host header. // Each entry corresponds with an express' router.
module.exports.base_url = '(?:/api/:version|/user/:user/api/:version)'; // You must define at least one path. However, middlewares are optional.
api: [{
// Required: path where other "routers" or "controllers" will be attached to.
paths: [
// In case the path has a :user param the username will be the one specified in the URL,
// otherwise it will fallback to extract the username from the host header.
'/api/:version',
'/user/:user/api/:version',
],
// Optional: attach middlewares at the begining of the router
// to perform custom operations.
middlewares: [
function noop () {
return function noopMiddleware (req, res, next) {
next();
}
}
],
sql: [{
// Required
paths: [
'/sql'
],
// Optional
middlewares: []
}]
}]
};
// If useProfiler is true every response will be served with an // If useProfiler is true every response will be served with an
// X-SQLAPI-Profile header containing elapsed timing for various // X-SQLAPI-Profile header containing elapsed timing for various
// steps taken for producing the response. // steps taken for producing the response.

View File

@ -1,9 +1,36 @@
// Time in milliseconds to force GC cycle. // Time in milliseconds to force GC cycle.
// Disable by using <=0 value. // Disable by using <=0 value.
module.exports.gc_interval = 10000; module.exports.gc_interval = 10000;
// In case the base_url has a :user param the username will be the one specified in the URL, module.exports.routes = {
// otherwise it will fallback to extract the username from the host header. // Each entry corresponds with an express' router.
module.exports.base_url = '(?:/api/:version|/user/:user/api/:version)'; // You must define at least one path. However, middlewares are optional.
api: [{
// Required: path where other "routers" or "controllers" will be attached to.
paths: [
// In case the path has a :user param the username will be the one specified in the URL,
// otherwise it will fallback to extract the username from the host header.
'/api/:version',
'/user/:user/api/:version',
],
// Optional: attach middlewares at the begining of the router
// to perform custom operations.
middlewares: [
function noop () {
return function noopMiddleware (req, res, next) {
next();
}
}
],
sql: [{
// Required
paths: [
'/sql'
],
// Optional
middlewares: []
}]
}]
};
// If useProfiler is true every response will be served with an // If useProfiler is true every response will be served with an
// X-SQLAPI-Profile header containing elapsed timing for various // X-SQLAPI-Profile header containing elapsed timing for various
// steps taken for producing the response. // steps taken for producing the response.

View File

@ -1,9 +1,36 @@
// Time in milliseconds to force GC cycle. // Time in milliseconds to force GC cycle.
// Disable by using <=0 value. // Disable by using <=0 value.
module.exports.gc_interval = 10000; module.exports.gc_interval = 10000;
// In case the base_url has a :user param the username will be the one specified in the URL, module.exports.routes = {
// otherwise it will fallback to extract the username from the host header. // Each entry corresponds with an express' router.
module.exports.base_url = '(?:/api/:version|/user/:user/api/:version)'; // You must define at least one path. However, middlewares are optional.
api: [{
// Required: path where other "routers" or "controllers" will be attached to.
paths: [
// In case the path has a :user param the username will be the one specified in the URL,
// otherwise it will fallback to extract the username from the host header.
'/api/:version',
'/user/:user/api/:version',
],
// Optional: attach middlewares at the begining of the router
// to perform custom operations.
middlewares: [
function noop () {
return function noopMiddleware (req, res, next) {
next();
}
}
],
sql: [{
// Required
paths: [
'/sql'
],
// Optional
middlewares: []
}]
}]
};
// If useProfiler is true every response will be served with an // If useProfiler is true every response will be served with an
// X-SQLAPI-Profile header containing elapsed timing for various // X-SQLAPI-Profile header containing elapsed timing for various
// steps taken for producing the response. // steps taken for producing the response.

64
lib/api/api-router.js Normal file
View File

@ -0,0 +1,64 @@
'use strict';
const { Router: router } = require('express');
const SqlRouter = require('./sql/sql-router');
const HealthCheckController = require('./health-check-controller');
const VersionController = require('./version-controller');
const JobsWipController = require('./jobs-wip-controller');
const BatchLogger = require('../batch/batch-logger');
const JobPublisher = require('../batch/pubsub/job-publisher');
const JobQueue = require('../batch/job-queue');
const JobBackend = require('../batch/job-backend');
const JobCanceller = require('../batch/job-canceller');
const JobService = require('../batch/job-service');
module.exports = class ApiRouter {
constructor ({ redisPool, metadataBackend, statsClient, dataIngestionLogger }) {
const logger = new BatchLogger(global.settings.batch_log_filename, 'batch-queries');
const jobPublisher = new JobPublisher(redisPool);
const jobQueue = new JobQueue(metadataBackend, jobPublisher, logger);
const jobBackend = new JobBackend(metadataBackend, jobQueue, logger);
const jobCanceller = new JobCanceller();
const jobService = new JobService(jobBackend, jobCanceller, logger);
this.healthCheckController = new HealthCheckController();
this.versionController = new VersionController();
this.jobsWipController = new JobsWipController({ jobService });
this.sqlRouter = new SqlRouter({
metadataBackend,
statsClient,
dataIngestionLogger,
jobService
});
}
route (app, routes) {
routes.forEach(route => {
const apiRouter = router({ mergeParams: true });
const paths = route.paths;
const middlewares = route.middlewares || [];
middlewares.forEach(middleware => apiRouter.use(middleware()));
// FIXME: version controller should be atached to the main entry point: "/"
// instead of "/api/:version" or "/user/:user/api/:version"
this.healthCheckController.route(apiRouter);
// FIXME: version controller should be atached to the main entry point: "/"
// instead of "/api/:version" or "/user/:user/api/:version"
this.versionController.route(apiRouter);
this.jobsWipController.route(apiRouter);
this.sqlRouter.route(apiRouter, route.sql);
paths.forEach(path => app.use(path, apiRouter));
});
}
};

View File

@ -1,14 +1,14 @@
'use strict'; 'use strict';
const HealthCheckBackend = require('../monitoring/health_check'); const HealthCheckBackend = require('../monitoring/health-check');
module.exports = class HealthCheckController { module.exports = class HealthCheckController {
constructor () { constructor () {
this.healthCheckBackend = new HealthCheckBackend(global.settings.disabled_file); this.healthCheckBackend = new HealthCheckBackend(global.settings.disabled_file);
} }
route (app) { route (apiRouter) {
app.get(`${global.settings.base_url}/health`, healthCheck({ healthCheckBackend: this.healthCheckBackend })); apiRouter.get('/health', healthCheck({ healthCheckBackend: this.healthCheckBackend }));
} }
}; };

View File

@ -0,0 +1,39 @@
'use strict';
const bodyParser = require('./middlewares/body-parser');
const error = require('./middlewares/error');
module.exports = class JobsWipController {
constructor ({ jobService }) {
this.jobService = jobService;
}
route (apiRouter) {
apiRouter.get('/jobs-wip', [
bodyParser(),
listWorkInProgressJobs(this.jobService),
sendResponse(),
error()
]);
}
};
function listWorkInProgressJobs (jobService) {
return function listWorkInProgressJobsMiddleware (req, res, next) {
jobService.listWorkInProgressJobs((err, list) => {
if (err) {
return next(err);
}
res.body = list;
next();
});
};
}
function sendResponse () {
return function sendResponseMiddleware (req, res) {
res.status(res.statusCode || 200).send(res.body);
};
}

View File

@ -1,6 +1,6 @@
'use strict'; 'use strict';
const pgEntitiesAccessValidator = require('../services/pg-entities-access-validator'); const pgEntitiesAccessValidator = require('../../services/pg-entities-access-validator');
module.exports = function accessValidator () { module.exports = function accessValidator () {
return function accessValidatorMiddleware (req, res, next) { return function accessValidatorMiddleware (req, res, next) {

View File

@ -1,6 +1,6 @@
'use strict'; 'use strict';
const AuthApi = require('../auth/auth_api'); const AuthApi = require('../../auth/auth-api');
const basicAuth = require('basic-auth'); const basicAuth = require('basic-auth');
module.exports = function authorization (metadataBackend, forceToBeMaster = false) { module.exports = function authorization (metadataBackend, forceToBeMaster = false) {

View File

@ -1,6 +1,6 @@
'use strict'; 'use strict';
const getContentDisposition = require('../utils/content_disposition'); const getContentDisposition = require('../../utils/content-disposition');
module.exports = function content () { module.exports = function content () {
return function contentMiddleware (req, res, next) { return function contentMiddleware (req, res, next) {

View File

@ -1,7 +1,7 @@
'use strict'; 'use strict';
const errorHandlerFactory = require('../services/error_handler_factory'); const errorHandlerFactory = require('../../services/error-handler-factory');
const { stringifyForLogs } = require('../utils/logs'); const { stringifyForLogs } = require('../../utils/logs');
const MAX_ERROR_STRING_LENGTH = 1024; const MAX_ERROR_STRING_LENGTH = 1024;
module.exports = function error() { module.exports = function error() {

View File

@ -1,6 +1,6 @@
'use strict'; 'use strict';
const formats = require('../models/formats'); const formats = require('../../models/formats');
module.exports = function formatter () { module.exports = function formatter () {
return function formatterMiddleware (req, res, next) { return function formatterMiddleware (req, res, next) {

View File

@ -1,6 +1,6 @@
'use strict'; 'use strict';
const { stringifyForLogs } = require('../utils/logs'); const { stringifyForLogs } = require('../../utils/logs');
const MAX_SQL_LENGTH = (global.settings.logQueries && global.settings.maxQueriesLogLength) || 1024; const MAX_SQL_LENGTH = (global.settings.logQueries && global.settings.maxQueriesLogLength) || 1024;

View File

@ -1,7 +1,7 @@
'use strict'; 'use strict';
const sanitizeFilename = require('../utils/filename_sanitizer'); const sanitizeFilename = require('../../utils/filename-sanitizer');
const formats = require('../models/formats'); const formats = require('../../models/formats');
module.exports = function params ({ strategy = 'query' } = {}) { module.exports = function params ({ strategy = 'query' } = {}) {
const getParams = getParamsFromStrategy(strategy); const getParams = getParamsFromStrategy(strategy);

View File

@ -1,6 +1,6 @@
'use strict'; 'use strict';
const Profiler = require('../stats/profiler-proxy'); const Profiler = require('../../stats/profiler-proxy');
module.exports = function profiler ({ statsClient }) { module.exports = function profiler ({ statsClient }) {
return function profilerMiddleware (req, res, next) { return function profilerMiddleware (req, res, next) {
@ -13,7 +13,7 @@ module.exports = function profiler ({ statsClient }) {
}; };
}; };
module.exports.initializeProfilerMiddleware = function initializeProfiler (label) { module.exports.initializeProfiler = function initializeProfiler (label) {
return function initializeProfilerMiddleware (req, res, next) { return function initializeProfilerMiddleware (req, res, next) {
if (req.profiler) { if (req.profiler) {
req.profiler.start(`sqlapi.${label}`); req.profiler.start(`sqlapi.${label}`);
@ -23,7 +23,7 @@ module.exports.initializeProfilerMiddleware = function initializeProfiler (label
}; };
}; };
module.exports.finishProfilerMiddleware = function finishProfiler () { module.exports.finishProfiler = function finishProfiler () {
return function finishProfilerMiddleware (req, res, next) { return function finishProfilerMiddleware (req, res, next) {
if (req.profiler) { if (req.profiler) {
req.profiler.end(); req.profiler.end();

View File

@ -1,6 +1,6 @@
'use strict'; 'use strict';
const queryMayWrite = require('../utils/query_may_write'); const queryMayWrite = require('../../utils/query-may-write');
module.exports = function mayWrite () { module.exports = function mayWrite () {
return function mayWriteMiddleware (req, res, next) { return function mayWriteMiddleware (req, res, next) {

View File

@ -1,8 +1,8 @@
'use strict'; 'use strict';
const CdbRequest = require('../models/cartodb_request'); const CdbRequest = require('../../models/cartodb-request');
module.exports = function user(metadataBackend) { module.exports = function user (metadataBackend) {
const cdbRequest = new CdbRequest(); const cdbRequest = new CdbRequest();
return function userMiddleware (req, res, next) { return function userMiddleware (req, res, next) {
@ -23,16 +23,16 @@ module.exports = function user(metadataBackend) {
}; };
}; };
function getUserNameFromRequest(req, cdbRequest) { function getUserNameFromRequest (req, cdbRequest) {
return cdbRequest.userByReq(req); return cdbRequest.userByReq(req);
} }
function checkUserExists(metadataBackend, userName, callback) { function checkUserExists (metadataBackend, userName, callback) {
metadataBackend.getUserId(userName, function(err) { metadataBackend.getUserId(userName, function(err) {
callback(err, !err); callback(err, !err);
}); });
} }
function errorUserNotFoundMessageTemplate(user) { function errorUserNotFoundMessageTemplate (user) {
return `Sorry, we can't find CARTO user '${user}'. Please check that you have entered the correct domain.`; return `Sorry, we can't find CARTO user '${user}'. Please check that you have entered the correct domain.`;
} }

View File

@ -1,21 +1,21 @@
'use strict'; 'use strict';
const userMiddleware = require('../middlewares/user'); const user = require('../middlewares/user');
const errorMiddleware = require('../middlewares/error'); const error = require('../middlewares/error');
const authorizationMiddleware = require('../middlewares/authorization'); const authorization = require('../middlewares/authorization');
const connectionParamsMiddleware = require('../middlewares/connection-params'); const connectionParams = require('../middlewares/connection-params');
const { initializeProfilerMiddleware } = require('../middlewares/profiler'); const { initializeProfiler } = require('../middlewares/profiler');
const rateLimitsMiddleware = require('../middlewares/rate-limit'); const dbQuota = require('../middlewares/db-quota');
const dbQuotaMiddleware = require('../middlewares/db-quota'); const bodyParser = require('../middlewares/body-parser');
const { RATE_LIMIT_ENDPOINTS_GROUPS } = rateLimitsMiddleware; const rateLimits = require('../middlewares/rate-limit');
const errorHandlerFactory = require('../services/error_handler_factory'); const { RATE_LIMIT_ENDPOINTS_GROUPS } = rateLimits;
const StreamCopy = require('../services/stream_copy'); const errorHandlerFactory = require('../../services/error-handler-factory');
const StreamCopyMetrics = require('../services/stream_copy_metrics'); const StreamCopy = require('../../services/stream-copy');
const Throttler = require('../services/throttler-stream'); const StreamCopyMetrics = require('../../services/stream-copy-metrics');
const Throttler = require('../../services/throttler-stream');
const zlib = require('zlib'); const zlib = require('zlib');
const { PassThrough } = require('stream'); const { PassThrough } = require('stream');
const params = require('../middlewares/params'); const params = require('../middlewares/params');
const bodyParserMiddleware = require('../middlewares/body-parser');
module.exports = class CopyController { module.exports = class CopyController {
constructor (metadataBackend, userDatabaseService, userLimitsService, logger) { constructor (metadataBackend, userDatabaseService, userLimitsService, logger) {
@ -25,40 +25,40 @@ module.exports = class CopyController {
this.logger = logger; this.logger = logger;
} }
route (apiRouter) { route (sqlRouter) {
const copyFromMiddlewares = endpointGroup => { const copyFromMiddlewares = endpointGroup => {
return [ return [
initializeProfilerMiddleware('copyfrom'), initializeProfiler('copyfrom'),
userMiddleware(this.metadataBackend), user(this.metadataBackend),
rateLimitsMiddleware(this.userLimitsService, endpointGroup), rateLimits(this.userLimitsService, endpointGroup),
authorizationMiddleware(this.metadataBackend), authorization(this.metadataBackend),
connectionParamsMiddleware(this.userDatabaseService), connectionParams(this.userDatabaseService),
dbQuotaMiddleware(), dbQuota(),
params({ strategy: 'copyfrom' }), params({ strategy: 'copyfrom' }),
handleCopyFrom(this.logger), handleCopyFrom(this.logger),
errorHandler(this.logger), errorHandler(this.logger),
errorMiddleware() error()
]; ];
}; };
const copyToMiddlewares = endpointGroup => { const copyToMiddlewares = endpointGroup => {
return [ return [
bodyParserMiddleware(), bodyParser(),
initializeProfilerMiddleware('copyto'), initializeProfiler('copyto'),
userMiddleware(this.metadataBackend), user(this.metadataBackend),
rateLimitsMiddleware(this.userLimitsService, endpointGroup), rateLimits(this.userLimitsService, endpointGroup),
authorizationMiddleware(this.metadataBackend), authorization(this.metadataBackend),
connectionParamsMiddleware(this.userDatabaseService), connectionParams(this.userDatabaseService),
params({ strategy: 'copyto' }), params({ strategy: 'copyto' }),
handleCopyTo(this.logger), handleCopyTo(this.logger),
errorHandler(this.logger), errorHandler(this.logger),
errorMiddleware() error()
]; ];
}; };
apiRouter.post('/sql/copyfrom', copyFromMiddlewares(RATE_LIMIT_ENDPOINTS_GROUPS.COPY_FROM)); sqlRouter.post('/copyfrom', copyFromMiddlewares(RATE_LIMIT_ENDPOINTS_GROUPS.COPY_FROM));
apiRouter.get('/sql/copyto', copyToMiddlewares(RATE_LIMIT_ENDPOINTS_GROUPS.COPY_TO)); sqlRouter.get('/copyto', copyToMiddlewares(RATE_LIMIT_ENDPOINTS_GROUPS.COPY_TO));
apiRouter.post('/sql/copyto', copyToMiddlewares(RATE_LIMIT_ENDPOINTS_GROUPS.COPY_TO)); sqlRouter.post('/copyto', copyToMiddlewares(RATE_LIMIT_ENDPOINTS_GROUPS.COPY_TO));
} }
}; };

View File

@ -2,16 +2,16 @@
const util = require('util'); const util = require('util');
const bodyParserMiddleware = require('../middlewares/body-parser'); const bodyParser = require('../middlewares/body-parser');
const userMiddleware = require('../middlewares/user'); const user = require('../middlewares/user');
const { initializeProfilerMiddleware, finishProfilerMiddleware } = require('../middlewares/profiler'); const { initializeProfiler, finishProfiler } = require('../middlewares/profiler');
const authorizationMiddleware = require('../middlewares/authorization'); const authorization = require('../middlewares/authorization');
const connectionParamsMiddleware = require('../middlewares/connection-params'); const connectionParams = require('../middlewares/connection-params');
const errorMiddleware = require('../middlewares/error'); const error = require('../middlewares/error');
const rateLimitsMiddleware = require('../middlewares/rate-limit'); const rateLimits = require('../middlewares/rate-limit');
const { RATE_LIMIT_ENDPOINTS_GROUPS } = rateLimitsMiddleware; const { RATE_LIMIT_ENDPOINTS_GROUPS } = rateLimits;
const params = require('../middlewares/params'); const params = require('../middlewares/params');
const logMiddleware = require('../middlewares/log'); const log = require('../middlewares/log');
module.exports = class JobController { module.exports = class JobController {
constructor (metadataBackend, userDatabaseService, jobService, statsdClient, userLimitsService) { constructor (metadataBackend, userDatabaseService, jobService, statsdClient, userLimitsService) {
@ -22,7 +22,7 @@ module.exports = class JobController {
this.userLimitsService = userLimitsService; this.userLimitsService = userLimitsService;
} }
route (apiRouter) { route (sqlRouter) {
const jobMiddlewares = composeJobMiddlewares( const jobMiddlewares = composeJobMiddlewares(
this.metadataBackend, this.metadataBackend,
this.userDatabaseService, this.userDatabaseService,
@ -31,55 +31,44 @@ module.exports = class JobController {
this.userLimitsService this.userLimitsService
); );
apiRouter.get( sqlRouter.post('/job', [
'/jobs-wip', bodyParser(),
bodyParserMiddleware(),
listWorkInProgressJobs(this.jobService),
sendResponse(),
errorMiddleware()
);
apiRouter.post(
'/sql/job',
bodyParserMiddleware(),
checkBodyPayloadSize(), checkBodyPayloadSize(),
params({ strategy: 'job' }), params({ strategy: 'job' }),
logMiddleware(logMiddleware.TYPES.JOB), log(log.TYPES.JOB),
jobMiddlewares('create', createJob, RATE_LIMIT_ENDPOINTS_GROUPS.JOB_CREATE) jobMiddlewares('create', createJob, RATE_LIMIT_ENDPOINTS_GROUPS.JOB_CREATE)
); ]);
apiRouter.get( sqlRouter.get('/job/:job_id', [
'/sql/job/:job_id', bodyParser(),
bodyParserMiddleware(),
jobMiddlewares('retrieve', getJob, RATE_LIMIT_ENDPOINTS_GROUPS.JOB_GET) jobMiddlewares('retrieve', getJob, RATE_LIMIT_ENDPOINTS_GROUPS.JOB_GET)
); ]);
apiRouter.delete( sqlRouter.delete('/job/:job_id', [
'/sql/job/:job_id', bodyParser(),
bodyParserMiddleware(),
jobMiddlewares('cancel', cancelJob, RATE_LIMIT_ENDPOINTS_GROUPS.JOB_DELETE) jobMiddlewares('cancel', cancelJob, RATE_LIMIT_ENDPOINTS_GROUPS.JOB_DELETE)
); ]);
} }
}; };
function composeJobMiddlewares (metadataBackend, userDatabaseService, jobService, statsdClient, userLimitsService) { function composeJobMiddlewares (metadataBackend, userDatabaseService, jobService, statsdClient, userLimitsService) {
return function jobMiddlewares (action, jobMiddleware, endpointGroup) { return function jobMiddlewares (action, job, endpointGroup) {
const forceToBeMaster = true; const forceToBeMaster = true;
return [ return [
initializeProfilerMiddleware('job'), initializeProfiler('job'),
userMiddleware(metadataBackend), user(metadataBackend),
rateLimitsMiddleware(userLimitsService, endpointGroup), rateLimits(userLimitsService, endpointGroup),
authorizationMiddleware(metadataBackend, forceToBeMaster), authorization(metadataBackend, forceToBeMaster),
connectionParamsMiddleware(userDatabaseService), connectionParams(userDatabaseService),
jobMiddleware(jobService), job(jobService),
setServedByDBHostHeader(), setServedByDBHostHeader(),
finishProfilerMiddleware(), finishProfiler(),
logJobResult(action), logJobResult(action),
incrementSuccessMetrics(statsdClient), incrementSuccessMetrics(statsdClient),
sendResponse(), sendResponse(),
incrementErrorMetrics(statsdClient), incrementErrorMetrics(statsdClient),
errorMiddleware() error()
]; ];
}; };
} }
@ -155,21 +144,6 @@ function createJob (jobService) {
}; };
} }
function listWorkInProgressJobs (jobService) {
return function listWorkInProgressJobsMiddleware (req, res, next) {
jobService.listWorkInProgressJobs((err, list) => {
if (err) {
return next(err);
}
res.body = list;
next();
});
};
}
function checkBodyPayloadSize () { function checkBodyPayloadSize () {
return function checkBodyPayloadSizeMiddleware(req, res, next) { return function checkBodyPayloadSizeMiddleware(req, res, next) {
const payload = JSON.stringify(req.body); const payload = JSON.stringify(req.body);

View File

@ -1,7 +1,7 @@
'use strict'; 'use strict';
const bodyParser = require('../middlewares/body-parser'); const bodyParser = require('../middlewares/body-parser');
const { initializeProfilerMiddleware: initializeProfiler } = require('../middlewares/profiler'); const { initializeProfiler } = require('../middlewares/profiler');
const user = require('../middlewares/user'); const user = require('../middlewares/user');
const rateLimits = require('../middlewares/rate-limit'); const rateLimits = require('../middlewares/rate-limit');
const authorization = require('../middlewares/authorization'); const authorization = require('../middlewares/authorization');
@ -32,7 +32,7 @@ module.exports = class QueryController {
this.userLimitsService = userLimitsService; this.userLimitsService = userLimitsService;
} }
route (apiRouter) { route (sqlRouter) {
const forceToBeMaster = false; const forceToBeMaster = false;
const queryMiddlewares = () => { const queryMiddlewares = () => {
@ -61,8 +61,8 @@ module.exports = class QueryController {
]; ];
}; };
apiRouter.all('/sql', queryMiddlewares()); sqlRouter.all('/', queryMiddlewares());
apiRouter.all('/sql.:f', queryMiddlewares()); sqlRouter.all('.:f', queryMiddlewares());
} }
}; };

73
lib/api/sql/sql-router.js Normal file
View File

@ -0,0 +1,73 @@
'use strict';
const { Router: router } = require('express');
const UserDatabaseService = require('../../services/user-database-service');
const UserLimitsService = require('../../services/user-limits');
const socketTimeout = require('../middlewares/socket-timeout');
const logger = require('../middlewares/logger');
const profiler = require('../middlewares/profiler');
const cors = require('../middlewares/cors');
const servedByHostHeader = require('../middlewares/served-by-host-header');
const QueryController = require('./query-controller');
const CopyController = require('./copy-controller');
const JobController = require('./job-controller');
module.exports = class SqlRouter {
constructor ({ metadataBackend, statsClient, dataIngestionLogger, jobService }) {
const userLimitsServiceOptions = {
limits: {
rateLimitsEnabled: global.settings.ratelimits.rateLimitsEnabled
}
};
const userDatabaseService = new UserDatabaseService(metadataBackend);
const userLimitsService = new UserLimitsService(metadataBackend, userLimitsServiceOptions);
this.queryController = new QueryController(
metadataBackend,
userDatabaseService,
statsClient,
userLimitsService
);
this.copyController = new CopyController(
metadataBackend,
userDatabaseService,
userLimitsService,
dataIngestionLogger
);
this.jobController = new JobController(
metadataBackend,
userDatabaseService,
jobService,
statsClient,
userLimitsService
);
}
route (apiRouter, routes) {
routes.forEach(route => {
const sqlRouter = router({ mergeParams: true });
const paths = route.paths;
const middlewares = route.middlewares || [];
middlewares.forEach(middleware => sqlRouter.use(middleware()));
sqlRouter.use(socketTimeout());
sqlRouter.use(logger());
sqlRouter.use(profiler({ statsClient: this.statsClient }));
sqlRouter.use(cors());
sqlRouter.use(servedByHostHeader());
this.queryController.route(sqlRouter);
this.copyController.route(sqlRouter);
this.jobController.route(sqlRouter);
paths.forEach(path => apiRouter.use(path, sqlRouter));
});
}
};

View File

@ -5,8 +5,8 @@ const versions = {
}; };
module.exports = class VersionController { module.exports = class VersionController {
route (app) { route (apiRouter) {
app.get(`${global.settings.base_url}/version`, version()); apiRouter.get('/version', version());
} }
}; };

View File

@ -4,7 +4,7 @@
var _ = require('underscore'); var _ = require('underscore');
var OAuthUtil = require('oauth-client'); var OAuthUtil = require('oauth-client');
var step = require('step'); var step = require('step');
var CdbRequest = require('../models/cartodb_request'); var CdbRequest = require('../models/cartodb-request');
var cdbReq = new CdbRequest(); var cdbReq = new CdbRequest();
var oAuth = (function(){ var oAuth = (function(){

View File

@ -1,6 +1,6 @@
'use strict'; 'use strict';
const Logger = require('../app/services/logger'); const Logger = require('../services/logger');
class BatchLogger extends Logger { class BatchLogger extends Logger {
constructor (path, name) { constructor (path, name) {
@ -10,7 +10,6 @@ class BatchLogger extends Logger {
log (job) { log (job) {
return job.log(this.logger); return job.log(this.logger);
} }
} }
module.exports = BatchLogger; module.exports = BatchLogger;

View File

@ -1,14 +1,14 @@
'use strict'; 'use strict';
var JobRunner = require('./job_runner'); var JobRunner = require('./job-runner');
var QueryRunner = require('./query_runner'); var QueryRunner = require('./query-runner');
var JobCanceller = require('./job_canceller'); var JobCanceller = require('./job-canceller');
var JobSubscriber = require('./pubsub/job-subscriber'); var JobSubscriber = require('./pubsub/job-subscriber');
var UserDatabaseMetadataService = require('./user_database_metadata_service'); var UserDatabaseMetadataService = require('./user-database-metadata-service');
var JobPublisher = require('./pubsub/job-publisher'); var JobPublisher = require('./pubsub/job-publisher');
var JobQueue = require('./job_queue'); var JobQueue = require('./job-queue');
var JobBackend = require('./job_backend'); var JobBackend = require('./job-backend');
var JobService = require('./job_service'); var JobService = require('./job-service');
var BatchLogger = require('./batch-logger'); var BatchLogger = require('./batch-logger');
var Batch = require('./batch'); var Batch = require('./batch');

View File

@ -2,7 +2,7 @@
var REDIS_PREFIX = 'batch:jobs:'; var REDIS_PREFIX = 'batch:jobs:';
var REDIS_DB = 5; var REDIS_DB = 5;
var JobStatus = require('./job_status'); var JobStatus = require('./job-status');
var queue = require('queue-async'); var queue = require('queue-async');
function JobBackend(metadataBackend, jobQueue, logger) { function JobBackend(metadataBackend, jobQueue, logger) {

View File

@ -1,7 +1,7 @@
'use strict'; 'use strict';
var errorCodes = require('../app/postgresql/error_codes').codeToCondition; var errorCodes = require('../postgresql/error-codes').codeToCondition;
var jobStatus = require('./job_status'); var jobStatus = require('./job-status');
var Profiler = require('step-profiler'); var Profiler = require('step-profiler');
var _ = require('underscore'); var _ = require('underscore');

View File

@ -1,7 +1,7 @@
'use strict'; 'use strict';
var JobFactory = require('./models/job_factory'); var JobFactory = require('./models/job-factory');
var jobStatus = require('./job_status'); var jobStatus = require('./job-status');
function JobService(jobBackend, jobCanceller, logger) { function JobService(jobBackend, jobCanceller, logger) {
this.jobBackend = jobBackend; this.jobBackend = jobBackend;

View File

@ -2,8 +2,8 @@
var util = require('util'); var util = require('util');
var uuid = require('node-uuid'); var uuid = require('node-uuid');
var JobStateMachine = require('./job_state_machine'); var JobStateMachine = require('./job-state-machine');
var jobStatus = require('../job_status'); var jobStatus = require('../job-status');
var mandatoryProperties = [ var mandatoryProperties = [
'job_id', 'job_id',
'status', 'status',

View File

@ -1,8 +1,8 @@
'use strict'; 'use strict';
var JobSimple = require('./job_simple'); var JobSimple = require('./job-simple');
var JobMultiple = require('./job_multiple'); var JobMultiple = require('./job-multiple');
var JobFallback = require('./job_fallback'); var JobFallback = require('./job-fallback');
var Models = [ JobSimple, JobMultiple, JobFallback ]; var Models = [ JobSimple, JobMultiple, JobFallback ];

View File

@ -1,11 +1,11 @@
'use strict'; 'use strict';
var util = require('util'); var util = require('util');
var JobBase = require('./job_base'); var JobBase = require('./job-base');
var JobStatus = require('../job_status'); var JobStatus = require('../job-status');
var QueryFallback = require('./query/query_fallback'); var QueryFallback = require('./query/query-fallback');
var MainFallback = require('./query/main_fallback'); var MainFallback = require('./query/main-fallback');
var QueryFactory = require('./query/query_factory'); var QueryFactory = require('./query/query-factory');
function JobFallback(jobDefinition) { function JobFallback(jobDefinition) {
JobBase.call(this, jobDefinition); JobBase.call(this, jobDefinition);

View File

@ -1,8 +1,8 @@
'use strict'; 'use strict';
var util = require('util'); var util = require('util');
var JobBase = require('./job_base'); var JobBase = require('./job-base');
var jobStatus = require('../job_status'); var jobStatus = require('../job-status');
function JobMultiple(jobDefinition) { function JobMultiple(jobDefinition) {
JobBase.call(this, jobDefinition); JobBase.call(this, jobDefinition);

View File

@ -1,8 +1,8 @@
'use strict'; 'use strict';
var util = require('util'); var util = require('util');
var JobBase = require('./job_base'); var JobBase = require('./job-base');
var jobStatus = require('../job_status'); var jobStatus = require('../job-status');
function JobSimple(jobDefinition) { function JobSimple(jobDefinition) {
JobBase.call(this, jobDefinition); JobBase.call(this, jobDefinition);

View File

@ -1,7 +1,7 @@
'use strict'; 'use strict';
var assert = require('assert'); var assert = require('assert');
var JobStatus = require('../job_status'); var JobStatus = require('../job-status');
var validStatusTransitions = [ var validStatusTransitions = [
[JobStatus.PENDING, JobStatus.RUNNING], [JobStatus.PENDING, JobStatus.RUNNING],
[JobStatus.PENDING, JobStatus.CANCELLED], [JobStatus.PENDING, JobStatus.CANCELLED],

View File

@ -1,8 +1,8 @@
'use strict'; 'use strict';
var util = require('util'); var util = require('util');
var QueryBase = require('./query_base'); var QueryBase = require('./query-base');
var jobStatus = require('../../job_status'); var jobStatus = require('../../job-status');
function Fallback(index) { function Fallback(index) {
QueryBase.call(this, index); QueryBase.call(this, index);

View File

@ -1,8 +1,8 @@
'use strict'; 'use strict';
var util = require('util'); var util = require('util');
var QueryBase = require('./query_base'); var QueryBase = require('./query-base');
var jobStatus = require('../../job_status'); var jobStatus = require('../../job-status');
function MainFallback() { function MainFallback() {
QueryBase.call(this); QueryBase.call(this);

View File

@ -1,7 +1,7 @@
'use strict'; 'use strict';
var util = require('util'); var util = require('util');
var JobStateMachine = require('../job_state_machine'); var JobStateMachine = require('../job-state-machine');
function QueryBase(index) { function QueryBase(index) {
JobStateMachine.call(this); JobStateMachine.call(this);

View File

@ -1,6 +1,6 @@
'use strict'; 'use strict';
var QueryFallback = require('./query_fallback'); var QueryFallback = require('./query-fallback');
function QueryFactory() { function QueryFactory() {
} }

View File

@ -1,10 +1,10 @@
'use strict'; 'use strict';
var util = require('util'); var util = require('util');
var QueryBase = require('./query_base'); var QueryBase = require('./query-base');
var Query = require('./query'); var Query = require('./query');
var Fallback = require('./fallback'); var Fallback = require('./fallback');
var jobStatus = require('../../job_status'); var jobStatus = require('../../job-status');
function QueryFallback(job, index) { function QueryFallback(job, index) {
QueryBase.call(this, index); QueryBase.call(this, index);

View File

@ -1,8 +1,8 @@
'use strict'; 'use strict';
var util = require('util'); var util = require('util');
var QueryBase = require('./query_base'); var QueryBase = require('./query-base');
var jobStatus = require('../../job_status'); var jobStatus = require('../../job-status');
function Query(index) { function Query(index) {
QueryBase.call(this, index); QueryBase.call(this, index);

View File

@ -3,7 +3,7 @@
var _ = require('underscore'); var _ = require('underscore');
var Pg = require('./../pg'); var Pg = require('./../pg');
var ArrayBufferSer = require("../../bin_encoder"); var ArrayBufferSer = require("../../bin-encoder");
function BinaryFormat() {} function BinaryFormat() {}

View File

@ -2,8 +2,8 @@
var _ = require('underscore'); var _ = require('underscore');
var Pg = require('./../pg'); var Pg = require('./../pg');
const errorHandlerFactory = require('../../../services/error_handler_factory'); const errorHandlerFactory = require('../../../services/error-handler-factory');
function GeoJsonFormat() { function GeoJsonFormat() {
this.buffer = ''; this.buffer = '';

View File

@ -3,7 +3,7 @@
var _ = require('underscore'); var _ = require('underscore');
var Pg = require('./../pg'); var Pg = require('./../pg');
const errorHandlerFactory = require('../../../services/error_handler_factory'); const errorHandlerFactory = require('../../../services/error-handler-factory');
function JsonFormat() { function JsonFormat() {
this.buffer = ''; this.buffer = '';

View File

@ -1,5 +1,7 @@
'use strict'; 'use strict';
// jshint ignore:start
var Pg = require('./../pg'); var Pg = require('./../pg');
var _ = require('underscore'); var _ = require('underscore');
var geojson = require('./geojson'); var geojson = require('./geojson');
@ -133,6 +135,6 @@ TopoJsonFormat.prototype.cancel = function() {
} }
}; };
module.exports = TopoJsonFormat; module.exports = TopoJsonFormat;
// jshint ignore:end

32
lib/server-options.js Normal file
View File

@ -0,0 +1,32 @@
'use strict';
module.exports = function getServerOptions () {
const defaults = {
routes: {
// Each entry corresponds with an express' router.
// You must define at least one path. However, middlewares are optional.
api: [{
// Required: path where other "routers" or "controllers" will be attached to.
paths: [
// In case the path has a :user param the username will be the one specified in the URL,
// otherwise it will fallback to extract the username from the host header.
'/api/:version',
'/user/:user/api/:version',
],
// Optional: attach middlewares at the begining of the router
// to perform custom operations.
middlewares: [],
sql: [{
// Required
paths: [
'/sql'
],
// Optional
middlewares: []
}]
}]
}
};
return Object.assign({}, defaults, global.settings);
};

Some files were not shown because too many files have changed in this diff Show More