- Added optional 'level' property to TransportPipelineOptions interface

- A level can now be defined for pipelines defined inside 'targets'
- Added UT in 'pipeline.test.js' to check expected behaviour with 'dedupe'

lib/transport.js

@@ -97,6 +97,7 @@ function transport (fullOptions) {
       return dest.pipeline.map((t) => {
         return {
           ...t,
+          level: dest.level, // duplicate the pipeline `level` property defined in the upper level
           target: fixTarget(t.target)
         }
       })

lib/worker.js

@@ -94,15 +94,21 @@ module.exports = async function ({ targets, pipelines, levels, dedupe }) {
   if (pipelines && pipelines.length) {
     pipelines = await Promise.all(
       pipelines.map(async (p) => {
+        let level
         const pipeDests = await Promise.all(
           p.map(async (t) => {
+            // level assigned to pipeline is duplicated over all its targets, just store it
+            level = t.level
             const fn = await loadTransportStreamBuilder(t.target)
             const stream = await fn(t.options)
             return stream
           }
           ))
 
-        return { stream: createPipeline(pipeDests) }
+        return {
+          level,
+          stream: createPipeline(pipeDests)
+        }
       })
     )
     targetStreams.push(...pipelines)

pino.d.ts

@@ -259,6 +259,7 @@ declare namespace pino {
 
     interface TransportPipelineOptions<TransportOptions = Record<string, any>> extends TransportBaseOptions<TransportOptions>{
         pipeline: TransportSingleOptions<TransportOptions>[]
+        level?: LevelWithSilentOrString
     }
 
     interface TransportMultiOptions<TransportOptions = Record<string, any>> extends TransportBaseOptions<TransportOptions>{

test/transport/pipeline.test.js

@@ -6,6 +6,7 @@ const { readFile } = require('fs').promises
 const { watchFileCreated, file } = require('../helper')
 const { test } = require('tap')
 const pino = require('../../')
+const { DEFAULT_LEVELS } = require('../../lib/constants')
 
 const { pid } = process
 const hostname = os.hostname()
@@ -29,21 +30,15 @@ test('pino.transport with a pipeline', async ({ same, teardown }) => {
   same(result, {
     pid,
     hostname,
-    level: 30,
+    level: DEFAULT_LEVELS.info,
     msg: 'hello',
     service: 'pino' // this property was added by the transform
   })
 })
 
 test('pino.transport with targets containing pipelines', async ({ same, teardown }) => {
-  const destinationA = join(
-    os.tmpdir(),
-    '_' + Math.random().toString(36).substr(2, 9)
-  )
-  const destinationB = join(
-    os.tmpdir(),
-    '_' + Math.random().toString(36).substr(2, 9)
-  )
+  const destinationA = file()
+  const destinationB = file()
   const transport = pino.transport({
     targets: [
       {
@@ -76,14 +71,65 @@ test('pino.transport with targets containing pipelines', async ({ same, teardown
   same(resultA, {
     pid,
     hostname,
-    level: 30,
+    level: DEFAULT_LEVELS.info,
     msg: 'hello'
   })
   same(resultB, {
     pid,
     hostname,
-    level: 30,
+    level: DEFAULT_LEVELS.info,
     msg: 'hello',
     service: 'pino' // this property was added by the transform
   })
 })
+
+test('pino.transport with targets containing pipelines with levels defined and dedupe', async ({ same, teardown }) => {
+  const destinationA = file()
+  const destinationB = file()
+  const transport = pino.transport({
+    targets: [
+      {
+        target: join(__dirname, '..', 'fixtures', 'to-file-transport.js'),
+        options: { destination: destinationA },
+        level: DEFAULT_LEVELS.info
+      },
+      {
+        pipeline: [
+          {
+            target: join(__dirname, '..', 'fixtures', 'transport-transform.js')
+          },
+          {
+            target: join(__dirname, '..', 'fixtures', 'to-file-transport.js'),
+            options: { destination: destinationB }
+          }
+        ],
+        level: DEFAULT_LEVELS.error
+      }
+    ],
+    dedupe: true
+  })
+
+  teardown(transport.end.bind(transport))
+  const instance = pino(transport)
+  instance.info('hello info')
+  instance.error('hello error')
+  await watchFileCreated(destinationA)
+  await watchFileCreated(destinationB)
+  const resultA = JSON.parse(await readFile(destinationA))
+  const resultB = JSON.parse(await readFile(destinationB))
+  delete resultA.time
+  delete resultB.time
+  same(resultA, {
+    pid,
+    hostname,
+    level: DEFAULT_LEVELS.info,
+    msg: 'hello info'
+  })
+  same(resultB, {
+    pid,
+    hostname,
+    level: DEFAULT_LEVELS.error,
+    msg: 'hello error',
+    service: 'pino' // this property was added by the transform
+  })
+})