YunMai/msa/js-executor/api/jsInvokeMessageProcessor.ts

///
/// Copyright © 2016-2023 The Thingsboard Authors
///
/// Licensed under the Apache License, Version 2.0 (the "License");
/// you may not use this file except in compliance with the License.
/// You may obtain a copy of the License at
///
///     http://www.apache.org/licenses/LICENSE-2.0
///
/// Unless required by applicable law or agreed to in writing, software
/// distributed under the License is distributed on an "AS IS" BASIS,
/// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
/// See the License for the specific language governing permissions and
/// limitations under the License.
///

import config from 'config';
import { _logger } from '../config/logger';
import { JsExecutor, TbScript } from './jsExecutor';
import { performance } from 'perf_hooks';
import { isString, parseJsErrorDetails, toUUIDString, UUIDFromBuffer, UUIDToBits, isNotUUID } from './utils';
import { IQueue } from '../queue/queue.models';
import {
    JsCompileRequest,
    JsCompileResponse,
    JsInvokeRequest,
    JsInvokeResponse,
    JsReleaseRequest,
    JsReleaseResponse,
    RemoteJsRequest,
    RemoteJsResponse,
    TbMessage
} from './jsExecutor.models';
import Long from 'long';

const COMPILATION_ERROR = 0;
const RUNTIME_ERROR = 1;
const TIMEOUT_ERROR = 2;
const NOT_FOUND_ERROR = 3;

const statFrequency = Number(config.get('script.stat_print_frequency'));
const memoryUsageTraceFrequency = Number(config.get('script.memory_usage_trace_frequency'));
const scriptBodyTraceFrequency = Number(config.get('script.script_body_trace_frequency'));
const useSandbox = config.get('script.use_sandbox') === 'true';
const maxActiveScripts = Number(config.get('script.max_active_scripts'));
const slowQueryLogMs = Number(config.get('script.slow_query_log_ms'));
const slowQueryLogBody = config.get('script.slow_query_log_body') === 'true';
const maxResultSize = Number(config.get('js.max_result_size'));

export class JsInvokeMessageProcessor {

    private logger = _logger(`JsInvokeMessageProcessor`);
    private producer: IQueue;
    private executor = new JsExecutor(useSandbox);
    private scriptMap = new Map<string, TbScript>();
    private scriptIds: string[] = [];
    private executedScriptIdsCounter: number[] = [];
    private executedScriptsCounter = 0;
    private lastStatTime = performance.now();
    private compilationTime = 0;

    constructor(produced: IQueue) {
        this.producer = produced;
    }

    onJsInvokeMessage(message: any) {
        const tStart = performance.now();
        let requestId = '';
        let responseTopic: string;
        let expireTs;
        let headers;
        let request: RemoteJsRequest = {};
        let buf: Buffer;
        try {
            request = JSON.parse(Buffer.from(message.data).toString('utf8'));
            headers = message.headers;
            buf = Buffer.from(headers.data['requestId']);
            requestId = UUIDFromBuffer(buf);
            buf = Buffer.from(headers.data['responseTopic']);
            responseTopic = buf.toString('utf8');
            buf = Buffer.from(headers.data['expireTs']);
            expireTs = Long.fromBytes(Array.from(buf), false, false).toNumber();

            const now = Date.now();

            if (expireTs && expireTs <= now) {
                if (this.logger.isDebugEnabled()) {
                    this.logger.debug('Message expired! expireTs is %s, buf is %s. Now is %s, ms to expire left %s', expireTs, buf.toString('hex'), now, expireTs - now)
                }
                return;
            }

            this.logger.debug('[%s] Received request, responseTopic: [%s]', requestId, responseTopic);

            if (request.compileRequest) {
                this.processCompileRequest(requestId, responseTopic, headers, request.compileRequest);
            } else if (request.invokeRequest) {
                this.processInvokeRequest(requestId, responseTopic, headers, request.invokeRequest);
            } else if (request.releaseRequest) {
                this.processReleaseRequest(requestId, responseTopic, headers, request.releaseRequest);
            } else {
                this.logger.error('[%s] Unknown request received!', requestId);
            }

        } catch (err: any) {
            this.logger.error('[%s] Failed to process request: %s', requestId, err.message);
            this.logger.error(err.stack);
        }

        const tFinish = performance.now();
        const tTook = tFinish - tStart;

        if (tTook > slowQueryLogMs) {
            let functionName;
            if (request.invokeRequest) {
                try {
                    buf = Buffer.from(request.invokeRequest['functionName']);
                    functionName = buf.toString('utf8');
                } catch (err: any) {
                    this.logger.error('[%s] Failed to read functionName from message header: %s', requestId, err.message);
                    this.logger.error(err.stack);
                }
            }
            this.logger.warn('[%s] SLOW PROCESSING [%s]ms, functionName [%s]', requestId, tTook, functionName);
            if (slowQueryLogBody) {
                this.logger.info('Slow request body: %s', JSON.stringify(request, null, 4))
            }
        }
    }

