Handle different input formats in one Logstash pipeline

Give each input a routing marker, then use conditional filters and outputs for its format. The input option type adds a scalar event field; tags supports multiple labels. Neither is an Elasticsearch mapping type.

Parse and route three sources

This example keeps all three sources, including syslog. Port 5514 avoids the privileges normally required to bind port 514; configure the sender accordingly.

input {
  file {
    type => "web_server_logs"
    path => "/var/log/apache/access.log"
  }
  file {
    type => "application_logs"
    path => "/var/log/app/application.log"
  }
  syslog {
    type => "syslog"
    port => 5514
  }
}
filter {
  if [type] == "web_server_logs" {
    grok { match => { "message" => "^%{COMBINEDAPACHELOG}$" } }
    mutate { add_field => { "[@metadata][index_prefix]" => "web-logs" } }
  } else if [type] == "application_logs" {
    grok {
      match => { "message" => "^%{TIMESTAMP_ISO8601:log_time} %{LOGLEVEL:level} %{GREEDYDATA:msg}$" }
    }
    mutate { add_field => { "[@metadata][index_prefix]" => "app-logs" } }
  } else if [type] == "syslog" {
    mutate { add_field => { "[@metadata][index_prefix]" => "syslog" } }
  } else {
    mutate { add_field => { "[@metadata][index_prefix]" => "other-logs" } }
  }
}
output {
  elasticsearch {
    hosts => ["http://localhost:9200"]
    ilm_enabled => false
    data_stream => false
    index => "%{[@metadata][index_prefix]}-%{+yyyy.MM.dd}"
  }
}

Use the connection security required by your cluster. These are ordinary daily indices; configure templates and retention separately. The syslog input already attempts its own parsing, so inspect its fields and failure tags instead of applying an unrelated application grok pattern.

The application pattern does not extract a response time. Add an explicit field capture before attempting to convert such a field to a number. A date filter is likewise required if log_time should replace @timestamp.

Keep routing predictable

An input's type does not overwrite a type already present in an incoming event. Use tags or another controlled field when upstream producers set their own types. if / else if selects one matching branch; separate if blocks can all run on the same event.

For growing configurations, define separate pipelines in pipelines.yml. Merely splitting a configuration into several files under one path.config combines them into one pipeline. Separate pipelines can isolate blocked outputs and tune workers or queues per flow, but they still share a process and host resources.

References

• Elastic: inputs file
• Elastic: multiple pipelines
• Elastic: inputs syslog
• Elastic: outputs elasticsearch