Capture pino logs in Fastify

Author: itssimon

src/common/pino.ts

@@ -0,0 +1,92 @@
+import { AsyncLocalStorage } from "async_hooks";
+import pino from "pino";
+
+import { LogRecord } from "./requestLogger.js";
+
+const logLevelMap: Record<number, string> = {
+  10: "trace",
+  20: "debug",
+  30: "info",
+  40: "warn",
+  50: "error",
+  60: "fatal",
+};
+
+export function patchPino(
+  logger: pino.BaseLogger,
+  logsContext: AsyncLocalStorage<LogRecord[]>,
+  filterLogs: (obj: any) => boolean = () => true,
+): void {
+  const loggerInternal = logger as any; // Cast to access internal pino properties
+  const originalStream = loggerInternal[pino.symbols.streamSym];
+
+  if (originalStream) {
+    const messageKey = loggerInternal[pino.symbols.messageKeySym];
+    const captureStream = new ApitallyLogCaptureStream(
+      logsContext,
+      messageKey,
+      filterLogs,
+    );
+    loggerInternal[pino.symbols.streamSym] = pino.multistream(
+      [
+        { level: 0, stream: originalStream },
+        { level: 0, stream: captureStream },
+      ],
+      {
+        levels: loggerInternal.levels,
+      },
+    );
+  }
+}
+
+class ApitallyLogCaptureStream implements pino.DestinationStream {
+  private logsContext: AsyncLocalStorage<LogRecord[]>;
+  private messageKey: string;
+  private filterLogs: (obj: any) => boolean;
+
+  constructor(
+    logsContext: AsyncLocalStorage<LogRecord[]>,
+    messageKey: string,
+    filterLogs: (obj: any) => boolean = () => true,
+  ) {
+    this.logsContext = logsContext;
+    this.messageKey = messageKey;
+    this.filterLogs = filterLogs;
+  }
+
+  write(msg: string): void {
+    const logs = this.logsContext.getStore();
+    if (!logs || !msg) {
+      return;
+    }
+
+    let obj: any;
+    try {
+      obj = JSON.parse(msg);
+    } catch (e) {
+      return;
+    }
+    if (obj === null || typeof obj !== "object" || !this.filterLogs(obj)) {
+      return;
+    }
+
+    try {
+      if (typeof obj[this.messageKey] === "string") {
+        logs.push({
+          timestamp: this.convertTime(obj.time),
+          level: logLevelMap[obj.level] || "info",
+          message: obj[this.messageKey],
+        });
+      }
+    } catch (e) {
+      // ignore
+    }
+  }
+
+  private convertTime(time: any): number {
+    if (typeof time === "number" && !isNaN(time)) {
+      return time / 1000; // Convert milliseconds to seconds
+    }
+    return Date.now() / 1000; // Fallback to current time
+  }
+}

src/fastify/plugin.ts

@@ -12,6 +12,7 @@ import { patchConsole } from "../common/consoleCapture.js";
 import { consumerFromStringOrObject } from "../common/consumerRegistry.js";
 import { parseContentLength } from "../common/headers.js";
 import { getPackageVersion } from "../common/packageVersions.js";
+import { patchPino } from "../common/pino.js";
 import {
   convertBody,
   convertHeaders,
@@ -52,6 +53,7 @@ const apitallyPlugin: FastifyPluginAsync<ApitallyConfig> = async (
 
   if (client.requestLogger.config.captureLogs) {
     patchConsole(logsContext);
+    patchPino(fastify.log, logsContext, filterLogs);
   }
 
   fastify.decorateRequest("apitallyConsumer", null);
@@ -315,6 +317,11 @@ function extractNestValidationErrors(message: any[]): ValidationError[] {
   }
 }
 
+function filterLogs(obj: any) {
+  // Filter out "request completed" logs
+  return !(obj.res && obj.responseTime);
+}
+
 export { apitallyPlugin };
 
 export default fp(apitallyPlugin, {

tests/fastify/app.test.ts

@@ -84,6 +84,8 @@ describe("Plugin for Fastify", () => {
     ]);
     expect(call[1].body).toBeInstanceOf(Buffer);
     expect(call[1].body!.toString()).toMatch(/^Hello John!/);
+    expect(call[3]).toBeDefined();
+    expect(call[3]).toHaveLength(2);
     expect(call[3]![0].level).toBe("log");
     expect(call[3]![0].message).toBe("Test 1");
     expect(call[3]![1].level).toBe("warn");
@@ -103,6 +105,10 @@ describe("Plugin for Fastify", () => {
     expect(call[0].body!.toString()).toMatch(/^{"name":"John","age":20}$/);
     expect(call[1].body).toBeInstanceOf(Buffer);
     expect(call[1].body!.toString()).toMatch(/^Hello John!/);
+    expect(call[3]).toBeDefined();
+    expect(call[3]).toHaveLength(1);
+    expect(call[3]![0].level).toBe("info");
+    expect(call[3]![0].message).toBe("Test 3");
   });
 
   it("Validation error counter", async () => {

tests/fastify/app.ts

@@ -6,6 +6,7 @@ import { CLIENT_ID, ENV } from "../utils.js";
 export const getApp = async () => {
   const app = Fastify({
     ajv: { customOptions: { allErrors: true } },
+    logger: true,
   });
 
   await app.register(apitallyPlugin, {
@@ -74,6 +75,7 @@ export const getApp = async () => {
     },
     async function (request) {
       const { name, age } = request.body;
+      request.log.info("Test 3");
       return `Hello ${name}! You are ${age} years old!`;
     },
   );