Detect blocking sync operations during function executions #558

package.json

@@ -7,12 +7,14 @@
     "dependencies": {
         "@grpc/grpc-js": "^1.2.7",
         "@grpc/proto-loader": "^0.6.4",
+        "blocked-at": "^1.2.0",
         "fs-extra": "^10.0.1",
         "long": "^4.0.0",
         "minimist": "^1.2.5",
         "uuid": "^8.3.0"
     },
     "devDependencies": {
+        "@types/blocked-at": "^1.0.1",
         "@types/chai": "^4.2.22",
         "@types/chai-as-promised": "^7.1.5",
         "@types/fs-extra": "^9.0.13",
@@ -76,4 +78,4 @@
         "dist"
     ],
     "main": "dist/src/nodejsWorker.js"
-}
+}

src/Worker.ts

@@ -6,9 +6,11 @@ import { FunctionLoader } from './FunctionLoader';
 import { CreateGrpcEventStream } from './GrpcClient';
 import { setupCoreModule } from './setupCoreModule';
 import { setupEventStream } from './setupEventStream';
+import { startBlockedMonitor } from './utils/blockedMonitor';
 import { ensureErrorType } from './utils/ensureErrorType';
 import { InternalException } from './utils/InternalException';
 import { systemError, systemLog } from './utils/Logger';
+import { isEnvironmentVariableSet } from './utils/util';
 import { WorkerChannel } from './WorkerChannel';
 
 export function startNodeWorker(args) {
@@ -69,4 +71,8 @@ export function startNodeWorker(args) {
     process.on('exit', (code) => {
         systemLog(`Worker ${workerId} exited with code ${code}`);
     });
+
+    if (isEnvironmentVariableSet(process.env.AZURE_FUNCTIONS_NODE_BLOCK_LOG)) {
+        startBlockedMonitor(channel);
+    }
 }

src/utils/blockedMonitor.ts

@@ -0,0 +1,46 @@
+// Copyright (c) .NET Foundation. All rights reserved.
+// Licensed under the MIT License.
+
+import { AzureFunctionsRpcMessages as rpc } from './../../azure-functions-language-worker-protobuf/src/rpc';
+import LogCategory = rpc.RpcLog.RpcLogCategory;
+import LogLevel = rpc.RpcLog.Level;
+import blockedAt = require('blocked-at');
+
+export function startBlockedMonitor(
+    channel: { log: (log: rpc.IRpcLog) => void },
+    threshold = 500,
+    intreval = 10000
+): NodeJS.Timer {
+    function logBlockedWarning(message: string) {
+        channel.log({
+            message,
+            level: LogLevel.Warning,
+            logCategory: LogCategory.System,
+        });
+    }
+
+    logBlockedWarning(
+        `Monitoring for blocking code is turned on, with a threshold of ${threshold} ms. This will have a negative impact on performance. Adjust "AZURE_FUNCTIONS_NODE_BLOCK_LOG" to turn it off. ` +
+            'IMPORTANT NOTE: The stack traces are only an approximation and you should analyze all synchronous operations'
+    );
+
+    let blockedHistory: { time: string; duration: number; stack: string[] }[] = [];
+
+    //threshold - minimum miliseconds of blockage to report.
+    //other parameters are default, more details on https://github.com/naugtur/blocked-at.
+    blockedAt(
+        (ms, stack) => {
+            const date = new Date();
+            blockedHistory.push({ time: date.toISOString(), duration: ms, stack: stack });
+        },
+        { threshold: threshold }
+    );
+
+    // Log blockedHistory if it's not empty each 10 seconds
+    return setInterval(() => {
+        if (blockedHistory.length > 0) {
+            logBlockedWarning(`Blocking code monitoring history: ${JSON.stringify(blockedHistory)}`);
+            blockedHistory = [];
+        }
+    }, intreval);
+}

src/utils/util.ts

@@ -0,0 +1,6 @@
+// Copyright (c) .NET Foundation. All rights reserved.
+// Licensed under the MIT License.
+
+export function isEnvironmentVariableSet(val: string | boolean | number | undefined | null): boolean {
+    return !/^(false|0)?$/i.test(val === undefined || val === null ? '' : String(val));
+}

test/blockMonitorTest.ts

@@ -0,0 +1,44 @@
+// Copyright (c) .NET Foundation. All rights reserved.
+// Licensed under the MIT License.
+
+import { expect } from 'chai';
+import 'mocha';
+import { AzureFunctionsRpcMessages as rpc } from './../azure-functions-language-worker-protobuf/src/rpc';
+import { startBlockedMonitor } from './../src/utils/blockedMonitor';
+import LogLevel = rpc.RpcLog.Level;
+
+describe('Event loop blocking operation monitor', () => {
+    it('startBlockMonitor logs warning', async () => {
+        console.log('start ' + new Date().getSeconds() + ':' + new Date().getMilliseconds());
+        let timer: NodeJS.Timer | null = null;
+        let isTimerDestroyed = false;
+        const logFun = function (log: rpc.IRpcLog): void {
+            expect(log.level).to.equal(LogLevel.Warning);
+            if (log.message && log.message.startsWith('Blocking code monitoring history')) {
+                if (timer) {
+                    clearInterval(timer);
+                    isTimerDestroyed = true;
+                }
+            }
+        };
+
+        timer = startBlockedMonitor({ log: logFun }, 100, 100);
+        await new Promise((resolve) => {
+            //Adding new event to event loop to start monitoring
+            setTimeout(() => {
+                resolve(true);
+            }, 1);
+        });
+        const end = Date.now() + 500;
+        while (Date.now() < end) {} // blocking code
+
+        await new Promise((resolve) => {
+            //assert
+            setTimeout(() => {
+                if (isTimerDestroyed) {
+                    resolve(true);
+                }
+            }, 500);
+        });
+    });
+});