    processCompileRequest(requestId: string, responseTopic: string, headers: any, compileRequest: JsCompileRequest) {
        const scriptId = JsInvokeMessageProcessor.getScriptId(compileRequest);
        this.logger.debug('[%s] Processing compile request, scriptId: [%s]', requestId, scriptId);
        if (this.scriptMap.has(scriptId)) {
            const compileResponse = JsInvokeMessageProcessor.createCompileResponse(scriptId, true);
            this.logger.debug('[%s] Script was already compiled, scriptId: [%s]', requestId, scriptId);
            this.sendResponse(requestId, responseTopic, headers, scriptId, compileResponse);
            return;
        }
        this.executor.compileScript(compileRequest.scriptBody).then(
            (script) => {
                this.cacheScript(scriptId, script);
                const compileResponse = JsInvokeMessageProcessor.createCompileResponse(scriptId, true);
                this.logger.debug('[%s] Sending success compile response, scriptId: [%s]', requestId, scriptId);
                this.sendResponse(requestId, responseTopic, headers, scriptId, compileResponse);
            },
            (err) => {
                const compileResponse = JsInvokeMessageProcessor.createCompileResponse(scriptId, false, COMPILATION_ERROR, err);
                this.logger.debug('[%s] Sending failed compile response, scriptId: [%s]', requestId, scriptId);
                this.sendResponse(requestId, responseTopic, headers, scriptId, compileResponse);
            }
        );
    }

    processInvokeRequest(requestId: string, responseTopic: string, headers: any, invokeRequest: JsInvokeRequest) {
        const scriptId = JsInvokeMessageProcessor.getScriptId(invokeRequest);
        this.logger.debug('[%s] Processing invoke request, scriptId: [%s]', requestId, scriptId);
        this.executedScriptsCounter++;
        if (this.executedScriptsCounter % statFrequency == 0) {
            const nowMs = performance.now();
            const msSinceLastStat = nowMs - this.lastStatTime;
            const requestsPerSec = msSinceLastStat == 0 ? statFrequency : statFrequency / msSinceLastStat * 1000;
            this.lastStatTime = nowMs;
            this.logger.info('STAT[%s]: requests [%s], took [%s]ms, request/s [%s], compilation [%s]ms', this.executedScriptsCounter, statFrequency, msSinceLastStat, requestsPerSec, this.compilationTime);
            this.compilationTime = 0;
        }

        if (this.executedScriptsCounter % scriptBodyTraceFrequency == 0) {
            this.logger.info('[%s] Executing script body: [%s]', scriptId, invokeRequest.scriptBody);
        }
        if (this.executedScriptsCounter % memoryUsageTraceFrequency == 0) {
            this.logger.info('Current memory usage: [%s]', process.memoryUsage());
        }

        this.getOrCompileScript(scriptId, invokeRequest.scriptBody).then(
            (script) => {
                this.executor.executeScript(script, invokeRequest.args, invokeRequest.timeout).then(
                    (result: string | undefined) => {
                        if (!result || result.length <= maxResultSize) {
                            const invokeResponse = JsInvokeMessageProcessor.createInvokeResponse(result, true);
                            this.logger.debug('[%s] Sending success invoke response, scriptId: [%s]', requestId, scriptId);
                            this.sendResponse(requestId, responseTopic, headers, scriptId, undefined, invokeResponse);
                        } else {
                            const err = {
                                name: 'Error',
                                message: 'script invocation result exceeds maximum allowed size of ' + maxResultSize + ' symbols'
                            }
                            const invokeResponse = JsInvokeMessageProcessor.createInvokeResponse("", false, RUNTIME_ERROR, err);
                            this.logger.debug('[%s] Script invocation result exceeds maximum allowed size of %s symbols, scriptId: [%s]', requestId, maxResultSize, scriptId);
                            this.sendResponse(requestId, responseTopic, headers, scriptId, undefined, invokeResponse);
                        }
                    },
                    (err: any) => {
                        let errorCode;
                        if (err && isString(err.message) && err.message.includes('Script execution timed out')) {
                            errorCode = TIMEOUT_ERROR;
                        } else {
                            errorCode = RUNTIME_ERROR;
                        }
                        const invokeResponse = JsInvokeMessageProcessor.createInvokeResponse("", false, errorCode, err);
                        this.logger.debug('[%s] Sending failed invoke response, scriptId: [%s], errorCode: [%s]', requestId, scriptId, errorCode);
                        this.sendResponse(requestId, responseTopic, headers, scriptId, undefined, invokeResponse);
                    }
                )
            },
            (err: any) => {
                let errorCode = COMPILATION_ERROR;
                if (err?.name === 'script body not found') {
                    errorCode = NOT_FOUND_ERROR;
                }
                const invokeResponse = JsInvokeMessageProcessor.createInvokeResponse("", false, errorCode, err);
                this.logger.debug('[%s] Sending failed invoke response, scriptId: [%s], errorCode: [%s]', requestId, scriptId, errorCode);
                this.sendResponse(requestId, responseTopic, headers, scriptId, undefined, invokeResponse);
            }
        );
    }

