Route Logs to Elasticsearch Indices with Fluentd or Logstash

Route logs to different Elasticsearch indices only when the separation serves retention, access or query requirements. Parse each event, classify it with an allowlisted routing value and preserve unmatched events in a fallback stream. A parse failure is not the same as an application error.

Separate parsing from routing

Start with known fixtures: a valid access event, an application error and malformed input. Keep the raw message until parsing succeeds. The original Nginx grok patterns describe a specific custom format; validate them against your actual logs before reuse.

Logstash: use bounded metadata

Save this local classification example as routing.conf. With Logstash 9.5.5, send JSON Lines on standard input and inspect the output. The metadata field can subsequently select an Elasticsearch output index.

input { stdin { codec => json_lines } }
filter {
  if [kind] == "access" {
    mutate { add_field => { "[@metadata][target_index]" => "cloudhero-lab-access" } }
  } else if [kind] == "error" {
    mutate { add_field => { "[@metadata][target_index]" => "cloudhero-lab-error" } }
  } else {
    mutate { add_field => { "[@metadata][target_index]" => "cloudhero-lab-unparsed" } }
  }
}
output { stdout { codec => rubydebug { metadata => true } } }

printf '%s\n' '{"kind":"access","message":"GET / 200"}' \
  '{"kind":"error","message":"upstream timeout"}' \
  '{"message":"unknown format"}' \
  | docker run --rm -i \
      -v "$PWD/routing.conf:/usr/share/logstash/pipeline/routing.conf:ro" \
      -e LS_JAVA_OPTS='-Xms256m -Xmx512m' \
      docker.elastic.co/logstash/logstash:9.5.5 \
      -f /usr/share/logstash/pipeline/routing.conf

Expect three distinct target indices in metadata. This local test checks classification, including the fallback; it does not connect Logstash to Elasticsearch.

Fluentd: rewrite a tag once, then match the result

The equivalent Fluentd approach uses the rewrite-tag-filter plugin. Ensure the plugin is installed in a pinned image, match only the original tag and send each new tag to one destination. The fragment below illustrates rewriting; the live full pipeline also normalized the routing field before it reached these rules.

<match input.logs>
  @type rewrite_tag_filter
  <rule>
    key kind
    pattern /^access$/
    tag routed.access
  </rule>
  <rule>
    key kind
    pattern /^error$/
    tag routed.error
  </rule>
  <rule>
    key message
    pattern /.+/
    tag routed.unparsed
  </rule>
</match>

Provide a source-level fallback for records lacking both kind and message; unmatched rewrite rules can drop records. Avoid matching rewritten tags with the original rewrite rule, which can cause loops.

Validate delivery and lifecycle separately

Configure authenticated TLS, buffering, retry behavior and duplicate handling. Run a downstream outage test and confirm recovery without silent loss. Use current vendor installation methods rather than the retired Helm stable repository or archived Elastic charts. See index creation and templates before enabling new destinations.

Sources and further reading

• Fluentd rewrite-tag-filter
• Logstash Elasticsearch output
• Archived Elastic Helm charts

Live routing and fallback results

The EKS run tested Logstash 9.5.5 with Filebeat input and real Elasticsearch delivery, including a destination outage. Logstash 9 required api.http.host rather than the removed http.host setting. An explicit pipeline file avoided accidentally loading the image’s stock pipeline alongside the test configuration.

The Fluentd alternative used Fluentd 1.20.1, rewrite-tag-filter 2.4.1 and fluent-config-regexp-type 1.0.0. The plugin needs its dependency available in the same Ruby gem search path; build and test a pinned image rather than installing dependencies during production Pod startup. Access and error fixtures reached their respective indices. An unknown message and an empty object reached the unparsed index. The complete pipeline assigned one of three allowed route values before rewriting; this closed the missing-field drop case in the illustrative fragment.

<filter input.logs>
  @type record_transformer
  enable_ruby true
  <record>
    route ${["access","error"].include?(record["kind"]) ? record["kind"] : "unparsed"}
  </record>
</filter>

Use that normalized field in explicit access/error/unparsed rewrite rules, with one matching output per routed tag. The local HTTP-to-Elasticsearch outputs used one-record chunks for the fixture. Production output authentication, TLS, buffering capacity and delivery semantics still need configuration for the actual environment.