diff --git a/lib/api.js b/lib/api.js index bccf917f8..df3206b1e 100644 --- a/lib/api.js +++ b/lib/api.js @@ -108,7 +108,7 @@ nThen(function (w) { }; // spawn ws server and attach netflux event handlers - let Server = NetfluxSrv.create(new WebSocketServer({ server: Env.httpServer})) + let Server = Env.Server = NetfluxSrv.create(new WebSocketServer({ server: Env.httpServer})) .on('channelClose', historyKeeper.channelClose) .on('channelMessage', historyKeeper.channelMessage) .on('channelOpen', historyKeeper.channelOpen) diff --git a/lib/hk-util.js b/lib/hk-util.js index 93c3f9c0e..167ee5ea0 100644 --- a/lib/hk-util.js +++ b/lib/hk-util.js @@ -321,6 +321,9 @@ const storeMessage = function (Env, channel, msg, isCp, optionalMessageHash, tim // Store the message first, and update the index only once it's stored. // store.messageBin can be async so updating the index first may // result in a wrong cpIndex + try { + Env.plugins.MONITORING.increment(`storeMessage`); + } catch (e) {} nThen((waitFor) => { Env.store.messageBin(id, msgBin, waitFor(function (err) { if (err) { @@ -392,6 +395,10 @@ const storeMessage = function (Env, channel, msg, isCp, optionalMessageHash, tim var msgLength = msgBin.length; index.size += msgLength; + try { + Env.plugins.MONITORING.increment('broadcastMessage', (channel.length-1)); + } catch (e) {} + // handle the next element in the queue next(); @@ -522,6 +529,9 @@ const getHistoryAsync = (Env, channelName, lastKnownHash, beforeHash, handler, c const store = Env.store; let offset = -1; + try { + Env.plugins.MONITORING.increment(`getHistoryAsync`); + } catch (e) {} nThen((waitFor) => { getHistoryOffset(Env, channelName, lastKnownHash, waitFor((err, os) => { if (err) { diff --git a/lib/http-worker.js b/lib/http-worker.js index b93120d74..0d4484d98 100644 --- a/lib/http-worker.js +++ b/lib/http-worker.js @@ -35,7 +35,7 @@ const guid = () => { return Util.guid(response._pending); }; -const sendMessage = (msg, cb, opt) => { +const sendMessage = Env.sendMessage = (msg, cb, opt) => { var txid = guid(); var timeout = (opt && opt.timeout) || DEFAULT_QUERY_TIMEOUT; var obj = { @@ -91,6 +91,23 @@ EVENTS.FLUSH_CACHE = function (data) { }); }; +Object.keys(plugins || {}).forEach(name => { + let plugin = plugins[name]; + if (!plugin.addHttpEvents) { return; } + try { + let events = plugin.addHttpEvents(Env); + Object.keys(events || {}).forEach(cmd => { + // Uppercase event name? + if (cmd !== cmd.toUpperCase()) { return; } + // Event is a function? + if (typeof(events[cmd]) !== "function") { return; } + // Event doesn't already exists? + if (EVENTS[cmd]) { return; } + EVENTS[cmd] = events[cmd]; + }); + } catch (e) {} +}); + process.on('message', msg => { if (!(msg && msg.txid)) { return; } if (msg.type === 'REPLY') { @@ -799,7 +816,12 @@ nThen(function (w) { })); }).nThen(function () { // TODO inform the parent process that this worker is ready - + Object.keys(Env.plugins || {}).forEach(name => { + let plugin = plugins[name]; + if (!plugin.initialize) { return; } + try { plugin.initialize(Env, "http-worker"); } + catch (e) {} + }); }); process.on('uncaughtException', function (err) { diff --git a/lib/rpc.js b/lib/rpc.js index 571272ab1..286604e6d 100644 --- a/lib/rpc.js +++ b/lib/rpc.js @@ -34,6 +34,10 @@ var isUnauthenticateMessage = function (msg) { var handleUnauthenticatedMessage = function (Env, msg, respond, Server, netfluxId) { Env.Log.silly('LOG_RPC', msg[0]); + try { + Env.plugins.MONITORING.increment(`rpc_${msg[0]}`); + } catch (e) {} + var method = UNAUTHENTICATED_CALLS[msg[0]]; method(Env, msg[1], function (err, value) { if (err) { @@ -170,6 +174,10 @@ var rpc = function (Env, Server, userId, data, respond) { var command = msg[1]; + try { + Env.plugins.MONITORING.increment(`rpc_${command}`); + } catch (e) {} + if (command === 'UPLOAD') { // UPLOAD is a special case that skips signature validation // intentional fallthrough behaviour diff --git a/lib/workers/db-worker.js b/lib/workers/db-worker.js index 398260d61..1a6971059 100644 --- a/lib/workers/db-worker.js +++ b/lib/workers/db-worker.js @@ -17,11 +17,18 @@ const Tasks = require("../storage/tasks"); const Nacl = require('tweetnacl/nacl-fast'); const Eviction = require("../eviction"); const CPCrypto = require('../crypto'); +const plugins = require("../plugin-manager"); const Env = { Log: {}, }; +const Monitoring = plugins && plugins.MONITORING; +const monitoringIncrement = key => { + if (!Monitoring || !Monitoring.increment) { return; } + Monitoring.increment(key); +}; + // support the usual log API but pass it to the main process Logger.levels.forEach(function (level) { Env.Log[level] = function (label, info) { @@ -57,6 +64,17 @@ const init = function (config, _cb) { Env.archiveRetentionTime = config.archiveRetentionTime; Env.accountRetentionTime = config.accountRetentionTime; + Env.sendMessage = data => { + process.send(data); + }; + + Object.keys(Env.plugins || {}).forEach(name => { + let plugin = plugins[name]; + if (!plugin.initialize) { return; } + try { plugin.initialize(Env, "db-worker"); } + catch (e) {} + }); + nThen(function (w) { Store.create(config, w(function (err, _store) { if (err) { @@ -277,6 +295,8 @@ const computeIndex = function (data, cb) { const channelName = data.channel; const CB = Util.once(cb); + monitoringIncrement('computeIndex'); + var start = 0; nThen(function (w) { store.getOffset(channelName, w(function (err, obj) { @@ -298,6 +318,7 @@ const computeIndex = function (data, cb) { }); } w.abort(); + monitoringIncrement('computeIndexFromOffset'); CB(err, index); })); }).nThen(function (w) { @@ -306,6 +327,7 @@ const computeIndex = function (data, cb) { store.clearOffset(channelName, w()); }).nThen(function () { // now get the history as though it were the first time + monitoringIncrement('computeIndexFromStart'); computeIndexFromOffset(channelName, 0, CB); }); }; @@ -313,6 +335,7 @@ const computeIndex = function (data, cb) { const computeMetadata = function (data, cb) { const ref = {}; const lineHandler = Meta.createLineHandler(ref, Env.Log.error); + monitoringIncrement('computeMetadata'); return void store.readChannelMetadata(data.channel, lineHandler, function (err) { if (err) { // stream errors? @@ -393,6 +416,7 @@ const getPinState = function (data, cb) { var lineHandler = Pins.createLineHandler(ref, Env.Log.error); // if channels aren't in memory. load them from disk + monitoringIncrement('getPin'); pinStore.readMessagesBin(safeKey, 0, (msgObj, readMore) => { lineHandler(msgObj.buff.toString('utf8')); readMore(); @@ -449,6 +473,7 @@ const _iterateFiles = function (channels, handler, cb) { const getTotalSize = function (data, cb) { var bytes = 0; + monitoringIncrement('getTotalSize'); _iterateFiles(data.channels, function (channel, next) { _getFileSize(channel, function (err, size) { if (!err) { bytes += size; } @@ -494,6 +519,7 @@ const getHashOffset = function (data, cb) { const lastKnownHash = data.hash; if (typeof(lastKnownHash) !== 'string') { return void cb("INVALID_HASH"); } + monitoringIncrement('getHashOffset'); var offset = -1; store.readMessagesBin(channelName, 0, (msgObj, readMore, abort) => { // tryParse return a parsed message or undefined @@ -592,6 +618,8 @@ const completeUpload = function (data, cb) { var arg = data.arg; var size = data.size; + monitoringIncrement('uploadedBlob'); + var method; var label; if (owned) { @@ -671,6 +699,7 @@ const COMMANDS = { }; COMMANDS.INLINE = function (data, cb) { + monitoringIncrement('inlineValidation'); var signedMsg; try { signedMsg = Nacl.util.decodeBase64(data.msg); @@ -695,6 +724,7 @@ COMMANDS.INLINE = function (data, cb) { const checkDetachedSignature = function (signedMsg, signature, publicKey) { if (!(signedMsg && publicKey)) { return false; } + monitoringIncrement('detachedValidation'); var signedBuffer; var pubBuffer; @@ -758,10 +788,12 @@ COMMANDS.HASH_CHANNEL_LIST = function (data, cb) { }; COMMANDS.VALIDATE_ANCESTOR_PROOF = function (data, cb) { + monitoringIncrement('validateAncestorProof'); Block.validateAncestorProof(Env, data && data.proof, cb); }; COMMANDS.VALIDATE_LOGIN_BLOCK = function (data, cb) { + monitoringIncrement('validateLoginBlock'); Block.validateLoginBlock(Env, data.publicKey, data.signature, data.block, cb); }; diff --git a/lib/workers/index.js b/lib/workers/index.js index 55487c848..3e3367e1e 100644 --- a/lib/workers/index.js +++ b/lib/workers/index.js @@ -155,6 +155,20 @@ Workers.initialize = function (Env, config, _cb) { state.worker.send(msg); }; + const pluginsResponses = {}; + Object.keys(Env.plugins || {}).forEach(name => { + let plugin = Env.plugins[name]; + if (!plugin.addWorkerResponses) { return; } + try { + let res = plugin.addWorkerResponses(Env); + Object.keys(res || {}).forEach(key => { + if (typeof(res[key]) !== "function") { return; } + if (pluginsResponses[key]) { return; } + pluginsResponses[key] = res[key]; + }); + } catch (e) {} + }); + var handleResponse = function (state, res) { if (!res) { return; } // handle log messages before checking if it was addressed to your PID @@ -162,12 +176,23 @@ Workers.initialize = function (Env, config, _cb) { if (res.log) { return void handleLog(res.log, res.label, res.info); } + // but don't bother handling things addressed to other processes // since it's basically guaranteed not to work if (res.pid !== PID) { return void Log.error("WRONG_PID", res); } + // handle plugins + if (res.plugin) { + Object.keys(pluginsResponses).some(key => { + if (res.type !== key) { return; } + pluginsResponses[key](res.data); + return true; + }); + return; + } + if (!res.txid) { return; } response.handle(res.txid, [res.error, res.value]); delete state.tasks[res.txid]; @@ -226,7 +251,15 @@ Workers.initialize = function (Env, config, _cb) { handleResponse(state, res); }); + let pid = worker.pid; var substituteWorker = Util.once(function () { + Object.keys(Env.plugins || {}).forEach(name => { + let plugin = Env.plugins[name]; + if (!plugin.onWorkerClosed) { return; } + try { plugin.onWorkerClosed("db-worker", pid); } + catch (e) {} + }); + Env.Log.info("SUBSTITUTE_DB_WORKER", ''); var idx = workers.indexOf(state); if (idx !== -1) { diff --git a/package-lock.json b/package-lock.json index faa58a38e..fd1512409 100644 --- a/package-lock.json +++ b/package-lock.json @@ -47,6 +47,7 @@ "open-sans-fontface": "^1.4.0", "openid-client": "^5.4.2", "pako": "^2.1.0", + "prom-client": "^14.2.0", "prompt-confirm": "^2.0.4", "pull-stream": "^3.6.1", "require-css": "0.1.10", @@ -1215,6 +1216,11 @@ "node": ">= 0.6.0" } }, + "node_modules/bintrees": { + "version": "1.0.2", + "resolved": "https://registry.npmjs.org/bintrees/-/bintrees-1.0.2.tgz", + "integrity": "sha512-VOMgTMwjAaUG580SXn3LacVgjurrbMme7ZZNYGSSV7mmtY6QQRh0Eg3pwIcntQ77DErK1L0NxkbetjcoXzVwKw==" + }, "node_modules/body-parser": { "version": "1.20.3", "resolved": "https://registry.npmjs.org/body-parser/-/body-parser-1.20.3.tgz", @@ -4030,6 +4036,17 @@ "node": ">=0.4.0" } }, + "node_modules/prom-client": { + "version": "14.2.0", + "resolved": "https://registry.npmjs.org/prom-client/-/prom-client-14.2.0.tgz", + "integrity": "sha512-sF308EhTenb/pDRPakm+WgiN+VdM/T1RaHj1x+MvAuT8UiQP8JmOEbxVqtkbfR4LrvOg5n7ic01kRBDGXjYikA==", + "dependencies": { + "tdigest": "^0.1.1" + }, + "engines": { + "node": ">=10" + } + }, "node_modules/prompt-actions": { "version": "3.0.2", "resolved": "https://registry.npmjs.org/prompt-actions/-/prompt-actions-3.0.2.tgz", @@ -5384,6 +5401,14 @@ "node": ">=8" } }, + "node_modules/tdigest": { + "version": "0.1.2", + "resolved": "https://registry.npmjs.org/tdigest/-/tdigest-0.1.2.tgz", + "integrity": "sha512-+G0LLgjjo9BZX2MfdvPfH+MKLCrxlXSYec5DaPYP1fe6Iyhf0/fSmJ0bFiZ1F8BT6cGXl2LpltQptzjXKWEkKA==", + "dependencies": { + "bintrees": "1.0.2" + } + }, "node_modules/terminal-paginator": { "version": "2.0.2", "resolved": "https://registry.npmjs.org/terminal-paginator/-/terminal-paginator-2.0.2.tgz", diff --git a/package.json b/package.json index c7cc84e99..7091ad676 100644 --- a/package.json +++ b/package.json @@ -50,6 +50,7 @@ "open-sans-fontface": "^1.4.0", "openid-client": "^5.4.2", "pako": "^2.1.0", + "prom-client": "^14.2.0", "prompt-confirm": "^2.0.4", "pull-stream": "^3.6.1", "require-css": "0.1.10", diff --git a/server.js b/server.js index f240b1cd9..0bff20dca 100644 --- a/server.js +++ b/server.js @@ -25,6 +25,13 @@ var app = Express(); } }()); +Object.keys(Env.plugins || {}).forEach(name => { + let plugin = Env.plugins[name]; + if (!plugin.initialize) { return; } + try { plugin.initialize(Env, "main"); } + catch (e) {} +}); + var COMMANDS = {}; COMMANDS.LOG = function (msg, cb) { @@ -49,6 +56,25 @@ COMMANDS.GET_PROFILING_DATA = function (msg, cb) { cb(void 0, Env.bytesWritten); }; +Object.keys(Env.plugins || {}).forEach(name => { + let plugin = Env.plugins[name]; + if (!plugin.addMainCommands) { return; } + try { + let commands = plugin.addMainCommands(Env); + Object.keys(commands || {}).forEach(cmd => { + // Uppercase command name? + if (cmd !== cmd.toUpperCase()) { return; } + // Command is a function? + if (typeof(commands[cmd]) !== "function") { return; } + // Command doesn't already exists? + if (COMMANDS[cmd]) { return; } + + COMMANDS[cmd] = commands[cmd]; + }); + } catch (e) {} +}); + + nThen(function (w) { require("./lib/log").create(config, w(function (_log) { Env.Log = _log; @@ -93,6 +119,7 @@ nThen(function (w) { var launchWorker = (online) => { var worker = Cluster.fork(workerState); + var pid = worker.process.pid; worker.on('online', () => { online(); }); @@ -122,6 +149,13 @@ nThen(function (w) { }); worker.on('exit', (code, signal) => { + Object.keys(Env.plugins || {}).forEach(name => { + let plugin = Env.plugins[name]; + if (!plugin.onWorkerClosed) { return; } + try { plugin.onWorkerClosed("http-worker", pid); } + catch (e) {} + }); + if (!signal && code === 0) { return; } // relaunch http workers if they crash Env.Log.error('HTTP_WORKER_EXIT', { @@ -147,7 +181,7 @@ nThen(function (w) { }); }; - var broadcast = (command, data/*, cb*/) => { + var broadcast = Env.broadcast = (command, data/*, cb*/) => { for (const worker of Object.values(Cluster.workers)) { sendCommand(worker, command, data /*, cb */); }