Implement provider redundancy (#100)
This change introduces redundancy in the cloudant feed provider, allowing multiple instances to be deployed simultaneously such that if any one instance stops working, the remaining instances can be activated to take over the work of the failed instance.
This is accomplished by changing the feed actions to write directly to the trigger DB, rather than talking directly to the feed provider through REST endpoints. In this way, provider instances become aware of new and deleted triggers by monitoring the DB changes feed.
diff --git a/actions/changes.js b/actions/changes.js
index e332a5b..1c30ba8 100644
--- a/actions/changes.js
+++ b/actions/changes.js
@@ -3,124 +3,69 @@
function main(msg) {
console.log("cloudant trigger feed: ", msg);
- // lifecycleEvent is one of [ create, delete ]
- var lifecycleEvent = (msg.lifecycleEvent || 'CREATE').trim().toUpperCase();
+ // for creation -> CREATE
+ // for deletion -> DELETE
+ var lifecycleEvent = msg.lifecycleEvent;
- var namespace = process.env.__OW_NAMESPACE;
- var triggerName = parseQName(msg.triggerName).name;
- var triggerId = ':' + namespace + ':' + triggerName;
+ var endpoint = msg.apihost;
+ var webparams = createWebParams(msg);
- // configuration parameters
- var provider_endpoint = msg.package_endpoint;
- var dbname = msg.dbname;
- var user = msg.username;
- var pass = msg.password;
- var host = msg.host;
- var protocol = msg.protocol || 'https';
- var port = msg.port;
- var maxTriggers = msg.maxTriggers;
- var filter;
- var query_params;
+ var url = `https://${endpoint}/api/v1/web/whisk.system/cloudantWeb/changesWebAction.http`;
- if (lifecycleEvent === 'CREATE') {
-
- // check for parameter errors
- if (msg.includeDoc) {
- return Promise.reject('cloudant trigger feed: includeDoc parameter is no longer supported');
- }
- if (!dbname) {
- return Promise.reject('cloudant trigger feed: missing dbname parameter - ' + dbname);
- }
- if (!host) {
- return Promise.reject('cloudant trigger feed: missing host parameter - ' + host);
- }
- if (!user) {
- return Promise.reject('cloudant trigger feed: missing username parameter - ' + user);
- }
- if (!pass) {
- return Promise.reject('cloudant trigger feed: missing password parameter - ' + pass);
- }
- if (namespace === "_") {
- return Promise.reject('You must supply a non-default namespace.');
- }
-
- if (msg.filter) {
- filter = msg.filter;
-
- if (typeof msg.query_params === 'object') {
- query_params = msg.query_params;
- }
- else if (typeof msg.query_params === 'string') {
- try {
- query_params = JSON.parse(msg.query_params);
- }
- catch (e) {
- return Promise.reject('The query_params parameter cannot be parsed. Ensure it is valid JSON.');
- }
- }
- }
- else if (msg.query_params) {
- return Promise.reject('The query_params parameter is only allowed if the filter parameter is defined');
- }
-
- // auth key for trigger
- var apiKey = msg.authKey;
- var input = {};
- input.accounturl = "https://" + host;
- input.host = host;
- input.port = port;
- input.protocol = protocol;
- input.dbname = dbname;
- input.user = user;
- input.pass = pass;
- input.apikey = apiKey;
- input.maxTriggers = maxTriggers;
- input.filter = filter;
- input.query_params = query_params;
-
- return cloudantHelper(provider_endpoint, 'put', triggerId, input);
- } else if (lifecycleEvent === 'DELETE') {
-
- var jsonOptions = {};
- jsonOptions.apikey = msg.authKey;
-
- return cloudantHelper(provider_endpoint, 'delete', triggerId, jsonOptions);
- } else {
- return Promise.reject('operation is neither CREATE or DELETE');
+ if (lifecycleEvent !== 'CREATE' && lifecycleEvent !== 'DELETE') {
+ return Promise.reject('lifecycleEvent must be CREATE or DELETE');
+ }
+ else {
+ var method = lifecycleEvent === 'CREATE' ? 'put' : 'delete';
+ return requestHelper(url, webparams, method);
}
}
-function cloudantHelper(endpoint, verb, name, input) {
- var url = 'http://' + endpoint + '/cloudanttriggers/' + name;
- var promise = new Promise(function(resolve, reject) {
+function requestHelper(url, input, method) {
+
+ return new Promise(function(resolve, reject) {
+
request({
- method : verb,
+ method : method,
url : url,
- json: input
+ json: input,
+ rejectUnauthorized: false
}, function(error, response, body) {
- console.log('cloudant trigger feed: done http request');
+
if (!error && response.statusCode === 200) {
- console.log(body);
resolve();
}
else {
if (response) {
- console.log('response code:', response.statusCode);
+ console.log('cloudant: Error invoking whisk action:', response.statusCode, body);
reject(body);
- } else {
- console.log('no response');
+ }
+ else {
+ console.log('cloudant: Error invoking whisk action:', error);
reject(error);
}
}
});
});
-
- return promise;
}
-function parseQName(qname) {
+function createWebParams(rawParams) {
+ var namespace = process.env.__OW_NAMESPACE;
+ var triggerName = ':' + namespace + ':' + parseQName(rawParams.triggerName, '/').name;
+
+ var webparams = Object.assign({}, rawParams);
+ delete webparams.lifecycleEvent;
+ delete webparams.bluemixServiceName;
+ delete webparams.apihost;
+
+ webparams.triggerName = triggerName;
+
+ return webparams;
+}
+
+function parseQName(qname, separator) {
var parsed = {};
- var delimiter = '/';
+ var delimiter = separator || ':';
var defaultNamespace = '_';
if (qname && qname.charAt(0) === delimiter) {
var parts = qname.split(delimiter);
@@ -132,3 +77,7 @@
}
return parsed;
}
+
+exports.main = main;
+
+
diff --git a/actions/changesWebAction.js b/actions/changesWebAction.js
new file mode 100644
index 0000000..b923a5d
--- /dev/null
+++ b/actions/changesWebAction.js
@@ -0,0 +1,282 @@
+var request = require('request');
+
+function main(params) {
+
+ if (!params.authKey) {
+ return sendError(400, 'no authKey parameter was provided');
+ }
+ if (!params.triggerName) {
+ return sendError(400, 'no trigger name parameter was provided');
+ }
+
+ var triggerID = params.triggerName;
+ var triggerParts = parseQName(triggerID);
+
+ var triggerURL = `https://${params.apihost}/api/v1/namespaces/${triggerParts.namespace}/triggers/${triggerParts.name}`;
+
+ var nano = require('nano')(params.DB_URL);
+ var db = nano.db.use(params.DB_NAME);
+
+ if (params.__ow_method === "put") {
+
+ // check for parameter errors
+ if (!params.dbname) {
+ return sendError(400, 'cloudant trigger feed: missing dbname parameter');
+ }
+ if (!params.host) {
+ return sendError(400, 'cloudant trigger feed: missing host parameter');
+ }
+ if (!params.username) {
+ return sendError(400, 'cloudant trigger feed: missing username parameter');
+ }
+ if (!params.password) {
+ return sendError(400, 'cloudant trigger feed: missing password parameter');
+ }
+
+ var query_params;
+
+ if (params.filter) {
+ if (typeof params.query_params === 'object') {
+ query_params = params.query_params;
+ }
+ else if (typeof params.query_params === 'string') {
+ try {
+ query_params = JSON.parse(params.query_params);
+ }
+ catch (e) {
+ return sendError(400, 'The query_params parameter cannot be parsed. Ensure it is valid JSON.');
+ }
+ }
+ }
+ else if (params.query_params) {
+ return sendError(400, 'The query_params parameter is only allowed if the filter parameter is defined');
+ }
+
+ var newTrigger = {
+ id: triggerID,
+ host: params.host,
+ port: params.port,
+ protocol: params.protocol || 'https',
+ dbname: params.dbname,
+ user: params.username,
+ pass: params.password,
+ apikey: params.authKey,
+ since: params.since,
+ maxTriggers: params.maxTriggers || -1,
+ filter: params.filter,
+ query_params: query_params,
+ status: {
+ 'active': true,
+ 'dateChanged': new Date().toISOString(),
+ }
+ };
+
+ return new Promise(function (resolve, reject) {
+ verifyTriggerAuth(triggerURL, params.authKey, false)
+ .then(() => {
+ return verifyUserDB(newTrigger);
+ })
+ .then(() => {
+ return createTrigger(db, triggerID, newTrigger);
+ })
+ .then(resolve)
+ .catch(err => {
+ reject(err);
+ });
+ });
+
+ }
+ else if (params.__ow_method === "delete") {
+
+ return new Promise(function (resolve, reject) {
+ verifyTriggerAuth(triggerURL, params.authKey, true)
+ .then(() => {
+ return updateTrigger(db, triggerID, 0);
+ })
+ .then(() => {
+ return deleteTrigger(db, triggerID, 0);
+ })
+ .then(resolve)
+ .catch(err => {
+ reject(err);
+ });
+ });
+ }
+ else {
+ return sendError(400, 'lifecycleEvent must be CREATE or DELETE');
+ }
+}
+
+function createTrigger(triggerDB, triggerID, newTrigger) {
+
+ return new Promise(function(resolve, reject) {
+
+ triggerDB.insert(newTrigger, triggerID, function (err) {
+ if (!err) {
+ resolve();
+ }
+ else {
+ reject(sendError(err.statusCode, 'error creating cloudant trigger.', err.message));
+ }
+ });
+ });
+}
+
+function updateTrigger(triggerDB, triggerID, retryCount) {
+
+ return new Promise(function(resolve, reject) {
+
+ triggerDB.get(triggerID, function (err, existing) {
+ if (!err) {
+ var updatedTrigger = existing;
+ updatedTrigger.status = {'active': false};
+
+ triggerDB.insert(updatedTrigger, triggerID, function (err) {
+ if (err) {
+ if (err.statusCode === 409 && retryCount < 5) {
+ setTimeout(function () {
+ updateTrigger(triggerDB, triggerID, (retryCount + 1))
+ .then(() => {
+ resolve();
+ }).catch(err => {
+ reject(err);
+ });
+ }, 1000);
+ }
+ else {
+ reject(sendError(err.statusCode, 'there was an error while marking the trigger for delete in the database.', err.message));
+ }
+ }
+ else {
+ resolve();
+ }
+ });
+ }
+ else {
+ reject(sendError(err.statusCode, 'could not find the trigger in the database'));
+ }
+ });
+ });
+}
+
+function deleteTrigger(triggerDB, triggerID, retryCount) {
+
+ return new Promise(function(resolve, reject) {
+
+ triggerDB.get(triggerID, function (err, existing) {
+ if (!err) {
+ triggerDB.destroy(existing._id, existing._rev, function (err) {
+ if (err) {
+ if (err.statusCode === 409 && retryCount < 5) {
+ setTimeout(function () {
+ deleteTrigger(triggerDB, triggerID, (retryCount + 1))
+ .then(resolve)
+ .catch(err => {
+ reject(err);
+ });
+ }, 1000);
+ }
+ else {
+ reject(sendError(err.statusCode, 'there was an error while deleting the trigger from the database.', err.message));
+ }
+ }
+ else {
+ resolve();
+ }
+ });
+ }
+ else {
+ reject(sendError(err.statusCode, 'could not find the trigger in the database'));
+ }
+ });
+ });
+}
+
+function verifyTriggerAuth(triggerURL, authKey, isDelete) {
+ var auth = authKey.split(':');
+
+ return new Promise(function(resolve, reject) {
+
+ request({
+ method: 'get',
+ url: triggerURL,
+ auth: {
+ user: auth[0],
+ pass: auth[1]
+ },
+ rejectUnauthorized: false
+ }, function(err, response) {
+ if (err) {
+ reject(sendError(400, 'Trigger authentication request failed.', err.message));
+ }
+ else if(response.statusCode >= 400 && !(isDelete && response.statusCode === 404)) {
+ reject(sendError(response.statusCode, 'Trigger authentication request failed.'));
+ }
+ else {
+ resolve();
+ }
+ });
+ });
+}
+
+function verifyUserDB(triggerObj) {
+ var dbURL = `${triggerObj.protocol}://${triggerObj.user}:${triggerObj.pass}@${triggerObj.host}`;
+
+ // add port if specified
+ if (triggerObj.port) {
+ dbURL += ':' + triggerObj.port;
+ }
+
+ return new Promise(function(resolve, reject) {
+ try {
+ var nanoConnection = require('nano')(dbURL);
+ var userDB = nanoConnection.use(triggerObj.dbname);
+ userDB.info(function(err, body) {
+ if (!err) {
+ resolve();
+ }
+ else {
+ reject(sendError(err.statusCode, 'error connecting to database ' + triggerObj.dbname, err.message));
+ }
+ });
+ }
+ catch(err) {
+ reject(sendError(400, 'error connecting to database ' + triggerObj.dbname, err.message));
+ }
+
+ });
+}
+
+function sendError(statusCode, error, message) {
+ var params = {error: error};
+ if (message) {
+ params.message = message;
+ }
+
+ return {
+ statusCode: statusCode,
+ headers: { 'Content-Type': 'application/json' },
+ body: new Buffer(JSON.stringify(params)).toString('base64'),
+ };
+}
+
+
+function parseQName(qname, separator) {
+ var parsed = {};
+ var delimiter = separator || ':';
+ var defaultNamespace = '_';
+ if (qname && qname.charAt(0) === delimiter) {
+ var parts = qname.split(delimiter);
+ parsed.namespace = parts[1];
+ parsed.name = parts.length > 2 ? parts.slice(2).join(delimiter) : '';
+ } else {
+ parsed.namespace = defaultNamespace;
+ parsed.name = qname;
+ }
+ return parsed;
+}
+
+
+exports.main = main;
+
+
diff --git a/installCatalog.sh b/installCatalog.sh
index eeaba47..2083cc2 100755
--- a/installCatalog.sh
+++ b/installCatalog.sh
@@ -4,8 +4,7 @@
# automatically
#
# To run this command
-# ./installCatalog.sh <AUTH> <APIHOST> <CLOUDANT_TRIGGER_HOST> <CLOUDANT_TRIGGER_PORT>
-# AUTH and APIHOST are found in $HOME/.wskprops
+# ./installCatalog.sh <authkey> <edgehost> <dburl> <dbprefix> <apihost>
set -e
set -x
@@ -15,14 +14,14 @@
if [ $# -eq 0 ]
then
-echo "Usage: ./installCatalog.sh <authkey> <apihost> <cloudanttriggerhost> <cloudanttriggerport>"
+echo "Usage: ./installCatalog.sh <authkey> <edgehost> <dburl> <dbprefix> <apihost>"
fi
AUTH="$1"
-APIHOST="$2"
-CLOUDANT_TRIGGER_HOST="$3"
-CLOUDANT_TRIGGER_PORT="$4"
-
+EDGEHOST="$2"
+DB_URL="$3"
+DB_NAME="${4}cloudanttrigger"
+APIHOST="$5"
# If the auth key file exists, read the key in the file. Otherwise, take the
# first argument as the key itself.
@@ -30,129 +29,146 @@
AUTH=`cat $AUTH`
fi
+# Make sure that the EDGEHOST is not empty.
+: ${EDGEHOST:?"EDGEHOST must be set and non-empty"}
+
+# Make sure that the DB_URL is not empty.
+: ${DB_URL:?"DB_URL must be set and non-empty"}
+
+# Make sure that the DB_NAME is not empty.
+: ${DB_NAME:?"DB_NAME must be set and non-empty"}
+
# Make sure that the APIHOST is not empty.
: ${APIHOST:?"APIHOST must be set and non-empty"}
-CLOUDANT_PROVIDER_ENDPOINT=$CLOUDANT_TRIGGER_HOST':'$CLOUDANT_TRIGGER_PORT
-echo 'cloudant trigger package endpoint:' $CLOUDANT_PROVIDER_ENDPOINT
-
PACKAGE_HOME="$( cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd )"
export WSK_CONFIG_FILE= # override local property file to avoid namespace clashes
echo Installing Cloudant package.
-$WSK_CLI -i --apihost "$APIHOST" package update --auth "$AUTH" --shared yes cloudant \
+$WSK_CLI -i --apihost "$EDGEHOST" package update --auth "$AUTH" --shared yes cloudant \
-a description "Cloudant database service" \
-a parameters '[ {"name":"bluemixServiceName", "required":false, "bindTime":true}, {"name":"username", "required":true, "bindTime":true, "description": "Your Cloudant username"}, {"name":"password", "required":true, "type":"password", "bindTime":true, "description": "Your Cloudant password"}, {"name":"host", "required":true, "bindTime":true, "description": "This is usually your username.cloudant.com"}, {"name":"dbname", "required":false, "description": "The name of your Cloudant database"}, {"name":"overwrite", "required":false, "type": "boolean"} ]' \
- -p package_endpoint "$CLOUDANT_PROVIDER_ENDPOINT" \
-p bluemixServiceName 'cloudantNoSQLDB' \
-p host '' \
-p username '' \
-p password '' \
- -p dbname ''
+ -p dbname '' \
+ -p apihost "$APIHOST"
+
+$WSK_CLI -i --apihost "$EDGEHOST" package update --auth "$AUTH" --shared no cloudantWeb \
+ -p DB_URL "$DB_URL" \
+ -p DB_NAME "$DB_NAME" \
+ -p apihost "$APIHOST"
# Cloudant feed action
-$WSK_CLI -i --apihost "$APIHOST" action update --auth "$AUTH" cloudant/changes "$PACKAGE_HOME/actions/changes.js" \
+$WSK_CLI -i --apihost "$EDGEHOST" action update --auth "$AUTH" cloudant/changes "$PACKAGE_HOME/actions/changes.js" \
-t 90000 \
-a feed true \
-a description 'Database change feed' \
-a parameters '[ {"name":"dbname", "required":true}, {"name": "filter", "required":false, "type": "string", "description": "The name of your Cloudant database filter"}, {"name": "query_params", "required":false, "description": "JSON Object containing query parameters that are passed to the filter"} ]' \
-a sampleInput '{ "dbname": "mydb", "filter": "mailbox/by_status", "query_params": {"status": "new"} }'
+# Cloudant web feed action
+
+$WSK_CLI -i --apihost "$EDGEHOST" action update --auth "$AUTH" cloudantWeb/changesWebAction "$PACKAGE_HOME/actions/changesWebAction.js" \
+ -a description 'Create/Delete a trigger in cloudant provider Database' \
+ --web true
+
# Cloudant account actions
-$WSK_CLI -i --apihost "$APIHOST" action update --auth "$AUTH" cloudant/create-database \
+$WSK_CLI -i --apihost "$EDGEHOST" action update --auth "$AUTH" cloudant/create-database \
"$PACKAGE_HOME/actions/account-actions/create-database.js" \
-a description 'Create Cloudant database' \
-a parameters '[ {"name":"dbname", "required":true} ]'
-$WSK_CLI -i --apihost "$APIHOST" action update --auth "$AUTH" cloudant/read-database \
+$WSK_CLI -i --apihost "$EDGEHOST" action update --auth "$AUTH" cloudant/read-database \
"$PACKAGE_HOME/actions/account-actions/read-database.js" \
-a description 'Read Cloudant database' \
-a parameters '[ {"name":"dbname", "required":true} ]'
-$WSK_CLI -i --apihost "$APIHOST" action update --auth "$AUTH" cloudant/delete-database \
+$WSK_CLI -i --apihost "$EDGEHOST" action update --auth "$AUTH" cloudant/delete-database \
"$PACKAGE_HOME/actions/account-actions/delete-database.js" \
-a description 'Delete Cloudant database' \
-a parameters '[ {"name":"dbname", "required":true} ]'
-$WSK_CLI -i --apihost "$APIHOST" action update --auth "$AUTH" cloudant/list-all-databases \
+$WSK_CLI -i --apihost "$EDGEHOST" action update --auth "$AUTH" cloudant/list-all-databases \
"$PACKAGE_HOME/actions/account-actions/list-all-databases.js" \
-a description 'List all Cloudant databases'
-$WSK_CLI -i --apihost "$APIHOST" action update --auth "$AUTH" cloudant/read-updates-feed \
+$WSK_CLI -i --apihost "$EDGEHOST" action update --auth "$AUTH" cloudant/read-updates-feed \
"$PACKAGE_HOME/actions/account-actions/read-updates-feed.js" \
-a description 'Read updates feed from Cloudant account (non-continuous)' \
-a parameters '[ {"name":"dbname", "required":true}, {"name":"params", "required":false} ]'
# Cloudant database actions
-$WSK_CLI -i --apihost "$APIHOST" action update --auth "$AUTH" cloudant/create-document \
+$WSK_CLI -i --apihost "$EDGEHOST" action update --auth "$AUTH" cloudant/create-document \
"$PACKAGE_HOME/actions/database-actions/create-document.js" \
-a description 'Create document in database' \
-a parameters '[ {"name":"dbname", "required":true}, {"name":"doc", "required":true, "description": "The JSON document to insert"}, {"name":"params", "required":false} ]' \
-$WSK_CLI -i --apihost "$APIHOST" action update --auth "$AUTH" cloudant/read \
+$WSK_CLI -i --apihost "$EDGEHOST" action update --auth "$AUTH" cloudant/read \
"$PACKAGE_HOME/actions/database-actions/read-document.js" \
-a description 'Read document from database' \
-a parameters '[ {"name":"dbname", "required":true}, {"name":"id", "required":true, "description": "The Cloudant document id to fetch"}, {"name":"params", "required":false}]' \
-p id ''
-$WSK_CLI -i --apihost "$APIHOST" action update --auth "$AUTH" cloudant/read-document \
+$WSK_CLI -i --apihost "$EDGEHOST" action update --auth "$AUTH" cloudant/read-document \
"$PACKAGE_HOME/actions/database-actions/read-document.js" \
-a description 'Read document from database' \
-a parameters '[ {"name":"dbname", "required":true}, {"name":"docid", "required":true, "description": "The Cloudant document id to fetch"}, {"name":"params", "required":false}]' \
-p docid ''
-$WSK_CLI -i --apihost "$APIHOST" action update --auth "$AUTH" cloudant/write \
+$WSK_CLI -i --apihost "$EDGEHOST" action update --auth "$AUTH" cloudant/write \
"$PACKAGE_HOME/actions/database-actions/write-document.js" \
-a description 'Write document in database' \
-a parameters '[ {"name":"dbname", "required":true}, {"name":"doc", "required":true} ]' \
-p doc '{}'
-$WSK_CLI -i --apihost "$APIHOST" action update --auth "$AUTH" cloudant/update-document \
+$WSK_CLI -i --apihost "$EDGEHOST" action update --auth "$AUTH" cloudant/update-document \
"$PACKAGE_HOME/actions/database-actions/update-document.js" \
-a description 'Update document in database' \
-a parameters '[ {"name":"dbname", "required":true}, {"name":"doc", "required":true}, {"name":"params", "required":false} ]' \
-p doc '{}'
-$WSK_CLI -i --apihost "$APIHOST" action update --auth "$AUTH" cloudant/delete-document \
+$WSK_CLI -i --apihost "$EDGEHOST" action update --auth "$AUTH" cloudant/delete-document \
"$PACKAGE_HOME/actions/database-actions/delete-document.js" \
-a description 'Delete document from database' \
-a parameters '[ {"name":"dbname", "required":true}, {"name":"docid", "required":true, "description": "The Cloudant document id to delete"}, {"name":"docrev", "required":true, "description": "The document revision number"} ]' \
-p docid '' \
-p docrev ''
-$WSK_CLI -i --apihost "$APIHOST" action update --auth "$AUTH" cloudant/list-documents \
+$WSK_CLI -i --apihost "$EDGEHOST" action update --auth "$AUTH" cloudant/list-documents \
"$PACKAGE_HOME/actions/database-actions/list-documents.js" \
-a description 'List all docs from database' \
-a parameters '[ {"name":"dbname", "required":true}, {"name":"params", "required":false} ]'
-$WSK_CLI -i --apihost "$APIHOST" action update --auth "$AUTH" cloudant/list-design-documents \
+$WSK_CLI -i --apihost "$EDGEHOST" action update --auth "$AUTH" cloudant/list-design-documents \
"$PACKAGE_HOME/actions/database-actions/list-design-documents.js" \
-a description 'List design documents from database' \
-a parameters '[ {"name":"dbname", "required":true}, {"name":"includedocs", "required":false} ]' \
-$WSK_CLI -i --apihost "$APIHOST" action update --auth "$AUTH" cloudant/create-query-index \
+$WSK_CLI -i --apihost "$EDGEHOST" action update --auth "$AUTH" cloudant/create-query-index \
"$PACKAGE_HOME/actions/database-actions/create-query-index.js" \
-a description 'Create a Cloudant Query index into database' \
-a parameters '[ {"name":"dbname", "required":true}, {"name":"index", "required":true} ]' \
-p index ''
-$WSK_CLI -i --apihost "$APIHOST" action update --auth "$AUTH" cloudant/list-query-indexes \
+$WSK_CLI -i --apihost "$EDGEHOST" action update --auth "$AUTH" cloudant/list-query-indexes \
"$PACKAGE_HOME/actions/database-actions/list-query-indexes.js" \
-a description 'List Cloudant Query indexes from database' \
-a parameters '[ {"name":"dbname", "required":true} ]' \
-$WSK_CLI -i --apihost "$APIHOST" action update --auth "$AUTH" cloudant/exec-query-find \
+$WSK_CLI -i --apihost "$EDGEHOST" action update --auth "$AUTH" cloudant/exec-query-find \
"$PACKAGE_HOME/actions/database-actions/exec-query-find.js" \
-a description 'Execute query against Cloudant Query index' \
-a parameters '[ {"name":"dbname", "required":true}, {"name":"query", "required":true} ]' \
-p query ''
-$WSK_CLI -i --apihost "$APIHOST" action update --auth "$AUTH" cloudant/exec-query-search \
+$WSK_CLI -i --apihost "$EDGEHOST" action update --auth "$AUTH" cloudant/exec-query-search \
"$PACKAGE_HOME/actions/database-actions/exec-query-search.js" \
-a description 'Execute query against Cloudant search' \
-a parameters '[ {"name":"dbname", "required":true}, {"name":"docid", "required":true}, {"name":"indexname", "required":true}, {"name":"search", "required":true} ]' \
@@ -160,39 +176,39 @@
-p indexname '' \
-p search ''
-$WSK_CLI -i --apihost "$APIHOST" action update --auth "$AUTH" cloudant/exec-query-view \
+$WSK_CLI -i --apihost "$EDGEHOST" action update --auth "$AUTH" cloudant/exec-query-view \
"$PACKAGE_HOME/actions/database-actions/exec-query-view.js" \
-a description 'Call view in design document from database' \
-a parameters '[ {"name":"dbname", "required":true}, {"name":"docid", "required":true}, {"name":"viewname", "required":true}, {"name":"params", "required":false} ]' \
-p docid '' \
-p viewname ''
-$WSK_CLI -i --apihost "$APIHOST" action update --auth "$AUTH" cloudant/manage-bulk-documents \
+$WSK_CLI -i --apihost "$EDGEHOST" action update --auth "$AUTH" cloudant/manage-bulk-documents \
"$PACKAGE_HOME/actions/database-actions/manage-bulk-documents.js" \
-a description 'Create, Update, and Delete documents in bulk' \
-a parameters '[ {"name":"dbname", "required":true}, {"name":"docs", "required":true}, {"name":"params", "required":false} ]' \
-p docs '{}'
-$WSK_CLI -i --apihost "$APIHOST" action update --auth "$AUTH" cloudant/delete-view \
+$WSK_CLI -i --apihost "$EDGEHOST" action update --auth "$AUTH" cloudant/delete-view \
"$PACKAGE_HOME/actions/database-actions/delete-view.js" \
-a description 'Delete view from design document' \
-a parameters '[ {"name":"dbname", "required":true}, {"name":"docid", "required":true}, {"name":"viewname", "required":true}, {"name":"params", "required":false} ]' \
-p docid '' \
-p viewname ''
-$WSK_CLI -i --apihost "$APIHOST" action update --auth "$AUTH" cloudant/delete-query-index \
+$WSK_CLI -i --apihost "$EDGEHOST" action update --auth "$AUTH" cloudant/delete-query-index \
"$PACKAGE_HOME/actions/database-actions/delete-query-index.js" \
-a description 'Delete index from design document' \
-a parameters '[ {"name":"dbname", "required":true}, {"name":"docid", "required":true}, {"name":"indexname", "required":true}, {"name":"params", "required":false} ]' \
-p docid '' \
-p indexname ''
-$WSK_CLI -i --apihost "$APIHOST" action update --auth "$AUTH" cloudant/read-changes-feed \
+$WSK_CLI -i --apihost "$EDGEHOST" action update --auth "$AUTH" cloudant/read-changes-feed \
"$PACKAGE_HOME/actions/database-actions/read-changes-feed.js" \
-a description 'Read Cloudant database changes feed (non-continuous)' \
-a parameters '[ {"name":"dbname", "required":true}, {"name":"params", "required":false} ]'
-$WSK_CLI -i --apihost "$APIHOST" action update --auth "$AUTH" cloudant/create-attachment \
+$WSK_CLI -i --apihost "$EDGEHOST" action update --auth "$AUTH" cloudant/create-attachment \
"$PACKAGE_HOME/actions/database-actions/create-update-attachment.js" \
-a description 'Create document attachment in database' \
-a parameters '[ {"name":"dbname", "required":true}, {"name":"docid", "required":true}, {"name":"docrev", "required":true}, {"name":"attachment", "required":true}, {"name":"attachmentname", "required":true}, {"name":"contenttype", "required":true}, {"name":"params", "required":false} ]' \
@@ -202,14 +218,14 @@
-p attachmentname '' \
-p contenttype ''
-$WSK_CLI -i --apihost "$APIHOST" action update --auth "$AUTH" cloudant/read-attachment \
+$WSK_CLI -i --apihost "$EDGEHOST" action update --auth "$AUTH" cloudant/read-attachment \
"$PACKAGE_HOME/actions/database-actions/read-attachment.js" \
-a description 'Read document attachment from database' \
-a parameters '[ {"name":"dbname", "required":true}, {"name":"docid", "required":true}, {"name":"attachmentname", "required":true}, {"name":"params", "required":false} ]' \
-p docid '' \
-p attachmentname ''
-$WSK_CLI -i --apihost "$APIHOST" action update --auth "$AUTH" cloudant/update-attachment \
+$WSK_CLI -i --apihost "$EDGEHOST" action update --auth "$AUTH" cloudant/update-attachment \
"$PACKAGE_HOME/actions/database-actions/create-update-attachment.js" \
-a description 'Update document attachment in database' \
-a parameters '[ {"name":"dbname", "required":true}, {"name":"docid", "required":true}, {"name":"docrev", "required":true}, {"name":"attachment", "required":true}, {"name":"attachmentname", "required":true}, {"name":"contenttype", "required":true}, {"name":"params", "required":false} ]' \
@@ -219,7 +235,7 @@
-p attachmentname '' \
-p contenttype ''
-$WSK_CLI -i --apihost "$APIHOST" action update --auth "$AUTH" cloudant/delete-attachment \
+$WSK_CLI -i --apihost "$EDGEHOST" action update --auth "$AUTH" cloudant/delete-attachment \
"$PACKAGE_HOME/actions/database-actions/delete-attachment.js" \
-a description 'Delete document attachment from database' \
-a parameters '[ {"name":"dbname", "required":true}, {"name":"docid", "required":true}, {"name":"docrev", "required":true}, {"name":"attachmentname", "required":true}, {"name":"params", "required":false} ]' \
diff --git a/package.json b/package.json
index 187de1f..1932a43 100644
--- a/package.json
+++ b/package.json
@@ -14,6 +14,8 @@
"lodash": "^3.10.1",
"request": "2.69.0",
"nano": "^6.2.0",
- "json-stringify-safe": "^5.0.1"
+ "json-stringify-safe": "^5.0.1",
+ "http-status-codes": "^1.0.5",
+ "request-promise": "^1.0.2"
}
}
\ No newline at end of file
diff --git a/provider/app.js b/provider/app.js
index 4408b22..5af7424 100644
--- a/provider/app.js
+++ b/provider/app.js
@@ -12,8 +12,7 @@
var ProviderUtils = require('./lib/utils.js');
var ProviderHealth = require('./lib/health.js');
var ProviderRAS = require('./lib/ras.js');
-var ProviderUpdate = require('./lib/update.js');
-var ProviderDelete = require('./lib/delete.js');
+var ProviderActivation = require('./lib/active.js');
var constants = require('./lib/constants.js');
// Initialize the Express Application
@@ -22,23 +21,18 @@
app.use(bodyParser.urlencoded({ extended: false }));
app.set('port', process.env.PORT || 8080);
-
-// Whisk API Router Host
-var routerHost = process.env.ROUTER_HOST || 'localhost';
-
// Allow invoking servers with self-signed certificates.
process.env.NODE_TLS_REJECT_UNAUTHORIZED = "0";
// If it does not already exist, create the triggers database. This is the database that will
// store the managed triggers.
-//
var dbUsername = process.env.DB_USERNAME;
var dbPassword = process.env.DB_PASSWORD;
var dbHost = process.env.DB_HOST;
var dbProtocol = process.env.DB_PROTOCOL;
var dbPrefix = process.env.DB_PREFIX;
var databaseName = dbPrefix + constants.TRIGGER_DB_SUFFIX;
-var ddname = '_design/filters';
+var ddname = '_design/triggers';
// Create the Provider Server
var server = http.createServer(app);
@@ -47,34 +41,33 @@
});
function createDatabase (nanop) {
- logger.info('createDatabase', 'creating the trigger database');
+ var method = 'createDatabase';
+ logger.info(method, 'creating the trigger database');
+
return new Promise(function(resolve, reject) {
- nanop.db.create(databaseName, function (err) {
+ nanop.db.create(databaseName, function (err, body) {
if (!err) {
- logger.info('createDatabase', 'created trigger database:', databaseName);
+ logger.info(method, 'created trigger database:', databaseName);
}
else {
- logger.info('createDatabase', 'failed to create trigger database:', databaseName, err);
+ logger.info(method, 'failed to create trigger database:', databaseName, err);
}
var db = nanop.db.use(databaseName);
- var only_triggers = {
- map: function (doc) {
- if (doc.maxTriggers) {
- emit(doc._id, 1);
- }
- }.toString()
- };
- db.get(ddname, function (error) {
+ var only_triggers_by_worker = function(doc, req) {
+ return doc.maxTriggers && ((!doc.worker && req.query.worker === 'worker0') || (doc.worker === req.query.worker));
+ }.toString();
+
+ db.get(ddname, function (error, body) {
if (error) {
//new design doc
db.insert({
- views: {
- only_triggers: only_triggers
+ filters: {
+ only_triggers_by_worker: only_triggers_by_worker
},
- }, ddname, function (error) {
+ }, ddname, function (error, body) {
if (error) {
- reject("view could not be created: " + error);
+ reject("filter could not be created: " + error);
}
resolve(db);
});
@@ -99,24 +92,24 @@
// Initialize the Provider Server
function init(server) {
+ var method = 'init';
if (server !== null) {
var address = server.address();
if (address === null) {
- logger.error('init', 'Error initializing server. Perhaps port is already in use.');
+ logger.error(method, 'Error initializing server. Perhaps port is already in use.');
process.exit(-1);
}
}
createTriggerDb()
.then(nanoDb => {
- logger.info('init', 'trigger storage database details:', nanoDb);
+ logger.info(method, 'trigger storage database details:', nanoDb);
- var providerUtils = new ProviderUtils(logger, app, nanoDb, routerHost);
- var providerRAS = new ProviderRAS(logger);
- var providerHealth = new ProviderHealth(logger, providerUtils);
- var providerUpdate = new ProviderUpdate(logger, providerUtils);
- var providerDelete = new ProviderDelete(logger, providerUtils);
+ var providerUtils = new ProviderUtils(logger, nanoDb);
+ var providerRAS = new ProviderRAS();
+ var providerHealth = new ProviderHealth(providerUtils);
+ var providerActivation = new ProviderActivation(logger, providerUtils);
// RAS Endpoint
app.get(providerRAS.endPoint, providerRAS.ras);
@@ -124,16 +117,12 @@
// Health Endpoint
app.get(providerHealth.endPoint, providerHealth.health);
- // Endpoint for Update OR Create a Trigger
- app.put(providerUpdate.endPoint, providerUtils.authorize, providerUpdate.update);
-
- // Endpoint for Deleting an existing Trigger.
- app.delete(providerDelete.endPoint, providerUtils.authorize, providerDelete.delete);
+ // Activation Endpoint
+ app.get(providerActivation.endPoint, providerUtils.authorize, providerActivation.active);
providerUtils.initAllTriggers();
-
}).catch(err => {
- logger.error('init', 'an error occurred creating database:', err);
+ logger.error(method, 'an error occurred creating database:', err);
});
}
diff --git a/provider/lib/active.js b/provider/lib/active.js
new file mode 100644
index 0000000..82c29f0
--- /dev/null
+++ b/provider/lib/active.js
@@ -0,0 +1,33 @@
+module.exports = function(logger, providerUtils) {
+
+ // Active Endpoint
+ this.endPoint = '/active';
+
+ this.active = function (req, res) {
+ var method = 'active';
+ var response = {};
+
+ if (req.query && req.query.active) {
+ var errorMessage = "Invalid query string";
+ try {
+ var active = JSON.parse(req.query.active);
+ if (typeof active !== 'boolean') {
+ response.error = errorMessage;
+ }
+ else if (providerUtils.active !== active) {
+ var message = 'The active state has been changed';
+ logger.info(method, message, 'to', active);
+ providerUtils.active = active;
+ response.message = message;
+ }
+ }
+ catch (e) {
+ response.error = errorMessage;
+ }
+ }
+ response.active = providerUtils.active;
+ response.worker = providerUtils.worker;
+ res.send(response);
+ };
+
+};
diff --git a/provider/lib/delete.js b/provider/lib/delete.js
deleted file mode 100644
index ca62359..0000000
--- a/provider/lib/delete.js
+++ /dev/null
@@ -1,61 +0,0 @@
-var request = require('request');
-
-module.exports = function(logger, utils) {
-
- // Test Endpoint
- this.endPoint = '/cloudanttriggers/:id';
-
- // Delete Logic
- this.delete = function (req, res) {
-
- var method = 'DELETE /cloudanttriggers';
-
- var id = req.params.id;
- var args = typeof req.body === 'object' ? req.body : JSON.parse(req.body);
-
- //Check that user has access rights to delete a trigger
- var triggerObj = utils.parseQName(id);
- var host = 'https://' + utils.routerHost +':'+ 443;
- var triggerURL = host + '/api/v1/namespaces/' + triggerObj.namespace + '/triggers/' + triggerObj.name;
- var auth = args.apikey.split(':');
-
- logger.info(method, 'Checking if user has access rights to delete trigger', id);
- request({
- method: 'get',
- url: triggerURL,
- auth: {
- user: auth[0],
- pass: auth[1]
- }
- }, function(error, response, body) {
- //delete from database if user is authenticated (200) or if trigger has already been deleted (404)
- if (!error && (response.statusCode === 200 || response.statusCode === 404)) {
- utils.deleteTriggerFromDB(id, res);
- }
- else {
- var errorMsg = 'Cloudant data trigger ' + id + ' cannot be deleted.';
- logger.error(method, errorMsg, error);
- if (error) {
- res.status(400).json({
- message: errorMsg,
- error: error.message
- });
- }
- else {
- var info;
- try {
- info = JSON.parse(body);
- }
- catch (e) {
- info = 'Authentication request failed with status code ' + response.statusCode;
- }
- res.status(response.statusCode).json({
- message: errorMsg,
- error: typeof info === 'object' ? info.error : info
- });
- }
- }
- });
- };
-
-};
diff --git a/provider/lib/health.js b/provider/lib/health.js
index 690dc71..808295d 100644
--- a/provider/lib/health.js
+++ b/provider/lib/health.js
@@ -1,4 +1,4 @@
-module.exports = function(logger, providerUtils) {
+module.exports = function(providerUtils) {
// Health Endpoint
this.endPoint = '/health';
diff --git a/provider/lib/update.js b/provider/lib/update.js
deleted file mode 100644
index c07c0d4..0000000
--- a/provider/lib/update.js
+++ /dev/null
@@ -1,77 +0,0 @@
-var request = require('request');
-
-module.exports = function(logger, utils) {
-
- // Test Endpoint
- this.endPoint = '/cloudanttriggers/:id';
-
- // Update Logic
- this.update = function (req, res) {
-
- var method = 'PUT /cloudanttriggers';
- var args = typeof req.body === 'object' ? req.body : JSON.parse(req.body);
-
- //Check that user has access rights to create a trigger
- var id = req.params.id;
- var triggerObj = utils.parseQName(id);
- var host = 'https://' + utils.routerHost +':'+ 443;
- var triggerURL = host + '/api/v1/namespaces/' + triggerObj.namespace + '/triggers/' + triggerObj.name;
- var auth = args.apikey.split(':');
-
- logger.info(method, 'Checking if user has access rights to create trigger', id);
- request({
- method: 'get',
- url: triggerURL,
- auth: {
- user: auth[0],
- pass: auth[1]
- }
- }, function(error, response, body) {
- if (error || response.statusCode >= 400) {
- var errorMsg = 'Trigger authentication request failed.';
- logger.error(method, errorMsg, error);
- if (error) {
- res.status(400).json({
- message: errorMsg,
- error: error.message
- });
- }
- else {
- var info;
- try {
- info = JSON.parse(body);
- }
- catch (e) {
- info = 'Authentication request failed with status code ' + response.statusCode;
- }
- res.status(response.statusCode).json({
- message: errorMsg,
- error: typeof info === 'object' ? info.error : info
- });
- }
- }
- else {
- var id = req.params.id;
- var trigger = utils.initTrigger(args, id);
- // number of retries to create a trigger.
- utils.createTrigger(trigger, utils.retryAttempts)
- .then(newTrigger => {
- newTrigger.status = {
- 'active': true,
- 'dateChanged': new Date().toISOString(),
- };
- utils.addTriggerToDB(newTrigger, res);
- logger.info(method, 'Trigger was added and database is confirmed.', newTrigger.id);
- }).catch(err => {
- logger.error(method, 'Trigger', id, 'could not be created.', err);
- utils.deleteTrigger(id);
- res.status(400).json({
- message: 'Trigger could not be created.',
- error: err
- });
- });
- }
- });
- };
-
-};
diff --git a/provider/lib/utils.js b/provider/lib/utils.js
index 4828b9b..db5c18e 100644
--- a/provider/lib/utils.js
+++ b/provider/lib/utils.js
@@ -1,64 +1,44 @@
var _ = require('lodash');
var request = require('request');
var constants = require('./constants.js');
-
+var HttpStatus = require('http-status-codes');
module.exports = function(
logger,
- app,
- triggerDB,
- routerHost
+ triggerDB
) {
-
- this.logger = logger;
- this.app = app;
- this.triggerDB = triggerDB;
- this.routerHost = routerHost;
-
- // this is the default trigger fire limit (in the event that it was not set during trigger creation)
- this.defaultTriggerFireLimit = constants.DEFAULT_MAX_TRIGGERS;
- this.retryDelay = constants.RETRY_DELAY;
- this.retryAttempts = constants.RETRY_ATTEMPTS;
-
- // Log HTTP Requests
- app.use(function(req, res, next) {
- if (req.url.indexOf('/cloudanttriggers') === 0) {
- logger.info('HttpRequest', req.method, req.url);
- }
- next();
- });
-
this.module = 'utils';
this.triggers = {};
+ this.endpointAuth = process.env.ENDPOINT_AUTH;
+ this.routerHost = process.env.ROUTER_HOST || 'localhost';
+ this.active = !(process.env.ACTIVE && process.env.ACTIVE.toLowerCase() === 'false');
+ this.worker = process.env.WORKER || "worker0";
- var that = this;
+ this.retryAttempts = constants.RETRY_ATTEMPTS;
+
+ var ddname = 'triggers';
+ var filter = 'only_triggers_by_worker';
+
+ var utils = this;
// Add a trigger: listen for changes and dispatch.
- this.createTrigger = function(dataTrigger, retryCount) {
-
+ this.createTrigger = function(dataTrigger) {
var method = 'createTrigger';
- var sinceToUse = dataTrigger.since ? dataTrigger.since : "now";
- var dbProtocol = dataTrigger.protocol ? dataTrigger.protocol : 'https';
-
- // unless specified host will default to accounturl without the https:// in front
- var dbHost = dataTrigger.host ? dataTrigger.host : dataTrigger.accounturl.replace('https://','');
-
// both couch and cloudant should have their URLs in the username:password@host format
- var dbURL = dbProtocol + '://' + dataTrigger.user + ':' + dataTrigger.pass + '@' + dbHost;
+ var dbURL = `${dataTrigger.protocol}://${dataTrigger.user}:${dataTrigger.pass}@${dataTrigger.host}`;
// add port if specified
if (dataTrigger.port) {
- dbURL = dbURL + ':' + dataTrigger.port;
+ dbURL += ':' + dataTrigger.port;
}
- var nanoConnection = require('nano')(dbURL);
-
try {
+ var nanoConnection = require('nano')(dbURL);
var triggeredDB = nanoConnection.use(dataTrigger.dbname);
// Listen for changes on this database.
- var feed = triggeredDB.follow({since: sinceToUse, include_docs: false});
+ var feed = triggeredDB.follow({since: dataTrigger.since, include_docs: false});
if (dataTrigger.filter) {
feed.filter = dataTrigger.filter;
}
@@ -67,17 +47,19 @@
}
dataTrigger.feed = feed;
- that.triggers[dataTrigger.id] = dataTrigger;
+ utils.triggers[dataTrigger.id] = dataTrigger;
feed.on('change', function (change) {
- logger.info(method, 'Trigger', dataTrigger.id, 'got change from', dataTrigger.dbname);
+ if (utils.active) {
+ logger.info(method, 'Trigger', dataTrigger.id, 'got change from', dataTrigger.dbname);
- var triggerHandle = that.triggers[dataTrigger.id];
- if (triggerHandle && (triggerHandle.maxTriggers === -1 || triggerHandle.triggersLeft > 0)) {
- try {
- that.fireTrigger(dataTrigger.id, change);
- } catch (e) {
- logger.error(method, 'Exception occurred while firing trigger', dataTrigger.id, e);
+ var triggerHandle = utils.triggers[dataTrigger.id];
+ if (triggerHandle && (triggerHandle.maxTriggers === -1 || triggerHandle.triggersLeft > 0)) {
+ try {
+ utils.fireTrigger(dataTrigger.id, change);
+ } catch (e) {
+ logger.error(method, 'Exception occurred while firing trigger', dataTrigger.id, e);
+ }
}
}
});
@@ -86,34 +68,17 @@
return new Promise(function(resolve, reject) {
- feed.on('error', function (err) {
- logger.error(method,'Error occurred for trigger', dataTrigger.id, '(db ' + dataTrigger.dbname + '):', err);
- // revive the feed if an error occurred for now
- that.deleteTrigger(dataTrigger.id);
- dataTrigger.since = "now";
- if (retryCount > 0) {
- logger.info(method, 'attempting to recreate trigger', dataTrigger.id, 'Retry Count:', (retryCount - 1));
- that.createTrigger(dataTrigger, (retryCount - 1))
- .then(trigger => {
- resolve(trigger);
- }).catch(err => {
- reject(err);
- });
- } else {
- logger.error(method, "Trigger's feed produced too many errors. Deleting the trigger", dataTrigger.id, '(db ' + dataTrigger.dbname + ')');
- reject({
- error: err,
- message: "Trigger's feed produced too many errors. Deleting the trigger " + dataTrigger.id
- });
- }
- });
+ feed.on('error', function (err) {
+ logger.error(method,'Error occurred for trigger', dataTrigger.id, '(db ' + dataTrigger.dbname + '):', err);
+ reject(err);
+ });
- feed.on('confirm', function (dbObj) {
- logger.info(method, 'Added cloudant data trigger', dataTrigger.id, 'listening for changes in database', dataTrigger.dbname);
- resolve(dataTrigger);
- });
+ feed.on('confirm', function (dbObj) {
+ logger.info(method, 'Added cloudant data trigger', dataTrigger.id, 'listening for changes in database', dataTrigger.dbname);
+ resolve(dataTrigger.id);
+ });
- });
+ });
} catch (err) {
logger.info(method, 'caught an exception for trigger', dataTrigger.id, err);
@@ -122,226 +87,107 @@
};
- this.initTrigger = function (obj, id) {
-
+ this.initTrigger = function (newTrigger) {
var method = 'initTrigger';
- // validate parameters here
- logger.info(method, 'create trigger', id, 'with the following args', obj);
+ logger.info(method, 'create trigger', newTrigger.id, 'with the following args', newTrigger);
- // if the trigger creation request has not set the max trigger fire limit
- // we will set it here (default value can be updated in ./constants.js)
- if (!obj.maxTriggers) {
- obj.maxTriggers = that.defaultTriggerFireLimit;
- }
+ var maxTriggers = newTrigger.maxTriggers || constants.DEFAULT_MAX_TRIGGERS;
var trigger = {
- id: id,
- accounturl: obj.accounturl,
- dbname: obj.dbname,
- user: obj.user,
- pass: obj.pass,
- host: obj.host,
- port: obj.port,
- protocol: obj.protocol,
- apikey: obj.apikey,
- since: obj.since,
- maxTriggers: obj.maxTriggers,
- triggersLeft: obj.maxTriggers,
- filter: obj.filter,
- query_params: obj.query_params
+ id: newTrigger.id,
+ host: newTrigger.host,
+ port: newTrigger.port,
+ protocol: newTrigger.protocol || "https",
+ dbname: newTrigger.dbname,
+ user: newTrigger.user,
+ pass: newTrigger.pass,
+ apikey: newTrigger.apikey,
+ since: newTrigger.since || "now",
+ maxTriggers: maxTriggers,
+ triggersLeft: maxTriggers,
+ filter: newTrigger.filter,
+ query_params: newTrigger.query_params
};
return trigger;
};
- this.initAllTriggers = function () {
+ this.shouldDisableTrigger = function (statusCode) {
+ return ((statusCode >= 400 && statusCode < 500) &&
+ [HttpStatus.REQUEST_TIMEOUT, HttpStatus.TOO_MANY_REQUESTS].indexOf(statusCode) === -1);
+ };
- var method = 'initAllTriggers';
- logger.info(method, 'Initializing all cloudant triggers from database.');
+ this.disableTrigger = function (id, statusCode, message) {
+ var method = 'disableTrigger';
- triggerDB.view('filters', 'only_triggers', {include_docs: true}, function(err, body) {
- if (!err) {
- body.rows.forEach(function(trigger) {
- if (!trigger.doc.status || trigger.doc.status.active === true) {
- //check if trigger still exists in whisk db
- var triggerObj = that.parseQName(trigger.doc.id);
- var host = 'https://' + routerHost + ':' + 443;
- var triggerURL = host + '/api/v1/namespaces/' + triggerObj.namespace + '/triggers/' + triggerObj.name;
- var auth = trigger.doc.apikey.split(':');
+ //only active/master provider should update the database
+ if (utils.active) {
+ triggerDB.get(id, function (err, existing) {
+ if (!err) {
+ if (!existing.status || existing.status.active === true) {
+ var updatedTrigger = existing;
+ var status = {
+ 'active': false,
+ 'dateChanged': new Date().toISOString(),
+ 'reason': {'kind': 'AUTO', 'statusCode': statusCode, 'message': message}
+ };
+ updatedTrigger.status = status;
- logger.info(method, 'Checking if trigger', trigger.doc.id, 'still exists');
- request({
- method: 'get',
- url: triggerURL,
- auth: {
- user: auth[0],
- pass: auth[1]
- }
- }, function (error, response) {
- //disable trigger in database if trigger is dead
- if (!error && that.shouldDisableTrigger(response.statusCode)) {
- var message = 'Automatically disabled after receiving a ' + response.statusCode + ' status code when re-creating the trigger';
- that.disableTriggerInDB(trigger.doc.id, response.statusCode, message);
- logger.error(method, 'trigger', trigger.doc.id, 'has been disabled due to status code:', response.statusCode);
+ triggerDB.insert(updatedTrigger, id, function (err) {
+ if (err) {
+ logger.error(method, 'there was an error while disabling', id, 'in database. ' + err);
}
else {
- var cloudantTrigger = that.initTrigger(trigger.doc, trigger.doc.id);
- that.createTrigger(cloudantTrigger, that.retryAttempts)
- .then(newTrigger => {
- logger.info(method, 'Trigger was added.', newTrigger.id);
- }).catch(err => {
- var message = 'Automatically disabled after receiving an exception when re-creating the trigger';
- that.disableTriggerInDB(cloudantTrigger.id, undefined, message);
- logger.error(method, 'Disabled trigger', cloudantTrigger.id, 'due to exception:', err);
- });
+ logger.info(method, 'trigger', id, 'successfully disabled in database');
}
});
}
- else {
- logger.info(method, 'ignoring trigger', trigger.doc._id, 'since it is disabled.');
- }
- });
- } else {
- logger.error(method, 'could not get latest state from database');
- }
- });
-
- };
-
- this.addTriggerToDB = function (trigger, res) {
-
- var method = 'addTriggerToDB';
- triggerDB.insert(_.omit(trigger, 'feed'), trigger.id, function(err) {
- if (!err) {
- logger.info(method, 'trigger', trigger.id, 'was inserted into db.');
- res.status(200).json(_.omit(trigger, 'feed'));
- } else {
- that.deleteTrigger(trigger.id);
- res.status(err.statusCode).json({error: 'Cloudant trigger cannot be created. ' + err});
- }
- });
-
- };
-
- this.shouldDisableTrigger = function (statusCode) {
- return ((statusCode >= 400 && statusCode < 500) && [408, 429].indexOf(statusCode) === -1);
- };
-
- this.disableTriggerInDB = function (id, statusCode, message) {
-
- var method = 'disableTriggerInDB';
-
- triggerDB.get(id, function (err, existing) {
- if (!err) {
- if (!existing.status || existing.status.active === true) {
- var updatedTrigger = existing;
- var status = {
- 'active': false,
- 'dateChanged': new Date().toISOString(),
- 'reason': {'kind': 'AUTO', 'statusCode': statusCode, 'message': message}
- };
- updatedTrigger.status = status;
-
- triggerDB.insert(updatedTrigger, id, function (err) {
- if (err) {
- logger.error(method, 'there was an error while disabling', id, 'in database. ' + err);
- }
- else {
- that.deleteTrigger(id);
- logger.info(method, 'trigger', id, 'successfully disabled in database');
- }
- });
}
- }
- else {
- if (err.statusCode === 404) {
- logger.error(method, 'there was no trigger with id', id, 'in database.', err.error);
- } else {
+ else {
logger.error(method, 'there was an error while getting', id, 'from database', err);
}
- }
- });
- };
-
- this.deleteTriggerFromDB = function (id, res) {
-
- var method = 'deleteTriggerFromDB';
-
- triggerDB.get(id, function(err, existing) {
- if (!err) {
- triggerDB.destroy(existing._id, existing._rev, function(err) {
- if (err) {
- logger.error(method, 'there was an error while deleting', id, 'from database. ' + err);
- if (res) {
- res.status(err.statusCode).json({ error: 'Cloudant data trigger ' + id + 'cannot be deleted. ' + err } );
- }
- } else {
- that.deleteTrigger(id);
- logger.info(method, 'cloudant trigger', id, ' is successfully deleted');
- if (res) {
- res.send('Deleted cloudant trigger ' + id);
- }
- }
- });
- } else {
- if (err.statusCode === 404) {
- logger.info(method, 'there was no trigger with id', id, 'in database.', err.error);
- if (res) {
- res.status(404).json({ message: 'there was no trigger with id ' + id + ' in database.' } );
- res.end();
- }
- } else {
- logger.error(method, 'there was an error while getting', id, 'from database', err);
- if (res) {
- res.status(err.statusCode).json({ error: 'Cloudant data trigger ' + id + ' cannot be deleted. ' + err } );
- }
- }
- }
- });
-
- };
-
- // Delete a trigger: stop listening for changes and remove it.
- this.deleteTrigger = function (id) {
- var method = 'deleteTrigger';
- var trigger = that.triggers[id];
- if (trigger) {
- logger.info(method, 'Stopped cloudant trigger', id, 'listening for changes in database', trigger.dbname);
- trigger.feed.stop();
- delete that.triggers[id];
- } else {
- logger.info(method, 'trigger', id, 'could not be found in the trigger list.');
- return false;
+ });
}
};
- this.fireTrigger = function (id, change) {
+ // Delete a trigger: stop listening for changes and remove it.
+ this.deleteTrigger = function (triggerIdentifier) {
+ var method = 'deleteTrigger';
+
+ if (utils.triggers[triggerIdentifier].feed) {
+ utils.triggers[triggerIdentifier].feed.stop();
+ }
+
+ delete utils.triggers[triggerIdentifier];
+ logger.info(method, 'trigger', triggerIdentifier, 'successfully deleted from memory');
+ };
+
+ this.fireTrigger = function (triggerIdentifier, change) {
var method = 'fireTrigger';
- var dataTrigger = that.triggers[id];
- var apikey = dataTrigger.apikey;
- var triggerObj = that.parseQName(dataTrigger.id);
+ var dataTrigger = utils.triggers[triggerIdentifier];
+ var triggerObj = utils.parseQName(dataTrigger.id);
var form = change;
form.dbname = dataTrigger.dbname;
logger.info(method, 'firing trigger', dataTrigger.id, 'with db update');
- var host = 'https://' + routerHost + ':' + 443;
+ var host = 'https://' + utils.routerHost + ':' + 443;
var uri = host + '/api/v1/namespaces/' + triggerObj.namespace + '/triggers/' + triggerObj.name;
- var auth = apikey.split(':');
+ var auth = dataTrigger.apikey.split(':');
- that.postTrigger(dataTrigger, form, uri, auth, 0)
- .then(triggerId => {
- logger.info(method, 'Trigger', triggerId, 'was successfully fired');
- if (dataTrigger.triggersLeft === 0) {
- that.disableTriggerInDB(dataTrigger.id, undefined, 'Automatically disabled after reaching max triggers');
- logger.error(method, 'no more triggers left, disabled', dataTrigger.id);
- }
- }).catch(err => {
- logger.error(method, err);
- });
+ utils.postTrigger(dataTrigger, form, uri, auth, 0)
+ .then(triggerId => {
+ logger.info(method, 'Trigger', triggerId, 'was successfully fired');
+ if (dataTrigger.triggersLeft === 0) {
+ utils.disableTrigger(dataTrigger.id, undefined, 'Automatically disabled after reaching max triggers');
+ logger.error(method, 'no more triggers left, disabled', dataTrigger.id);
+ }
+ }).catch(err => {
+ logger.error(method, err);
+ });
};
this.postTrigger = function (dataTrigger, form, uri, auth, retryCount) {
@@ -362,18 +208,18 @@
logger.info(method, dataTrigger.id, 'http post request, STATUS:', response ? response.statusCode : response);
if (error || response.statusCode >= 400) {
logger.error(method, 'there was an error invoking', dataTrigger.id, response ? response.statusCode : error);
- if (!error && that.shouldDisableTrigger(response.statusCode)) {
+ if (!error && utils.shouldDisableTrigger(response.statusCode)) {
//disable trigger
var message = 'Automatically disabled after receiving a ' + response.statusCode + ' status code when firing the trigger';
- that.disableTriggerInDB(dataTrigger.id, response.statusCode, message);
+ utils.disableTrigger(dataTrigger.id, response.statusCode, message);
reject('Disabled trigger ' + dataTrigger.id + ' due to status code: ' + response.statusCode);
}
else {
- if (retryCount < that.retryAttempts ) {
+ if (retryCount < utils.retryAttempts ) {
var timeout = response && response.statusCode === 429 && retryCount === 0 ? 60000 : 1000 * Math.pow(retryCount + 1, 2);
logger.info(method, 'attempting to fire trigger again', dataTrigger.id, 'Retry Count:', (retryCount + 1));
setTimeout(function () {
- that.postTrigger(dataTrigger, form, uri, auth, (retryCount + 1))
+ utils.postTrigger(dataTrigger, form, uri, auth, (retryCount + 1))
.then(triggerId => {
resolve(triggerId);
}).catch(err => {
@@ -400,6 +246,139 @@
});
};
+ this.initAllTriggers = function () {
+ var method = 'initAllTriggers';
+
+ logger.info(method, 'resetting system from last state');
+
+ triggerDB.changes({ since: 0, include_docs: true, filter: ddname + '/' + filter, worker: utils.worker }, (err, changes) => {
+ if (!err) {
+ changes.results.forEach(function (change) {
+ var triggerIdentifier = change.id;
+ var doc = change.doc;
+
+ if (!doc.status || doc.status.active === true) {
+ //check if trigger still exists in whisk db
+ var triggerObj = utils.parseQName(triggerIdentifier);
+ var host = 'https://' + utils.routerHost + ':' + 443;
+ var triggerURL = host + '/api/v1/namespaces/' + triggerObj.namespace + '/triggers/' + triggerObj.name;
+ var auth = doc.apikey.split(':');
+
+ logger.info(method, 'Checking if trigger', triggerIdentifier, 'still exists');
+ request({
+ method: 'get',
+ url: triggerURL,
+ auth: {
+ user: auth[0],
+ pass: auth[1]
+ }
+ }, function (error, response) {
+ //disable trigger in database if trigger is dead
+ if (!error && utils.shouldDisableTrigger(response.statusCode)) {
+ var message = 'Automatically disabled after receiving a ' + response.statusCode + ' status code on init trigger';
+ utils.disableTrigger(triggerIdentifier, response.statusCode, message);
+ logger.error(method, 'trigger', triggerIdentifier, 'has been disabled due to status code:', response.statusCode);
+ }
+ else {
+ utils.createTrigger(utils.initTrigger(doc))
+ .then(triggerIdentifier => {
+ logger.info(method, 'Trigger was added.', triggerIdentifier);
+ }).catch(err => {
+ var message = 'Automatically disabled after receiving exception on init trigger: ' + err;
+ utils.disableTrigger(triggerIdentifier, undefined, message);
+ logger.error(method, 'Disabled trigger', triggerIdentifier, 'due to exception:', err);
+ });
+ }
+ });
+ }
+ else {
+ logger.info(method, 'ignoring trigger', triggerIdentifier, 'since it is disabled.');
+ }
+ });
+ utils.setupFollow(changes.last_seq);
+ } else {
+ logger.error(method, 'could not get latest state from database', err);
+ }
+ });
+ };
+
+ this.setupFollow = function setupFollow(seq) {
+ var method = 'setupFollow';
+
+ var feed = triggerDB.follow({ since: seq, include_docs: true, filter: ddname + '/' + filter, query_params: { worker: utils.worker } });
+
+ feed.on('change', (change) => {
+ var triggerIdentifier = change.id;
+ var doc = change.doc;
+
+ logger.info(method, 'got change for trigger', triggerIdentifier);
+
+ if (utils.triggers[triggerIdentifier]) {
+ if (doc.status && doc.status.active === false) {
+ utils.deleteTrigger(triggerIdentifier);
+ }
+ }
+ else {
+ //ignore changes to disabled triggers
+ if (!doc.status || doc.status.active === true) {
+ utils.createTrigger(utils.initTrigger(doc))
+ .then(triggerIdentifier => {
+ logger.info(method, triggerIdentifier, 'created successfully');
+ }).catch(err => {
+ var message = 'Automatically disabled after receiving exception on create trigger: ' + err;
+ utils.disableTrigger(triggerIdentifier, undefined, message);
+ logger.error(method, 'Disabled trigger', triggerIdentifier, 'due to exception:', err);
+ });
+ }
+ }
+ });
+ feed.follow();
+ };
+
+ this.authorize = function(req, res, next) {
+ var method = 'authorize';
+
+ if (utils.endpointAuth) {
+
+ if (!req.headers.authorization) {
+ return utils.sendError(method, HttpStatus.BAD_REQUEST, 'Malformed request, authentication header expected', res);
+ }
+
+ var parts = req.headers.authorization.split(' ');
+ if (parts[0].toLowerCase() !== 'basic' || !parts[1]) {
+ return utils.sendError(method, HttpStatus.BAD_REQUEST, 'Malformed request, basic authentication expected', res);
+ }
+
+ var auth = new Buffer(parts[1], 'base64').toString();
+ auth = auth.match(/^([^:]*):(.*)$/);
+ if (!auth) {
+ return utils.sendError(method, HttpStatus.BAD_REQUEST, 'Malformed request, authentication invalid', res);
+ }
+
+ var uuid = auth[1];
+ var key = auth[2];
+
+ var endpointAuth = utils.endpointAuth.split(':');
+
+ if (endpointAuth[0] === uuid && endpointAuth[1] === key) {
+ logger.info(method, 'Authentication successful');
+ next();
+ }
+ else {
+ logger.warn(method, 'Invalid key');
+ return utils.sendError(method, HttpStatus.UNAUTHORIZED, 'Invalid key', res);
+ }
+ }
+ else {
+ next();
+ }
+ };
+
+ this.sendError = function (method, code, message, res) {
+ logger.error(method, message);
+ res.status(code).json({error: message});
+ };
+
this.parseQName = function (qname, separator) {
var parsed = {};
var delimiter = separator || ':';
@@ -415,8 +394,4 @@
return parsed;
};
- this.authorize = function(req, res, next) {
- next();
- };
-
};
diff --git a/tests/src/test/scala/system/packages/CloudantFeedTests.scala b/tests/src/test/scala/system/packages/CloudantFeedTests.scala
index efb5b19..145e992 100644
--- a/tests/src/test/scala/system/packages/CloudantFeedTests.scala
+++ b/tests/src/test/scala/system/packages/CloudantFeedTests.scala
@@ -40,39 +40,6 @@
behavior of "Cloudant trigger service"
- it should "fail on create feed when includeDocs is set" in withAssetCleaner(wskprops) {
-
- (wp, assetHelper) =>
- implicit val wskprops = wp // shadow global props and make implicit
- val triggerName = s"dummyCloudantTrigger-${System.currentTimeMillis}"
- val packageName = "dummyCloudantPackage"
- val feed = "changes"
-
- val packageGetResult = wsk.pkg.get("/whisk.system/cloudant")
- println("Fetching cloudant package.")
- packageGetResult.stdout should include("ok")
-
- println("Creating cloudant package binding.")
- assetHelper.withCleaner(wsk.pkg, packageName) {
- (pkg, name) => pkg.bind("/whisk.system/cloudant", name)
- }
-
- println("Creating cloudant trigger feed.")
- val feedCreationResult = assetHelper.withCleaner(wsk.trigger, triggerName, confirmDelete = false) {
- (trigger, name) =>
- trigger.create(name, feed = Some(s"$packageName/$feed"), parameters = Map(
- "username" -> myCloudantCreds.user.toJson,
- "password" -> myCloudantCreds.password.toJson,
- "host" -> myCloudantCreds.host().toJson,
- "dbname" -> myCloudantCreds.dbname.toJson,
- "includeDoc" -> "true".toJson,
- "maxTriggers" -> 1.toJson),
- expectedExitCode = 246)
- }
- feedCreationResult.stderr should include("includeDoc parameter is no longer supported")
-
- }
-
it should "return useful error message when changes feed does not include host parameter" in withAssetCleaner(wskprops) {
(wp, assetHelper) =>
@@ -197,7 +164,7 @@
}
- it should "delete trigger if its Cloudant connection is not created" in withAssetCleaner(wskprops) {
+ it should "throw error if Cloudant connection is invalid" in withAssetCleaner(wskprops) {
(wp, assetHelper) =>
implicit val wskprops = wp // shadow global props and make implicit
@@ -230,7 +197,7 @@
}
- it should "should disable after reaching max triggers" in withAssetCleaner(wskprops) {
+ it should "disable after reaching max triggers" in withAssetCleaner(wskprops) {
(wp, assetHelper) =>
implicit val wskprops = wp // shadow global props and make implicit
val triggerName = s"dummyCloudantTrigger-${System.currentTimeMillis}"
@@ -267,7 +234,7 @@
response1.get("ok").getAsString() should be("true")
println("Checking for activations")
- val activations = wsk.activation.pollFor(N = 2, Some(triggerName), retries = 5).length
+ val activations = wsk.activation.pollFor(N = 2, Some(triggerName)).length
println(s"Found activation size (should be exactly 1): $activations")
activations should be(1)
diff --git a/tests/src/test/scala/system/packages/CloudantFeedWebTests.scala b/tests/src/test/scala/system/packages/CloudantFeedWebTests.scala
new file mode 100644
index 0000000..a3445d0
--- /dev/null
+++ b/tests/src/test/scala/system/packages/CloudantFeedWebTests.scala
@@ -0,0 +1,122 @@
+/*
+ * Copyright 2015-2016 IBM Corporation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package system.packages
+
+import com.jayway.restassured.RestAssured
+import com.jayway.restassured.config.SSLConfig
+import com.jayway.restassured.http.ContentType
+import common.TestUtils.FORBIDDEN
+import common.{Wsk, WskProps}
+import org.junit.runner.RunWith
+import org.scalatest.junit.JUnitRunner
+import org.scalatest.{BeforeAndAfter, FlatSpec, Matchers}
+import spray.json._
+
+@RunWith(classOf[JUnitRunner])
+class CloudantFeedWebTests
+ extends FlatSpec
+ with BeforeAndAfter
+ with Matchers {
+
+ val wskprops = WskProps()
+
+ val webAction = "/whisk.system/cloudantWeb/changesWebAction"
+ val webActionURL = s"https://${wskprops.apihost}/api/v1/web${webAction}.http"
+
+ val requiredParams = JsObject(
+ "triggerName" -> JsString("/invalidNamespace/invalidTrigger"),
+ "authKey" -> JsString("DoesNotWork"),
+ "dbname" -> JsString("DoesNotExist"),
+ "host" -> JsString("bad.hostname"),
+ "username" -> JsString("username"),
+ "password" -> JsString("password")
+ )
+
+ behavior of "Cloudant changes web action"
+
+ it should "not be obtainable using the CLI" in {
+ val wsk = new Wsk()
+ implicit val wp = wskprops
+
+ wsk.action.get(webAction, FORBIDDEN)
+ }
+
+ it should "reject put of a trigger due to missing triggerName argument" in {
+ val params = JsObject(requiredParams.fields - "triggerName")
+
+ makePutCallWithExpectedResult(params, JsObject("error" -> JsString("no trigger name parameter was provided")), 400)
+ }
+
+ it should "reject put of a trigger due to missing host argument" in {
+ val params = JsObject(requiredParams.fields - "host")
+
+ makePutCallWithExpectedResult(params, JsObject("error" -> JsString("cloudant trigger feed: missing host parameter")), 400)
+ }
+
+ it should "reject put of a trigger due to missing username argument" in {
+ val params = JsObject(requiredParams.fields - "username")
+
+ makePutCallWithExpectedResult(params, JsObject("error" -> JsString("cloudant trigger feed: missing username parameter")), 400)
+ }
+
+ it should "reject put of a trigger due to missing password argument" in {
+ val params = JsObject(requiredParams.fields - "password")
+
+ makePutCallWithExpectedResult(params, JsObject("error" -> JsString("cloudant trigger feed: missing password parameter")), 400)
+ }
+
+ it should "reject put of a trigger due to missing dbname argument" in {
+ val params = JsObject(requiredParams.fields - "dbname")
+
+ makePutCallWithExpectedResult(params, JsObject("error" -> JsString("cloudant trigger feed: missing dbname parameter")), 400)
+ }
+
+ it should "reject put of a trigger when authentication fails" in {
+
+ makePutCallWithExpectedResult(requiredParams, JsObject("error" -> JsString("Trigger authentication request failed.")), 401)
+ }
+
+ it should "reject delete of a trigger due to missing triggerName argument" in {
+ val params = JsObject(requiredParams.fields - "triggerName")
+
+ makeDeleteCallWithExpectedResult(params, JsObject("error" -> JsString("no trigger name parameter was provided")), 400)
+ }
+
+ it should "reject delete of a trigger when authentication fails" in {
+ makeDeleteCallWithExpectedResult(requiredParams, JsObject("error" -> JsString("Trigger authentication request failed.")), 401)
+ }
+
+ def makePutCallWithExpectedResult(params: JsObject, expectedResult: JsObject, expectedCode: Int) = {
+ val response = RestAssured.given()
+ .contentType(ContentType.JSON)
+ .config(RestAssured.config().sslConfig(new SSLConfig().relaxedHTTPSValidation()))
+ .body(params.toString())
+ .put(webActionURL)
+ assert(response.statusCode() == expectedCode)
+ response.body.asString.parseJson.asJsObject shouldBe expectedResult
+ }
+
+ def makeDeleteCallWithExpectedResult(params: JsObject, expectedResult: JsObject, expectedCode: Int) = {
+ val response = RestAssured.given()
+ .contentType(ContentType.JSON)
+ .config(RestAssured.config().sslConfig(new SSLConfig().relaxedHTTPSValidation()))
+ .body(params.toString())
+ .delete(webActionURL)
+ assert(response.statusCode() == expectedCode)
+ response.body.asString.parseJson.asJsObject shouldBe expectedResult
+ }
+
+}
