Logstash Pipelines: Inputs, Filters, Outputs & Grok

In this tutorial, you'll learn about Logstash Pipelines: Inputs, Filters, Outputs & Grok. We cover key concepts, practical examples, and best practices to help you understand and apply this topic effectively.

Logstash pipelines consist of three stages -- input, filter, and output -- that transform raw log data into structured, searchable documents ready for Elasticsearch indexing.

What You'll Learn

In this tutorial, you will create a Logstash pipeline that reads log files, parses them with grok filters, adds metadata with mutate filters, and sends the structured data to Elasticsearch.

Why It Matters

Raw log data is unstructured text. Without Logstash pipelines, you would index unparsed log lines, making it impossible to search by fields like IP address, status code, or error type. A well-designed pipeline extracts structured fields, normalizes timestamps, enriches data with geo-location, and ensures consistent field types across all log sources.

Real-World Use

DodaZIP generates access logs with fields embedded in a single text line. A Logstash pipeline uses a grok filter to extract IP address, timestamp, HTTP method, path, status code, and response size from each line. The mutate filter converts the status code to an integer and response size to a long. Elasticsearch receives perfectly structured documents ready for Kibana dashboards.

Pipeline Structure

A Logstash pipeline configuration file defines the three stages:

input {
  beats {
    port => 5044
  }
}

filter {
  grok {
    match => { "message" => "%{COMBINEDAPACHELOG}" }
  }
  date {
    match => [ "timestamp", "dd/MMM/yyyy:HH:mm:ss Z" ]
  }
}

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

Input Plugins

The input stage defines how Logstash receives data:

• beats -- receives from Filebeat (port 5044)
• file -- tails log files directly
• syslog -- listens for syslog messages (port 514)
• tcp/udp -- receives raw data over TCP or UDP
• http -- receives data via HTTP requests

Filter Plugins

Filters transform data. The most important filters are:

grok -- parses unstructured text into structured fields using pattern matching:

filter {
  grok {
    match => { "message" => "%{TIMESTAMP_ISO8601:timestamp} %{LOGLEVEL:level} %{GREEDYDATA:message_body}" }
  }
}

mutate -- transforms field values (convert types, rename, remove):

filter {
  mutate {
    convert => { "response_size" => "integer" }
    rename => { "message_body" => "log_message" }
    remove_field => [ "unwanted_field" ]
  }
}

date -- parse timestamps into Elasticsearch's @timestamp field:

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

Output Plugins

The output sends processed data to a destination:

output {
  elasticsearch {
    hosts => ["https://elasticsearch:9200"]
    user => "logstash"
    password => "${ES_PASSWORD}"
    index => "logs-%{+yyyy.MM.dd}"
    ssl => true
  }
  stdout { codec => rubydebug }
}

Adding stdout output with rubydebug codec is useful for debugging -- it prints the processed event to the console.

Running a Pipeline

Run Logstash with a pipeline configuration:

bin/logstash -f pipeline.conf --config.reload.automatic

The --config.reload.automatic flag reloads the config when it changes, avoiding restarts during development.

Common Mistakes

1. No Error Handling in Filters

If a grok pattern fails to match, the event proceeds without Parsing. Use _grokparsefailure tags to filter or handle unmatched events.

2. Missing Timestamp Parsing

Without a date filter, Logstash uses the ingestion time for @timestamp, not the original log timestamp. Always parse the original timestamp.

3. Hardcoding Elasticsearch Credentials

Passwords in config files are a security risk. Use environment variables or the Elasticsearch keystore for credentials.

4. Overly Complex Single Pipeline

Monolithic pipelines are hard to debug. Split into multiple pipelines for different log types using pipelines.yml.

5. Not Setting pipeline.workers

Logstash defaults to the number of CPU cores. For high-throughput environments, configure pipeline.workers to match your workload.

Practice Questions

1. What are the three stages of a Logstash pipeline? Input (receiving data), filter (transforming data), and output (sending data to destination).

2. What does the grok filter do? It parses unstructured text into structured fields using pattern matching with predefined or custom patterns.

3. Why use a date filter in a Logstash pipeline? To replace the default @timestamp (ingestion time) with the actual timestamp from the log event.

4. How do you debug a Logstash pipeline? Add stdout { codec => rubydebug } as an output to print processed events to the console.

5. Challenge: Create a pipeline that parses Nginx access logs, converts status code and bytes to integers, removes the user agent field, and indexes logs into daily indices.

What's Next

Visualize your Elasticsearch data with Kibana dashboards, creating charts and graphs for operational insights.

Built by the developers of Doda Browser, DodaZIP, and Durga Antivirus Pro.