Log Aggregation via Streams

In this lesson, you will learn how to orchestrate a real-time log aggregation pipeline by connecting distributed log streams into a single observable output.

The Architecture of Log Aggregation

When building scalable applications, logs are often generated by multiple processes or services simultaneously. Rather than writing to fragmented files, we use streams to funnel these logs into a unified aggregator. By using the pipeline helper, we can treat these separate log sources as a single continuous flow of data that is filtered, formatted, and shipped to a centralized observability tool.

Creating a Log Aggregation Pipeline

Imagine you have two log sources: an access log and an error log. You want to filter out "DEBUG" messages from both and send the remainder to a final destination, like a network socket or an analytics endpoint.

Using stream.pipeline, we can manage this cleanly. We represent each source as a Readable stream, pipe them through a Transform stream to perform the filtering, and finally direct them to an output destination:

const { pipeline, Transform } = require('stream');

const logFilter = new Transform({
  transform(chunk, encoding, callback) {
    const log = chunk.toString();
    // Only pass through logs that aren't DEBUG
    if (!log.includes('DEBUG')) {
      this.push(log);
    }
    callback();
  }
});

// Aggregate multiple streams into one destination
pipeline(
  myLogSource, // e.g., a file watcher or socket
  logFilter,
  process.stdout, // Or an observability tool like Winston/Datadog
  (err) => {
    if (err) console.error('Pipeline failed:', err);
    else console.log('Log aggregation pipeline closed successfully.');
  }
);

By leveraging Transform streams as filters, we ensure that our log aggregation happens in memory. This is highly efficient because it avoids writing intermediate files to the disk. Because we are using pipeline, if the destination (the observability tool) cannot keep up with the volume of logs, the backpressure will naturally propagate upstream to our log sources, preventing the application from crashing due to memory overflow.

Key Takeaway: Log aggregation via streams allows you to transform, filter, and consolidate distributed log events in memory, ensuring efficient observability while maintaining system stability through inherent backpressure.