loosen restrictions on redis values (#173)
diff --git a/provider/lib/active.js b/provider/lib/active.js
index 9f8a0fc..cbc5017 100644
--- a/provider/lib/active.js
+++ b/provider/lib/active.js
@@ -28,10 +28,10 @@
var activeHost = query === 'true' ? utils.host : redundantHost;
if (utils.activeHost !== activeHost) {
if (utils.redisClient) {
- utils.redisClient.hsetAsync(utils.redisHash, utils.redisKey, activeHost)
+ utils.redisClient.hsetAsync(utils.redisKey, utils.redisField, activeHost)
.then(() => {
response.active = 'swapping';
- utils.redisClient.publish(utils.redisHash, activeHost);
+ utils.redisClient.publish(utils.redisKey, activeHost);
logger.info(method, 'Active host swap in progress');
res.send(response);
})
diff --git a/provider/lib/constants.js b/provider/lib/constants.js
index 539fd99..305f1c7 100644
--- a/provider/lib/constants.js
+++ b/provider/lib/constants.js
@@ -2,7 +2,7 @@
const DEFAULT_MAX_TRIGGERS = -1;
const RETRY_ATTEMPTS = 12;
const RETRY_DELAY = 1000; //in milliseconds
-const REDIS_KEY = 'active';
+const REDIS_FIELD = 'active';
const FILTERS_DESIGN_DOC = 'triggerFilters';
const VIEWS_DESIGN_DOC = 'triggerViews';
const MONITOR_DESIGN_DOC = 'monitorFilters';
@@ -16,7 +16,7 @@
DEFAULT_MAX_TRIGGERS: DEFAULT_MAX_TRIGGERS,
RETRY_ATTEMPTS: RETRY_ATTEMPTS,
RETRY_DELAY: RETRY_DELAY,
- REDIS_KEY: REDIS_KEY,
+ REDIS_FIELD: REDIS_FIELD,
FILTERS_DESIGN_DOC: FILTERS_DESIGN_DOC,
VIEWS_DESIGN_DOC: VIEWS_DESIGN_DOC,
TRIGGERS_BY_WORKER: TRIGGERS_BY_WORKER,
diff --git a/provider/lib/health.js b/provider/lib/health.js
index b13d6ad..c2055a3 100644
--- a/provider/lib/health.js
+++ b/provider/lib/health.js
@@ -14,7 +14,6 @@
var canaryDocID;
var monitorStatus;
var monitorStages = ['triggerStarted', 'triggerFired', 'triggerStopped'];
- var healthMonitor = this;
// Health Logic
this.health = function (req, res) {
@@ -65,10 +64,10 @@
//delete the trigger
var uri = utils.uriHost + '/api/v1/namespaces/_/triggers/' + triggerName;
- healthMonitor.deleteTrigger(existingTriggerID, uri, auth, 0);
+ deleteTrigger(existingTriggerID, uri, auth, 0);
//delete the canary doc
- healthMonitor.deleteDocFromDB(existingCanaryID, 0);
+ deleteDocFromDB(existingCanaryID, 0);
}
//create new cloudant trigger and canary doc
@@ -82,17 +81,17 @@
var triggerURL = utils.uriHost + '/api/v1/namespaces/_/triggers/' + triggerName;
var triggerID = `:_:${triggerName}`;
- healthMonitor.createTrigger(triggerURL, auth)
+ createTrigger(triggerURL, auth)
.then(info => {
logger.info(method, triggerID, info);
- var newTrigger = healthMonitor.createCloudantTrigger(triggerID, apikey);
+ var newTrigger = createCloudantTrigger(triggerID, apikey);
utils.createTrigger(newTrigger);
setTimeout(function () {
var canaryDoc = {
isCanaryDoc: true,
host: utils.host
};
- healthMonitor.createDocInDB(canaryDocID, canaryDoc);
+ createDocInDB(canaryDocID, canaryDoc);
}, monitoringInterval / 3);
})
.catch(err => {
@@ -100,7 +99,7 @@
});
};
- this.createCloudantTrigger = function(triggerID, apikey) {
+ function createCloudantTrigger(triggerID, apikey) {
var method = 'createCloudantTrigger';
var dbURL = new URL(utils.db.config.url);
@@ -125,9 +124,9 @@
};
return newTrigger;
- };
+ }
- this.createTrigger = function(triggerURL, auth) {
+ function createTrigger(triggerURL, auth) {
var method = 'createTrigger';
return new Promise(function(resolve, reject) {
@@ -149,9 +148,9 @@
}
});
});
- };
+ }
- this.createDocInDB = function(docID, doc) {
+ function createDocInDB(docID, doc) {
var method = 'createDocInDB';
utils.db.insert(doc, docID, function (err) {
@@ -162,9 +161,9 @@
logger.error(method, docID, err);
}
});
- };
+ }
- this.deleteTrigger = function(triggerID, uri, auth, retryCount) {
+ function deleteTrigger(triggerID, uri, auth, retryCount) {
var method = 'deleteTrigger';
request({
@@ -180,7 +179,7 @@
if (!error && response.statusCode === 409 && retryCount < 5) {
logger.info(method, 'attempting to delete trigger again', triggerID, 'Retry Count:', (retryCount + 1));
setTimeout(function () {
- healthMonitor.deleteTrigger(triggerID, uri, auth, (retryCount + 1));
+ deleteTrigger(triggerID, uri, auth, (retryCount + 1));
}, 1000);
} else {
logger.error(method, triggerID, 'trigger delete request failed');
@@ -190,9 +189,9 @@
logger.info(method, triggerID, 'trigger delete request was successful');
}
});
- };
+ }
- this.deleteDocFromDB = function(docID, retryCount) {
+ function deleteDocFromDB(docID, retryCount) {
var method = 'deleteDocFromDB';
//delete from database
@@ -202,7 +201,7 @@
if (err) {
if (err.statusCode === 409 && retryCount < 5) {
setTimeout(function () {
- healthMonitor.deleteDocFromDB(docID, (retryCount + 1));
+ deleteDocFromDB(docID, (retryCount + 1));
}, 1000);
}
else {
@@ -218,6 +217,6 @@
logger.error(method, docID, 'could not be found in the database');
}
});
- };
+ }
};
diff --git a/provider/lib/utils.js b/provider/lib/utils.js
index fb5c3dc..1256d59 100644
--- a/provider/lib/utils.js
+++ b/provider/lib/utils.js
@@ -4,6 +4,13 @@
module.exports = function(logger, triggerDB, redisClient) {
+ var retryAttempts = constants.RETRY_ATTEMPTS;
+ var filterDDName = constants.FILTERS_DESIGN_DOC;
+ var viewDDName = constants.VIEWS_DESIGN_DOC;
+ var triggersByWorker = constants.TRIGGERS_BY_WORKER;
+ var redisKeyPrefix = process.env.REDIS_KEY_PREFIX || triggerDB.config.db;
+ var self = this;
+
this.triggers = {};
this.endpointAuth = process.env.ENDPOINT_AUTH;
this.routerHost = process.env.ROUTER_HOST || 'localhost';
@@ -12,17 +19,11 @@
this.activeHost = 'host0'; //default value on init (will be updated for existing redis)
this.db = triggerDB;
this.redisClient = redisClient;
- this.redisHash = this.db.config.db + '_' + this.worker;
- this.redisKey = constants.REDIS_KEY;
+ this.redisKey = redisKeyPrefix + '_' + this.worker;
+ this.redisField = constants.REDIS_FIELD;
this.uriHost ='https://' + this.routerHost;
this.monitorStatus = {};
- var retryAttempts = constants.RETRY_ATTEMPTS;
- var filterDDName = constants.FILTERS_DESIGN_DOC;
- var viewDDName = constants.VIEWS_DESIGN_DOC;
- var triggersByWorker = constants.TRIGGERS_BY_WORKER;
- var utils = this;
-
// Add a trigger: listen for changes and dispatch.
this.createTrigger = function(dataTrigger) {
var method = 'createTrigger';
@@ -49,14 +50,14 @@
}
dataTrigger.feed = feed;
- utils.triggers[dataTrigger.id] = dataTrigger;
+ self.triggers[dataTrigger.id] = dataTrigger;
feed.on('change', function (change) {
- var triggerHandle = utils.triggers[dataTrigger.id];
- if (triggerHandle && utils.shouldFireTrigger(triggerHandle) && utils.hasTriggersRemaining(triggerHandle)) {
+ var triggerHandle = self.triggers[dataTrigger.id];
+ if (triggerHandle && shouldFireTrigger(triggerHandle) && hasTriggersRemaining(triggerHandle)) {
logger.info(method, 'Trigger', dataTrigger.id, 'got change from', dataTrigger.dbname);
try {
- utils.fireTrigger(dataTrigger.id, change);
+ fireTrigger(dataTrigger.id, change);
} catch (e) {
logger.error(method, 'Exception occurred while firing trigger', dataTrigger.id, e);
}
@@ -73,8 +74,8 @@
feed.on('confirm', function () {
logger.info(method, 'Added cloudant data trigger', dataTrigger.id, 'listening for changes in database', dataTrigger.dbname);
- if (utils.isMonitoringTrigger(dataTrigger.monitor, dataTrigger.id)) {
- utils.monitorStatus.triggerStarted = "success";
+ if (isMonitoringTrigger(dataTrigger.monitor, dataTrigger.id)) {
+ self.monitorStatus.triggerStarted = "success";
}
resolve(dataTrigger.id);
});
@@ -87,7 +88,7 @@
};
- this.initTrigger = function(newTrigger) {
+ function initTrigger(newTrigger) {
var maxTriggers = newTrigger.maxTriggers || constants.DEFAULT_MAX_TRIGGERS;
var trigger = {
@@ -107,26 +108,26 @@
};
return trigger;
- };
+ }
- this.shouldDisableTrigger = function(statusCode) {
+ function shouldDisableTrigger(statusCode) {
return ((statusCode >= 400 && statusCode < 500) &&
[HttpStatus.REQUEST_TIMEOUT, HttpStatus.TOO_MANY_REQUESTS].indexOf(statusCode) === -1);
- };
+ }
- this.shouldFireTrigger = function(trigger) {
- return trigger.monitor || utils.activeHost === utils.host;
- };
+ function shouldFireTrigger(trigger) {
+ return trigger.monitor || self.activeHost === self.host;
+ }
- this.hasTriggersRemaining = function(trigger) {
+ function hasTriggersRemaining(trigger) {
return !trigger.maxTriggers || trigger.maxTriggers === -1 || trigger.triggersLeft > 0;
- };
+ }
- this.isMonitoringTrigger = function(monitor, triggerIdentifier) {
- return monitor && utils.monitorStatus.triggerName === utils.parseQName(triggerIdentifier).name;
- };
+ function isMonitoringTrigger(monitor, triggerIdentifier) {
+ return monitor && self.monitorStatus.triggerName === parseQName(triggerIdentifier).name;
+ }
- this.disableTrigger = function(id, statusCode, message) {
+ function disableTrigger(id, statusCode, message) {
var method = 'disableTrigger';
triggerDB.get(id, function (err, existing) {
@@ -153,56 +154,56 @@
else {
logger.info(method, 'could not find', id, 'in database');
//make sure it is removed from memory as well
- utils.deleteTrigger(id);
+ deleteTrigger(id);
}
});
- };
+ }
// Delete a trigger: stop listening for changes and remove it.
- this.deleteTrigger = function(triggerIdentifier, isMonitoringTrigger) {
+ function deleteTrigger(triggerIdentifier, monitorTrigger) {
var method = 'deleteTrigger';
- if (utils.triggers[triggerIdentifier]) {
- if (utils.triggers[triggerIdentifier].feed) {
- utils.triggers[triggerIdentifier].feed.stop();
+ if (self.triggers[triggerIdentifier]) {
+ if (self.triggers[triggerIdentifier].feed) {
+ self.triggers[triggerIdentifier].feed.stop();
}
- delete utils.triggers[triggerIdentifier];
+ delete self.triggers[triggerIdentifier];
logger.info(method, 'trigger', triggerIdentifier, 'successfully deleted from memory');
- if (utils.isMonitoringTrigger(isMonitoringTrigger, triggerIdentifier)) {
- utils.monitorStatus.triggerStopped = "success";
+ if (isMonitoringTrigger(monitorTrigger, triggerIdentifier)) {
+ self.monitorStatus.triggerStopped = "success";
}
}
- };
+ }
- this.fireTrigger = function(triggerIdentifier, change) {
+ function fireTrigger(triggerIdentifier, change) {
var method = 'fireTrigger';
- var dataTrigger = utils.triggers[triggerIdentifier];
- var triggerObj = utils.parseQName(dataTrigger.id);
+ var dataTrigger = self.triggers[triggerIdentifier];
+ var triggerObj = parseQName(dataTrigger.id);
var form = change;
form.dbname = dataTrigger.dbname;
logger.info(method, 'firing trigger', dataTrigger.id, 'with db update');
- var host = 'https://' + utils.routerHost + ':' + 443;
+ var host = 'https://' + self.routerHost + ':' + 443;
var uri = host + '/api/v1/namespaces/' + triggerObj.namespace + '/triggers/' + triggerObj.name;
var auth = dataTrigger.apikey.split(':');
- utils.postTrigger(dataTrigger, form, uri, auth, 0)
+ postTrigger(dataTrigger, form, uri, auth, 0)
.then(triggerId => {
logger.info(method, 'Trigger', triggerId, 'was successfully fired');
- if (utils.isMonitoringTrigger(dataTrigger.monitor, triggerId)) {
- utils.monitorStatus.triggerFired = "success";
+ if (isMonitoringTrigger(dataTrigger.monitor, triggerId)) {
+ self.monitorStatus.triggerFired = "success";
}
if (dataTrigger.triggersLeft === 0) {
if (dataTrigger.monitor) {
- utils.deleteTrigger(triggerId, dataTrigger.monitor);
+ deleteTrigger(triggerId, dataTrigger.monitor);
}
else {
- utils.disableTrigger(triggerId, undefined, 'Automatically disabled after reaching max triggers');
+ disableTrigger(triggerId, undefined, 'Automatically disabled after reaching max triggers');
logger.warn(method, 'no more triggers left, disabled', triggerId);
}
}
@@ -210,9 +211,9 @@
.catch(err => {
logger.error(method, err);
});
- };
+ }
- this.postTrigger = function(dataTrigger, form, uri, auth, retryCount) {
+ function postTrigger(dataTrigger, form, uri, auth, retryCount) {
var method = 'postTrigger';
return new Promise(function(resolve, reject) {
@@ -239,10 +240,10 @@
dataTrigger.triggersLeft++;
}
logger.error(method, 'there was an error invoking', dataTrigger.id, response ? response.statusCode : error);
- if (!error && utils.shouldDisableTrigger(response.statusCode)) {
+ if (!error && shouldDisableTrigger(response.statusCode)) {
//disable trigger
var message = 'Automatically disabled after receiving a ' + response.statusCode + ' status code when firing the trigger';
- utils.disableTrigger(dataTrigger.id, response.statusCode, message);
+ disableTrigger(dataTrigger.id, response.statusCode, message);
reject('Disabled trigger ' + dataTrigger.id + ' due to status code: ' + response.statusCode);
}
else {
@@ -250,7 +251,7 @@
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 () {
- utils.postTrigger(dataTrigger, form, uri, auth, (retryCount + 1))
+ postTrigger(dataTrigger, form, uri, auth, (retryCount + 1))
.then(triggerId => {
resolve(triggerId);
})
@@ -272,25 +273,25 @@
}
});
});
- };
+ }
this.initAllTriggers = function() {
var method = 'initAllTriggers';
//follow the trigger DB
- utils.setupFollow('now');
+ setupFollow('now');
logger.info(method, 'resetting system from last state');
- triggerDB.view(viewDDName, triggersByWorker, {reduce: false, include_docs: true, key: utils.worker}, function(err, body) {
+ triggerDB.view(viewDDName, triggersByWorker, {reduce: false, include_docs: true, key: self.worker}, function(err, body) {
if (!err) {
body.rows.forEach(function (trigger) {
var triggerIdentifier = trigger.id;
var doc = trigger.doc;
- if (!(triggerIdentifier in utils.triggers)) {
+ if (!(triggerIdentifier in self.triggers)) {
//check if trigger still exists in whisk db
- var triggerObj = utils.parseQName(triggerIdentifier);
- var host = 'https://' + utils.routerHost + ':' + 443;
+ var triggerObj = parseQName(triggerIdentifier);
+ var host = 'https://' + self.routerHost + ':' + 443;
var triggerURL = host + '/api/v1/namespaces/' + triggerObj.namespace + '/triggers/' + triggerObj.name;
var auth = doc.apikey.split(':');
@@ -304,19 +305,19 @@
}
}, function (error, response) {
//disable trigger in database if trigger is dead
- if (!error && utils.shouldDisableTrigger(response.statusCode)) {
+ if (!error && shouldDisableTrigger(response.statusCode)) {
var message = 'Automatically disabled after receiving a ' + response.statusCode + ' status code on init trigger';
- utils.disableTrigger(triggerIdentifier, response.statusCode, message);
+ 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))
+ self.createTrigger(initTrigger(doc))
.then(triggerIdentifier => {
logger.info(method, triggerIdentifier, 'created successfully');
})
.catch(err => {
var message = 'Automatically disabled after receiving exception on init trigger: ' + err;
- utils.disableTrigger(triggerIdentifier, undefined, message);
+ disableTrigger(triggerIdentifier, undefined, message);
logger.error(method, 'Disabled trigger', triggerIdentifier, 'due to exception:', err);
});
}
@@ -329,7 +330,7 @@
});
};
- this.setupFollow = function(seq) {
+ function setupFollow(seq) {
var method = 'setupFollow';
try {
@@ -337,28 +338,28 @@
since: seq,
include_docs: true,
filter: filterDDName + '/' + triggersByWorker,
- query_params: {worker: utils.worker}
+ query_params: {worker: self.worker}
});
feed.on('change', (change) => {
var triggerIdentifier = change.id;
var doc = change.doc;
- if (utils.triggers[triggerIdentifier]) {
+ if (self.triggers[triggerIdentifier]) {
if (doc.status && doc.status.active === false) {
- utils.deleteTrigger(triggerIdentifier);
+ deleteTrigger(triggerIdentifier);
}
}
else {
//ignore changes to disabled triggers
if (!doc.status || doc.status.active === true) {
- utils.createTrigger(utils.initTrigger(doc))
+ self.createTrigger(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);
+ disableTrigger(triggerIdentifier, undefined, message);
logger.error(method, 'Disabled trigger', triggerIdentifier, 'due to exception:', err);
});
}
@@ -374,12 +375,12 @@
catch (err) {
logger.error(method, err);
}
- };
+ }
this.authorize = function(req, res, next) {
var method = 'authorize';
- if (utils.endpointAuth) {
+ if (self.endpointAuth) {
if (!req.headers.authorization) {
res.set('www-authenticate', 'Basic realm="Private"');
res.status(HttpStatus.UNAUTHORIZED);
@@ -388,24 +389,24 @@
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);
+ return 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);
+ return sendError(method, HttpStatus.BAD_REQUEST, 'Malformed request, authentication invalid', res);
}
var uuid = auth[1];
var key = auth[2];
- var endpointAuth = utils.endpointAuth.split(':');
+ var endpointAuth = self.endpointAuth.split(':');
if (endpointAuth[0] === uuid && endpointAuth[1] === key) {
next();
}
else {
logger.warn(method, 'Invalid key');
- return utils.sendError(method, HttpStatus.UNAUTHORIZED, 'Invalid key', res);
+ return sendError(method, HttpStatus.UNAUTHORIZED, 'Invalid key', res);
}
}
else {
@@ -413,12 +414,12 @@
}
};
- this.sendError = function(method, code, message, res) {
+ function sendError(method, code, message, res) {
logger.error(method, message);
res.status(code).json({error: message});
- };
+ }
- this.parseQName = function(qname, separator) {
+ function parseQName(qname, separator) {
var parsed = {};
var delimiter = separator || ':';
var defaultNamespace = '_';
@@ -431,7 +432,7 @@
parsed.name = qname;
}
return parsed;
- };
+ }
this.initRedis = function() {
var method = 'initRedis';
@@ -443,10 +444,8 @@
//create a subscriber client that listens for requests to perform swap
subscriber.on('message', function (channel, message) {
- if (message === 'host0' || message === 'host1') {
- logger.info(method, message, 'set to active host in channel', channel);
- utils.activeHost = message;
- }
+ logger.info(method, message, 'set to active host in channel', channel);
+ self.activeHost = message;
});
subscriber.on('error', function (err) {
@@ -454,19 +453,19 @@
reject(err);
});
- subscriber.subscribe(utils.redisHash);
+ subscriber.subscribe(self.redisKey);
- redisClient.hgetAsync(utils.redisHash, utils.redisKey)
+ redisClient.hgetAsync(self.redisKey, self.redisField)
.then(activeHost => {
- return utils.initActiveHost(activeHost);
+ return initActiveHost(activeHost);
})
.then(() => {
process.on('SIGTERM', function onSigterm() {
- if (utils.activeHost === utils.host) {
- var redundantHost = utils.host === 'host0' ? 'host1' : 'host0';
- utils.redisClient.hsetAsync(utils.redisHash, utils.redisKey, redundantHost)
+ if (self.activeHost === self.host) {
+ var redundantHost = self.host === 'host0' ? 'host1' : 'host0';
+ self.redisClient.hsetAsync(self.redisKey, self.redisField, redundantHost)
.then(() => {
- utils.redisClient.publish(utils.redisHash, redundantHost);
+ self.redisClient.publish(self.redisKey, redundantHost);
})
.catch(err => {
logger.error(method, err);
@@ -485,18 +484,18 @@
});
};
- this.initActiveHost = function(activeHost) {
+ function initActiveHost(activeHost) {
var method = 'initActiveHost';
if (activeHost === null) {
//initialize redis key with active host
- logger.info(method, 'redis hset', utils.redisHash, utils.redisKey, utils.activeHost);
- return redisClient.hsetAsync(utils.redisHash, utils.redisKey, utils.activeHost);
+ logger.info(method, 'redis hset', self.redisKey, self.redisField, self.activeHost);
+ return redisClient.hsetAsync(self.redisKey, self.redisField, self.activeHost);
}
else {
- utils.activeHost = activeHost;
+ self.activeHost = activeHost;
return Promise.resolve();
}
- };
+ }
};
