The_Great_Archive/cryptpad
1
0
Fork 0
mirror of https://github.com/cryptpad/cryptpad.git synced 2026-09-14 11:05:41 +05:00
Code Issues Packages Projects Releases Wiki Activity

Improve plugins options and support monitoring plugin

Browse Source
This commit is contained in:
yflory 2024-10-17 17:09:08 +02:00
parent fe481eda29
commit 1aecb4fbab
9 changed files with 169 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
lib/api.js
View File

@ -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)

10
lib/hk-util.js
View File

@ -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) {

26
lib/http-worker.js
View File

@ -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) {

8
lib/rpc.js
View File

@ -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

32
lib/workers/db-worker.js
View File

@ -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);
};

33
lib/workers/index.js
View File

@ -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) {

25
package-lock.json generated
View File

@ -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",

1
package.json
View File

@ -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",

36
server.js
View File

@ -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 */);
}