From d46d1338787a94a237138db17e2a6b95ae5b5285 Mon Sep 17 00:00:00 2001 From: Shweelan Samson Date: Tue, 5 Feb 2019 16:30:59 +0300 Subject: [PATCH] Remove UI dependance on NBI --- lib/local-cache.js | 3 +- lib/ui/api-functions.js | 468 ++++++++++++------------------------ lib/ui/api.js | 33 ++- lib/ui/db.js | 114 ++++++++- lib/ui/mongodb-functions.js | 8 - 5 files changed, 285 insertions(+), 341 deletions(-) diff --git a/lib/local-cache.js b/lib/local-cache.js index 4876d54..b7d51df 100644 --- a/lib/local-cache.js +++ b/lib/local-cache.js @@ -434,7 +434,8 @@ export function getConfig(snapshotKey, key, context, now) { "cwmp.booleanLiteral": "BOOLEAN_LITERAL", "cwmp.connectionRequestAllowBasicAuth": "CONNECTION_REQUEST_ALLOW_BASIC_AUTH", - "cwmp.maxCommitIterations": "MAX_COMMIT_ITERATIONS" + "cwmp.maxCommitIterations": "MAX_COMMIT_ITERATIONS", + "cwmp.deviceOnlineThreshold": "DEVICE_ONLINE_THRESHOLD" }; if (!(key in snapshot.config)) { diff --git a/lib/ui/api-functions.js b/lib/ui/api-functions.js index 918f7c2..d4b2cb8 100644 --- a/lib/ui/api-functions.js +++ b/lib/ui/api-functions.js @@ -1,340 +1,184 @@ "use strict"; -import http from "http"; -import https from "https"; -import url from "url"; -import * as config from "../config"; import * as db from "./db"; -import * as cache from "../cache"; -import * as mongodbFunctions from "./mongodb-functions"; +import { del } from "../cache"; +import { ObjectID } from "mongodb"; +import { exec } from "child_process"; +import { + insertTasks, + watchTask, + connectionRequest, + deleteDevice +} from "../api-functions"; -export function deleteResource(resource, id) { +function deleteFault(id) { + const deviceId = id.split(":", 1)[0]; + const channel = id.slice(deviceId.length + 1); return new Promise((resolve, reject) => { - if (resource === "config") { - db.deleteConfig(id) - .then(() => { - cache.del("presets_hash", err => { - if (err) return void reject(err); - resolve(); - }); - }) - .catch(reject); - return; - } - - const resourceUrl = - resource === "virtualParameters" ? "virtual_parameters" : resource; - const options = url.parse( - `http://127.0.0.1:${config.get( - "NBI_PORT" - )}/${resourceUrl}/${encodeURIComponent(id)}` - ); - - options.method = "DELETE"; - - const _http = options.protocol === "https:" ? https : http; - - _http - .request(options, res => { - res.resume(); - if (res.statusCode === 200) resolve(true); - else if (res.statusCode === 404) resolve(false); - else reject(new Error(`Unexpected status code ${res.statusCode}`)); - }) - .end(); - }); -} - -function postTask(task, connectionRequest, callback) { - const options = url.parse( - `http://127.0.0.1:${config.get("NBI_PORT")}/devices/${encodeURIComponent( - task.device - )}/tasks${connectionRequest ? "?connection_request" : ""}` - ); - options.method = "POST"; - const _http = options.protocol === "https:" ? https : http; - - _http - .request(options, res => { - if (res.statusCode !== 200 && res.statusCode !== 202) { - callback(new Error(`Unexpected status code ${res.statusCode}`)); - res.resume(); - return; - } - const chunks = []; - let bytes = 0; - res.on("data", chunk => { - chunks.push(chunk); - bytes += chunk.length; - }); - - res.on("end", () => { - const buf = Buffer.allocUnsafe(bytes); - let o = 0; - for (const c of chunks) { - c.copy(buf, o, 0, c.length); - o += c.length; - } - try { - const t = JSON.parse(buf); - let connectionRequestStatus; - if (connectionRequest) { - connectionRequestStatus = "OK"; - if ( - res.statusCode === 202 && - res.statusMessage !== "Task queued but not processed" && - res.statusMessage !== "Task faulted" - ) - connectionRequestStatus = res.statusMessage; - } - callback(null, t, connectionRequestStatus); - } catch (err) { - callback(err); - } - }); - }) - .end(JSON.stringify(task)); -} - -export function postTasks(deviceId, tasks) { - return new Promise((resolve, reject) => { - let connectionRequestStatus; - const promises = []; - for (const [idx, task] of tasks.entries()) { - task.device = deviceId; - promises.push( - new Promise((res, rej) => { - const conReq = idx === tasks.length - 1; - delete task._id; - task.expiry = 5; - postTask(task, conReq, (err, t, crs) => { - if (err) return void rej(err); - if (conReq) connectionRequestStatus = crs; - res({ _id: t._id, status: "pending" }); - }); - }) - ); - } - - Promise.all(promises) - .then(statuses => { - if (connectionRequestStatus !== "OK") { - return void resolve({ - connectionRequest: connectionRequestStatus, - tasks: statuses - }); - } - - const promises2 = []; - for (const s of statuses) { - promises2.push(db.query("tasks", ["=", ["PARAM", "_id"], s._id])); - promises2.push( - db.query("faults", [ - "=", - ["PARAM", "_id"], - `${deviceId}:task_${s._id}` - ]) - ); - } - - Promise.all(promises2) - .then(res => { - for (const [i, r] of statuses.entries()) { - if (res[i * 2].length === 0) { - r.status = "done"; - } else if (res[i * 2 + 1].length === 1) { - r.status = "fault"; - r.fault = res[i * 2 + 1][0]; - } - deleteResource("tasks", r._id); - } - - resolve({ - connectionRequest: connectionRequestStatus, - tasks: statuses - }); - }) - .catch(reject); + Promise.all([ + db.deleteFault(id), + channel.startsWith("task_") + ? db.deleteTask(new ObjectID(channel.slice(5))) + : null + ]) + .then(() => { + del(`${deviceId}_tasks_faults_operations`, err => { + err ? reject(err) : resolve(); + }); }) .catch(reject); }); } -export function updateTags(deviceId, tags) { - return Promise.all( - Object.entries(tags).map( - ([tag, onOff]) => - new Promise((resolve, reject) => { - const options = url.parse( - `http://127.0.0.1:${config.get( - "NBI_PORT" - )}/devices/${encodeURIComponent( - deviceId - )}/tags/${encodeURIComponent(tag)}` - ); - if (onOff) options.method = "POST"; - else options.method = "DELETE"; - const _http = options.protocol === "https:" ? https : http; +export function deleteResource(resource, id) { + return new Promise((resolve, reject) => { + let promise; - _http - .request(options, res => { - res.resume(); - if (res.statusCode === 200) resolve(); - else - reject(new Error(`Unexpected status code ${res.statusCode}`)); - }) - .end(); - }) - ) - ); + switch (resource) { + case "devices": + promise = new Promise((res, rej) => { + deleteDevice(id, err => { + if (err) return void rej(err); + res(); + }); + }); + break; + + case "files": + promise = db.deleteFile(id); + break; + + case "faults": + promise = deleteFault(id); + break; + + case "provisions": + promise = db.deleteProvision(id); + break; + + case "presets": + promise = db.deletePreset(id); + break; + + case "virtualParameters": + promise = db.deleteVirtualParameter(id); + break; + + case "config": + promise = db.deleteConfig(id); + break; + } + + promise + .then(() => { + del("presets_hash", err => { + if (err) return void reject(err); + resolve(); + }); + }) + .catch(reject); + }); +} + +export function postTasks(deviceId, tasks, timeout) { + return new Promise((resolve, reject) => { + for (const task of tasks) task.device = deviceId; + + insertTasks(tasks, err => { + if (err) return void reject(err); + const statuses = tasks.map(t => { + return { _id: t._id, status: "pending" }; + }); + del(`${deviceId}_tasks_faults_operations`, err => { + if (err) return void reject(err); + connectionRequest(deviceId, err => { + if (err) { + return void resolve({ + connectionRequest: err.message, + tasks: statuses + }); + } + + const sample = tasks[tasks.length - 1]; + + // Waiting for session to finish or timeout + watchTask(deviceId, sample._id, timeout, err => { + if (err) return void reject(err); + + const promises2 = []; + for (const s of statuses) { + promises2.push(db.query("tasks", ["=", ["PARAM", "_id"], s._id])); + promises2.push( + db.query("faults", [ + "=", + ["PARAM", "_id"], + `${deviceId}:task_${s._id}` + ]) + ); + } + + Promise.all(promises2) + .then(res => { + for (const [i, r] of statuses.entries()) { + if (res[i * 2].length === 0) { + r.status = "done"; + } else if (res[i * 2 + 1].length === 1) { + r.status = "fault"; + r.fault = res[i * 2 + 1][0]; + } + db.deleteTask(r._id); + } + + resolve({ connectionRequest: "OK", tasks: statuses }); + }) + .catch(reject); + }); + }); + }); + }); + }); } export function ping(host) { return new Promise((resolve, reject) => { - const options = url.parse( - `http://127.0.0.1:${config.get("NBI_PORT")}/ping/${encodeURIComponent( - host - )}` - ); + exec(`ping -w 1 -i 0.2 -c 3 ${host}`, (err, stdout) => { + if (err) return void reject(err); - const _http = options.protocol === "https:" ? https : http; - _http - .request(options, res => { - if (res.statusCode === 404) { - res.resume(); - return void resolve({}); - } + const m = stdout.match( + /(\d) packets transmitted, (\d) received, ([\d.%]+) packet loss[^]*([\d.]+)\/([\d.]+)\/([\d.]+)\/([\d.]+)/ + ); + if (!m) return void reject(new Error("Could not parse ping response")); - if (res.statusCode !== 200) { - res.resume(); - return void reject( - new Error(`Unexpected status code ${res.statusCode}`) - ); - } - - const chunks = []; - let bytes = 0; - res.on("data", chunk => { - chunks.push(chunk); - bytes += chunk.length; - }); - - res.on("end", () => { - const buf = Buffer.allocUnsafe(bytes); - let o = 0; - for (const c of chunks) { - c.copy(buf, o, 0, c.length); - o += c.length; - } - const m = buf - .toString() - .match( - /(\d) packets transmitted, (\d) received, ([\d.%]+) packet loss[^]*([\d.]+)\/([\d.]+)\/([\d.]+)\/([\d.]+)/ - ); - if (!m) - return void reject(new Error("Could not parse ping response")); - - resolve({ - packetsTransmitted: +m[1], - packetsReceived: +m[2], - packetLoss: m[3], - min: +m[4], - avg: +m[5], - max: +m[6], - mdev: +m[7] - }); - }); - }) - .end(); + resolve({ + packetsTransmitted: +m[1], + packetsReceived: +m[2], + packetLoss: m[3], + min: +m[4], + avg: +m[5], + max: +m[6], + mdev: +m[7] + }); + }); }); } export function putResource(resource, id, data) { return new Promise((resolve, reject) => { - if (resource === "config") { - db.putConfig(id, data) - .then(() => { - cache.del("presets_hash", err => { - if (err) return void reject(err); - resolve(); - }); - }) - .catch(reject); - return; - } + let promise; - const resourceUrl = - resource === "virtualParameters" ? "virtual_parameters" : resource; - const options = url.parse( - `http://127.0.0.1:${config.get( - "NBI_PORT" - )}/${resourceUrl}/${encodeURIComponent(id)}` - ); + if (resource === "presets") promise = db.putPreset(id, data); + else if (resource === "provisions") promise = db.putProvision(id, data); + else if (resource === "virtualParameters") + promise = db.putVirtualParameter(id, data); + else if (resource === "config") promise = db.putConfig(id, data); - options.method = "PUT"; - - const _http = options.protocol === "https:" ? https : http; - - if (resource === "presets") data = mongodbFunctions.preProcessPreset(data); - - if (resource === "provisions") - data = mongodbFunctions.preProcessProvision(data); - - if (resource === "virtualParameters") - data = mongodbFunctions.preProcessVirtualParameters(data); - - const body = typeof data !== "string" ? JSON.stringify(data) : data; - _http - .request(options, res => { - const chunks = []; - let bytes = 0; - res.on("data", chunk => { - chunks.push(chunk); - bytes += chunk.length; - }); - - res.on("end", () => { - if (res.statusCode === 200) { - return void resolve(); - } else if (res.statusCode !== 400) { - return void reject( - new Error(`Unexpected status code ${res.statusCode}`) - ); - } - const buf = Buffer.allocUnsafe(bytes); - let o = 0; - for (const c of chunks) { - c.copy(buf, o, 0, c.length); - o += c.length; - } - resolve(buf.toString()); + promise + .then(() => { + del("presets_hash", err => { + if (err) return void reject(err); + resolve(); }); }) - .end(body); - }); -} - -export function putFile(id, metadata, fileStream) { - return new Promise((resolve, reject) => { - const options = url.parse( - `http://127.0.0.1:${config.get("NBI_PORT")}/files/${encodeURIComponent( - id - )}` - ); - - options.method = "PUT"; - options.headers = Object.assign({}, metadata); - const _http = options.protocol === "https:" ? https : http; - const req = _http.request(options, res => { - res.resume(); - if (res.statusCode === 201) { - return void resolve(); - } else { - return void reject( - new Error(`Unexpected status code ${res.statusCode}`) - ); - } - }); - fileStream.pipe(req); + .catch(reject); }); } diff --git a/lib/ui/api.js b/lib/ui/api.js index 725ec86..0f8b6cc 100644 --- a/lib/ui/api.js +++ b/lib/ui/api.js @@ -7,6 +7,7 @@ import * as apiFunctions from "./api-functions"; import { evaluate, and, extractParams } from "../common/expression"; import { parse } from "../common/expression-parser"; import * as logger from "../logger"; +import { getConfig } from "../local-cache"; const router = new Router(); export default router; @@ -400,15 +401,7 @@ router.put("/files/:id", async (ctx, next) => { return void (ctx.status = 403); } - const err = await apiFunctions.putFile(id, metadata, ctx.req); - - if (err) { - log.message += " failed"; - logger.accessWarn(log); - ctx.body = err; - return void (ctx.status = 400); - } - + await db.putFile(id, metadata, ctx.req); log.metadata = metadata; logger.accessInfo(log); @@ -439,7 +432,17 @@ router.post("/devices/:id/tasks", async (ctx, next) => { } } - const res = await apiFunctions.postTasks(ctx.params.id, ctx.request.body); + // TODO Pass context and timestamp + const onlineThreshold = getConfig( + ctx.state.configSnapshot, + "cwmp.deviceOnlineThreshold" + ); + + const res = await apiFunctions.postTasks( + ctx.params.id, + ctx.request.body, + onlineThreshold + ); log.tasks = res.tasks.map(t => t._id).join(","); @@ -472,7 +475,7 @@ router.post("/devices/:id/tags", async (ctx, next) => { return void (ctx.status = 403); } - await apiFunctions.updateTags(ctx.params.id, ctx.request.body); + await db.updateDeviceTags(ctx.params.id, ctx.request.body); logger.accessInfo(log); @@ -480,5 +483,11 @@ router.post("/devices/:id/tags", async (ctx, next) => { }); router.get("/ping/:host", async ctx => { - ctx.body = await apiFunctions.ping(ctx.params.host); + try { + ctx.body = await apiFunctions.ping(ctx.params.host); + } catch (err) { + ctx.status = 404; + ctx.response.headers["Cache-Control"] = "no-cache"; + ctx.body = `${err.name}: ${err.message}`; + } }); diff --git a/lib/ui/db.js b/lib/ui/db.js index 39b2ce0..2f1d62f 100644 --- a/lib/ui/db.js +++ b/lib/ui/db.js @@ -1,9 +1,10 @@ "use strict"; -import { MongoClient } from "mongodb"; +import { MongoClient, GridFSBucket } from "mongodb"; import * as config from "../config"; import * as mongodbFunctions from "./mongodb-functions"; import * as expression from "../common/expression"; +import { preProcessPreset } from "./mongodb-functions"; const CACHE_TTL = 300000; @@ -187,14 +188,23 @@ export function count(resource, filter) { }); } -export function deleteConfig(id) { +export function updateDeviceTags(deviceId, tags) { return new Promise((resolve, reject) => { + const add = []; + const pull = []; + + for (const [tag, onOff] of Object.entries(tags)) { + if (onOff) add.push(tag); + else pull.push(tag); + } getClient() .then(client => { - const collection = client - .db() - .collection(RESOURCE_COLLECTION["config"] || "config"); - collection.deleteOne({ _id: id }, err => { + const collection = client.db().collection("devices"); + + const object = {}; + if (add && add.length) object["$addToSet"] = { _tags: { $each: add } }; + if (pull && pull.length) object["$pullAll"] = { _tags: pull }; + collection.updateOne({ _id: deviceId }, object, err => { if (err) return void reject(err); resolve(); }); @@ -203,13 +213,13 @@ export function deleteConfig(id) { }); } -export function putConfig(id, object) { +function putResource(resource, id, object) { return new Promise((resolve, reject) => { getClient() .then(client => { const collection = client .db() - .collection(RESOURCE_COLLECTION["config"] || "config"); + .collection(RESOURCE_COLLECTION[resource] || resource); collection.replaceOne({ _id: id }, object, { upsert: true }, err => { if (err) return void reject(err); resolve(); @@ -219,6 +229,94 @@ export function putConfig(id, object) { }); } +function deleteResource(resource, id) { + return new Promise((resolve, reject) => { + getClient() + .then(client => { + const collection = client + .db() + .collection(RESOURCE_COLLECTION[resource] || resource); + collection.deleteOne({ _id: id }, err => { + if (err) return void reject(err); + resolve(); + }); + }) + .catch(reject); + }); +} + +export function putPreset(id, object) { + object = preProcessPreset(object); + return putResource("presets", id, object); +} + +export function deletePreset(id) { + return deleteResource("presets", id); +} + +export function putProvision(id, object) { + return putResource("provisions", id, object); +} + +export function deleteProvision(id) { + return deleteResource("provisions", id); +} + +export function putVirtualParameter(id, object) { + return putResource("virtualParameters", id, object); +} + +export function deleteVirtualParameter(id) { + return deleteResource("virtualParameters", id); +} + +export function putConfig(id, object) { + return putResource("config", id, object); +} + +export function deleteConfig(id) { + return deleteResource("config", id); +} + +export function putFile(filename, metadata, contentStream) { + return new Promise((resolve, reject) => { + getClient() + .then(client => { + const bucket = new GridFSBucket(client.db()); + const uploadStream = bucket.openUploadStreamWithId(filename, filename, { + metadata: metadata + }); + uploadStream.on("error", reject); + contentStream.on("error", reject); + uploadStream.on("finish", resolve); + contentStream.pipe(uploadStream); + }) + .catch(reject); + }); +} + +export function deleteFile(filename) { + return new Promise((resolve, reject) => { + getClient() + .then(client => { + const bucket = new GridFSBucket(client.db()); + bucket.delete(filename, err => { + if (err) return void reject(err); + resolve(); + }); + }) + .catch(reject); + }); +} + +export function deleteFault(id) { + return deleteResource("faults", id); +} + +export function deleteTask(id) { + return deleteResource("tasks", id); +} + export function disconnect() { if (_clientPromise) { _clientPromise.then(client => { diff --git a/lib/ui/mongodb-functions.js b/lib/ui/mongodb-functions.js index a6bc095..04d2e9e 100644 --- a/lib/ui/mongodb-functions.js +++ b/lib/ui/mongodb-functions.js @@ -572,11 +572,3 @@ export function preProcessPreset(data) { preset.configurations = [configuration]; return preset; } - -export function preProcessProvision(data) { - return data.script; -} - -export function preProcessVirtualParameters(data) { - return data.script; -}