A Logstash pipeline from input to Elasticsearch
A Logstash pipeline is three blocks: where events come from, how they are transformed, and where they go. This tool assembles a common shape — Beats in, Grok and Date and Mutate filters, Elasticsearch out — into a ready pipeline.conf.
How it works
The input block opens a beats listener on a port that Filebeat ships to over the lumberjack protocol. Events then flow through the filter block in written order, which is why sequence matters. grok matches a pattern against the raw message field and extracts named fields; for example %{COMBINEDAPACHELOG} turns an access-log line into status, response time, and client IP. Because the timestamp only exists after parsing, the date filter runs next, parsing a field with a Joda format string and writing the result into @timestamp so dashboards align with when events actually happened rather than when Logstash saw them. The mutate filter then tidies up, using remove_field to drop the now-redundant raw message and noisy host fields.
The output block sends each event to Elasticsearch. The index setting uses %{+YYYY.MM.dd} date math to route events into daily indices, which keeps retention manageable since an entire day can be deleted as one index.
Tips and example
Keep Grok before Date, and validate patterns in Kibana’s Grok Debugger before deploying — a mismatch tags events with _grokparsefailure and leaves message unparsed. A minimal pipeline:
input { beats { port => 5044 } }
filter {
grok { match => { "message" => "%{COMBINEDAPACHELOG}" } }
date { match => [ "timestamp", "dd/MMM/yyyy:HH:mm:ss Z" ] target => "@timestamp" }
}
output {
elasticsearch {
hosts => [ "http://localhost:9200" ]
index => "app-logs-%{+YYYY.MM.dd}"
}
}Drop the file into the Logstash conf.d directory and reload; pipeline reloads are supported without a full restart when config.reload.automatic is on.