Skip to main content
Mapagam

Managing Logstash Pipelines

1. Understanding Pipeline Configuration

Property Description Values
input Defines the source of the log data file, stdin, tcp, udp, http
filter Processes and transforms log data grok, mutate, date, json
output Defines where to send the processed logs elasticsearch, file, stdout, http

Example: Basic Pipeline Configuration

input {
  file {
    path => "/var/log/myapp.log"
    start_position => "beginning"
  }
}

filter {
  grok {
    match => { "message" => "%{TIMESTAMP_ISO8601:timestamp} %{LOGLEVEL:loglevel} %{MESSAGE:message}" }
  }
  date {
    match => [ "timestamp", "ISO8601" ]
  }
}

output {
  elasticsearch {
    hosts => [ "localhost:9200" ]
    index => "myapp-%{+YYYY.MM.dd}"
  }
}

2. Creating Pipeline Files

Property Description Example
input Defines the source of the data to be processed {<br> "file": {<br> "path": "/var/log/app.log"<br> }<br>}
filter Transforms and enriches the data {<br> "grok": {<br> "match": {<br> "message": "%{TIMESTAMP}"<br> }<br> }<br>}
output Defines where to send the processed data {<br> "elasticsearch": {<br> "hosts": ["localhost:9200"]<br> }<br>}

Example: Basic Pipeline Configuration

{<br>  "pipeline": {<br>    "id": "my-pipeline",<br>    "config": "input { ... } filter { ... } output { ... }"<br>  }<br>}
Note: Pipeline files are written in Ruby-based syntax and must be placed in the pipelines directory of Logstash.

3. Defining Multiple Pipelines

Pipeline ID Configuration File Description
main pipelines.yml Primary pipeline for main log processing
audit pipelines.yml Dedicated pipeline for audit logs
metrics pipelines.yml Pipeline for metrics collection and processing

Example: pipelines.yml configuration

pipeline.1:
  id: "main"
  path.config: "/etc/logstash/conf.d/main.conf"

pipeline.2:
  id: "audit"
  path.config: "/etc/logstash/conf.d/audit.conf"

pipeline.3:
  id: "metrics"
  path.config: "/etc/logstash/conf.d/metrics.conf"
Note: Each pipeline must have a unique ID and its own configuration file.
Warning: Logstash will not start if any pipeline configuration is invalid.

4. Using Conditional Statements

Logstash supports conditional statements to control the flow of events through a pipeline. These conditions allow you to apply different filters or actions based on specific criteria.

Condition Type Syntax Description Example
if if [field] == "value" { ... } Applies filter if field matches value 
if [http_response_code] == 200 {
  mutate { add_tag => "success" }
}
else if else if [field] == "value" { ... } Alternative condition when previous is false 
if [http_response_code] == 200 {
  mutate { add_tag => "success" }
} else if [http_response_code] == 404 {
  mutate { add_tag => "not_found" }
}
unless unless [field] == "value" { ... } Applies filter if condition is NOT met 
unless [user_agent] =~ /bot/ {
  mutate { add_tag => "human_traffic" }
}

Example: Complex Conditional Pipeline

# Example pipeline with nested conditions
filter {
  if [type] == "nginx" {
    grok { match => { "message" => "%{IP:client_ip} %{WORD:method} %{URIPATH:uri} %{NUMBER:status_code}" } }
    
    if [status_code] == "404" {
      mutate { add_tag => "not_found_error" }
    } else if [status_code] == "500" {
      mutate { add_tag => "server_error" }
    } else {
      mutate { add_tag => "success" }
    }
  }
}
Note: Conditions in Logstash support a wide range of operators including ==, !=, <, >, =~ (regex match), and !~ (negated regex). Use parentheses to group complex conditions.
Warning: Complex conditional logic can impact pipeline performance. Avoid deeply nested conditions or excessive use of regex matching in high-volume pipelines.

5. Setting Pipeline Workers

Property Values Description
pipeline.workers 1 to 32 Number of worker threads to use for processing events. Default is 1.
pipeline.batch.size 125 to 2048 Number of events to process in a batch. Larger batches can improve throughput but increase latency.
pipeline.batch.delay 5 to 5000 Delay in milliseconds before processing a batch. Helps reduce CPU usage by batching events.

Example: Configuring Pipeline Workers

pipeline.workers: 4
pipeline.batch.size: 512
pipeline.batch.delay: 50
Note: Increasing pipeline.workers can improve throughput but may increase memory usage and contention. Adjust based on your system's capabilities.

6. Configuring Batch Size

Property Values Description
pipeline.batch.size 125 Number of events processed in a single batch. Default is 125.
pipeline.batch.delay 50 Delay in milliseconds before processing a batch. Default is 50ms.

Example: Setting Batch Size in Logstash Configuration

{
  "pipeline": {
    "batch": {
      "size": 250,
      "delay": 100
    }
  }
}
Note: Increasing batch size can improve throughput but may increase memory usage and latency.

7. Using Pipeline-to-Pipeline Communication

Logstash supports communication between pipelines using the pipeline output plugin. This allows one pipeline to send events to another pipeline running on the same or different Logstash instance.

