feat: TRAC-31-Pino-for-Node-logging (#531)

* feat: Support Pino for Node logging in the OTel Distro

---------

Co-authored-by: Eugene Orlovsky <eugene.orlovsky@staircase.co>

package.json

@@ -39,6 +39,7 @@
     "@opentelemetry/instrumentation-mongodb": "0.39.0",
     "@opentelemetry/instrumentation-nestjs-core": "^0.34.0",
     "@opentelemetry/instrumentation-pg": "^0.38.0",
+    "@opentelemetry/instrumentation-pino": "0.42.0",
     "@opentelemetry/instrumentation-redis-4": "^0.35.3",
     "@opentelemetry/instrumentation-winston": "0.36.0",
     "@opentelemetry/resource-detector-aws": "1.3.2",

src/bootstrap.ts

@@ -41,6 +41,7 @@ import {
 } from './instrumentations/aws-sdk';
 import LumigoWinstonInstrumentation from './instrumentations/winston/WinstonInstrumentation';
 import LumigoBunyanInstrumentation from './instrumentations/bunyan/BunyanInstrumentation';
+import LumigoPinoInstrumentation from './instrumentations/pino/PinoInstrumentation';
 
 import { LumigoW3CTraceContextPropagator } from './propagator/w3cTraceContextPropagator';
 import {
@@ -142,6 +143,7 @@ export const init = async (): Promise<LumigoSdkInitialization> => {
       // Loggers
       new LumigoWinstonInstrumentation(),
       new LumigoBunyanInstrumentation(),
+      new LumigoPinoInstrumentation(),
     ].filter((i) => i.isApplicable());
 
     /*

src/instrumentations/@grpc/grpc-js/wrapGrpcServer.ts

@@ -21,9 +21,11 @@ const newHandleServerFunction = (originalPatcher) => {
 
     const serverStreamAndBidiHandlerPrefix = (): void => {
       let serverStreamAggData = '';
+      // @ts-ignore
       call.on('data', (res) => {
         serverStreamAggData = concatenatePayload(serverStreamAggData, res);
       });
+      // @ts-ignore
       call.on('finish', () => {
         safeExecute(() => {
           span.setAttribute(
@@ -36,6 +38,7 @@ const newHandleServerFunction = (originalPatcher) => {
     };
 
     const clientStreamAndUnaryHandlerPrefix = (): void => {
+      // @ts-ignore
       call.on('data', (res) => {
         clientStreamAggData = concatenatePayload(clientStreamAggData, res);
       });

src/instrumentations/pino/PinoInstrumentation.ts

@@ -0,0 +1,12 @@
+import { PinoInstrumentation } from '@opentelemetry/instrumentation-pino';
+import { LoggingInstrumentor } from '../instrumentor';
+
+export default class LumigoPinoInstrumentation extends LoggingInstrumentor<PinoInstrumentation> {
+  getInstrumentedModule(): string {
+    return 'pino';
+  }
+
+  getInstrumentation(): PinoInstrumentation {
+    return new PinoInstrumentation();
+  }
+}

src/instrumentations/pino/tested_versions/14/pino

@@ -0,0 +1 @@
+!9.4.0

src/instrumentations/pino/tested_versions/16/pino

@@ -0,0 +1 @@
+!9.4.0

src/instrumentations/pino/tested_versions/18/pino

@@ -0,0 +1 @@
+9.4.0

src/instrumentations/pino/tested_versions/20/pino

@@ -0,0 +1 @@
+9.4.0

test/instrumentations/pino/.gitignore

@@ -0,0 +1 @@
+logs

test/instrumentations/pino/app/.gitignore

@@ -0,0 +1 @@
+package-lock.json

test/instrumentations/pino/app/package.json

@@ -0,0 +1,14 @@
+{
+  "name": "lumigo-pino-test",
+  "version": "1.0.0",
+  "description": "",
+  "scripts": {
+    "start": "node -r @lumigo/opentelemetry pino_app.js"
+  },
+  "author": "",
+  "license": "ISC",
+  "dependencies": {
+    "@lumigo/opentelemetry": "file:../../../../distro.tgz",
+    "pino": "^9.4.0"
+  }
+}

test/instrumentations/pino/app/pino_app.js

@@ -0,0 +1,57 @@
+const http = require('http');
+const url = require('url');
+const { init } = require("@lumigo/opentelemetry")
+
+require('log-timestamp');
+
+const host = 'localhost';
+let httpServer;
+
+function respond(res, status, body) {
+  console.log(`responding with ${status} ${JSON.stringify(body)}`);
+  res.setHeader('Content-Type', 'application/json');
+  res.setHeader('access-control-allow-origin', '*');
+  res.writeHead(status);
+  res.end(JSON.stringify(body));
+}
+
+const requestListener = async function (req, res) {
+  await init;
+
+  const pinoLogger = require('pino')()
+
+  console.error(`Received request: ${req.method} ${req.url}`);
+
+  const requestUrl = url.parse(req.url, true);
+
+  switch (requestUrl.pathname) {
+    case '/write-log-line':
+      try {
+        const logLine = JSON.parse(requestUrl?.query?.logLine)
+        pinoLogger.info(logLine);
+        respond(res, 200, {})
+      } catch (err) {
+        console.error(`Error writing log line`, err);
+        respond(res, 500, { error: err });
+      }
+      break;
+
+    case '/quit':
+      console.error('Received quit command');
+      respond(res, 200, {});
+      httpServer.close();
+      break;
+
+    default:
+      respond(res, 404, { error: 'Resource not found' });
+  }
+};
+
+httpServer = http.createServer(requestListener);
+httpServer.listen(0, host, () => {
+  const port = httpServer.address().port;
+  console.error(`HTTP server listening on port ${port}`);
+  if (process.send) {
+    process.send(port);
+  }
+});

test/instrumentations/pino/pino.test.ts

@@ -0,0 +1,110 @@
+import fs from 'fs';
+import { join } from 'path';
+import { itTest } from '../../integration/setup';
+import { TestApp } from '../../utils/test-apps';
+import { installPackage, reinstallPackages, uninstallPackage } from '../../utils/test-setup';
+import { versionsToTest } from '../../utils/versions';
+import { FakeEdge } from '../../utils/fake-edge';
+
+const INSTRUMENTATION_NAME = 'pino';
+const LOGS_DIR = join(__dirname, 'logs');
+const TEST_APP_DIR = join(__dirname, 'app');
+
+describe.each(versionsToTest(INSTRUMENTATION_NAME, INSTRUMENTATION_NAME))(
+  `Instrumentation tests for the ${INSTRUMENTATION_NAME} package`,
+  function (versionToTest) {
+    let testApp: TestApp;
+    const fakeEdge = new FakeEdge();
+
+    beforeAll(async () => {
+      await fakeEdge.start();
+
+      reinstallPackages({ appDir: TEST_APP_DIR });
+      fs.mkdirSync(LOGS_DIR, { recursive: true });
+      installPackage({
+        appDir: TEST_APP_DIR,
+        packageName: INSTRUMENTATION_NAME,
+        packageVersion: versionToTest,
+      });
+    });
+
+    afterEach(async () => {
+      fakeEdge.reset();
+
+      if (testApp) {
+        await testApp.kill();
+      }
+    });
+
+    afterAll(async () => {
+      await fakeEdge.stop();
+
+      uninstallPackage({
+        appDir: TEST_APP_DIR,
+        packageName: INSTRUMENTATION_NAME,
+        packageVersion: versionToTest,
+      });
+    });
+
+    itTest(
+      {
+        testName: `${INSTRUMENTATION_NAME} logger: ${versionToTest}`,
+        packageName: INSTRUMENTATION_NAME,
+        version: versionToTest,
+        timeout: 20_000,
+      },
+      async function () {
+        const logDumpPath = `${LOGS_DIR}/${INSTRUMENTATION_NAME}.${INSTRUMENTATION_NAME}-logs@${versionToTest}.json`;
+
+        testApp = new TestApp(TEST_APP_DIR, INSTRUMENTATION_NAME, {
+          logDumpPath,
+          env: {
+            LUMIGO_ENABLE_LOGS: 'true',
+            LUMIGO_SECRET_MASKING_REGEX: '[".*sekret.*"]',
+            LUMIGO_LOGS_ENDPOINT: fakeEdge.logsUrl,
+            LUMIGO_ENDPOINT: fakeEdge.tracesUrl,
+            LUMIGO_TRACER_TOKEN: 't_123456789',
+          },
+        });
+
+        await writeLogLine('Hello Pino!');
+        await writeLogLine({ a: 1, sekret: 'this is secret!' });
+
+        await fakeEdge.waitFor(({ resources }) => resources.length > 1, 'waiting for resources to be processed');
+        await fakeEdge.waitFor(({ logs } ) => logs.length == 2, 'waiting for logs to be processed');
+
+        console.log('fakeEdge.resources', JSON.stringify(fakeEdge.resources, null, 2));
+
+        expect(fakeEdge.resources[0].attributes).toIncludeAllMembers([
+          {
+            key: 'service.name',
+            value: {
+              stringValue: 'pino',
+            },
+          },
+        ]);
+
+        expect(fakeEdge.logs[0].body).toEqual({ stringValue: 'Hello Pino!' });
+        // Span context is available since the test app is an instrumented HTTP server
+        expect(fakeEdge.logs[0]['traceId']).toHaveLength(32);
+        expect(fakeEdge.logs[0]['spanId']).toHaveLength(16);
+
+        // Logging an object in Pino produces attributes, as opposed to making the body an object
+        expect(fakeEdge.logs[1].attributes).toIncludeAllMembers([
+          { key: 'a', value: { intValue: 1 } },
+          { key: 'sekret', value: { stringValue: '****' } },
+        ]);
+        expect(fakeEdge.logs[1]['traceId']).toHaveLength(32);
+        expect(fakeEdge.logs[1]['spanId']).toHaveLength(16);
+
+        // Test the log-dump functionality
+        await testApp.getFinalLogs(2);
+      }
+    );
+
+    const writeLogLine = async (logLine: any) =>
+      testApp.invokeGetPath(
+        `/write-log-line?logLine=${encodeURIComponent(JSON.stringify(logLine))}`
+      );
+  }
+);

test/utils/fake-edge.ts

@@ -26,11 +26,11 @@ export class FakeEdge {
     this.app.use('/v1/traces', (req: Request, res: Response) => {
       try {
         const scopeSpans = req.body.resourceSpans.flatMap(rl => rl.scopeSpans.flatMap(sl => sl.scopeSpans))
-        console.log(`Received ${scopeSpans.length} logs in edge`);
+        console.log(`Received ${scopeSpans.length} spans in edge on /v1/traces`);
         this.spans.push(...scopeSpans)
 
         const resources = req.body.resourceSpans.map(rl => rl.resource)
-        console.log(`Received ${resources.length} resources in edge`);
+        console.log(`Received ${resources.length} resources in edge on /v1/traces`);
         this.resources.push(...resources)
       } catch (e) {
         console.error('Error parsing spans in edge: ', e);
@@ -41,11 +41,11 @@ export class FakeEdge {
     this.app.use('/v1/logs', (req: Request, res: Response) => {
       try {
         const logRecords = req.body.resourceLogs.flatMap(rl => rl.scopeLogs.flatMap(sl => sl.logRecords))
-        console.log(`Received ${logRecords.length} logs in edge`);
+        console.log(`Received ${logRecords.length} logs in edge on /v1/logs`);
         this.logs.push(...logRecords)
 
         const resources = req.body.resourceLogs.map(rl => rl.resource)
-        console.log(`Received ${resources.length} resources in edge`);
+        console.log(`Received ${resources.length} resources in edge /v1/logs`);
         this.resources.push(...resources)
 
       } catch (e) {