[system-utility] Profiler writes to file system instead of mongodb. new endpoint: take-heap-snapshot

Author: Cihad Tekin

CHANGELOG.md

@@ -1,7 +1,12 @@
+## Version xx.xx.xx
+Features:
+- [system-utility] New endpoint: /take-heap-snapshot.
+- [system-utility] Using nodejs fs to write profiler files instead of gridfs.
+
 ## Version 24.05.22
 Features:
 - [core] Add self tracking capability
-  
+
 Fixes:
 - [push] Using apns-id header as message result in debug mode
 - [server-stats] Fix data point calculation in job

plugins/system-utility/api/api.js

@@ -31,11 +31,7 @@ function handleMessage(msg) {
         if (msg.cmd === "stopProfiler" || msg.cmd === "startProfiler") {
             args = [processName];
         }
-
-        systemUtility[msg.cmd](...args).catch(err => {
-            log.e(err);
-            console.error(err);
-        });
+        systemUtility[msg.cmd](...args).catch(err => log.e(err));
     }
     else if (typeof msg === "object" && msg.cmd === "setNumberOfWorkers") {
         numberOfWorkers = msg.params.numberOfWorkers;
@@ -178,29 +174,29 @@ function stopWithTimeout(type, fromTimeout = false) {
             });
             return true;
 
