ELK Stack — Multiple Pipelines

Multiple Pipelines

By default, Logstash runs a single pipeline. In production, you often need to process different data sources with different parsing rules, performance settings, and destinations. Multiple pipelines let you isolate these data flows.

┌──────────────────────────────────────────────────────────────────┐
│                        LOGSTASH INSTANCE                        │
│                                                                  │
│  ┌──────────────────────────────────┐                           │
│  │  Pipeline: apache-logs           │                           │
│  │  beats:5044 → grok → ES         │──▶ apache-access-*        │
│  │  workers: 2                      │                           │
│  └──────────────────────────────────┘                           │
│                                                                  │
│  ┌──────────────────────────────────┐                           │
│  │  Pipeline: app-logs              │                           │
│  │  beats:5045 → json → ES         │──▶ app-logs-*             │
│  │  workers: 4                      │                           │
│  └──────────────────────────────────┘                           │
│                                                                  │
│  ┌──────────────────────────────────┐                           │
│  │  Pipeline: metrics               │                           │
│  │  kafka → mutate → ES            │──▶ metrics-*              │
│  │  workers: 1                      │                           │
│  └──────────────────────────────────┘                           │
│                                                                  │
│  ┌──────────────────────────────────┐                           │
│  │  Pipeline: dead-letter           │                           │
│  │  DLQ → reprocess → ES           │──▶ failed-events-*        │
│  │  workers: 1                      │                           │
│  └──────────────────────────────────┘                           │
└──────────────────────────────────────────────────────────────────┘

pipelines.yml Configuration

Define multiple pipelines in /etc/logstash/pipelines.yml:

# /etc/logstash/pipelines.yml

# Pipeline 1: Apache Access Logs
- pipeline.id: apache-logs
  path.config: "/etc/logstash/conf.d/apache-access.conf"
  pipeline.workers: 2
  pipeline.batch.size: 250
  pipeline.batch.delay: 50
  queue.type: persisted
  queue.max_bytes: 512mb

# Pipeline 2: Application Logs
- pipeline.id: app-logs
  path.config: "/etc/logstash/conf.d/app-logs.conf"
  pipeline.workers: 4
  pipeline.batch.size: 125
  queue.type: memory

# Pipeline 3: Metrics
- pipeline.id: metrics
  path.config: "/etc/logstash/conf.d/metrics.conf"
  pipeline.workers: 1
  pipeline.batch.size: 500
  queue.type: memory

# Pipeline 4: Dead Letter Queue Reprocessing
- pipeline.id: dead-letter-reprocess
  path.config: "/etc/logstash/conf.d/dlq-reprocess.conf"
  pipeline.workers: 1
  dead_letter_queue.enable: false

Each pipeline gets its own:

• Config file — separate input/filter/output definitions
• Worker threads — tuned for the workload
• Queue — independent buffering
• Metrics — separate monitoring stats

Pipeline Configuration Files

apache-access.conf:

input {
  beats {
    port => 5044
    type => "apache-access"
  }
}

filter {
  grok {
    match => { "message" => "%{COMBINEDAPACHELOG}" }
  }

  date {
    match => ["timestamp", "dd/MMM/yyyy:HH:mm:ss Z"]
    target => "@timestamp"
    remove_field => ["timestamp"]
  }

  mutate {
    convert => {
      "response" => "integer"
      "bytes" => "integer"
    }
  }

  geoip {
    source => "clientip"
  }
}

output {
  elasticsearch {
    hosts => ["https://localhost:9200"]
    index => "apache-access-%{+YYYY.MM.dd}"
    user => "elastic"
    password => "changeme"
  }
}

app-logs.conf:

input {
  beats {
    port => 5045
    type => "app-logs"
  }
}

filter {
  json {
    source => "message"
  }

  date {
    match => ["timestamp", "ISO8601"]
    target => "@timestamp"
    remove_field => ["timestamp"]
  }
}

output {
  elasticsearch {
    hosts => ["https://localhost:9200"]
    index => "app-logs-%{+YYYY.MM.dd}"
    user => "elastic"
    password => "changeme"
  }
}

Pipeline-to-Pipeline Communication

Logstash supports sending events between pipelines using virtual pipeline input and output plugins:

┌──────────────┐    ┌──────────────┐    ┌──────────────┐
│   Upstream   │    │  Distributor │    │  Downstream  │
│   Pipeline   │───▶│   Pipeline   │───▶│  Pipeline(s) │
│              │    │              │    │              │
│ beats:5044   │    │ conditionals │    │ → ES index   │
└──────────────┘    └──────────────┘    └──────────────┘

pipelines.yml for pipeline-to-pipeline:

- pipeline.id: upstream
  path.config: "/etc/logstash/conf.d/upstream.conf"
  pipeline.workers: 2

- pipeline.id: distributor
  path.config: "/etc/logstash/conf.d/distributor.conf"
  pipeline.workers: 1

- pipeline.id: apache-downstream
  path.config: "/etc/logstash/conf.d/apache-downstream.conf"
  pipeline.workers: 2

- pipeline.id: app-downstream
  path.config: "/etc/logstash/conf.d/app-downstream.conf"
  pipeline.workers: 4

upstream.conf:

input {
  beats {
    port => 5044
  }
}

output {
  pipeline {
    send_to => ["distributor"]
  }
}

distributor.conf:

input {
  pipeline {
    address => "distributor"
  }
}

output {
  if [type] == "apache" {
    pipeline {
      send_to => ["apache-sink"]
    }
  } else if [type] == "application" {
    pipeline {
      send_to => ["app-sink"]
    }
  }
}

