rodrigo/genieacs
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Support caching in MongoDB instead of Redis

Browse Source
This commit is contained in:
Zaid Abdulla committed 2017-05-08 10:44:42 -07:00
1 parent c6328ce2f4
commit 624cf01270
10 files changed
+808 -528
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

.gitignore
-1
View File
@@ -2,7 +2,6 @@
lib/*.map
lib/api-functions.js
lib/auth.js
lib/cache.js
lib/common.js
lib/config.js
lib/cwmp.js
lib/api-functions.coffee
+2 -6
View File
@@ -27,6 +27,7 @@ common = require './common'
util = require 'util'
URL = require 'url'
auth = require './auth'
cache = require './cache'
udpConReq = (address, un, key, callback) ->
@@ -242,12 +243,7 @@ deleteDevice = (deviceId, callback) ->
return callback(err) if err
db.operationsCollection.remove({'_id' : {'$regex' : "^#{common.escapeRegExp(deviceId)}\\:"}}, (err) ->
return callback(err) if err
db.redisClient.del("#{deviceId}_presets_hash",
"#{deviceId}_inform_hash",
"#{deviceId}_faults",
"#{deviceId}_tasks",
"#{deviceId}_operations",
callback)
cache.del("#{deviceId}_tasks_faults_operations", callback)
)
)
)
lib/cache.coffee
-310
View File
@@ -1,310 +0,0 @@
###
# Copyright 2013-2017 Zaid Abdulla
#
# This file is part of GenieACS.
#
# GenieACS is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# GenieACS is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with GenieACS. If not, see <http://www.gnu.org/licenses/>.
###
vm = require 'vm'
crypto = require 'crypto'
later = require 'later'
db = require './db'
query = require './query'
REFRESH = 3000
nextRefresh = 0
hash = null
presets = null
provisions = null
virtualParameters = null
files = null
UNLOCK_SCRIPT = 'if redis.call("get", KEYS[1]) == ARGV[1] then return redis.call("del", KEYS[1]) else return 0 end'
EXTEND_SCRIPT = 'if redis.call("get", KEYS[1]) == ARGV[1] then return redis.call("pexpire", KEYS[1], ARGV[2]) else return 0 end'
lock = (lockName, ttl, callback) ->
token = Math.random().toString(36).slice(2)
unlockOrExtend = (ttl) ->
if not ttl
db.redisClient.eval(UNLOCK_SCRIPT, 1, lockName, token, (err, res) ->
throw err or new Error('Lock expired') if err or not res
)
else
db.redisClient.eval(EXTEND_SCRIPT, 1, lockName, token, ttl, (err, res) ->
throw err or new Error('Lock expired') if err or not res
)
return
db.redisClient.set(lockName, token, 'NX', 'PX', ttl, (err, res) ->
if err or res
return callback(err, unlockOrExtend)
setTimeout(() ->
return lock(lockName, ttl, callback)
, 200)
)
computeHash = () ->
# MD5 hash for presets, provisions, virtual parameters for detecting changes
h = crypto.createHash('md5')
for p in presets
h.update(JSON.stringify(p.name))
h.update(JSON.stringify(p.channel))
h.update(JSON.stringify(p.schedule))
h.update(JSON.stringify(p.events))
h.update(JSON.stringify(p.precondition))
h.update(JSON.stringify(p.provisions))
keys = Object.keys(provisions).sort()
h.update(JSON.stringify(keys))
h.update(provisions[k].md5) for k in keys
keys = Object.keys(virtualParameters).sort()
h.update(JSON.stringify(keys))
h.update(virtualParameters[k].md5) for k in keys
hash = h.digest('hex')
refresh = (callback) ->
now = Date.now()
db.redisClient.get('presets_hash', (err, res) ->
return callback(err) if err
if hash? and res == hash
nextRefresh = now + REFRESH
return callback()
lock('presets_hash_lock', 3000, (err, unlockOrExtend) ->
return callback(err) if err
counter = 3
counter += 2
db.presetsCollection.find().toArray((err, res) ->
if err
callback(err) if counter & 1
return counter = 0
db.objectsCollection.find().toArray((err, objects) ->
if err
callback(err) if (counter & 1)
return counter = 0
res.sort((a, b) ->
if a.weight == b.weight
return a._id > b._id
else
return a.weight - b.weight
)
presets = []
for preset in res
schedule = null
if preset.schedule
parts = preset.schedule.trim().split(/\s+/)
schedule = {
md5: crypto.createHash('md5').update(preset.schedule).digest('hex')
}
try
schedule.duration = +(parts.shift()) * 1000
parts.unshift('0') if parts.length == 5
# TODO later.js doesn't throw erorr if expression is invalid!
schedule.schedule = later.schedule(later.parse.cron(parts.join(' '), true))
catch err
# TODO show a warning
schedule.schedule = false
events = preset.events ? {}
precondition = query.convertMongoQueryToFilters(JSON.parse(preset.precondition))
_provisions = preset.provisions or []
# Generate provisions from the old configuration format
for c in preset.configurations
switch c.type
when 'age'
_provisions.push(['refresh', c.name, c.age])
when 'value'
_provisions.push(['value', c.name, c.value])
when 'add_tag'
_provisions.push(['tag', c.tag, true])
when 'delete_tag'
_provisions.push(['tag', c.tag, false])
when 'provision'
_provisions.push([c.name].concat(c.args or []))
when 'add_object'
for obj in objects
if obj['_id'] == c.object
alias = ("#{k}:#{JSON.stringify(obj[k])}" for k in obj['_keys']).join(',')
p = "#{c.name}.[#{alias}]"
_provisions.push(['instances', p, 1])
for k of obj
if k[0] != '_' and k not in obj['_keys']
_provisions.push(['value', "#{p}.#{k}", obj[k]])
when 'delete_object'
for obj in objects
if obj['_id'] == c.object
alias = ("#{k}:#{JSON.stringify(obj[k])}" for k in obj['_keys']).join(',')
p = "#{c.name}.[#{alias}]"
_provisions.push(['instances', p, 0])
else
callback(new Error("Unknown configuration type #{c.type}")) if counter & 1
return counter = 0
presets.push({name: preset._id, channel: preset.channel or 'default', schedule: schedule, events: events, precondition: precondition, provisions: _provisions})
if (counter -= 2) == 1
computeHash()
db.redisClient.setex("presets_hash", 300, hash, (err) ->
unlockOrExtend(0)
nextRefresh = now + REFRESH
return callback()
)
)
)
counter += 2
db.provisionsCollection.find().toArray((err, res) ->
if err
callback(err) if counter & 1
return counter = 0
provisions = {}
for r in res
provisions[r._id] = {}
provisions[r._id].md5 = crypto.createHash('md5').update(r.script).digest('hex')
provisions[r._id].script = new vm.Script("\"use strict\";(function(){\n#{r.script}\n})();", {filename: r._id, lineOffset: -1, timeout: 50})
if (counter -= 2) == 1
computeHash()
db.redisClient.setex("presets_hash", 300, hash, (err) ->
unlockOrExtend(0)
nextRefresh = now + REFRESH
return callback()
)
)
counter += 2
db.virtualParametersCollection.find().toArray((err, res) ->
if err
callback(err) if counter & 1
return counter = 0
virtualParameters = {}
for r in res
virtualParameters[r._id] = {}
virtualParameters[r._id].md5 = crypto.createHash('md5').update(r.script).digest('hex')
virtualParameters[r._id].script = new vm.Script("\"use strict\";(function(){\n#{r.script}\n})();", {filename: r._id, lineOffset: -1, timeout: 50})
if (counter -= 2) == 1
computeHash()
db.redisClient.setex("presets_hash", 300, hash, (err) ->
unlockOrExtend(0)
nextRefresh = now + REFRESH
return callback()
)
)
counter += 2
db.filesCollection.find().toArray((err, res) ->
if err
callback(err) if counter & 1
return counter = 0
files = {}
for r in res
id = r.filename or r._id.toString()
files[id] = {}
files[id].length = r.length
files[id].md5 = r.md5
files[id].contentType = r.contentType
if (counter -= 2) == 1
computeHash()
db.redisClient.setex("presets_hash", 300, hash, (err) ->
unlockOrExtend(0)
nextRefresh = now + REFRESH
return callback()
)
)
if (counter -= 2) == 1
computeHash()
db.redisClient.setex("presets_hash", 300, hash, (err) ->
unlockOrExtend(0)
nextRefresh = now + REFRESH
return callback()
)
)
)
getPresets = (callback) ->
if Date.now() < nextRefresh
return callback(null, hash, presets)
refresh((err) ->
return callback(err, hash, presets)
)
getProvisions = (callback) ->
if Date.now() < nextRefresh
return callback(null, hash, provisions)
refresh((err) ->
return callback(err, hash, provisions)
)
getVirtualParameters = (callback) ->
if Date.now() < nextRefresh
return callback(null, hash, virtualParameters)
refresh((err) ->
return callback(err, hash, virtualParameters)
)
getFiles = (callback) ->
if Date.now() < nextRefresh
return callback(null, hash, files)
refresh((err) ->
return callback(err, hash, files)
)
getProvisionsAndVirtualParameters = (callback) ->
if Date.now() < nextRefresh
return callback(null, hash, provisions, virtualParameters)
refresh((err) ->
return callback(err, hash, provisions, virtualParameters)
)
exports.getPresets = getPresets
exports.getProvisions = getProvisions
exports.getFiles = getFiles
exports.getProvisionsAndVirtualParameters = getProvisionsAndVirtualParameters
lib/cache.js
+257
View File
@@ -0,0 +1,257 @@
/**
* Copyright 2013-2017 Zaid Abdulla
*
* This file is part of GenieACS.
*
* GenieACS is free software: you can redistribute it and/or modify
* it under the terms of the GNU Affero General Public License as
* published by the Free Software Foundation, either version 3 of the
* License, or (at your option) any later version.
*
* GenieACS is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU Affero General Public License for more details.
*
* You should have received a copy of the GNU Affero General Public License
* along with GenieACS. If not, see <http://www.gnu.org/licenses/>.
*/
"use strict";
const config = require("./config");
const MAX_CACHE_TTL = config.get("MAX_CACHE_TTL");
let redisClient;
let mongoDb;
let mongoCollection;
let mongoTimeOffset = 0;
function connect(callback) {
const REDIS_HOST = config.get("REDIS_HOST");
if (REDIS_HOST) {
const redis = require("redis");
const REDIS_PORT = config.get("REDIS_PORT");
const REDIS_DB = config.get("REDIS_DB");
redisClient = redis.createClient(REDIS_PORT, REDIS_HOST);
redisClient.select(REDIS_DB, function(err) {
exports.get = redisGet;
exports.set = redisSet;
exports.del = redisDel;
exports.pop = redisPop;
exports.lock = redisLock;
callback(err);
});
}
else {
const mongodb = require("mongodb");
const MONGODB_CONNECTION_URL = config.get("MONGODB_CONNECTION_URL");
mongodb.MongoClient.connect(MONGODB_CONNECTION_URL, function(err, db) {
if (err)
return callback(err);
mongoDb = db;
mongoCollection = db.collection("cache");
mongoCollection.ensureIndex({"expire": 1}, {expireAfterSeconds: 0});
const now = Date.now();
mongoDb.command({hostInfo: 1}, function(err, res) {
if (err)
return callback(err);
mongoTimeOffset = res.system.currentTime.getTime() - now;
exports.get = mongoGet;
exports.set = mongoSet;
exports.del = mongoDel;
exports.pop = mongoPop;
exports.lock = mongoLock;
callback();
});
});
}
}
function disconnect() {
if (redisClient)
redisClient.quit();
if (mongoDb)
mongoDb.close();
}
function redisGet(key, callback) {
if (Array.isArray(key))
redisClient.mget(key, callback);
else
redisClient.get(key, callback);
}
function redisDel(key, callback) {
redisClient.del(key, callback);
}
function redisSet(key, value, ttl, callback) {
if (!callback && typeof v === "function") {
callback = ttl;
ttl = null;
}
if (!ttl)
ttl = MAX_CACHE_TTL;
redisClient.setex(key, ttl, value, callback);
}
function redisPop(key, callback) {
const POP_SCRIPT = "local v=redis.call(\"get\",KEYS[1]);redis.call(\"del\",KEYS[1]);return v;";
redisClient.eval(POP_SCRIPT, 1, key, callback);
}
function redisLock(lockName, ttl, callback) {
const UNLOCK_SCRIPT = "if redis.call(\"get\", KEYS[1]) == ARGV[1] then return redis.call(\"del\", KEYS[1]) else return 0 end";
const EXTEND_SCRIPT = "if redis.call(\"get\", KEYS[1]) == ARGV[1] then return redis.call(\"pexpire\", KEYS[1], ARGV[2]) else return 0 end";
const token = Math.random().toString(36).slice(2);
function unlockOrExtend(ttl) {
if (!ttl) {
redisClient.eval(UNLOCK_SCRIPT, 1, lockName, token, function(err, res) {
if (err || !res)
throw err || new Error("Lock expired");
});
}
else {
redisClient.eval(EXTEND_SCRIPT, 1, lockName, token, ttl, function(err, res) {
if (err || !res)
throw err || new Error("Lock expired");
});
}
}
redisClient.set(lockName, token, "NX", "PX", ttl, function(err, res) {
if (err || res)
return callback(err, unlockOrExtend);
setTimeout(function() {
redisLock(lockName, ttl, callback);
}, 200);
});
}
function mongoGet(key, callback) {
let expire = new Date(Date.now() - mongoTimeOffset);
if (Array.isArray(key))
mongoCollection.find({"_id": {"$in": key}}).toArray(function(err, res) {
if (err)
return callback(err);
let indices = {};
key.forEach(function(v, i) {
indices[v] = i;
});
let values = [];
res.forEach(function(r) {
if (r["expire"] > expire)
values[indices[r["_id"]]] = r["value"];
});
callback(null, values);
});
else
mongoCollection.findOne({"_id": {"$in": [key]}}, function(err, res) {
if (err || !res)
return callback(err);
if (res["expire"] > expire)
return callback(null, res["value"]);
callback();
});
}
function mongoDel(key, callback) {
if (Array.isArray(key))
mongoCollection.remove({"_id": {"$in": key}}, callback);
else
mongoCollection.remove({"_id": key}, callback);
}
function mongoSet(key, value, ttl, callback) {
if (!callback && typeof v === "function") {
callback = ttl;
ttl = null;
}
if (!ttl)
ttl = MAX_CACHE_TTL;
let expire = new Date((Date.now() - mongoTimeOffset) + (ttl * 1000));
mongoCollection.save({"_id": key, "value": value, "expire": expire}, callback);
}
function mongoPop(key, callback) {
mongoCollection.findAndModify({"_id": key}, null, null, {remove: true}, function(err, res) {
if (err || !res["value"])
return callback(err);
if (res["value"]["expire"] > new Date(Date.now() - mongoTimeOffset))
return callback(null, res["value"]["value"]);
callback();
});
}
function mongoLock(lockName, ttl, callback) {
const token = Math.random().toString(36).slice(2);
function unlockOrExtend(ttl) {
if (!ttl) {
mongoCollection.remove({"_id": lockName, "value": token}, function(err, res) {
if (err || res["result"]["n"] !== 1)
throw err || new Error("Lock expired");
});
}
else {
let expire = new Date((Date.now() - mongoTimeOffset) + (ttl * 1000));
mongoCollection.update(
{"_id": lockName, "value": token},
{"expire": expire},
function(err, res) {
if (err || res["result"]["n"] !== 1)
throw err | new Error("Lock expired");
}
);
}
}
let expireTest = new Date(Date.now() - mongoTimeOffset);
let expireSet = new Date((Date.now() - mongoTimeOffset) + (ttl * 1000));
mongoCollection.update(
{"_id": lockName, "expire": {"$lte": expireTest}},
{"$set": {"value" : token, "expire": expireSet}},
{upsert: true},
function(err, res) {
if (err && err.code === 11000)
return setTimeout(function() {mongoLock(lockName, ttl, callback)}, 200);
return callback(err, unlockOrExtend);
}
);
}
exports.connect = connect;
exports.disconnect = disconnect;
lib/cwmp.coffee
+152 -38
View File
@@ -28,6 +28,7 @@ session = require './session'
query = require './query'
device = require './device'
cache = require './cache'
localCache = require './local-cache'
db = require './db'
logger = require './logger'
@@ -213,7 +214,7 @@ appendProvisions = (original, toAppend) ->
applyPresets = (sessionContext) ->
cache.getPresets((err, presetsHash, presets) ->
localCache.getPresets((err, presetsHash, presets) ->
return throwError(err, sessionContext.httpResponse) if err
# Filter presets based on existing faults
@@ -437,50 +438,100 @@ nextRpc = (sessionContext) ->
)
endSession = (sessionContext, callback) ->
delete currentSessions.get(sessionContext.httpRequest.connection)[sessionContext.sessionId]
saveCache = sessionContext.cacheUntil?
counter = 3
counter += 2
db.saveDevice(sessionContext.deviceId, sessionContext.deviceData, sessionContext.new, (err) ->
if err
callback(err) if counter & 1
return counter = 0
return callback(null, sessionContext.new) if (counter -= 2) == 1
)
for k of sessionContext.operationsTouched
counter += 2
saveCache = true
if sessionContext.operations[k]?
db.saveOperation(sessionContext.deviceId, k, sessionContext.operations[k], (err) ->
if err
callback(err) if counter & 1
return counter = 0
return callback(null, sessionContext.new) if (counter -= 2) == 1
)
else
db.deleteOperation(sessionContext.deviceId, k, (err) ->
if err
callback(err) if counter & 1
return counter = 0
return callback(null, sessionContext.new) if (counter -= 2) == 1
)
if sessionContext.doneTasks?.length
counter += 2
saveCache = true
db.clearTasks(sessionContext.deviceId, sessionContext.doneTasks, (err) ->
if err
callback(err) if counter & 1
return counter = 0
return callback(null, sessionContext.new) if (counter -= 2) == 1
)
for k of sessionContext.faultsTouched
counter += 2
saveCache = true
if sessionContext.faults[k]
sessionContext.faults[k].retries = sessionContext.retries[k]
db.saveFault(sessionContext.deviceId, k, sessionContext.faults[k], (err) ->
if err
callback(err) if counter & 1
return counter = 0
return callback(null, sessionContext.new) if (counter -= 2) == 1
)
else
db.deleteFault(sessionContext.deviceId, k, (err) ->
if err
callback(err) if counter & 1
return counter = 0
return callback(null, sessionContext.new) if (counter -= 2) == 1
)
if saveCache
counter += 2
cacheDueTasksAndFaultsAndOperations(sessionContext.deviceId, sessionContext.tasks, sessionContext.faults, sessionContext.operations, sessionContext.cacheUntil, (err) ->
if err
callback(err) if counter & 1
return counter = 0
return callback(null, sessionContext.new) if (counter -= 2) == 1
)
callback(null, sessionContext.new) if (counter -= 2) == 1
sendAcsRequest = (sessionContext, id, acsRequest) ->
if not acsRequest?
session.end(sessionContext, (err, isNew) ->
return endSession(sessionContext, (err, isNew) ->
return throwError(err, sessionContext.httpResponse) if err
delete currentSessions.get(sessionContext.httpRequest.connection)[sessionContext.sessionId]
if isNew
logger.accessInfo({
sessionContext: sessionContext
message: 'New device registered'
})
db.clearTasks(sessionContext.deviceId, sessionContext.doneTasks, (err) ->
return throwError(err, sessionContext.httpResponse) if err
counter = 3
for k of sessionContext.faultsTouched
counter += 2
if sessionContext.faults[k]
sessionContext.faults[k].retries = sessionContext.retries[k]
db.saveFault(sessionContext.deviceId, k, sessionContext.faults[k], (err) ->
if err
throwError(err, sessionContext.httpResponse) if counter & 1
return counter = 0
if (counter -= 2) == 1
return writeResponse(sessionContext, soap.response(null))
writeResponse(sessionContext, soap.response(null))
)
else
db.deleteFault(sessionContext.deviceId, k, (err) ->
if err
throwError(err, sessionContext.httpResponse) if counter & 1
return counter = 0
if (counter -= 2) == 1
return writeResponse(sessionContext, soap.response(null))
)
if (counter -= 2) == 1
return writeResponse(sessionContext, soap.response(null))
)
)
return
if acsRequest.name is 'Download'
if not acsRequest.url?
@@ -493,7 +544,7 @@ sendAcsRequest = (sessionContext, id, acsRequest) ->
acsRequest.url += "/#{encodeURIComponent(acsRequest.fileName)}"
if not acsRequest.fileSize?
return cache.getFiles((err, hash, files) ->
return localCache.getFiles((err, hash, files) ->
return throwError(err, sessionContext.httpResponse) if err
if acsRequest.fileName of files
@@ -535,7 +586,7 @@ getSession = (httpRequest, callback) ->
return callback(null, sessionContext)
setTimeout(() ->
db.redisClient.eval('local v=redis.call("get",KEYS[1]);redis.call("del",KEYS[1]);return v;', 1, "session_#{sessionId}", (err, sessionContextString) ->
cache.pop("session_#{sessionId}", (err, sessionContextString) ->
return callback(err) if err or not sessionContextString
session.deserialize(sessionContextString, (err, sessionContext) ->
return callback(err) if err
@@ -560,13 +611,75 @@ onConnection = (socket) ->
session.serialize(sessionContext, (err, sessionContextString) ->
return throwError(err) if err
# TODO don't set if process is shutting down
db.redisClient.setex("session_#{sessionId}", sessionContext.timeout, sessionContextString, (err) ->
cache.set("session_#{sessionId}", sessionContextString, sessionContext.timeout, (err) ->
return throwError(err) if err
)
)
)
getDueTasksAndFaultsAndOperations = (deviceId, timestamp, callback) ->
cache.get("#{deviceId}_tasks_faults_operations", (err, res) ->
return callback(err) if err
if res
res = JSON.parse(res)
return callback(null, res.tasks or [], res.faults or {}, res.operations or {})
faults = tasks = operations = cacheUntil = null
db.getFaults(deviceId, (err, _faults) ->
if err
callback?(err)
return callback = null
faults = _faults
if tasks? and operations?
return callback(null, tasks, faults, operations, cacheUntil)
)
db.getDueTasks(deviceId, timestamp, (err, dueTasks, nextTimestamp) ->
if err
callback?(err)
return callback = null
tasks = dueTasks
cacheUntil = nextTimestamp or 0
if faults? and operations?
return callback(null, tasks, faults, operations, cacheUntil)
)
db.getOperations(deviceId, (err, _operations) ->
if err
callback?(err)
return callback = null
operations = _operations
if faults? and tasks?
return callback(null, tasks, faults, operations, cacheUntil)
)
if faults? and tasks? and operations?
return callback(null, tasks, faults, operations, cacheUntil)
)
cacheDueTasksAndFaultsAndOperations = (deviceId, tasks, faults, operations, cacheUntil, callback) ->
v = {}
v.tasks = tasks if tasks.length
v.faults = faults if Object.keys(faults).length
v.operations = operations if Object.keys(operations).length
if cacheUntil
ttl = Math.trunc((Date.now() - cacheUntil) / 1000)
else
ttl = config.get('MAX_CACHE_TTL', deviceId)
cache.set("#{deviceId}_tasks_faults_operations", JSON.stringify(v), ttl, callback)
listener = (httpRequest, httpResponse) ->
if httpRequest.method != 'POST'
httpResponse.writeHead 405, {'Allow': 'POST'}
@@ -626,7 +739,7 @@ listener = (httpRequest, httpResponse) ->
httpRequest.connection.setTimeout(sessionContext.timeout * 1000)
db.getDueTasksAndFaultsAndOperations(deviceId, sessionContext.timestamp, (err, dueTasks, faults, operations) ->
getDueTasksAndFaultsAndOperations(deviceId, sessionContext.timestamp, (err, dueTasks, faults, operations, cacheUntil) ->
return throwError(err, httpResponse) if err
sessionContext.tasks = dueTasks
sessionContext.faults = faults
@@ -634,6 +747,7 @@ listener = (httpRequest, httpResponse) ->
for k, v of faults
sessionContext.retries[k] = v.retries
sessionContext.operations = operations
sessionContext.cacheUntil = cacheUntil
# Delete expired faults
for k, v of sessionContext.faults
lib/db.coffee
+35 -108
View File
@@ -19,10 +19,8 @@
config = require './config'
mongodb = require 'mongodb'
redis = require('redis')
common = require './common'
redisClient = null
tasksCollection = null
devicesCollection = null
presetsCollection = null
@@ -35,7 +33,7 @@ operationsCollection = null
connect = (callback) ->
callbackCounter = 10
callbackCounter = 9
mongodb.MongoClient.connect(config.get('MONGODB_CONNECTION_URL'), (err, db) ->
return callback(err) if err
exports.mongoDb = db
@@ -105,18 +103,11 @@ connect = (callback) ->
return callback(err)
)
exports.redisClient = redisClient = redis.createClient(config.get('REDIS_PORT'), config.get('REDIS_HOST'))
redisClient.select(config.get('REDIS_DB'), (err) ->
if --callbackCounter == 0 or err
callbackCounter = 0
return callback(err)
)
)
disconnect = () ->
exports.mongoDb?.close()
redisClient?.quit()
# Optimize projection by removing overlaps
@@ -461,71 +452,6 @@ saveDevice = (deviceId, deviceData, isNew, callback) ->
)
getDueTasksAndFaultsAndOperations = (deviceId, timestamp, callback) ->
redisClient.mget("#{deviceId}_tasks", "#{deviceId}_faults", "#{deviceId}_operations", (err, res) ->
return callback(err) if err
if res[0]
tasks = JSON.parse(res[0])
if res[1]?
faults = JSON.parse(res[1])
if res[2]?
operations = JSON.parse(res[2])
if tasks? and faults? and operations?
return callback(null, tasks, faults, operations)
MAX_CACHE_TTL = config.get('MAX_CACHE_TTL', deviceId)
if not faults?
getFaults(deviceId, (err, flts) ->
if err
callback?(err)
return callback = null
faults = flts
if tasks? and operations?
return callback(null, tasks, faults, operations)
)
if not tasks?
getDueTasks(deviceId, timestamp, (err, dueTasks, nextTimestamp) ->
if err
callback?(err)
return callback = null
if nextTimestamp?
exp = Math.min(0, Math.trunc((nextTimestamp - Date.now()) / 1000))
else
exp = MAX_CACHE_TTL
redisClient.setex("#{deviceId}_tasks", exp, JSON.stringify(dueTasks), (err) ->
if err
callback?(err)
return callback = null
tasks = dueTasks
if faults? and operations?
return callback(null, tasks, faults, operations)
)
)
if not operations?
getOperations(deviceId, (err, _operations) ->
if err
callback?(err)
return callback = null
operations = _operations
if tasks? and faults?
return callback(null, tasks, faults, operations)
)
)
getFaults = (deviceId, callback) ->
faultsCollection.find({'_id' : {'$regex' : "^#{common.escapeRegExp(deviceId)}\\:"}}).toArray((err, res) ->
return callback(err) if err
@@ -540,32 +466,22 @@ getFaults = (deviceId, callback) ->
r.provisions = JSON.parse(r.provisions)
faults[channel] = r
MAX_CACHE_TTL = config.get('MAX_CACHE_TTL', deviceId)
redisClient.setex("#{deviceId}_faults", MAX_CACHE_TTL, JSON.stringify(faults), (err) ->
callback(err, faults)
)
)
saveFault = (deviceId, channel, fault, callback) ->
fault = Object.assign({}, fault)
fault._id = "#{deviceId}:#{channel}"
fault.device = deviceId
fault.channel = channel
fault.timestamp = new Date(fault.timestamp)
fault.provisions = JSON.stringify(fault.provisions)
faultsCollection.save(fault, (err) ->
return callback(err) if err
redisClient.del("#{deviceId}_faults", callback)
)
faultsCollection.save(fault, callback)
deleteFault = (deviceId, channel, callback) ->
faultsCollection.remove({_id: "#{deviceId}:#{channel}"}, (err) ->
return callback(err) if err
redisClient.del("#{deviceId}_faults", callback)
)
faultsCollection.remove({_id: "#{deviceId}:#{channel}"}, callback)
getDueTasks = (deviceId, timestamp, callback) ->
@@ -608,13 +524,7 @@ getDueTasks = (deviceId, timestamp, callback) ->
clearTasks = (deviceId, taskIds, callback) ->
if not taskIds?.length
return callback()
tasksCollection.remove({'_id' : {'$in' : (new mongodb.ObjectID(id) for id in taskIds)}}, (err) ->
return callback(err) if err
redisClient.del("#{deviceId}_tasks", callback)
)
tasksCollection.remove({'_id' : {'$in' : (new mongodb.ObjectID(id) for id in taskIds)}}, callback)
getOperations = (deviceId, callback) ->
@@ -631,41 +541,58 @@ getOperations = (deviceId, callback) ->
r.retries = JSON.parse(r.retries)
operations[commandKey] = r
MAX_CACHE_TTL = config.get('MAX_CACHE_TTL', deviceId)
redisClient.setex("#{deviceId}_operations", MAX_CACHE_TTL, JSON.stringify(operations), (err) ->
callback(err, operations)
)
)
saveOperation = (deviceId, commandKey, operation, callback) ->
operation = Object.assign({}, operation)
operation._id = "#{deviceId}:#{commandKey}"
operation.timestamp = new Date(operation.timestamp)
operation.provisions = JSON.stringify(operation.provisions)
operation.retries = JSON.stringify(operation.retries)
operation.args = JSON.stringify(operation.args)
operationsCollection.save(operation, (err) ->
return callback(err) if err
redisClient.del("#{deviceId}_operations", callback)
)
operationsCollection.save(operation, callback)
deleteOperation = (deviceId, commandKey, callback) ->
operationsCollection.remove({_id: "#{deviceId}:#{commandKey}"}, (err) ->
return callback(err) if err
redisClient.del("#{deviceId}_operations", callback)
)
operationsCollection.remove({_id: "#{deviceId}:#{commandKey}"}, callback)
getPresets = (callback) ->
presetsCollection.find().toArray(callback)
getObjects = (callback) ->
objectsCollection.find().toArray(callback)
getProvisions = (callback) ->
provisionsCollection.find().toArray(callback)
getVirtualParameters = (callback) ->
virtualParametersCollection.find().toArray(callback)
getFiles = (callback) ->
filesCollection.find().toArray(callback)
exports.connect = connect
exports.disconnect = disconnect
exports.fetchDevice = fetchDevice
exports.saveDevice = saveDevice
exports.getDueTasksAndFaultsAndOperations = getDueTasksAndFaultsAndOperations
exports.getFaults = getFaults
exports.saveFault = saveFault
exports.deleteFault = deleteFault
exports.clearTasks = clearTasks
exports.saveOperation = saveOperation
exports.deleteOperation = deleteOperation
exports.getPresets = getPresets
exports.getObjects = getObjects
exports.getProvisions = getProvisions
exports.getVirtualParameters = getVirtualParameters
exports.getFiles = getFiles
exports.getDueTasks = getDueTasks
exports.getOperations = getOperations
lib/local-cache.js
+339
View File
@@ -0,0 +1,339 @@
/**
* Copyright 2013-2017 Zaid Abdulla
*
* This file is part of GenieACS.
*
* GenieACS is free software: you can redistribute it and/or modify
* it under the terms of the GNU Affero General Public License as
* published by the Free Software Foundation, either version 3 of the
* License, or (at your option) any later version.
*
* GenieACS is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU Affero General Public License for more details.
*
* You should have received a copy of the GNU Affero General Public License
* along with GenieACS. If not, see <http://www.gnu.org/licenses/>.
*/
"use strict";
const vm = require("vm");
const crypto = require("crypto");
const later = require("later");
const db = require("./db");
const cache = require("./cache");
const query = require("./query");
const logger = require("./logger");
const REFRESH = 3000;
let nextRefresh = 1;
let hash = null;
let presets, provisions, virtualParameters, files;
function computeHash() {
// MD5 hash for presets, provisions, virtual parameters for detecting changes
let h = crypto.createHash("md5");
for (let p of presets) {
h.update(JSON.stringify(p.name));
h.update(JSON.stringify(p.channel));
h.update(JSON.stringify(p.schedule));
h.update(JSON.stringify(p.events));
h.update(JSON.stringify(p.precondition));
h.update(JSON.stringify(p.provisions));
}
let keys;
keys = Object.keys(provisions).sort();
h.update(JSON.stringify(keys));
for (let k of keys)
h.update(provisions[k].md5);
keys = Object.keys(virtualParameters).sort();
h.update(JSON.stringify(keys));
for (let k of keys)
h.update(virtualParameters[k].md5);
hash = h.digest("hex");
}
function refresh(callback) {
if (!nextRefresh)
return setTimeout(function() {refresh(callback);}, 20);
nextRefresh = 0;
const now = Date.now();
cache.get("presets_hash", function(err, res) {
if (err)
return callback(err);
if (hash && res === hash) {
nextRefresh = now + (REFRESH - (now % REFRESH));
return callback();
}
cache.lock("presets_hash_lock", 3000, function(err, unlockOrExtend) {
if (err)
return callback(err);
let counter = 3;
counter += 2;
db.getPresets(function(err, res) {
if (err) {
if (counter & 1)
callback(err);
return counter = 0;
}
db.getObjects(function(err, objects) {
if (err) {
if (counter & 1)
callback(err);
return counter = 0;
}
res.sort((a, b) => {
if (a.weight === b.weight)
return a._id > b._id;
else
return a.weight - b.weight;
});
presets = [];
for (let preset of res) {
let schedule = null;
if (preset.schedule) {
let parts = preset.schedule.trim().split(/\s+/);
schedule = {
md5: crypto.createHash("md5").update(preset.schedule).digest("hex")
};
try {
schedule.duration = +(parts.shift()) * 1000;
if (parts.length === 5) {
parts.unshift("0");
// TODO later.js doesn't throw erorr if expression is invalid!
schedule.schedule = later.schedule(later.parse.cron(parts.join(" "), true));
}
}
catch (err) {
logger.warn({
message: "Invalid preset schedule",
preset: preset._id,
schedule: preset.schedule
});
schedule.schedule = false;
}
}
let events = preset.events || {};
let precondition = query.convertMongoQueryToFilters(JSON.parse(preset.precondition));
let _provisions = preset.provisions || [];
// Generate provisions from the old configuration format
for (let c of preset.configurations) {
switch (c.type) {
case "age":
_provisions.push(["refresh", c.name, c.age]);
break;
case "value":
_provisions.push(["value", c.name, c.value]);
break;
case "add_tag":
_provisions.push(["tag", c.tag, true]);
break;
case "delete_tag":
_provisions.push(["tag", c.tag, false]);
break;
case "provision":
_provisions.push([c.name].concat(c.args || []));
break;
case "add_object":
for (let obj of objects) {
if (obj["_id"] === c.object) {
let alias = obj["_keys"].map((k) => `${k}:${JSON.stringify(obj[k])}`).join(",");
let p = `${c.name}.[${alias}]`;
_provisions.push(["instances", p, 1]);
for (let k in obj)
if (k[0] !== "_" && !(obj["_keys"]))
_provisions.push(["value", `${p}.${k}`, obj[k]]);
}
}
break;
case "delete_object":
for (let obj of objects) {
if (obj["_id"] === c.object) {
let alias = obj["_keys"].map((k) => `${k}:${JSON.stringify(obj[k])}`).join(",");
let p = `${c.name}.[${alias}]`;
_provisions.push(["instances", p, 0]);
}
}
break;
default:
if (counter & 1)
callback(new Error(`Unknown configuration type ${c.type}`));
return counter = 0;
break;
}
}
presets.push({
name: preset._id,
channel: preset.channel || "default",
schedule: schedule,
events: events,
precondition: precondition,
provisions: _provisions
});
}
if ((counter -= 2) === 1) {
computeHash();
cache.set("presets_hash", hash, 300, function(err) {
unlockOrExtend(0);
nextRefresh = now + (REFRESH - (now % REFRESH));
callback(err);
});
}
});
});
counter += 2;
db.getProvisions(function(err, res) {
if (err) {
if (counter & 1)
callback(err);
return counter = 0;
}
provisions = {};
for (let r of res) {
provisions[r._id] = {};
provisions[r._id].md5 = crypto.createHash("md5").update(r.script).digest("hex");
provisions[r._id].script = new vm.Script(`"use strict";(function(){\n${r.script}\n})();`, {filename: r._id, lineOffset: -1, timeout: 50});
}
if ((counter -= 2) === 1) {
computeHash();
cache.set("presets_hash", hash, 300, function(err) {
unlockOrExtend(0);
nextRefresh = now + (REFRESH - (now % REFRESH));
callback(err);
});
}
});
counter += 2;
db.getVirtualParameters(function(err, res) {
if (err) {
if (counter & 1)
callback(err);
return counter = 0;
}
virtualParameters = {};
for (let r of res) {
virtualParameters[r._id] = {}
virtualParameters[r._id].md5 = crypto.createHash("md5").update(r.script).digest("hex");
virtualParameters[r._id].script = new vm.Script(`"use strict";(function(){\n${r.script}\n})();`, {filename: r._id, lineOffset: -1, timeout: 50});
}
if ((counter -= 2) === 1) {
computeHash();
cache.set("presets_hash", hash, 300, function(err) {
unlockOrExtend(0);
nextRefresh = now + (REFRESH - (now % REFRESH));
callback(err);
});
}
});
counter += 2;
db.getFiles(function(err, res) {
if (err) {
if (counter & 1)
callback(err);
return counter = 0;
}
files = {};
for (let r of res) {
const id = r.filename || r._id.toString();
files[id] = {};
files[id].length = r.length;
files[id].md5 = r.md5;
files[id].contentType = r.contentType;
}
if ((counter -= 2) === 1) {
computeHash();
cache.set("presets_hash", hash, 300, function(err) {
unlockOrExtend(0);
nextRefresh = now + (REFRESH - (now % REFRESH));
callback(err);
});
}
});
if ((counter -= 2) === 1) {
computeHash();
cache.set("presets_hash", hash, 300, function(err) {
unlockOrExtend(0);
nextRefresh = now + (REFRESH - (now % REFRESH));
callback(err);
});
}
});
});
}
function getPresets(callback) {
if (Date.now() < nextRefresh)
return callback(null, hash, presets);
refresh(function(err) {
callback(err, hash, presets);
});
}
function getFiles(callback) {
if (Date.now() < nextRefresh)
return callback(null, hash, files);
refresh(function(err) {
callback(err, hash, files);
})
}
function getProvisionsAndVirtualParameters(callback) {
if (Date.now() < nextRefresh)
return callback(null, hash, provisions, virtualParameters);
refresh(function(err) {
callback(err, hash, provisions, virtualParameters);
});
}
exports.getPresets = getPresets;
exports.getFiles = getFiles;
exports.getProvisionsAndVirtualParameters = getProvisionsAndVirtualParameters;
lib/nbi.coffee
+14 -19
View File
@@ -50,6 +50,7 @@ common = require './common'
db = require './db'
query = require './query'
apiFunctions = require './api-functions'
cache = require './cache'
VERSION = require('../package.json').version
@@ -108,7 +109,7 @@ listener = (request, response) ->
db.presetsCollection.save(preset, (err) ->
return throwError(err, response) if err
db.redisClient.del('presets_hash', (err) ->
cache.del('presets_hash', (err) ->
return throwError(err, response) if err
response.writeHead(200)
response.end()
@@ -117,7 +118,7 @@ listener = (request, response) ->
else if request.method == 'DELETE'
db.presetsCollection.remove({'_id' : presetName}, (err) ->
return throwError(err, response) if err
db.redisClient.del('presets_hash', (err) ->
cache.del('presets_hash', (err) ->
return throwError(err, response) if err
response.writeHead(200)
response.end()
@@ -134,7 +135,7 @@ listener = (request, response) ->
db.objectsCollection.save(object, (err) ->
return throwError(err, response) if err
db.redisClient.del('objects', 'presets_hash', (err) ->
cache.del('presets_hash', (err) ->
return throwError(err, response) if err
response.writeHead(200)
response.end()
@@ -143,7 +144,7 @@ listener = (request, response) ->
else if request.method == 'DELETE'
db.objectsCollection.remove({'_id' : objectName}, (err) ->
return throwError(err, response) if err
db.redisClient.del('objects', 'presets_hash', (err) ->
cache.del('presets_hash', (err) ->
return throwError(err, response) if err
response.writeHead(200)
response.end()
@@ -168,7 +169,7 @@ listener = (request, response) ->
db.provisionsCollection.save(object, (err) ->
return throwError(err, response) if err
db.redisClient.del('presets_hash', (err) ->
cache.del('presets_hash', (err) ->
return throwError(err, response) if err
response.writeHead(200)
response.end()
@@ -177,7 +178,7 @@ listener = (request, response) ->
else if request.method == 'DELETE'
db.provisionsCollection.remove({'_id' : provisionName}, (err) ->
return throwError(err, response) if err
db.redisClient.del('presets_hash', (err) ->
cache.del('presets_hash', (err) ->
return throwError(err, response) if err
response.writeHead(200)
response.end()
@@ -202,7 +203,7 @@ listener = (request, response) ->
db.virtualParametersCollection.save(object, (err) ->
return throwError(err, response) if err
db.redisClient.del('presets_hash', (err) ->
cache.del('presets_hash', (err) ->
return throwError(err, response) if err
response.writeHead(200)
response.end()
@@ -211,7 +212,7 @@ listener = (request, response) ->
else if request.method == 'DELETE'
db.virtualParametersCollection.remove({'_id' : virtualParameterName}, (err) ->
return throwError(err, response) if err
db.redisClient.del('presets_hash', (err) ->
cache.del('presets_hash', (err) ->
return throwError(err, response) if err
response.writeHead(200)
response.end()
@@ -226,22 +227,16 @@ listener = (request, response) ->
tag = querystring.unescape(r[2])
if request.method == 'POST'
db.devicesCollection.update({'_id' : deviceId}, {'$addToSet' : {'_tags' : tag}}, {safe: true}, (err) ->
return throwError(err, response) if err
db.redisClient.del("#{deviceId}_presets_hash", (err) ->
return throwError(err, response) if err
response.writeHead(200)
response.end()
)
)
else if request.method == 'DELETE'
db.devicesCollection.update({'_id' : deviceId}, {'$pull' : {'_tags' : tag}}, {safe: true}, (err) ->
return throwError(err, response) if err
db.redisClient.del("#{deviceId}_presets_hash", (err) ->
return throwError(err, response) if err
response.writeHead(200)
response.end()
)
)
else
response.writeHead 405, {'Allow': 'POST, DELETE'}
response.end('405 Method Not Allowed')
@@ -255,14 +250,14 @@ listener = (request, response) ->
if channel.startsWith('task_')
return db.tasksCollection.remove({_id : new mongodb.ObjectID(channel.slice(5))}, (err) ->
return throwError(err, response) if err
db.redisClient.del("#{deviceId}_tasks", "#{deviceId}_faults", (err) ->
cache.del("#{deviceId}_tasks_faults_operations", (err) ->
return throwError(err, response) if err
response.writeHead(200)
response.end()
)
)
db.redisClient.del("#{deviceId}_faults", (err) ->
cache.del("#{deviceId}_tasks_faults_operations", (err) ->
return throwError(err, response) if err
response.writeHead(200)
response.end()
@@ -279,7 +274,7 @@ listener = (request, response) ->
task.device = deviceId
apiFunctions.insertTasks(task, (err) ->
return throwError(err, response) if err
db.redisClient.del("#{deviceId}_tasks", (err) ->
cache.del("#{deviceId}_tasks_faults_operations", (err) ->
return throwError(err, response) if err
if urlParts.query.connection_request?
@@ -340,7 +335,7 @@ listener = (request, response) ->
return throwError(err, response) if err
db.faultsCollection.remove({_id : "#{deviceId}:task_#{String(taskId)}"}, (err) ->
return throwError(err, response) if err
db.redisClient.del("#{deviceId}_tasks", "#{deviceId}_faults", (err) ->
cache.del("#{deviceId}_tasks_faults_operations", (err) ->
return throwError(err, response) if err
response.writeHead(200)
response.end()
@@ -358,7 +353,7 @@ listener = (request, response) ->
deviceId = task.device
db.faultsCollection.remove({_id : "#{deviceId}:task_#{String(taskId)}"}, (err) ->
return throwError(err, response) if err
db.redisClient.del("#{deviceId}_faults", (err) ->
cache.del("#{deviceId}_tasks_faults_operations", (err) ->
return throwError(err, response) if err
response.writeHead(200)
response.end()
lib/server.coffee
+6
View File
@@ -27,6 +27,7 @@ logger = require './logger'
logger.init(service, require('../package.json').version)
config = require './config'
db = require './db'
cache = require './cache'
extensions = require './extensions'
networkInterface = config.get("#{service.toUpperCase()}_INTERFACE")
@@ -47,12 +48,14 @@ exit = () ->
if not server
db.disconnect()
cache.disconnect()
extensions.killAll()
logger.close()
return
server.close(() ->
db.disconnect()
cache.disconnect()
extensions.killAll()
logger.close()
)
@@ -90,9 +93,12 @@ else
server.on('connection', onConnection) if onConnection?
db.connect((err) ->
throw err if err
cache.connect((err) ->
throw err if err
server.listen(port, networkInterface)
)
)
process.on('SIGINT', exit)
lib/session.coffee
+3 -46
View File
@@ -22,7 +22,7 @@ common = require './common'
db = require './db'
device = require './device'
sandbox = require './sandbox'
cache = require './cache'
localCache = require './local-cache'
PathSet = require './path-set'
VersionedMap = require './versioned-map'
InstanceSet = require './instance-set'
@@ -65,7 +65,7 @@ init = (deviceId, cwmpVersion, timeout, callback) ->
declarations: []
}
cache.getProvisionsAndVirtualParameters((err, hash, provisions, virtualParameters) ->
localCache.getProvisionsAndVirtualParameters((err, hash, provisions, virtualParameters) ->
return callback(err) if err
sessionContext.presetsHash = hash
@@ -1515,7 +1515,7 @@ rpcFault = (sessionContext, id, faultResponse, callback) ->
deserialize = (sessionContextString, callback) ->
cache.getProvisionsAndVirtualParameters((err, hash, provisions, virtualParameters) ->
localCache.getProvisionsAndVirtualParameters((err, hash, provisions, virtualParameters) ->
return callback(err) if err
sessionContext = JSON.parse(sessionContextString)
@@ -1581,48 +1581,6 @@ serialize = (sessionContext, callback) ->
return callback(null, sessionContextString)
end = (sessionContext, callback) ->
counter = 3
counter += 2
db.saveDevice(sessionContext.deviceId, sessionContext.deviceData, sessionContext.new, (err) ->
if err
callback(err) if counter & 1
return counter = 0
return callback(null, sessionContext.new) if (counter -= 2) == 1
)
counter += 2
db.redisClient.del("session_#{sessionContext.id}", (err) ->
if err
callback(err) if counter & 1
return counter = 0
return callback(null, sessionContext.new) if (counter -= 2) == 1
)
for k of sessionContext.operationsTouched
counter += 2
if sessionContext.operations[k]?
db.saveOperation(sessionContext.deviceId, k, sessionContext.operations[k], (err) ->
if err
callback(err) if counter & 1
return counter = 0
return callback(null, sessionContext.new) if (counter -= 2) == 1
)
else
db.deleteOperation(sessionContext.deviceId, k, (err) ->
if err
callback(err) if counter & 1
return counter = 0
return callback(null, sessionContext.new) if (counter -= 2) == 1
)
callback(null, sessionContext.new) if (counter -= 2) == 1
exports.init = init
exports.timeoutOperations = timeoutOperations
exports.inform = inform
@@ -1632,6 +1590,5 @@ exports.clearProvisions = clearProvisions
exports.rpcRequest = rpcRequest
exports.rpcResponse = rpcResponse
exports.rpcFault = rpcFault
exports.end = end
exports.serialize = serialize
exports.deserialize = deserialize