Property Description Default
pipeline Name of the target pipeline to send events to none
host Hostname or IP address of the Logstash instance running the target pipeline localhost
port Port number to connect to on the target Logstash instance 5044
timeout Timeout in seconds for connecting to the target pipeline 30
ssl_enabled Enable SSL/TLS encryption for communication false
ssl_certificate Path to the SSL certificate file null
ssl_key Path to the SSL private key file null

Example: Sending Events to Another Pipeline

output {
  pipeline {
    host => "logstash-2.example.com"
    port => 5044
    pipeline => "pipeline-2"
  }
}
Note: Ensure that the target pipeline is configured to accept incoming events via the pipeline input plugin.
Warning: When using pipeline output, make sure that the target Logstash instance is configured to handle incoming events properly to avoid data loss or performance issues.

8. Enabling Dead Letter Queue

Dead Letter Queues (DLQ) in Logstash help manage failed events by routing them to a separate queue for further analysis or retry processing.

Configuration Option Description Default Value Notes
dead_letter_queue.enabled Enables or disables the dead letter queue functionality. false Set to true to enable DLQ.
dead_letter_queue.path Path to the directory where dead letter queue files are stored. /var/lib/logstash/dead_letter_queue Must be a writable directory.
dead_letter_queue.max_retention_time Maximum time (in seconds) that events can remain in the DLQ before being purged. 3600 Set to -1 for infinite retention.

Example: Enabling Dead Letter Queue in Logstash Configuration

input {
  beats {
    port => 5044
  }
}

filter {
  grok {
    match => { "message" => "%{TIMESTAMP_ISO8601:timestamp} %{LOGLEVEL:loglevel} %{MESSAGE:message}" }
  }

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

output {
  elasticsearch {
    hosts => [ "localhost:9200" ]
    index => "logs-%{+YYYY.MM.dd}"
  }

  dead_letter_queue {
    enabled => true
    path => "/var/lib/logstash/dead_letter_queue"
    max_retention_time => 3600
  }
}
Note: Events in the DLQ are stored as JSON files and can be reprocessed or analyzed separately.

9. Configuring Persistent Queue

Property Description Default Value Notes
queue.type Queue type to use for persistent queue persisted Must be persisted for persistent queue
queue.path Directory where queue files are stored /var/lib/logstash/queue Must be writable by Logstash process
queue.max_events Maximum number of events to queue 1024 Set to 0 for unlimited
queue.max_bytes Maximum size of queue in bytes 1073741824 Default is 1GB
queue.checkpoint.warnings Enable warnings when checkpoint files are too large true Set to false to disable warnings

Example: Persistent Queue Configuration

queue.type: persisted
queue.path: /var/lib/logstash/queue
queue.max_events: 2048
queue.max_bytes: 2147483648
queue.checkpoint.warnings: false
Note: Persistent queue is enabled by default in Logstash 8.x. It ensures that events are not lost during restarts or system failures.

10. Setting Pipeline Ordering

Pipeline Order Description
pipeline.order 0 Defines the order in which pipelines are processed. Lower numbers are processed first.
pipeline.name main-pipeline Name of the pipeline for identification in logs and monitoring tools.

Example: Configuring Pipeline Order

pipeline:
  order: 1
  name: "main-pipeline"
Note: The pipeline.order setting ensures that pipelines are processed in a specific sequence, which is crucial for maintaining data flow integrity.

11. Using Environment Variables

Variable Description Default Value Usage
LOGSTASH_HOME Root directory of Logstash installation /usr/share/logstash Used to locate Logstash configuration and plugin directories
LS_JAVA_OPTS Java options for Logstash process -Xmx1g -Xms256m Set memory limits and JVM options
LS_SETTINGS_DIR Directory containing Logstash settings files /etc/logstash Overrides default settings directory
LS_LOG_DIR Directory for Logstash log files /var/log/logstash Configures where logs are written to

Example: Setting Environment Variables in systemd Service File

# /etc/systemd/system/logstash.service
[Service]
Environment=LS_JAVA_OPTS="-Xmx2g -Xms512m"
Environment=LOGSTASH_HOME=/opt/logstash
Environment=LS_SETTINGS_DIR=/etc/logstash-custom
Note: Environment variables are loaded before Logstash starts and can be overridden by command-line arguments or configuration files.

12. Reloading Configuration

Logstash supports reloading configurations without restarting the entire service. This is useful for applying changes to pipelines dynamically.

Method Description Notes
SIGUSR1 signal Send SIGUSR1 to the Logstash process to trigger a configuration reload. Works for configurations managed via logstash.yml or file-based pipelines.
API endpoint Use the /_node/hot_reload API to reload configurations. Available in Logstash 7.10+.

Example: Reloading via Signal

kill -SIGUSR1 <logstash_pid>

Example: Reloading via API

curl -X POST "http://localhost:9600/_node/hot_reload"
Note: Configuration reloads are asynchronous and may take a few seconds to complete.
Warning: If a configuration is invalid, Logstash will continue running the previous valid configuration and log an error.

Was this article helpful?

Related Cheat Sheets