triggerdotdev--trigger.dev/apps/webapp/app/eventLoopMonitor.server.ts

import { createHook } from "node:async_hooks";
import { singleton } from "./utils/singleton";
import { tracer } from "./v3/tracer.server";
import { env } from "./env.server";
import { context, Context } from "@opentelemetry/api";
import { performance } from "node:perf_hooks";
import { logger } from "./services/logger.server";
import { signalsEmitter } from "./services/signals.server";

const THRESHOLD_NS = env.EVENT_LOOP_MONITOR_THRESHOLD_MS * 1e6;

// ANSI color codes for terminal output
const RED = "\x1b[31m";
const YELLOW = "\x1b[33m";
const RESET = "\x1b[0m";

function notifyEventLoopBlocked(timeMs: number, asyncType: string): void {
  if (env.EVENT_LOOP_MONITOR_NOTIFY_ENABLED !== "1") {
    return;
  }

  console.warn(
    `${RED}⚠️  Event loop blocked${RESET} for ${YELLOW}${timeMs.toFixed(
      1
    )}ms${RESET} (${asyncType})`
  );
}

const cache = new Map<number, { type: string; start?: [number, number]; parentCtx?: Context }>();

function init(asyncId: number, type: string, triggerAsyncId: number, resource: any) {
  cache.set(asyncId, {
    type,
  });
}

function destroy(asyncId: number) {
  cache.delete(asyncId);
}

function before(asyncId: number) {
  const cached = cache.get(asyncId);

  if (!cached) {
    return;
  }

  cache.set(asyncId, {
    ...cached,
    start: process.hrtime(),
    parentCtx: context.active(),
  });
}

function after(asyncId: number) {
  const cached = cache.get(asyncId);

  if (!cached) {
    return;
  }

  cache.delete(asyncId);

  if (!cached.start) {
    return;
  }

  const diff = process.hrtime(cached.start);
  const diffNs = diff[0] * 1e9 + diff[1];
  if (diffNs > THRESHOLD_NS) {
    const time = diffNs / 1e6; // in ms

    const newSpan = tracer.startSpan(
      "event-loop-blocked",
      {
        startTime: new Date(new Date().getTime() - time),
        attributes: {
          asyncType: cached.type,
          label: "EventLoopMonitor",
        },
      },
      cached.parentCtx
    );

    newSpan.end();

    notifyEventLoopBlocked(time, cached.type);
  }
}

export const eventLoopMonitor = singleton("eventLoopMonitor", () => {
  const hook = createHook({ init, before, after, destroy });

  let stopEventLoopUtilizationMonitoring: () => void;

  return {
    enable: () => {
      console.log("🥸  Initializing event loop monitor");

      hook.enable();

      stopEventLoopUtilizationMonitoring = startEventLoopUtilizationMonitoring();
    },
    disable: () => {
      console.log("🥸  Disabling event loop monitor");

      hook.disable();

      stopEventLoopUtilizationMonitoring?.();
    },
  };
});

function startEventLoopUtilizationMonitoring() {
  let lastEventLoopUtilization = performance.eventLoopUtilization();

  const interval = setInterval(() => {
    const currentEventLoopUtilization = performance.eventLoopUtilization();

    const diff = performance.eventLoopUtilization(
      currentEventLoopUtilization,
      lastEventLoopUtilization
    );
    const utilization = Number.isFinite(diff.utilization) ? diff.utilization : 0;

    if (Math.random() < env.EVENT_LOOP_MONITOR_UTILIZATION_SAMPLE_RATE) {
      logger.info("nodejs.event_loop.utilization", { utilization });
    }

    lastEventLoopUtilization = currentEventLoopUtilization;
  }, env.EVENT_LOOP_MONITOR_UTILIZATION_INTERVAL_MS);

  signalsEmitter.on("SIGTERM", () => {
    clearInterval(interval);
  });
  signalsEmitter.on("SIGINT", () => {
    clearInterval(interval);
  });

  return () => {
    clearInterval(interval);
  };
}