    processReleaseRequest(requestId: string, responseTopic: string, headers: any, releaseRequest: JsReleaseRequest) {
        const scriptId = JsInvokeMessageProcessor.getScriptId(releaseRequest);
        this.logger.debug('[%s] Processing release request, scriptId: [%s]', requestId, scriptId);
        if (this.scriptMap.has(scriptId)) {
            const index = this.scriptIds.indexOf(scriptId);
            if (index > -1) {
                this.scriptIds.splice(index, 1);
                this.executedScriptIdsCounter.splice(index, 1);
            }
            this.scriptMap.delete(scriptId);
        }
        const releaseResponse = JsInvokeMessageProcessor.createReleaseResponse(scriptId, true);
        this.logger.debug('[%s] Sending success release response, scriptId: [%s]', requestId, scriptId);
        this.sendResponse(requestId, responseTopic, headers, scriptId, undefined, undefined, releaseResponse);
    }

    sendResponse(requestId: string, responseTopic: string, headers: any, scriptId: string,
                 compileResponse?: JsCompileResponse, invokeResponse?: JsInvokeResponse, releaseResponse?: JsReleaseResponse) {
        const tStartSending = performance.now();
        const remoteResponse = JsInvokeMessageProcessor.createRemoteResponse(requestId, compileResponse, invokeResponse, releaseResponse);
        const rawResponse = Buffer.from(JSON.stringify(remoteResponse), 'utf8');
        this.logger.debug('[%s] Sending response to queue, scriptId: [%s]', requestId, scriptId);
        this.producer.send(responseTopic, requestId, rawResponse, headers).then(
            () => {
                this.logger.debug('[%s] Response sent to queue, took [%s]ms, scriptId: [%s]', requestId, (performance.now() - tStartSending), scriptId);
            },
            (err: any) => {
                if (err) {
                    this.logger.error('[%s] Failed to send response to queue: %s', requestId, err.message);
                    this.logger.error(err.stack);
                }
            }
        );
    }

    getOrCompileScript(scriptId: string, scriptBody: string): Promise<TbScript> {
        const self = this;
        return new Promise(function (resolve, reject) {
            const script = self.scriptMap.get(scriptId);
            if (script) {
                self.incrementUseScriptId(scriptId);
                resolve(script);
            } else if (scriptBody) {
                const startTime = performance.now();
                self.executor.compileScript(scriptBody).then(
                    (compiledScript) => {
                        self.compilationTime += (performance.now() - startTime);
                        self.cacheScript(scriptId, compiledScript);
                        resolve(compiledScript);
                    },
                    (err) => {
                        self.compilationTime += (performance.now() - startTime);
                        reject(err);
                    }
                );
            } else {
                const err = {
                    name: 'script body not found',
                    message: ''
                }
                reject(err);
            }
        });
    }