-        case 'list-files':
-            validate(params, () => {
-                systemUtility.listProfilerFiles()
-                    .then(res => common.returnMessage(params, 200, res))
-                    .catch(err => {
-                        log.e(err);
-                        common.returnMessage(params, 404, "Profiler files not found");
+        case "take-heap-snapshot":
+            validate(params, async () => {
+                try {
+                    params.res.writeHead(200, {
+                        "Content-Type": "plain/text; charset=utf-8",
+                        "Content-Disposition": "attachment; filename=heap.heapsnapshot"
                     });
+                    systemUtility.takeHeapSnapshot(params.res);
+                }
+                catch (err) {
+                    log.e(err);
+                    common.returnMessage(params, 500, err.toString());
+                }
             });
             return true;
 
-        case 'download':
+        case 'list-files':
             validate(params, () => {
-                systemUtility.downloadProfilerFile(params.qstring.filename)
-                    .then(({ data, filename }) => {
-                        common.returnRaw(params, 200, data, {
-                            'Content-Type': 'plain/text; charset=utf-8',
-                            'Content-disposition': 'attachment; filename=' + filename
-                        });
-                    })
+                systemUtility.listProfilerFiles()
+                    .then(res => common.returnMessage(params, 200, res))
                     .catch(err => {
                         log.e(err);
-                        common.returnMessage(params, 404, "File not found");
+                        common.returnMessage(params, 404, "Profiler files couldn't be found");
                     });
             });
             return true;
@@ -210,7 +206,7 @@ function stopWithTimeout(type, fromTimeout = false) {
                 try {
                     const tarStream = await systemUtility.profilerFilesTarStream();
                     if (tarStream === null) {
-                        common.returnMessage(params, 404, "Profiler files not found");
+                        common.returnMessage(params, 404, "Profiler files couldn't be found");
                     }
                     else {
                         params.res.writeHead(200, {
@@ -223,8 +219,7 @@ function stopWithTimeout(type, fromTimeout = false) {
                 }
                 catch (err) {
                     log.e(err);
-                    console.error(err);
-                    common.returnMessage(params, 500, "Server error");
+                    common.returnMessage(params, 500, err.toString());
                 }
             });
             return true;

plugins/system-utility/api/system.utility.js

@@ -1,11 +1,13 @@
 var common = require('../../../api/utils/common.js');
 const inspector = require('inspector');
-const countlyFs = require('../../../api/utils/countlyFs.js');
 var exec = require('child_process').exec;
 const tar = require("tar-stream");
 const session = new inspector.Session();
+const path = require("path");
+const fs = require("fs/promises");
+const { Writable } = require('stream');
 
-const PROFILER_DIR = "nodeprofile";
+const PROFILER_DIR = path.join(__dirname, "../../../log/nodeprofile");
 
 var _id = null;
 
@@ -424,23 +426,20 @@ function sessionPost(cmd) {
 }
 
 /**
- * Saves the result to gridfs
- * @param {string} filename file name with extension
- * @param {object} result result object returned by the profiler
- * @returns {Promise<string>} filename
+ * Takes a snapshot and writes its content to the passed stream.
+ * IMPORTANT: it will call the "end" for the passed stream object when it's done
+ * @param {Writable} stream Writable stream to write snapshot content to
+ * @returns {void}
  */
-function saveProfilerResult(filename, result) {
-    return new Promise((res, rej) => {
-        countlyFs.gridfs.saveData(
-            PROFILER_DIR, filename, JSON.stringify(result),
-            { writeMode: "overwrite" },
-            function(err) {
-                if (err) {
-                    return rej(err);
-                }
-                res(filename);
-            }
-        );
+function takeHeapSnapshot(stream) {
+    session.connect();
+    session.on("HeapProfiler.addHeapSnapshotChunk", m => stream.write(m.params.chunk));
+    session.post("HeapProfiler.takeHeapSnapshot", null, (err) => {
+        session.disconnect();
+        stream.end();
+        if (err) {
+            console.error(err);
+        }
     });
 }
 
@@ -470,30 +469,22 @@ async function startProfiler() {
 async function stopProfiler(processName) {
     const errors = [];
 
-    // clear old files
     try {
-        await new Promise(
-            (res, rej) => countlyFs.gridfs.deleteAll(
-                PROFILER_DIR,
-                null,
-                err => err ? rej(err) : res()
-            )
-        );
+        await fs.rm(PROFILER_DIR, { force: true, recursive: true }); // clear old files
+        await fs.mkdir(PROFILER_DIR, { recursive: true });
     }
     catch (err) {
-        if (err.code === 26) { // NamespaceNotFound: thrown when there's no collection initially
-            // do nothing...
-        }
-        else {
-            throw err;
-        }
+        errors.push(err);
     }
 
     // coverage
     try {
         const coverage = await sessionPost("Profiler.takePreciseCoverage");
-        await saveProfilerResult(processName + ".coverage", coverage?.result);
         await sessionPost("Profiler.stopPreciseCoverage");
+        await fs.writeFile(
+            path.join(PROFILER_DIR, processName + ".coverage"),
+            JSON.stringify(coverage?.result)
+        );
     }
     catch (err) {
         errors.push(err);
@@ -502,8 +493,11 @@ async function stopProfiler(processName) {
     // cpu profiler
     try {
         const cpuProfile = await sessionPost("Profiler.stop");
-        await saveProfilerResult(processName + ".cpuprofile", cpuProfile?.profile);
         await sessionPost("Profiler.disable");
+        await fs.writeFile(
+            path.join(PROFILER_DIR, processName + ".cpuprofile"),
+            JSON.stringify(cpuProfile?.profile)
+        );
     }
     catch (err) {
         errors.push(err);
@@ -512,8 +506,11 @@ async function stopProfiler(processName) {
     // heap profiler
     try {
         const heapProfile = await sessionPost("HeapProfiler.stopSampling");
-        await saveProfilerResult(processName + ".heapprofile", heapProfile?.profile);
         await sessionPost("HeapProfiler.disable");
+        await fs.writeFile(
+            path.join(PROFILER_DIR, processName + ".heapprofile"),
+            JSON.stringify(heapProfile?.profile)
+        );
     }
     catch (err) {
         errors.push(err);
@@ -526,39 +523,24 @@ async function stopProfiler(processName) {
     }
 }
 
-/**
- * Returns the data of a file in PROFILER_DIR collection
- * @param {string} filename file name with extension
- * @returns {Promise<{data:string, filename:string}>} file object with name and content
- */
-function downloadProfilerFile(filename) {
-    return new Promise((resolve, reject) => {
-        countlyFs.gridfs.getData(PROFILER_DIR, filename, {}, (err, data) => {
-            if (err) {
-                return reject("File not found");
-            }
-            resolve({ data, filename });
-        });
-    });
-}
 /**
  * Returns the names, creation dates and size of all files in the PROFILER_DIR collection
- * @returns {Promise<Array<{createdOn: Date, filename: string, size: number }>>} file info
+ * @returns {Promise<Array<{createdOn: Date, filename: string, size: number, path: string }>>} file info
  */
-function listProfilerFiles() {
-    return new Promise((resolve, reject) => {
-        countlyFs.gridfs.listFiles(PROFILER_DIR, (err, files) => {
-            if (err) {
-                return reject(err);
-            }
-            resolve(files);
-        });
-    });
+async function listProfilerFiles() {
+    const files = await fs.readdir(PROFILER_DIR);
+    return Promise.all(
+        files.map(async (filename) => {
+            const fullPath = path.join(PROFILER_DIR, filename);
+            const { birthtime, size } = await fs.stat(fullPath);
+            return { createdOn: birthtime, size, filename, path: fullPath }
+        })
+    );
 }
 
 /**
  * Returns the tarball read stream for all profiler files
- * @returns {tar.Pack} tar stream
+ * @returns {Promise<tar.Pack>} tar stream
  */
 async function profilerFilesTarStream() {
     const files = await listProfilerFiles();
@@ -569,14 +551,8 @@ async function profilerFilesTarStream() {
     const pack = tar.pack();
     for (let i = 0; i < files.length; i++) {
         const entry = pack.entry({ name: files[i].filename, size: files[i].size });
-        const stream = await new Promise((res, rej) => {
-            countlyFs.gridfs.getStream(
-                PROFILER_DIR,
-                files[i].filename,
-                {},
-                (err, fileStream) => err ? rej(err) : res(fileStream)
-            );
-        });
+        const handle = await fs.open(files[i].path);
+        const stream = handle.createReadStream();
         stream.pipe(entry);
         stream.on("end", () => {
             entry.end();
@@ -622,9 +598,9 @@ function stopInspector() {
     });
 }
 
+exports.takeHeapSnapshot = takeHeapSnapshot;
 exports.startProfiler = startProfiler;
 exports.stopProfiler = stopProfiler;
-exports.downloadProfilerFile = downloadProfilerFile;
 exports.listProfilerFiles = listProfilerFiles;
 exports.startInspector = startInspector;
 exports.stopInspector = stopInspector;