Decoupling Logstash Workloads with Pipeline-to-Pipeline Communication

The Bottleneck of the Monolithic Pipeline

In a standard Logstash configuration, a single pipeline handles everything: ingestion, parsing, enrichment, and output. While simple, this creates a performance ceiling. If you have a high-volume stream of logs and a few CPU-intensive filters—such as complex grok patterns or ruby scripts—the slow filters create a bottleneck that throttles the entire ingestion process. You cannot scale the parsing logic independently of the ingestion logic because they share the same worker threads.

The solution is to break the monolith into modular pipelines using the distributer pattern. By separating concerns into isolated pipelines within a single Logstash instance, you can allocate resources more effectively and prevent a single slow filter from stalling your entire data stream.

How Pipeline-to-Pipeline Communication Works

Logstash allows you to define multiple independent pipelines in a pipelines.yml file. To move data between them without sending it over the network, you use two specific plugins: the pipeline input and the distributer output.

Unlike a traditional pipeline where data flows linearly through input → filter → output, the distributer output acts as a hand-off mechanism. It sends the event directly to the internal queue of another pipeline. Because this happens within the same Java Virtual Machine (JVM), there is no network overhead or serialization cost associated with external brokers like Redis or Kafka.

Resource Isolation and Backpressure

Each pipeline operates with its own thread pool. This means you can configure a high-throughput "Ingest" pipeline with many workers to handle raw TCP or Beats traffic, and a separate "Enrichment" pipeline with a smaller, more focused set of workers for heavy processing.

Crucially, this architecture supports backpressure. If the downstream enrichment pipeline cannot keep up and its internal queue fills, the pressure propagates back to the distributer output in the ingest pipeline. This eventually signals the input plugin to slow down the source (e.g., telling Filebeat to stop sending data), preventing the JVM from crashing due to out-of-memory errors.

Implementation Example: Ingest and Enrich

Consider a scenario where you ingest raw system logs but need to perform a heavy lookup or complex regex parsing on only a subset of that data. Instead of slowing down the entire stream, you can split the workload.

1. Define the Pipelines (pipelines.yml)

Run this configuration on your Logstash node. Ensure the user running Logstash has read permissions for these paths.

- pipeline.id: ingest_pipeline
  path.config: "/etc/logstash/conf.d/ingest.conf"
- pipeline.id: enrich_pipeline
  path.config: "/etc/logstash/conf.d/enrich.conf"

2. The Ingest Configuration (ingest.conf)

This pipeline focuses on fast ingestion and routing.

input {
  beats {
    port => 5044
  }
}

filter {
  # Simple routing logic to decide what needs enrichment
  if [event_type] == "complex_log" {
    mutate { add_tag => ["needs_enrichment"] }
  }
}

output {
  if "needs_enrichment" in [tags] {
    distributer {
      target_pipeline => "enrich_pipeline"
    }
  } else {
    elasticsearch {
      hosts => ["localhost:9200"]
      index => "logs-simple"
    }
  }
}

3. The Enrichment Configuration (enrich.conf)

This pipeline handles the CPU-heavy lifting.

input {
  pipeline {
    # This listens for events sent by the distributer
  }
}

filter {
  grok {
    match => { "message" => "%{GREEDYDATA:complex_field}" } # Imagine a very heavy regex here
  }
}

output {
  elasticsearch {
    hosts => ["localhost:9200"]
    index => "logs-enriched"
  }
}

Trade-offs and Operational Limits

While modular pipelines reduce complexity compared to external message brokers, they introduce specific constraints:

• Shared JVM Heap: All pipelines share the same memory pool. If the enrich_pipeline consumes massive amounts of RAM for large events, it can trigger Garbage Collection (GC) pauses that affect the ingest_pipeline.
• Single-Node Bound: This communication is internal. If you need to scale horizontally across multiple physical servers, the distributer plugin cannot send events to a different node; you must return to a distributed broker like Kafka.
• Static Routing: The distributer output routes based on the pipeline ID. You cannot dynamically calculate a destination pipeline name based on an event field; you must use if/else logic in the output block.

Verification and Monitoring

To verify that your pipelines are communicating and to check for bottlenecks, use the Logstash Stats API. Run the following command from your terminal (adjusting for your Logstash host/port):

curl -XGET 'localhost:9600/_node/stats/pipelines?pretty'

Look for the queue and events sections for both ingest_pipeline and enrich_pipeline. If the enrich_pipeline queue is consistently full while the ingest_pipeline throughput is low, you have identified a processing bottleneck and can now increase the worker count for that specific pipeline without risking the stability of your ingestion layer.