    cacheScript(scriptId: string, script: TbScript) {
        if (!this.scriptMap.has(scriptId)) {
            this.scriptIds.push(scriptId);
            this.executedScriptIdsCounter.push(0);
            while (this.scriptIds.length > maxActiveScripts) {
                this.logger.info('Active scripts count [%s] exceeds maximum limit [%s]', this.scriptIds.length, maxActiveScripts);
                this.deleteMinUsedScript();
            }
        }
        this.scriptMap.set(scriptId, script);
        this.logger.info("scriptMap size is [%s]", this.scriptMap.size);
    }

    private static createRemoteResponse(requestId: string, compileResponse?: JsCompileResponse,
                                        invokeResponse?: JsInvokeResponse, releaseResponse?: JsReleaseResponse): RemoteJsResponse {
        const requestIdBits = UUIDToBits(requestId);
        return {
            requestIdMSB: requestIdBits[0],
            requestIdLSB: requestIdBits[1],
            compileResponse: compileResponse,
            invokeResponse: invokeResponse,
            releaseResponse: releaseResponse
        };
    }

    private static createCompileResponse(scriptId: string, success: boolean, errorCode?: number, err?: any): JsCompileResponse {
        if (isNotUUID(scriptId)) {
            return {
                errorCode: errorCode,
                success: success,
                errorDetails: parseJsErrorDetails(err),
                scriptIdMSB: "0",
                scriptIdLSB: "0",
                scriptHash: scriptId
            };
        } else { // this is for backward compatibility (to be able to work with tb-node of previous version) - todo: remove in the next release
            let scriptIdBits = UUIDToBits(scriptId);
            return {
                errorCode: errorCode,
                success: success,
                errorDetails: parseJsErrorDetails(err),
                scriptIdMSB: scriptIdBits[0],
                scriptIdLSB: scriptIdBits[1],
                scriptHash: ""
            };
        }
    }

    private static createInvokeResponse(result: string | undefined, success: boolean, errorCode?: number, err?: any): JsInvokeResponse {
        return {
            errorCode: errorCode,
            success: success,
            errorDetails: parseJsErrorDetails(err),
            result: result
        };
    }

    private static createReleaseResponse(scriptId: string, success: boolean): JsReleaseResponse {
        if (isNotUUID(scriptId)) {
            return {
                success: success,
                scriptIdMSB: "0",
                scriptIdLSB: "0",
                scriptHash: scriptId,
            };
        } else { // todo: remove in the next release
            let scriptIdBits = UUIDToBits(scriptId);
            return {
                success: success,
                scriptIdMSB: scriptIdBits[0],
                scriptIdLSB: scriptIdBits[1],
                scriptHash: ""
            }
        }
    }

    private static getScriptId(request: TbMessage): string {
        return request.scriptHash ? request.scriptHash : toUUIDString(request.scriptIdMSB, request.scriptIdLSB);
    }

    private incrementUseScriptId(scriptId: string) {
        const index = this.scriptIds.indexOf(scriptId);
        if (this.executedScriptIdsCounter[index] < Number.MAX_SAFE_INTEGER) {
            this.executedScriptIdsCounter[index]++;
        }
    }

    private deleteMinUsedScript() {
        let min = Infinity;
        let minIndex = 0;
        const scriptIdsLength = this.executedScriptIdsCounter.length - 1; // ignored last added script
        for (let i = 0; i < scriptIdsLength; i++) {
            if (this.executedScriptIdsCounter[i] < min) {
                min = this.executedScriptIdsCounter[i];
                minIndex = i;
            }
        }
        const prevScriptId = this.scriptIds.splice(minIndex, 1)[0];
        this.executedScriptIdsCounter.splice(minIndex, 1)
        this.logger.info('Removing active script with id [%s]', prevScriptId);
        this.scriptMap.delete(prevScriptId);
    }
}