joe/kami-parse-server
Public Access
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Adds jobs endpoint protected by masterKey (#2560)

Browse Source 
* Adds jobs endpoint protected by masterKey

* Adds connection timeout for 15 minutes in jobs

* Refactors pushStatusHandler into StatusHandler

* Adds reporting of _JobStatus

* Only accept strings as messages

* Adds test for masterKey basic auth

* Adds CloudCodeRouter for cloud_code endpoint of job status, enable Jobs feature on dashboard

* xit racing test
This commit is contained in:
Florent Vilmart
2016-08-30 07:19:21 -04:00
committed by GitHub
parent 4b2a780e03
commit 10ace495d8
12 changed files with 414 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files

166
spec/CloudCode.spec.js
View File

@@ -3,6 +3,7 @@ const Parse = require("parse/node");
const request = require('request');
const rp = require('request-promise');
const InMemoryCacheAdapter = require('../src/Adapters/Cache/InMemoryCacheAdapter').InMemoryCacheAdapter;
const triggers = require('../src/triggers');
describe('Cloud Code', () => {
it('can load absolute cloud code file', done => {
@@ -211,7 +212,8 @@ describe('Cloud Code', () => {
});
});
it('test afterSave ignoring promise, object not found', function(done) {
// TODO: Fails on CI randomly as racing
xit('test afterSave ignoring promise, object not found', function(done) {
Parse.Cloud.afterSave('AfterSaveTest2', function(req) {
let obj = req.object;
if(!obj.existed())
@@ -1005,4 +1007,166 @@ it('beforeSave should not affect fetched pointers', done => {
done();
})
});
describe('cloud jobs', () => {
it('should define a job', (done) => {
expect(() => {
Parse.Cloud.job('myJob', (req, res) => {
res.success();
});
}).not.toThrow();
rp.post({
url: 'http://localhost:8378/1/jobs/myJob',
headers: {
'X-Parse-Application-Id': Parse.applicationId,
'X-Parse-Master-Key': Parse.masterKey,
},
}).then((result) => {
done();
}, (err) =>  {
fail(err);
done();
});
});
it('should not run without master key', (done) => {
expect(() => {
Parse.Cloud.job('myJob', (req, res) => {
res.success();
});
}).not.toThrow();
rp.post({
url: 'http://localhost:8378/1/jobs/myJob',
headers: {
'X-Parse-Application-Id': Parse.applicationId,
'X-Parse-REST-API-Key': 'rest',
},
}).then((result) => {
fail('Expected to be unauthorized');
done();
}, (err) =>  {
expect(err.statusCode).toBe(403);
done();
});
});
it('should run with master key', (done) => {
expect(() => {
Parse.Cloud.job('myJob', (req, res) => {
expect(req.functionName).toBeUndefined();
expect(req.jobName).toBe('myJob');
expect(typeof req.jobId).toBe('string');
expect(typeof res.success).toBe('function');
expect(typeof res.error).toBe('function');
expect(typeof res.message).toBe('function');
res.success();
done();
});
}).not.toThrow();
rp.post({
url: 'http://localhost:8378/1/jobs/myJob',
headers: {
'X-Parse-Application-Id': Parse.applicationId,
'X-Parse-Master-Key': Parse.masterKey,
},
}).then((response) => {
}, (err) =>  {
fail(err);
done();
});
});
it('should run with master key basic auth', (done) => {
expect(() => {
Parse.Cloud.job('myJob', (req, res) => {
expect(req.functionName).toBeUndefined();
expect(req.jobName).toBe('myJob');
expect(typeof req.jobId).toBe('string');
expect(typeof res.success).toBe('function');
expect(typeof res.error).toBe('function');
expect(typeof res.message).toBe('function');
res.success();
done();
});
}).not.toThrow();
rp.post({
url: `http://${Parse.applicationId}:${Parse.masterKey}@localhost:8378/1/jobs/myJob`,
}).then((response) => {
}, (err) =>  {
fail(err);
done();
});
});
it('should set the message / success on the job', (done) => {
Parse.Cloud.job('myJob', (req, res) => {
res.message('hello');
res.message().then(() => {
return getJobStatus(req.jobId);
}).then((jobStatus) => {
expect(jobStatus.get('message')).toEqual('hello');
expect(jobStatus.get('status')).toEqual('running');
return res.success().then(() => {
return getJobStatus(req.jobId);
});
}).then((jobStatus) => {
expect(jobStatus.get('message')).toEqual('hello');
expect(jobStatus.get('status')).toEqual('succeeded');
done();
}).catch(err => {
console.error(err);
jfail(err);
done();
});
});
rp.post({
url: 'http://localhost:8378/1/jobs/myJob',
headers: {
'X-Parse-Application-Id': Parse.applicationId,
'X-Parse-Master-Key': Parse.masterKey,
},
}).then((response) => {
}, (err) =>  {
fail(err);
done();
});
});
it('should set the failure on the job', (done) => {
Parse.Cloud.job('myJob', (req, res) => {
res.error('Something went wrong').then(() => {
return getJobStatus(req.jobId);
}).then((jobStatus) => {
expect(jobStatus.get('message')).toEqual('Something went wrong');
expect(jobStatus.get('status')).toEqual('failed');
done();
}).catch(err => {
jfail(err);
done();
});
});
rp.post({
url: 'http://localhost:8378/1/jobs/myJob',
headers: {
'X-Parse-Application-Id': Parse.applicationId,
'X-Parse-Master-Key': Parse.masterKey,
},
}).then((response) => {
}, (err) =>  {
fail(err);
done();
});
});
function getJobStatus(jobId) {
let q = new Parse.Query('_JobStatus');
return q.get(jobId, {useMasterKey: true});
}
});
});

4
spec/PushController.spec.js
View File

@@ -1,6 +1,6 @@
"use strict";
var PushController = require('../src/Controllers/PushController').PushController;
var pushStatusHandler = require('../src/pushStatusHandler');
var StatusHandler = require('../src/StatusHandler');
var Config = require('../src/Config');
const successfulTransmissions = function(body, installations) {
@@ -439,7 +439,7 @@ describe('PushController', () => {
});
it('should flatten', () => {
var res = pushStatusHandler.flatten([1, [2], [[3, 4], 5], [[[6]]]])
var res = StatusHandler.flatten([1, [2], [[3, 4], 5], [[[6]]]])
expect(res).toEqual([1,2,3,4,5,6]);
})
});

12
src/Adapters/Storage/Postgres/PostgresStorageAdapter.js
View File

@@ -534,7 +534,7 @@ export class PostgresStorageAdapter {
let joins = results.reduce((list, schema) => {
return list.concat(joinTablesForSchema(schema.schema));
}, []);
const classes = ['_SCHEMA','_PushStatus','_Hooks','_GlobalConfig', ...results.map(result => result.className), ...joins];
const classes = ['_SCHEMA','_PushStatus','_JobStatus','_Hooks','_GlobalConfig', ...results.map(result => result.className), ...joins];
return this._client.tx(t=>t.batch(classes.map(className=>t.none('DROP TABLE IF EXISTS $<className:name>', { className }))));
}, error => {
if (error.code === PostgresRelationDoesNotExistError) {
@@ -783,7 +783,11 @@ export class PostgresStorageAdapter {
for (let fieldName in update) {
let fieldValue = update[fieldName];
if (fieldName == 'authData') {
if (fieldValue === null) {
updatePatterns.push(`$${index}:name = NULL`);
values.push(fieldName);
index += 1;
} else if (fieldName == 'authData') {
// This recursively sets the json_object
// Only 1 level deep
let generate = (jsonb, key, value) => {
@@ -848,6 +852,10 @@ export class PostgresStorageAdapter {
updatePatterns.push(`$${index}:name = $${index + 1}`);
values.push(fieldName, toPostgresValue(fieldValue));
index += 2;
} else if (fieldValue instanceof Date) {
updatePatterns.push(`$${index}:name = $${index + 1}`);
values.push(fieldName, fieldValue);
index += 2;
} else if (fieldValue.__type === 'File') {
updatePatterns.push(`$${index}:name = $${index + 1}`);
values.push(fieldName, toPostgresValue(fieldValue));

25
src/Controllers/PushController.js
View File

@@ -1,13 +1,13 @@
import { Parse } from 'parse/node';
import PromiseRouter from '../PromiseRouter';
import rest from '../rest';
import AdaptableController from './AdaptableController';
import { PushAdapter } from '../Adapters/Push/PushAdapter';
import deepcopy from 'deepcopy';
import RestQuery from '../RestQuery';
import RestWrite from '../RestWrite';
import { master } from '../Auth';
import pushStatusHandler from '../pushStatusHandler';
import { Parse } from 'parse/node';
import PromiseRouter from '../PromiseRouter';
import rest from '../rest';
import AdaptableController from './AdaptableController';
import { PushAdapter } from '../Adapters/Push/PushAdapter';
import deepcopy from 'deepcopy';
import RestQuery from '../RestQuery';
import RestWrite from '../RestWrite';
import { master } from '../Auth';
import { pushStatusHandler } from '../StatusHandler';
const FEATURE_NAME = 'push';
const UNSUPPORTED_BADGE_KEY = "unsupported";
@@ -98,8 +98,9 @@ export class PushController extends AdaptableController {
}).then((results) => {
return pushStatus.complete(results);
}).catch((err) => {
pushStatus.fail(err);
return Promise.reject(err);
return pushStatus.fail(err).then(() => {
throw err;
});
});
}

19
src/Controllers/SchemaController.js
View File

@@ -87,6 +87,14 @@ const defaultColumns = Object.freeze({
"sentPerType": {type:'Object'},
"failedPerType":{type:'Object'},
},
_JobStatus: {
"jobName": {type: 'String'},
"source": {type: 'String'},
"status": {type: 'String'},
"message": {type: 'String'},
"params": {type: 'Object'}, // params received when calling the job
"finishedAt": {type: 'Date'}
},
_Hooks: {
"functionName": {type:'String'},
"className": {type:'String'},
@@ -104,9 +112,9 @@ const requiredColumns = Object.freeze({
_Role: ["name", "ACL"]
});
const systemClasses = Object.freeze(['_User', '_Installation', '_Role', '_Session', '_Product', '_PushStatus']);
const systemClasses = Object.freeze(['_User', '_Installation', '_Role', '_Session', '_Product', '_PushStatus', '_JobStatus']);
const volatileClasses = Object.freeze(['_PushStatus', '_Hooks', '_GlobalConfig']);
const volatileClasses = Object.freeze(['_JobStatus', '_PushStatus', '_Hooks', '_GlobalConfig']);
// 10 alpha numberic chars + uppercase
const userIdRegex = /^[a-zA-Z0-9]{10}$/;
@@ -275,7 +283,12 @@ const _PushStatusSchema = convertSchemaToAdapterSchema(injectDefaultSchema({
fields: {},
classLevelPermissions: {}
}));
const VolatileClassesSchemas = [_HooksSchema, _PushStatusSchema, _GlobalConfigSchema];
const _JobStatusSchema = convertSchemaToAdapterSchema(injectDefaultSchema({
className: "_JobStatus",
fields: {},
classLevelPermissions: {}
}));
const VolatileClassesSchemas = [_HooksSchema, _JobStatusSchema, _PushStatusSchema, _GlobalConfigSchema];
const dbTypeMatchesObjectType = (dbType, objectType) => {
if (dbType.type !== objectType.type) return false;

4
src/ParseServer.js
View File

@@ -45,6 +45,7 @@ import { ParseLiveQueryServer } from './LiveQuery/ParseLiveQueryServer';
import { PublicAPIRouter } from './Routers/PublicAPIRouter';
import { PushController } from './Controllers/PushController';
import { PushRouter } from './Routers/PushRouter';
import { CloudCodeRouter } from './Routers/CloudCodeRouter';
import { randomString } from './cryptoUtils';
import { RolesRouter } from './Routers/RolesRouter';
import { SchemasRouter } from './Routers/SchemasRouter';
@@ -285,7 +286,8 @@ class ParseServer {
new FeaturesRouter(),
new GlobalConfigRouter(),
new PurgeRouter(),
new HooksRouter()
new HooksRouter(),
new CloudCodeRouter()
];
let routes = routers.reduce((memo, router) => {

20
src/Routers/CloudCodeRouter.js Normal file
View File

@@ -0,0 +1,20 @@
import PromiseRouter from '../PromiseRouter';
const triggers = require('../triggers');
export class CloudCodeRouter extends PromiseRouter {
mountRoutes() {
this.route('GET',`/cloud_code/jobs`, CloudCodeRouter.getJobs);
}
static getJobs(req) {
let config = req.config;
let jobs = triggers.getJobs(config.applicationId) || {};
return Promise.resolve({
response: Object.keys(jobs).map((jobName) => {
return {
jobName,
}
})
});
}
}

3
src/Routers/FeaturesRouter.js
View File

@@ -18,6 +18,9 @@ export class FeaturesRouter extends PromiseRouter {
update: false,
delete: false,
},
cloudCode: {
jobs: true,
},
logs: {
level: true,
size: true,

68
src/Routers/FunctionsRouter.js
View File

@@ -5,6 +5,8 @@ var express = require('express'),
triggers = require('../triggers');
import PromiseRouter from '../PromiseRouter';
import { promiseEnforceMasterKeyAccess } from '../middlewares';
import { jobStatusHandler } from '../StatusHandler';
import _ from 'lodash';
import { logger } from '../logger';
@@ -32,9 +34,51 @@ export class FunctionsRouter extends PromiseRouter {
mountRoutes() {
this.route('POST', '/functions/:functionName', FunctionsRouter.handleCloudFunction);
this.route('POST', '/jobs/:jobName', promiseEnforceMasterKeyAccess, function(req) {
return FunctionsRouter.handleCloudJob(req);
});
this.route('POST', '/jobs', promiseEnforceMasterKeyAccess, function(req) {
return FunctionsRouter.handleCloudJob(req);
});
}
static createResponseObject(resolve, reject) {
static handleCloudJob(req) {
const jobName = req.params.jobName || req.body.jobName;
const applicationId = req.config.applicationId;
const jobHandler = jobStatusHandler(req.config);
const jobFunction = triggers.getJob(jobName, applicationId);
if (!jobFunction) {
throw new Parse.Error(Parse.Error.SCRIPT_FAILED, 'Invalid job.');
}
let params = Object.assign({}, req.body, req.query);
params = parseParams(params);
const request = {
params: params,
log: req.config.loggerController,
headers: req.headers,
jobName
};
const status = {
success: jobHandler.setSucceeded.bind(jobHandler),
error: jobHandler.setFailed.bind(jobHandler),
message: jobHandler.setMessage.bind(jobHandler)
}
return jobHandler.setRunning(jobName, params).then((jobStatus) => {
request.jobId = jobStatus.objectId
// run the function async
process.nextTick(() => {
jobFunction(request, status);
});
return {
headers: {
'X-Parse-Job-Status-Id': jobStatus.objectId
},
response: {}
}
});
}
static createResponseObject(resolve, reject, message) {
return {
success: function(result) {
resolve({
@@ -49,15 +93,17 @@ export class FunctionsRouter extends PromiseRouter {
code = Parse.Error.SCRIPT_FAILED;
}
reject(new Parse.Error(code, message));
}
},
message: message
}
}
static handleCloudFunction(req) {
var applicationId = req.config.applicationId;
var theFunction = triggers.getFunction(req.params.functionName, applicationId);
var theValidator = triggers.getValidator(req.params.functionName, applicationId);
if (theFunction) {
const functionName = req.params.functionName;
const applicationId = req.config.applicationId;
const theFunction = triggers.getFunction(functionName, applicationId);
const theValidator = triggers.getValidator(req.params.functionName, applicationId);
if (theFunction) {
let params = Object.assign({}, req.body, req.query);
params = parseParams(params);
var request = {
@@ -67,7 +113,7 @@ export class FunctionsRouter extends PromiseRouter {
installationId: req.info.installationId,
log: req.config.loggerController,
headers: req.headers,
functionName: req.params.functionName
functionName
};
if (theValidator && typeof theValidator === "function") {
@@ -83,9 +129,9 @@ export class FunctionsRouter extends PromiseRouter {
var response = FunctionsRouter.createResponseObject((result) => {
try {
const cleanResult = logger.truncateLogMessage(JSON.stringify(result.response.result));
logger.info(`Ran cloud function ${req.params.functionName} for user ${userString} `
logger.info(`Ran cloud function ${functionName} for user ${userString} `
+ `with:\n Input: ${cleanInput }\n Result: ${cleanResult }`, {
functionName: req.params.functionName,
functionName,
params,
user: userString,
});
@@ -95,10 +141,10 @@ export class FunctionsRouter extends PromiseRouter {
}
}, (error) => {
try {
logger.error(`Failed running cloud function ${req.params.functionName} for `
logger.error(`Failed running cloud function ${functionName} for `
+ `user ${userString} with:\n Input: ${cleanInput}\n Error: `
+ JSON.stringify(error), {
functionName: req.params.functionName,
functionName,
error,
params,
user: userString

115
src/pushStatusHandler.js → src/StatusHandler.js
View File

@@ -2,6 +2,7 @@ import { md5Hash, newObjectId } from './cryptoUtils';
import { logger } from './logger';
const PUSH_STATUS_COLLECTION = '_PushStatus';
const JOB_STATUS_COLLECTION = '_JobStatus';
export function flatten(array) {
return array.reduce((memo, element) => {
@@ -14,13 +15,91 @@ export function flatten(array) {
}, []);
}
export default function pushStatusHandler(config) {
function statusHandler(className, database) {
let lastPromise = Promise.resolve();
function create(object) {
lastPromise = lastPromise.then(() => {
return database.create(className, object).then(() => {
return Promise.resolve(object);
});
});
return lastPromise;
}
function update(where, object) {
lastPromise = lastPromise.then(() => {
return database.update(className, where, object);
});
return lastPromise;
}
return Object.freeze({
create,
update
})
}
export function jobStatusHandler(config) {
let jobStatus;
let objectId = newObjectId();
let database = config.database;
let lastPromise = Promise.resolve();
let handler = statusHandler(JOB_STATUS_COLLECTION, database);
let setRunning = function(jobName, params) {
let now = new Date();
jobStatus = {
objectId,
jobName,
params,
status: 'running',
source: 'api',
createdAt: now,
// lockdown!
ACL: {}
}
return handler.create(jobStatus);
}
let setMessage = function(message) {
if (!message || typeof message !== 'string') {
return Promise.resolve();
}
return handler.update({ objectId }, { message });
}
let setSucceeded = function(message) {
return setFinalStatus('succeeded', message);
}
let setFailed = function(message) {
return setFinalStatus('failed', message);
}
let setFinalStatus = function(status, message = undefined) {
let finishedAt = new Date();
let update = { status, finishedAt };
if (message && typeof message === 'string') {
update.message = message;
}
return handler.update({ objectId }, update);
}
return Object.freeze({
setRunning,
setSucceeded,
setMessage,
setFailed
});
}
export function pushStatusHandler(config) {
let initialPromise;
let pushStatus;
let objectId = newObjectId();
let database = config.database;
let lastPromise;
let handler = statusHandler(PUSH_STATUS_COLLECTION, database);
let setInitial = function(body = {}, where, options = {source: 'rest'}) {
let now = new Date();
let data = body.data || {};
@@ -48,25 +127,19 @@ export default function pushStatusHandler(config) {
// lockdown!
ACL: {}
}
lastPromise = Promise.resolve().then(() => {
return database.create(PUSH_STATUS_COLLECTION, object).then(() => {
pushStatus = {
objectId
};
return Promise.resolve(pushStatus);
});
return handler.create(object).then(() => {
pushStatus = {
objectId
};
return Promise.resolve(pushStatus);
});
return lastPromise;
}
let setRunning = function(installations) {
logger.verbose('sending push to %d installations', installations.length);
lastPromise = lastPromise.then(() => {
return database.update(PUSH_STATUS_COLLECTION,
{status:"pending", objectId: objectId},
return handler.update({status:"pending", objectId: objectId},
{status: "running", updatedAt: new Date() });
});
return lastPromise;
}
let complete = function(results) {
@@ -100,10 +173,7 @@ export default function pushStatusHandler(config) {
}, update);
}
logger.verbose('sent push! %d success, %d failures', update.numSent, update.numFailed);
lastPromise = lastPromise.then(() => {
return database.update(PUSH_STATUS_COLLECTION, {status:"running", objectId }, update);
});
return lastPromise;
return handler.update({status:"running", objectId }, update);
}
let fail = function(err) {
@@ -113,10 +183,7 @@ export default function pushStatusHandler(config) {
updatedAt: new Date()
}
logger.info('warning: error while sending push', err);
lastPromise = lastPromise.then(() => {
return database.update(PUSH_STATUS_COLLECTION, { objectId }, update);
});
return lastPromise;
return handler.update({ objectId }, update);
}
return Object.freeze({

4
src/cloud-code/Parse.Cloud.js
View File

@@ -21,6 +21,10 @@ ParseCloud.define = function(functionName, handler, validationHandler) {
triggers.addFunction(functionName, handler, validationHandler, Parse.applicationId);
};
ParseCloud.job = function(functionName, handler) {
triggers.addJob(functionName, handler, Parse.applicationId);
};
ParseCloud.beforeSave = function(parseClass, handler) {
var className = getClassName(parseClass);
triggers.addTrigger(triggers.Types.beforeSave, className, handler, Parse.applicationId);

30
src/triggers.js
View File

@@ -13,6 +13,7 @@ export const Types = {
const baseStore = function() {
let Validators = {};
let Functions = {};
let Jobs = {};
let Triggers = Object.keys(Types).reduce(function(base, key){
base[key] = {};
return base;
@@ -20,6 +21,7 @@ const baseStore = function() {
return Object.freeze({
Functions,
Jobs,
Validators,
Triggers
});
@@ -34,6 +36,12 @@ export function addFunction(functionName, handler, validationHandler, applicatio
_triggerStore[applicationId].Validators[functionName] = validationHandler;
}
export function addJob(jobName, handler, applicationId) {
applicationId = applicationId || Parse.applicationId;
_triggerStore[applicationId] = _triggerStore[applicationId] || baseStore();
_triggerStore[applicationId].Jobs[jobName] = handler;
}
export function addTrigger(type, className, handler, applicationId) {
applicationId = applicationId || Parse.applicationId;
_triggerStore[applicationId] = _triggerStore[applicationId] || baseStore();
@@ -45,6 +53,11 @@ export function removeFunction(functionName, applicationId) {
delete _triggerStore[applicationId].Functions[functionName]
}
export function removeJob(jobName, applicationId) {
applicationId = applicationId || Parse.applicationId;
delete _triggerStore[applicationId].Jobs[jobName]
}
export function removeTrigger(type, className, applicationId) {
applicationId = applicationId || Parse.applicationId;
delete _triggerStore[applicationId].Triggers[type][className]
@@ -89,6 +102,23 @@ export function getFunction(functionName, applicationId) {
return undefined;
}
export function getJob(jobName, applicationId) {
var manager = _triggerStore[applicationId];
if (manager && manager.Jobs) {
return manager.Jobs[jobName];
};
return undefined;
}
export function getJobs(applicationId) {
var manager = _triggerStore[applicationId];
if (manager && manager.Jobs) {
return manager.Jobs;
};
return undefined;
}
export function getValidator(functionName, applicationId) {
var manager = _triggerStore[applicationId];
if (manager && manager.Validators) {