MVP-9756: Added event handling for function node

This commit is contained in:
michaeltreyvaud 2024-01-11 08:48:40 +00:00
parent fe189e0f38
commit e8edb4437b

View File

@ -14,375 +14,420 @@
* limitations under the License. * limitations under the License.
**/ **/
const clone = require("clone"); const clone = require("clone");
const PayloadValidator = require("../../PayloadValidator"); const PayloadValidator = require("../../PayloadValidator");
module.exports = function (RED) { const processServisbotActions = (originalMessage, message) => {
"use strict"; if (message.servisbot && message.servisbot.actions && Array.isArray(message.servisbot.actions) && message.servisbot.actions.length > 0) {
var util = require("util"); message.servisbot.actions.forEach((action) => {
var vm2 = require("vm2"); const [func, args] = action;
if (originalMessage && originalMessage.servisbot && originalMessage.servisbot[func]) {
function sendResults(node, _msgid, msgs) { originalMessage.servisbot[func](args);
if (msgs == null) { }
return; });
} else if (!util.isArray(msgs)) { }
msgs = [msgs]; }
}
var msgCount = 0; const handleCodeFile = async (node, sendResults, {
for (var m = 0; m < msgs.length; m++) { logger, msg, codefile, afterVm2
if (msgs[m]) { }) => {
if (!util.isArray(msgs[m])) { const {
msgs[m] = [msgs[m]]; payload: {
} system: { organization },
for (var n = 0; n < msgs[m].length; n++) { },
var msg = msgs[m][n]; event: {
if (msg !== null && msg !== undefined) { workers: [{ id: workerId }],
if ( },
typeof msg === "object" && } = msg;
!Buffer.isBuffer(msg) && const workId = workerId.split(':::')[0];
!util.isArray(msg) const nodeId = node.id.split(`${organization}-${workId}-`)[1];
) { try {
msg._msgid = _msgid; const beforeCodefile = process.hrtime();
msgCount++; const {
} else { payload: {
var type = typeof msg; result,
if (type === "object") { error
type = Buffer.isBuffer(msg) },
? "Buffer" requestId
: util.isArray(msg) } = await codefile.run({
? "Array" srcCode: node.func,
: "Date"; context: {
} msg,
node.error( node: {
RED._("function.error.non-message-returned", { type: type }) id: nodeId,
); name: node.name
} }
} }
} });
}
} const afterCodefile = process.hrtime(beforeCodefile);
if (msgCount > 0) {
node.send(msgs); const metrics = {
} lambdaRequestId: requestId,
} action: 'codefile-success',
organization,
function FunctionNode(n) { workerId,
RED.nodes.createNode(this, n); nodeId,
var node = this; rawCode: node.func,
this.name = n.name; vm2Runtime: `${
this.func = n.func; Math.floor((afterVm2[0] * 1e9 + afterVm2[1]) / 10000) / 100
var functionText = }ms`,
"var results = null;" + codefileRuntime: `${
"results = (function(msg){ " + Math.floor(
"var __msgid__ = msg._msgid;" + (afterCodefile[0] * 1e9 + afterCodefile[1]) / 10000
"var node = {" + ) / 100
"log:__node__.log," + }ms`,
"error:__node__.error," + };
"warn:__node__.warn," + if (result) {
"debug:__node__.debug," + const payloadValidator = new PayloadValidator(msg, node.id);
"trace:__node__.trace," + payloadValidator.verify(result);
"on:__node__.on," + // Re-attach logger to msgs as they get lost when passing over to the lambda
"status:__node__.status," + let messageToForward = result;
"send:function(msgs){ __node__.send(__msgid__,msgs);}" + if (Array.isArray(result)) {
"};\n" + // Array result, re-attach logger and process any servisbot.log actions returned from the lambda
this.func + messageToForward = result.map((_res) => {
"\n" + if (_res !== null && typeof _res === 'object') {
"})(msg);"; _res.logger = logger;
this.topic = n.topic; processServisbotActions(msg, _res);
this.outstandingTimers = []; if (msg.servisbot) {
this.outstandingIntervals = []; _res.servisbot = msg.servisbot;
var sandbox = { }
console: console, }
util: util, return _res;
//Buffer:Buffer, });
//Date: Date, } else if (typeof result === 'object') {
RED: { result.logger = logger;
util: RED.util, processServisbotActions(msg, result);
}, if (msg.servisbot) {
__node__: { result.servisbot = msg.servisbot;
log: function () { }
node.log.apply(node, arguments); }
}, sendResults(node, msg._msgid, messageToForward);
error: function () { } else {
node.error.apply(node, arguments); metrics.error = error;
}, metrics.action = 'codefile-error';
warn: function () { }
node.warn.apply(node, arguments); logger.info(metrics);
}, } catch (e) {
debug: function () { logger.error(e);
node.debug.apply(node, arguments); logger.error({
}, message: 'Error running codefile',
trace: function () { action: 'codefile-error',
node.trace.apply(node, arguments); error: e.message,
}, organization,
send: function (id, msgs) { workerId,
sendResults(node, id, msgs); nodeId,
}, rawCode: node.func,
on: function () { });
if (arguments[0] === "input") { }
throw new Error(RED._("function.error.inputListener")); };
}
node.on.apply(node, arguments); module.exports = function (RED) {
}, "use strict";
status: function () { var util = require("util");
node.status.apply(node, arguments); var vm2 = require("vm2");
},
}, function sendResults(node, _msgid, msgs) {
context: { if (msgs == null) {
set: function () { return;
node.context().set.apply(node, arguments); } else if (!util.isArray(msgs)) {
}, msgs = [msgs];
get: function () { }
return node.context().get.apply(node, arguments); var msgCount = 0;
}, for (var m = 0; m < msgs.length; m++) {
keys: function () { if (msgs[m]) {
return node.context().keys.apply(node, arguments); if (!util.isArray(msgs[m])) {
}, msgs[m] = [msgs[m]];
get global() { }
return node.context().global; for (var n = 0; n < msgs[m].length; n++) {
}, var msg = msgs[m][n];
get flow() { if (msg !== null && msg !== undefined) {
return node.context().flow; if (
}, typeof msg === "object" &&
}, !Buffer.isBuffer(msg) &&
flow: { !util.isArray(msg)
set: function () { ) {
node.context().flow.set.apply(node, arguments); msg._msgid = _msgid;
}, msgCount++;
get: function () { } else {
return node.context().flow.get.apply(node, arguments); var type = typeof msg;
}, if (type === "object") {
keys: function () { type = Buffer.isBuffer(msg)
return node.context().flow.keys.apply(node, arguments); ? "Buffer"
}, : util.isArray(msg)
}, ? "Array"
// global: { : "Date";
// set: function() { }
// node.context().global.set.apply(node,arguments); node.error(
// }, RED._("function.error.non-message-returned", { type: type })
// get: function() { );
// return node.context().global.get.apply(node,arguments); }
// }, }
// keys: function() { }
// return node.context().global.keys.apply(node,arguments); }
// } }
// }, if (msgCount > 0) {
setTimeout: function () { node.send(msgs);
var func = arguments[0]; }
var timerId; }
arguments[0] = function () {
sandbox.clearTimeout(timerId); function FunctionNode(n) {
try { RED.nodes.createNode(this, n);
func.apply(this, arguments); var node = this;
} catch (err) { this.name = n.name;
node.error(err, {}); this.func = n.func;
} var functionText =
}; "var results = null;" +
timerId = setTimeout.apply(this, arguments); "results = (function(msg){ " +
node.outstandingTimers.push(timerId); "var __msgid__ = msg._msgid;" +
return timerId; "var node = {" +
}, "log:__node__.log," +
clearTimeout: function (id) { "error:__node__.error," +
clearTimeout(id); "warn:__node__.warn," +
var index = node.outstandingTimers.indexOf(id); "debug:__node__.debug," +
if (index > -1) { "trace:__node__.trace," +
node.outstandingTimers.splice(index, 1); "on:__node__.on," +
} "status:__node__.status," +
}, "send:function(msgs){ __node__.send(__msgid__,msgs);}" +
setInterval: function () { "};\n" +
var func = arguments[0]; this.func +
var timerId; "\n" +
arguments[0] = function () { "})(msg);";
try { this.topic = n.topic;
func.apply(this, arguments); this.outstandingTimers = [];
} catch (err) { this.outstandingIntervals = [];
node.error(err, {}); var sandbox = {
} console: console,
}; util: util,
timerId = setInterval.apply(this, arguments); //Buffer:Buffer,
node.outstandingIntervals.push(timerId); //Date: Date,
return timerId; RED: {
}, util: RED.util,
clearInterval: function (id) { },
clearInterval(id); __node__: {
var index = node.outstandingIntervals.indexOf(id); log: function () {
if (index > -1) { node.log.apply(node, arguments);
node.outstandingIntervals.splice(index, 1); },
} error: function () {
}, node.error.apply(node, arguments);
}; },
warn: function () {
if (util.hasOwnProperty("promisify")) { node.warn.apply(node, arguments);
sandbox.setTimeout[util.promisify.custom] = function (after, value) { },
return new Promise(function (resolve, reject) { debug: function () {
sandbox.setTimeout(function () { node.debug.apply(node, arguments);
resolve(value); },
}, after); trace: function () {
}); node.trace.apply(node, arguments);
}; },
} send: function (id, msgs) {
try { sendResults(node, id, msgs);
this.on("input", async function (msg) { },
try { on: function () {
const originalMessage = clone(msg); if (arguments[0] === "input") {
const payloadValidator = new PayloadValidator(msg, this.id); throw new Error(RED._("function.error.inputListener"));
var start = process.hrtime(); }
sandbox.msg = msg; node.on.apply(node, arguments);
const vm2Instance = new vm2.VM({ sandbox, timeout: 5000 }); },
const beforeVm2 = process.hrtime(); status: function () {
const result = vm2Instance.run(functionText); node.status.apply(node, arguments);
const afterVm2 = process.hrtime(beforeVm2); },
payloadValidator.verify(result); },
sendResults(this, msg._msgid, result); context: {
const logger = clone(msg.logger); set: function () {
let lambdaRequestId; node.context().set.apply(node, arguments);
let { },
payload: { get: function () {
system: { organization }, return node.context().get.apply(node, arguments);
}, },
event: { keys: function () {
workers: [{ id: workerId }], return node.context().keys.apply(node, arguments);
}, },
} = originalMessage; get global() {
return node.context().global;
const { },
settings: { get flow() {
api: { codefile = false }, return node.context().flow;
}, },
} = RED; },
flow: {
if (codefile) { set: function () {
workerId = workerId.split(":::")[0]; node.context().flow.set.apply(node, arguments);
const nodeId = this.id.split(`${organization}-${workerId}-`)[1]; },
try { get: function () {
const messageToSend = clone(msg); return node.context().flow.get.apply(node, arguments);
delete messageToSend.logger; },
keys: function () {
const beforeCodefile = process.hrtime(); return node.context().flow.keys.apply(node, arguments);
const { },
payload: { },
result, // global: {
error // set: function() {
}, // node.context().global.set.apply(node,arguments);
requestId // },
} = await codefile.run({ srcCode: this.func, context: { msg } }); // get: function() {
// return node.context().global.get.apply(node,arguments);
const afterCodefile = process.hrtime(beforeCodefile); // },
// keys: function() {
const metrics = { // return node.context().global.keys.apply(node,arguments);
lambdaRequestId: requestId, // }
action:'codefile-success', // },
organization, setTimeout: function () {
workerId: workerId, var func = arguments[0];
nodeId: nodeId, var timerId;
rawCode: this.func, arguments[0] = function () {
vm2Runtime: `${ sandbox.clearTimeout(timerId);
Math.floor((afterVm2[0] * 1e9 + afterVm2[1]) / 10000) / 100 try {
}ms`, func.apply(this, arguments);
codefileRuntime: `${ } catch (err) {
Math.floor( node.error(err, {});
(afterCodefile[0] * 1e9 + afterCodefile[1]) / 10000 }
) / 100 };
}ms`, timerId = setTimeout.apply(this, arguments);
}; node.outstandingTimers.push(timerId);
if(result){ return timerId;
// not required right now since we dont go via this path },
// const responseMessage = result.msg clearTimeout: function (id) {
// responseMessage.logger = logger; clearTimeout(id);
// payloadValidator.verify(responseMessage); var index = node.outstandingTimers.indexOf(id);
// sendResults(this,msg._msgid, responseMessage); if (index > -1) {
} node.outstandingTimers.splice(index, 1);
else{ }
metrics.error = error; },
metrics.action = 'codefile-error'; setInterval: function () {
} var func = arguments[0];
logger.info(metrics); var timerId;
} catch (e) { arguments[0] = function () {
logger.error(e) try {
logger.error({ func.apply(this, arguments);
message: "Error running codefile", } catch (err) {
action:'codefile-error', node.error(err, {});
error: e.message, }
organization, };
workerId: workerId, timerId = setInterval.apply(this, arguments);
nodeId: nodeId, node.outstandingIntervals.push(timerId);
rawCode: this.func, return timerId;
}); },
} clearInterval: function (id) {
} clearInterval(id);
var index = node.outstandingIntervals.indexOf(id);
// sendResults(this,msg._msgid, responseMessage); if (index > -1) {
var duration = process.hrtime(start); node.outstandingIntervals.splice(index, 1);
var converted = }
Math.floor((duration[0] * 1e9 + duration[1]) / 10000) / 100; },
this.metric("duration", msg, converted); };
if (process.env.NODE_RED_FUNCTION_TIME) {
this.status({ fill: "yellow", shape: "dot", text: "" + converted }); if (util.hasOwnProperty("promisify")) {
} sandbox.setTimeout[util.promisify.custom] = function (after, value) {
} catch (err) { return new Promise(function (resolve, reject) {
//remove unwanted part sandbox.setTimeout(function () {
var index = err.stack.search( resolve(value);
/\n\s*at ContextifyScript.Script.runInContext/ }, after);
); });
err.stack = err.stack };
.slice(0, index) }
.split("\n") try {
.slice(0, -1) this.on("input", async function (msg) {
.join("\n"); try {
var stack = err.stack.split(/\r?\n/); const originalMessage = clone(msg);
const payloadValidator = new PayloadValidator(msg, this.id);
//store the error in msg to be used in flows var start = process.hrtime();
msg.error = err; sandbox.msg = msg;
// const vm2Instance = new vm2.VM({ sandbox, timeout: 5000 });
var line = 0; const beforeVm2 = process.hrtime();
var errorMessage; // const result = vm2Instance.run(functionText);
var stack = err.stack.split(/\r?\n/); const afterVm2 = process.hrtime(beforeVm2);
if (stack.length > 0) { // payloadValidator.verify(result);
while ( // sendResults(this, msg._msgid, result);
line < stack.length && const logger = clone(msg.logger);
stack[line].indexOf("ReferenceError") !== 0
) { const {
line++; settings: {
} api: { codefile = false },
},
if (line < stack.length) { } = RED;
errorMessage = stack[line];
var m = /:(\d+):(\d+)$/.exec(stack[line + 1]); if (codefile) {
if (m) { await handleCodeFile(this, sendResults, {
var lineno = Number(m[1]) - 1; logger,
var cha = m[2]; msg: originalMessage,
errorMessage += " (line " + lineno + ", col " + cha + ")"; codefile,
} afterVm2,
} });
} }
if (!errorMessage) {
errorMessage = err.toString(); var duration = process.hrtime(start);
} var converted =
Math.floor((duration[0] * 1e9 + duration[1]) / 10000) / 100;
// gives access to the msg object in custom logger this.metric("duration", msg, converted);
const temp = errorMessage; if (process.env.NODE_RED_FUNCTION_TIME) {
errorMessage = msg; this.status({ fill: "yellow", shape: "dot", text: "" + converted });
errorMessage.toString = () => temp; // preserve original error message in logs }
msg.errorMessage = temp; } catch (err) {
//remove unwanted part
this.error(errorMessage, msg); var index = err.stack.search(
} /\n\s*at ContextifyScript.Script.runInContext/
}); );
this.on("close", function () { err.stack = err.stack
while (node.outstandingTimers.length > 0) { .slice(0, index)
clearTimeout(node.outstandingTimers.pop()); .split("\n")
} .slice(0, -1)
while (node.outstandingIntervals.length > 0) { .join("\n");
clearInterval(node.outstandingIntervals.pop()); var stack = err.stack.split(/\r?\n/);
}
this.status({}); //store the error in msg to be used in flows
}); msg.error = err;
} catch (err) {
// eg SyntaxError - which v8 doesn't include line number information var line = 0;
// so we can't do better than this var errorMessage;
this.error(err); var stack = err.stack.split(/\r?\n/);
} if (stack.length > 0) {
} while (
RED.nodes.registerType("function", FunctionNode); line < stack.length &&
RED.library.register("functions"); stack[line].indexOf("ReferenceError") !== 0
}; ) {
line++;
}
if (line < stack.length) {
errorMessage = stack[line];
var m = /:(\d+):(\d+)$/.exec(stack[line + 1]);
if (m) {
var lineno = Number(m[1]) - 1;
var cha = m[2];
errorMessage += " (line " + lineno + ", col " + cha + ")";
}
}
}
if (!errorMessage) {
errorMessage = err.toString();
}
// gives access to the msg object in custom logger
const temp = errorMessage;
errorMessage = msg;
errorMessage.toString = () => temp; // preserve original error message in logs
msg.errorMessage = temp;
this.error(errorMessage, msg);
}
});
this.on("close", function () {
while (node.outstandingTimers.length > 0) {
clearTimeout(node.outstandingTimers.pop());
}
while (node.outstandingIntervals.length > 0) {
clearInterval(node.outstandingIntervals.pop());
}
this.status({});
});
} catch (err) {
// eg SyntaxError - which v8 doesn't include line number information
// so we can't do better than this
this.error(err);
}
}
RED.nodes.registerType("function", FunctionNode);
RED.library.register("functions");
};