Load config from DB
This commit is contained in:
1 parent
4530967b6b
commit
819e8a52a3
4 files changed
+203
-65
No files matched your search
+76
-40
@@ -57,7 +57,7 @@ function writeResponse(sessionContext, res, close) {
|
||||
// Close connection after last request in session
|
||||
if (close) res.headers["Connection"] = "close";
|
||||
|
||||
if (config.get("DEBUG", sessionContext.deviceId)) {
|
||||
if (sessionContext.debug) {
|
||||
const dump =
|
||||
`# RESPONSE ${new Date(Date.now())}\n` +
|
||||
JSON.stringify(res.headers) +
|
||||
@@ -195,7 +195,12 @@ function inform(sessionContext, rpc) {
|
||||
cwmpVersion: sessionContext.cwmpVersion
|
||||
});
|
||||
|
||||
const cookiesPath = config.get("COOKIES_PATH", sessionContext.deviceId);
|
||||
const cookiesPath = localCache.getConfig(
|
||||
sessionContext.cacheSnapshot,
|
||||
"cwmp.cookiesPath",
|
||||
sessionContext.configContext
|
||||
);
|
||||
|
||||
if (cookiesPath) {
|
||||
res.headers["Set-Cookie"] = `session=${
|
||||
sessionContext.sessionId
|
||||
@@ -284,7 +289,11 @@ function applyPresets(sessionContext) {
|
||||
const blackList = {};
|
||||
let whiteList = null;
|
||||
let whiteListProvisions = null;
|
||||
const RETRY_DELAY = config.get("RETRY_DELAY", sessionContext.deviceId);
|
||||
const RETRY_DELAY = localCache.getConfig(
|
||||
sessionContext.cacheSnapshot,
|
||||
"cwmp.retryDelay",
|
||||
sessionContext.configContext
|
||||
);
|
||||
|
||||
if (sessionContext.faults) {
|
||||
for (const [channel, fault] of Object.entries(sessionContext.faults)) {
|
||||
@@ -1062,7 +1071,7 @@ function listener(httpRequest, httpResponse) {
|
||||
logger.accessWarn(w);
|
||||
}
|
||||
|
||||
if (config.get("DEBUG", sessionContext.deviceId)) {
|
||||
if (sessionContext.debug) {
|
||||
const dump =
|
||||
`# REQUEST ${new Date(Date.now())}\n` +
|
||||
JSON.stringify(httpRequest.headers) +
|
||||
@@ -1165,45 +1174,72 @@ function listener(httpRequest, httpResponse) {
|
||||
stats.initiatedSessions += 1;
|
||||
const deviceId = common.generateDeviceId(rpc.cpeRequest.deviceId);
|
||||
|
||||
session.init(
|
||||
deviceId,
|
||||
rpc.cwmpVersion,
|
||||
rpc.sessionTimeout || config.get("SESSION_TIMEOUT", deviceId),
|
||||
(err, _sessionContext) => {
|
||||
if (err) return void throwError(err, httpResponse);
|
||||
localCache.getCurrentSnapshot((err, cacheSnapshot) => {
|
||||
if (err) return void throwError(err, httpResponse);
|
||||
|
||||
_sessionContext.httpRequest = httpRequest;
|
||||
_sessionContext.httpResponse = httpResponse;
|
||||
_sessionContext.sessionId = crypto.randomBytes(8).toString("hex");
|
||||
httpRequest.connection.setTimeout(_sessionContext.timeout * 1000);
|
||||
const configContext = {
|
||||
id: deviceId,
|
||||
serialNumber: rpc.cpeRequest.deviceId["SerialNumber"],
|
||||
productClass: rpc.cpeRequest.deviceId["ProductClass"],
|
||||
oui: rpc.cpeRequest.deviceId["OUI"]
|
||||
};
|
||||
|
||||
getDueTasksAndFaultsAndOperations(
|
||||
deviceId,
|
||||
_sessionContext.timestamp,
|
||||
(err, dueTasks, faults, operations, cacheUntil) => {
|
||||
if (err) return void throwError(err, httpResponse);
|
||||
|
||||
_sessionContext.tasks = dueTasks;
|
||||
_sessionContext.operations = operations;
|
||||
_sessionContext.cacheUntil = cacheUntil;
|
||||
_sessionContext.faults = faults;
|
||||
_sessionContext.retries = {};
|
||||
for (const [k, v] of Object.entries(_sessionContext.faults)) {
|
||||
if (v.expiry >= _sessionContext.timestamp) {
|
||||
// Delete expired faults
|
||||
delete _sessionContext.faults[k];
|
||||
if (!_sessionContext.faultsTouched)
|
||||
_sessionContext.faultsTouched = {};
|
||||
_sessionContext.faultsTouched[k] = true;
|
||||
} else {
|
||||
_sessionContext.retries[k] = v.retries;
|
||||
}
|
||||
}
|
||||
parsedRpc(_sessionContext, rpc, parseWarnings);
|
||||
}
|
||||
const sessionTimeout =
|
||||
rpc.sessionTimeout ||
|
||||
localCache.getConfig(
|
||||
cacheSnapshot,
|
||||
"cwmp.sessionTimeout",
|
||||
configContext
|
||||
);
|
||||
}
|
||||
);
|
||||
|
||||
session.init(
|
||||
deviceId,
|
||||
rpc.cwmpVersion,
|
||||
sessionTimeout,
|
||||
(err, _sessionContext) => {
|
||||
if (err) return void throwError(err, httpResponse);
|
||||
|
||||
_sessionContext.cacheSnapshot = cacheSnapshot;
|
||||
_sessionContext.configContext = configContext;
|
||||
_sessionContext.debug = localCache.getConfig(
|
||||
cacheSnapshot,
|
||||
"cwmp.debug",
|
||||
configContext
|
||||
);
|
||||
|
||||
_sessionContext.httpRequest = httpRequest;
|
||||
_sessionContext.httpResponse = httpResponse;
|
||||
_sessionContext.sessionId = crypto.randomBytes(8).toString("hex");
|
||||
httpRequest.connection.setTimeout(_sessionContext.timeout * 1000);
|
||||
|
||||
getDueTasksAndFaultsAndOperations(
|
||||
deviceId,
|
||||
_sessionContext.timestamp,
|
||||
(err, dueTasks, faults, operations, cacheUntil) => {
|
||||
if (err) return void throwError(err, httpResponse);
|
||||
|
||||
_sessionContext.tasks = dueTasks;
|
||||
_sessionContext.operations = operations;
|
||||
_sessionContext.cacheUntil = cacheUntil;
|
||||
_sessionContext.faults = faults;
|
||||
_sessionContext.retries = {};
|
||||
for (const [k, v] of Object.entries(_sessionContext.faults)) {
|
||||
if (v.expiry >= _sessionContext.timestamp) {
|
||||
// Delete expired faults
|
||||
delete _sessionContext.faults[k];
|
||||
if (!_sessionContext.faultsTouched)
|
||||
_sessionContext.faultsTouched = {};
|
||||
_sessionContext.faultsTouched[k] = true;
|
||||
} else {
|
||||
_sessionContext.retries[k] = v.retries;
|
||||
}
|
||||
}
|
||||
parsedRpc(_sessionContext, rpc, parseWarnings);
|
||||
}
|
||||
);
|
||||
}
|
||||
);
|
||||
});
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
@@ -21,6 +21,7 @@
|
||||
const mongodb = require("mongodb");
|
||||
const config = require("./config");
|
||||
const common = require("./common");
|
||||
const expressionParser = require("./common/expression-parser");
|
||||
|
||||
let tasksCollection,
|
||||
devicesCollection,
|
||||
@@ -30,7 +31,8 @@ let tasksCollection,
|
||||
virtualParametersCollection,
|
||||
faultsCollection,
|
||||
filesCollection,
|
||||
operationsCollection;
|
||||
operationsCollection,
|
||||
configCollection;
|
||||
|
||||
function connect(callback) {
|
||||
mongodb.MongoClient.connect(
|
||||
@@ -58,6 +60,7 @@ function connect(callback) {
|
||||
exports.operationsCollection = operationsCollection = db.collection(
|
||||
"operations"
|
||||
);
|
||||
configCollection = db.collection("config");
|
||||
|
||||
callback();
|
||||
}
|
||||
@@ -873,6 +876,20 @@ function getFiles(callback) {
|
||||
filesCollection.find().toArray(callback);
|
||||
}
|
||||
|
||||
function getConfig(callback) {
|
||||
configCollection.find().toArray((err, res) => {
|
||||
if (err) return void callback(err);
|
||||
|
||||
callback(
|
||||
null,
|
||||
res.map(c => ({
|
||||
id: c["_id"],
|
||||
value: expressionParser.parse(c["value"])
|
||||
}))
|
||||
);
|
||||
});
|
||||
}
|
||||
|
||||
exports.connect = connect;
|
||||
exports.disconnect = disconnect;
|
||||
exports.fetchDevice = fetchDevice;
|
||||
@@ -890,3 +907,4 @@ exports.getVirtualParameters = getVirtualParameters;
|
||||
exports.getFiles = getFiles;
|
||||
exports.getDueTasks = getDueTasks;
|
||||
exports.getOperations = getOperations;
|
||||
exports.getConfig = getConfig;
|
||||
+50
-1
@@ -21,11 +21,13 @@
|
||||
const vm = require("vm");
|
||||
const crypto = require("crypto");
|
||||
|
||||
const config = require("./config");
|
||||
const db = require("./db");
|
||||
const cache = require("./cache");
|
||||
const query = require("./query");
|
||||
const logger = require("./logger");
|
||||
const scheduling = require("./scheduling");
|
||||
const expression = require("./common/expression");
|
||||
|
||||
const REFRESH = 3000;
|
||||
const EVICT_TIMEOUT = 60000;
|
||||
@@ -56,6 +58,10 @@ function computeHash(snapshot) {
|
||||
h.update(JSON.stringify(keys));
|
||||
for (const k of keys) h.update(snapshot.virtualParameters[k].md5);
|
||||
|
||||
keys = Object.keys(snapshot.config).sort();
|
||||
h.update(JSON.stringify(keys));
|
||||
for (const k of keys) h.update(JSON.stringify(snapshot.config[k]));
|
||||
|
||||
return h.digest("hex");
|
||||
}
|
||||
|
||||
@@ -296,13 +302,27 @@ function refresh(callback) {
|
||||
})
|
||||
);
|
||||
|
||||
promises.push(
|
||||
new Promise((resolve, reject) => {
|
||||
db.getConfig((err, conf) => {
|
||||
if (err) return void reject(err);
|
||||
|
||||
const _config = {};
|
||||
for (const c of conf) _config[c.id] = c.value;
|
||||
|
||||
resolve(_config);
|
||||
});
|
||||
})
|
||||
);
|
||||
|
||||
Promise.all(promises)
|
||||
.then(res => {
|
||||
const snapshot = {
|
||||
presets: res[0],
|
||||
provisions: res[1],
|
||||
virtualParameters: res[2],
|
||||
files: res[3]
|
||||
files: res[3],
|
||||
config: res[4]
|
||||
};
|
||||
|
||||
if (currentSnapshot) {
|
||||
@@ -362,9 +382,38 @@ function getFiles(snapshotKey) {
|
||||
return snapshot.files;
|
||||
}
|
||||
|
||||
function getConfig(snapshotKey, key, context, now) {
|
||||
const snapshot = snapshots.get(snapshotKey);
|
||||
if (!snapshot) throw new Error("Cache snapshot does not exist");
|
||||
|
||||
const oldOpts = {
|
||||
"cwmp.downloadTimeout": "DOWNLOAD_TIMEOUT",
|
||||
"cwmp.debug": "DEBUG",
|
||||
"cwmp.retryDelay": "RETRY_DELAY",
|
||||
"cwmp.sessionTimeout": "SESSION_TIMEOUT",
|
||||
"cwmp.connectionRequestTimeout": "CONNECTION_REQUEST_TIMEOUT",
|
||||
"cwmp.gpnNextLevel": "GPN_NEXT_LEVEL",
|
||||
"cwmp.gpvBatchSize": "GPV_BATCH_SIZE",
|
||||
"cwmp.cookiesPath": "COOKIES_PATH",
|
||||
"cwmp.datetimeMilliseconds": "DATETIME_MILLISECONDS",
|
||||
"cwmp.booleanLiteral": "BOOLEAN_LITERAL",
|
||||
"cwmp.connectionRequestAllowBasicAuth":
|
||||
"CONNECTION_REQUEST_ALLOW_BASIC_AUTH",
|
||||
"cwmp.maxCommitIterations": "MAX_COMMIT_ITERATIONS"
|
||||
};
|
||||
|
||||
if (!(key in snapshot.config)) {
|
||||
if (key in oldOpts)
|
||||
return config.get(oldOpts[key], context ? context.id : null);
|
||||
else return null;
|
||||
}
|
||||
return expression.evaluate(snapshot.config[key], context, now || Date.now());
|
||||
}
|
||||
|
||||
exports.getCurrentSnapshot = getCurrentSnapshot;
|
||||
exports.hasSnapshot = hasSnapshot;
|
||||
exports.getPresets = getPresets;
|
||||
exports.getProvisions = getProvisions;
|
||||
exports.getVirtualParameters = getVirtualParameters;
|
||||
exports.getFiles = getFiles;
|
||||
exports.getConfig = getConfig;
|
||||
+58
-23
@@ -30,10 +30,6 @@ const InstanceSet = require("./instance-set");
|
||||
const defaultProvisions = require("./default-provisions");
|
||||
const gpnHeuristic = require("./gpn-heuristic");
|
||||
|
||||
// Multiply by two because every iteration is two
|
||||
// phases: read and update
|
||||
const MAX_ITERATIONS = config.get("MAX_COMMIT_ITERATIONS") * 2;
|
||||
|
||||
function initDeviceData() {
|
||||
return {
|
||||
paths: new PathSet(),
|
||||
@@ -64,11 +60,7 @@ function init(deviceId, cwmpVersion, timeout, callback) {
|
||||
declarations: []
|
||||
};
|
||||
|
||||
localCache.getCurrentSnapshot((err, cacheSnapshot) => {
|
||||
if (err) return void callback(err);
|
||||
sessionContext.cacheSnapshot = cacheSnapshot;
|
||||
callback(null, sessionContext);
|
||||
});
|
||||
callback(null, sessionContext);
|
||||
}
|
||||
|
||||
function loadParameters(sessionContext, callback) {
|
||||
@@ -435,7 +427,11 @@ function timeoutOperations(sessionContext, callback) {
|
||||
}
|
||||
|
||||
const DOWNLOAD_TIMEOUT =
|
||||
config.get("DOWNLOAD_TIMEOUT", sessionContext.deviceId) * 1000;
|
||||
localCache.getConfig(
|
||||
sessionContext.cacheSnapshot,
|
||||
"cwmp.downloadTimeout",
|
||||
sessionContext.configContext
|
||||
) * 1000;
|
||||
|
||||
if (sessionContext.timestamp > operation.timestamp + DOWNLOAD_TIMEOUT) {
|
||||
delete sessionContext.operations[commandKey];
|
||||
@@ -465,6 +461,15 @@ function timeoutOperations(sessionContext, callback) {
|
||||
}
|
||||
|
||||
function addProvisions(sessionContext, channel, provisions) {
|
||||
// Multiply by two because every iteration is two
|
||||
// phases: read and update
|
||||
const MAX_ITERATIONS =
|
||||
localCache.getConfig(
|
||||
sessionContext.cacheSnapshot,
|
||||
"cwmp.maxCommitIterations",
|
||||
sessionContext.configContext
|
||||
) * 2;
|
||||
|
||||
delete sessionContext.syncState;
|
||||
delete sessionContext.rpcRequest;
|
||||
sessionContext.declarations = [];
|
||||
@@ -509,6 +514,15 @@ function addProvisions(sessionContext, channel, provisions) {
|
||||
}
|
||||
|
||||
function clearProvisions(sessionContext) {
|
||||
// Multiply by two because every iteration is two
|
||||
// phases: read and update
|
||||
const MAX_ITERATIONS =
|
||||
localCache.getConfig(
|
||||
sessionContext.cacheSnapshot,
|
||||
"cwmp.maxCommitIterations",
|
||||
sessionContext.configContext
|
||||
) * 2;
|
||||
|
||||
if (sessionContext.revisions[sessionContext.revisions.length - 1] > 0) {
|
||||
sessionContext.deviceData.timestamps.collapse(1);
|
||||
sessionContext.deviceData.attributes.collapse(1);
|
||||
@@ -1080,6 +1094,15 @@ function rpcRequest(sessionContext, _declarations, callback) {
|
||||
});
|
||||
}
|
||||
|
||||
// Multiply by two because every iteration is two
|
||||
// phases: read and update
|
||||
const MAX_ITERATIONS =
|
||||
localCache.getConfig(
|
||||
sessionContext.cacheSnapshot,
|
||||
"cwmp.maxCommitIterations",
|
||||
sessionContext.configContext
|
||||
) * 2;
|
||||
|
||||
if (sessionContext.iteration >= MAX_ITERATIONS * (sessionContext.cycle + 1)) {
|
||||
return void callback(null, {
|
||||
code: "too_many_commits",
|
||||
@@ -1504,10 +1527,12 @@ function generateGetRpcRequest(sessionContext) {
|
||||
syncState.refreshAttributes.writable.clear();
|
||||
|
||||
if (syncState.gpn.size) {
|
||||
const GPN_NEXT_LEVEL = config.get(
|
||||
"GPN_NEXT_LEVEL",
|
||||
sessionContext.deviceId
|
||||
const GPN_NEXT_LEVEL = localCache.getConfig(
|
||||
sessionContext.cacheSnapshot,
|
||||
"cwmp.gpnNextLevel",
|
||||
sessionContext.configContext
|
||||
);
|
||||
|
||||
const paths = Array.from(syncState.gpn.keys()).sort(
|
||||
(a, b) => b.length - a.length
|
||||
);
|
||||
@@ -1561,10 +1586,12 @@ function generateGetRpcRequest(sessionContext) {
|
||||
}
|
||||
|
||||
if (syncState.refreshAttributes.value.size) {
|
||||
const GPV_BATCH_SIZE = config.get(
|
||||
"GPV_BATCH_SIZE",
|
||||
sessionContext.deviceId
|
||||
const GPV_BATCH_SIZE = localCache.getConfig(
|
||||
sessionContext.cacheSnapshot,
|
||||
"cwmp.gpvBatchSize",
|
||||
sessionContext.configContext
|
||||
);
|
||||
|
||||
const parameterNames = [];
|
||||
for (const path of syncState.refreshAttributes.value) {
|
||||
syncState.refreshAttributes.value.delete(path);
|
||||
@@ -1620,14 +1647,22 @@ function generateSetRpcRequest(sessionContext) {
|
||||
}
|
||||
|
||||
// Set values
|
||||
const GPV_BATCH_SIZE = config.get("GPV_BATCH_SIZE", sessionContext.deviceId);
|
||||
const DATETIME_MILLISECONDS = config.get(
|
||||
"DATETIME_MILLISECONDS",
|
||||
sessionContext.deviceId
|
||||
const GPV_BATCH_SIZE = localCache.getConfig(
|
||||
sessionContext.cacheSnapshot,
|
||||
"cwmp.gpvBatchSize",
|
||||
sessionContext.configContext
|
||||
);
|
||||
const BOOLEAN_LITERAL = config.get(
|
||||
"BOOLEAN_LITERAL",
|
||||
sessionContext.deviceId
|
||||
|
||||
const DATETIME_MILLISECONDS = localCache.getConfig(
|
||||
sessionContext.cacheSnapshot,
|
||||
"cwmp.datetimeMilliseconds",
|
||||
sessionContext.configContext
|
||||
);
|
||||
|
||||
const BOOLEAN_LITERAL = localCache.getConfig(
|
||||
sessionContext.cacheSnapshot,
|
||||
"cwmp.booleanLiteral",
|
||||
sessionContext.configContext
|
||||
);
|
||||
|
||||
const parameterValues = [];
|
||||
|
||||
Reference in new issue
Block a user