apache-downstream.conf:

input {
  pipeline {
    address => "apache-sink"
  }
}

filter {
  grok {
    match => { "message" => "%{COMBINEDAPACHELOG}" }
  }
}

output {
  elasticsearch {
    hosts => ["https://localhost:9200"]
    index => "apache-%{+YYYY.MM.dd}"
    user => "elastic"
    password => "changeme"
  }
}

Pipeline Workers and Batch Settings

Tuning workers and batch settings affects throughput and latency:

Setting	Description	Default	Guidance
pipeline.workers	Number of parallel filter/output threads	CPU cores	Set to CPU cores for CPU-heavy filters; reduce for I/O-heavy outputs
pipeline.batch.size	Events per worker batch	125	Increase for throughput (e.g., 250–500); decrease for latency
pipeline.batch.delay	Max wait (ms) to fill a batch	50	Lower = less latency; higher = more batching efficiency
pipeline.ordered	Maintain event order	auto	Set true if order matters (reduces parallelism)

# High-throughput pipeline
- pipeline.id: high-volume
  path.config: "/etc/logstash/conf.d/high-volume.conf"
  pipeline.workers: 8
  pipeline.batch.size: 500
  pipeline.batch.delay: 10

# Low-latency pipeline
- pipeline.id: low-latency
  path.config: "/etc/logstash/conf.d/low-latency.conf"
  pipeline.workers: 2
  pipeline.batch.size: 50
  pipeline.batch.delay: 5

Dead Letter Queue (DLQ)

When an event can’t be processed (e.g., mapping conflict in Elasticsearch), the DLQ captures it instead of dropping it:

Enable DLQ in logstash.yml:

dead_letter_queue.enable: true
dead_letter_queue.max_bytes: 1024mb
path.dead_letter_queue: /var/lib/logstash/dead_letter_queue

Or per-pipeline in pipelines.yml:

- pipeline.id: main
  path.config: "/etc/logstash/conf.d/main.conf"
  dead_letter_queue.enable: true
  dead_letter_queue.max_bytes: 512mb

DLQ reprocessing pipeline:

# dlq-reprocess.conf
input {
  dead_letter_queue {
    path => "/var/lib/logstash/dead_letter_queue"
    pipeline_id => "main"
    commit_offsets => true
  }
}

filter {
  # The original event is in the "message" field
  # DLQ metadata is available:
  #   [@metadata][dead_letter_queue][plugin_type]
  #   [@metadata][dead_letter_queue][reason]

  # Fix the issue that caused the failure
  mutate {
    remove_field => ["problematic_field"]
  }
}

output {
  elasticsearch {
    hosts => ["https://localhost:9200"]
    index => "recovered-events-%{+YYYY.MM.dd}"
    user => "elastic"
    password => "changeme"
  }
}

Persistent Queues

Persistent queues write events to disk, protecting against data loss during crashes:

# logstash.yml or pipelines.yml
queue.type: persisted            # Default: memory
queue.max_bytes: 1024mb          # Max disk usage for queue
queue.checkpoint.acks: 1024      # ACK checkpoint interval
queue.checkpoint.writes: 1024    # Write checkpoint interval
queue.drain: false               # Drain queue on shutdown

When to use persistent queues:

Scenario	Queue Type
Development / testing	memory
Production with Beats backpressure	memory (Beats handle retries)
Production without upstream buffering	persisted
Critical data, no data loss tolerance	persisted
High-throughput, low-latency required	memory

Monitoring Pipeline Performance

Logstash Monitoring API (default port 9600):

# Node info
curl -s localhost:9600/_node | jq .

# Pipeline stats
curl -s localhost:9600/_node/stats/pipelines | jq .

# Hot threads
curl -s localhost:9600/_node/hot_threads

# Specific pipeline
curl -s localhost:9600/_node/stats/pipelines/apache-logs | jq .

Key metrics to monitor:

Metric	What It Tells You
events.in	Total events received
events.filtered	Total events processed by filters
events.out	Total events sent to outputs
events.duration_in_millis	Total processing time
queue.events_count	Events currently in queue
reloads.successes / failures	Config reload status

Pipeline Viewer in Kibana

Kibana provides a visual pipeline viewer under Stack Monitoring → Logstash → Pipelines:

1. Enable monitoring in logstash.yml:

monitoring.enabled: true
monitoring.elasticsearch.hosts: ["https://localhost:9200"]
monitoring.elasticsearch.username: "logstash_system"
monitoring.elasticsearch.password: "changeme"

1. Open Kibana → Stack Monitoring → Logstash
2. Select a pipeline to see:
   • Visual flow diagram (input → filter → output)
   • Events per second at each stage
   • Processing time per plugin
   • Queue depth and throughput

Quick Reference

Concept	Description
pipelines.yml	Defines multiple independent pipelines with separate configs and settings
pipeline.id	Unique identifier for each pipeline
pipeline.workers	Number of filter/output worker threads per pipeline
pipeline.batch.size	Events processed per batch per worker
Pipeline-to-pipeline	Use pipeline input/output plugins to chain pipelines
Dead Letter Queue	Captures events that fail processing instead of dropping them
Persistent queue	Disk-backed queue to prevent data loss during crashes
queue.type	memory (default, fast) or persisted (durable)
Monitoring API	localhost:9600/_node/stats/pipelines for pipeline metrics
Kibana Pipeline Viewer	Visual flow diagram with per-plugin metrics

Tip: Start with a single pipeline and split into multiple pipelines only when you need different performance settings or isolation between data sources. Over-splitting adds complexity without benefit.

👉 Next: Logstash Exercises