Improve logging pipeline in @kbn/legacy-logging (#84629)

packages/kbn-legacy-logging/src/log_reporter.test.ts

@@ -0,0 +1,142 @@
+/*
+ * Licensed to Elasticsearch B.V. under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch B.V. licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import os from 'os';
+import path from 'path';
+import fs from 'fs';
+
+import stripAnsi from 'strip-ansi';
+
+import { getLogReporter } from './log_reporter';
+
+const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));
+
+describe('getLogReporter', () => {
+  it('should log to stdout (not json)', async () => {
+    const lines: string[] = [];
+    const origWrite = process.stdout.write;
+    process.stdout.write = (buffer: string | Uint8Array): boolean => {
+      lines.push(stripAnsi(buffer.toString()).trim());
+      return true;
+    };
+
+    const loggerStream = getLogReporter({
+      config: {
+        json: false,
+        dest: 'stdout',
+        filter: {},
+      },
+      events: { log: '*' },
+    });
+
+    loggerStream.end({ event: 'log', tags: ['foo'], data: 'hello world' });
+
+    await sleep(500);
+
+    process.stdout.write = origWrite;
+    expect(lines.length).toBe(1);
+    expect(lines[0]).toMatch(/^log   \[[^\]]*\] \[foo\] hello world$/);
+  });
+
+  it('should log to stdout (as json)', async () => {
+    const lines: string[] = [];
+    const origWrite = process.stdout.write;
+    process.stdout.write = (buffer: string | Uint8Array): boolean => {
+      lines.push(JSON.parse(buffer.toString().trim()));
+      return true;
+    };
+
+    const loggerStream = getLogReporter({
+      config: {
+        json: true,
+        dest: 'stdout',
+        filter: {},
+      },
+      events: { log: '*' },
+    });
+
+    loggerStream.end({ event: 'log', tags: ['foo'], data: 'hello world' });
+
+    await sleep(500);
+
+    process.stdout.write = origWrite;
+    expect(lines.length).toBe(1);
+    expect(lines[0]).toMatchObject({
+      type: 'log',
+      tags: ['foo'],
+      message: 'hello world',
+    });
+  });
+
+  it('should log to custom file (not json)', async () => {
+    const dir = os.tmpdir();
+    const logfile = `dest-${Date.now()}.log`;
+    const dest = path.join(dir, logfile);
+
+    const loggerStream = getLogReporter({
+      config: {
+        json: false,
+        dest,
+        filter: {},
+      },
+      events: { log: '*' },
+    });
+
+    loggerStream.end({ event: 'log', tags: ['foo'], data: 'hello world' });
+
+    await sleep(500);
+
+    const lines = stripAnsi(fs.readFileSync(dest, { encoding: 'utf8' }))
+      .trim()
+      .split(os.EOL);
+    expect(lines.length).toBe(1);
+    expect(lines[0]).toMatch(/^log   \[[^\]]*\] \[foo\] hello world$/);
+  });
+
+  it('should log to custom file (as json)', async () => {
+    const dir = os.tmpdir();
+    const logfile = `dest-${Date.now()}.log`;
+    const dest = path.join(dir, logfile);
+
+    const loggerStream = getLogReporter({
+      config: {
+        json: true,
+        dest,
+        filter: {},
+      },
+      events: { log: '*' },
+    });
+
+    loggerStream.end({ event: 'log', tags: ['foo'], data: 'hello world' });
+
+    await sleep(500);
+
+    const lines = fs
+      .readFileSync(dest, { encoding: 'utf8' })
+      .trim()
+      .split(os.EOL)
+      .map((data) => JSON.parse(data));
+    expect(lines.length).toBe(1);
+    expect(lines[0]).toMatchObject({
+      type: 'log',
+      tags: ['foo'],
+      message: 'hello world',
+    });
+  });
+});

packages/kbn-legacy-logging/src/log_reporter.ts

@@ -17,9 +17,11 @@
  * under the License.
  */
 
+import { createWriteStream } from 'fs';
+import { pipeline } from 'stream';
+
 // @ts-expect-error missing type def
 import { Squeeze } from '@hapi/good-squeeze';
-import { createWriteStream as writeStr, WriteStream } from 'fs';
 
 import { KbnLoggerJsonFormat } from './log_format_json';
 import { KbnLoggerStringFormat } from './log_format_string';
@@ -31,21 +33,28 @@ export function getLogReporter({ events, config }: { events: any; config: LogFor
   const format = config.json ? new KbnLoggerJsonFormat(config) : new KbnLoggerStringFormat(config);
   const logInterceptor = new LogInterceptor();
 
-  let dest: WriteStream | NodeJS.WritableStream;
   if (config.dest === 'stdout') {
-    dest = process.stdout;
+    pipeline(logInterceptor, squeeze, format, onFinished);
+    // The `pipeline` function is used to properly close all streams in the
+    // pipeline in case one of them ends or fails. Since stdout obviously
+    // shouldn't be closed in case of a failure in one of the other streams,
+    // we're not including that in the call to `pipeline`, but rely on the old
+    // `pipe` function instead.
+    format.pipe(process.stdout);
   } else {
-    dest = writeStr(config.dest, {
+    const dest = createWriteStream(config.dest, {
       flags: 'a',
       encoding: 'utf8',
     });
-
-    logInterceptor.on('end', () => {
-      dest.end();
-    });
+    pipeline(logInterceptor, squeeze, format, dest, onFinished);
   }
 
-  logInterceptor.pipe(squeeze).pipe(format).pipe(dest);
-
   return logInterceptor;
 }
+
+function onFinished(err: NodeJS.ErrnoException | null) {
+  if (err) {
+    // eslint-disable-next-line no-console
+    console.error('An unexpected error occurred in the logging pipeline:', err.stack);
+  }
+}