adapt to transformer

probot · Apr 16, 2022 · 7e8c6be · 7e8c6be
1 parent a130f26
commit 7e8c6be
Show file tree

Hide file tree

Showing 4 changed files with 172 additions and 77 deletions.
diff --git a/cli.js b/cli.js
@@ -1,5 +1,4 @@
 const pump = require("pump");
-const split = require("split2");
 
 const { getTransformStream } = require("./");
 
@@ -9,4 +8,5 @@ const options = {
   sentryDsn: process.env.SENTRY_DSN,
 };
 
-pump(process.stdin, split(), getTransformStream(options), process.stdout);
+const res = getTransformStream(opts)
+pump(process.stdin, res)
diff --git a/index.js b/index.js
@@ -1,8 +1,11 @@
 module.exports = { getTransformStream };
 
 const { Transform } = require("readable-stream");
-
-const prettyFactory = require("pino-pretty");
+const { isMainThread } = require('worker_threads');
+const { prettyFactory } = require("pino-pretty");
+const abstractTransport = require('pino-abstract-transport');
+const pump = require('pump');
+const SonicBoom = require('sonic-boom');
 const Sentry = require("@sentry/node");
 
 const LEVEL_MAP = {
@@ -14,6 +17,70 @@ const LEVEL_MAP = {
   60: "fatal",
 };
 
+function noop() { }
+
+/**
+ * Creates a safe SonicBoom instance
+ *
+ * @param {object} opts Options for SonicBoom
+ *
+ * @returns {object} A new SonicBoom stream
+ */
+function buildSafeSonicBoom(opts) {
+  const stream = new SonicBoom(opts)
+  stream.on('error', filterBrokenPipe)
+  // if we are sync: false, we must flush on exit
+  if (!opts.sync && isMainThread) {
+    setupOnExit(stream)
+  }
+  return stream
+
+  function filterBrokenPipe(err) {
+    if (err.code === 'EPIPE') {
+      stream.write = noop
+      stream.end = noop
+      stream.flushSync = noop
+      stream.destroy = noop
+      return
+    }
+    stream.removeListener('error', filterBrokenPipe)
+  }
+}
+
+function setupOnExit(stream) {
+  /* istanbul ignore next */
+  if (global.WeakRef && global.WeakMap && global.FinalizationRegistry) {
+    // This is leak free, it does not leave event handlers
+    const onExit = require('on-exit-leak-free')
+
+    onExit.register(stream, autoEnd)
+
+    stream.on('close', function () {
+      onExit.unregister(stream)
+    })
+  }
+}
+
+/* istanbul ignore next */
+function autoEnd(stream, eventName) {
+  // This check is needed only on some platforms
+
+  if (stream.destroyed) {
+    return
+  }
+
+  if (eventName === 'beforeExit') {
+    // We still have an event loop, let's use it
+    stream.flush()
+    stream.on('drain', function () {
+      stream.end()
+    })
+  } else {
+    // We do not have an event loop, so flush synchronously
+    stream.flushSync()
+  }
+}
+
 /**
  * Implements Probot's default logging formatting and error captionaing using Sentry.
  *
@@ -52,85 +119,109 @@ function getTransformStream(options = {}) {
     ),
   });
 
-  return new Transform({
-    objectMode: true,
-    transform(chunk, enc, cb) {
-      const line = chunk.toString().trim();
-
-      /* istanbul ignore if */
-      if (line === undefined) return cb();
-
-      const data = sentryEnabled ? JSON.parse(line) : null;
-
-      if (!sentryEnabled || data.level < 50) {
-        if (formattingEnabled) {
-          return cb(null, pretty(line));
-        }
+  return abstractTransport(function (source) {
+    const stream = new Transform({
+      objectMode: true,
+      autoDestroy: true,
+      transform(chunk, enc, cb) {
+        const line = chunk.toString().trim();
 
-        if (levelAsString) {
-          return cb(null, stringifyLogLevel(JSON.parse(line)));
-        }
+        /* istanbul ignore if */
+        if (line === undefined) return cb();
 
-        cb(null, line + "\n");
-        return;
-      }
+        const data = sentryEnabled ? JSON.parse(line) : null;
 
-      Sentry.withScope(function (scope) {
-        const sentryLevelName =
-          data.level === 50 ? Sentry.Severity.Error : Sentry.Severity.Fatal;
-        scope.setLevel(sentryLevelName);
+        if (!sentryEnabled || data.level < 50) {
+          if (formattingEnabled) {
+            return cb(null, pretty(line));
+          }
 
-        for (const extra of ["event", "headers", "request", "status"]) {
-          if (!data[extra]) continue;
+          if (levelAsString) {
+            return cb(null, stringifyLogLevel(JSON.parse(line)));
+          }
 
-          scope.setExtra(extra, data[extra]);
+          cb(null, line + "\n");
+          return;
         }
 
-        // set user id and username to installation ID and account login
-        if (data.event && data.event.payload) {
-          const {
-            // When GitHub App is installed organization wide
-            installation: { id, account: { login: account } = {} } = {},
-
-            // When the repository belongs to an organization
-            organization: { login: organization } = {},
-            // When the repository belongs to a user
-            repository: { owner: { login: owner } = {} } = {},
-          } = data.event.payload;
-
-          scope.setUser({
-            id: id,
-            username: account || organization || owner,
-          });
-        }
+        Sentry.withScope(function (scope) {
+          const sentryLevelName =
+            data.level === 50 ? Sentry.Severity.Error : Sentry.Severity.Fatal;
+          scope.setLevel(sentryLevelName);
+
+          for (const extra of ["event", "headers", "request", "status"]) {
+            if (!data[extra]) continue;
+
+            scope.setExtra(extra, data[extra]);
+          }
+
+          // set user id and username to installation ID and account login
+          if (data.event && data.event.payload) {
+            const {
+              // When GitHub App is installed organization wide
+              installation: { id, account: { login: account } = {} } = {},
+
+              // When the repository belongs to an organization
+              organization: { login: organization } = {},
+              // When the repository belongs to a user
+              repository: { owner: { login: owner } = {} } = {},
+            } = data.event.payload;
+
+            scope.setUser({
+              id: id,
+              username: account || organization || owner,
+            });
+          }
+
+          const sentryEventId = Sentry.captureException(toSentryError(data));
+
+          // reduce logging data and add reference to sentry event instead
+          if (data.event) {
+            data.event = { id: data.event.id };
+          }
+          if (data.request) {
+            data.request = {
+              method: data.request.method,
+              url: data.request.url,
+            };
+          }
+          data.sentryEventId = sentryEventId;
+
+          if (formattingEnabled) {
+            return cb(null, pretty(data));
+          }
+
+          // istanbul ignore if
+          if (levelAsString) {
+            return cb(null, stringifyLogLevel(data));
+          }
+
+          cb(null, JSON.stringify(data) + "\n");
+        });
+      },
+    });
 
-        const sentryEventId = Sentry.captureException(toSentryError(data));
+    let destination
 
-        // reduce logging data and add reference to sentry event instead
-        if (data.event) {
-          data.event = { id: data.event.id };
-        }
-        if (data.request) {
-          data.request = {
-            method: data.request.method,
-            url: data.request.url,
-          };
-        }
-        data.sentryEventId = sentryEventId;
+    if (typeof options.destination === 'object' && typeof options.destination.write === 'function') {
+      destination = options.destination
+    } else {
+      destination = buildSafeSonicBoom({
+        dest: options.destination || 1,
+        append: options.append,
+        mkdir: options.mkdir,
+        sync: options.sync // by default sonic will be async
+      })
+    }
 
-        if (formattingEnabled) {
-          return cb(null, pretty(data));
-        }
+    source.on('unknown', function (line) {
+      destination.write(line + '\n')
+    })
 
-        // istanbul ignore if
-        if (levelAsString) {
-          return cb(null, stringifyLogLevel(data));
-        }
+    pump(source, stream, destination)
 
-        cb(null, JSON.stringify(data) + "\n");
-      });
-    },
-  });
+    return stream;
+  }, { parse: 'lines' });
 }
 
 function stringifyLogLevel(data) {

diff --git a/package-lock.json b/package-lock.json
diff --git a/package.json b/package.json
@@ -30,9 +30,11 @@
   },
   "dependencies": {
     "@sentry/node": "^6.0.0",
+    "pino-abstract-transport": "^0.5.0",
     "pino-pretty": "^7.0.0",
     "pump": "^3.0.0",
     "readable-stream": "^3.6.0",
+    "sonic-boom": "^2.7.0",
     "split2": "^4.0.0"
   },
   "release": {