inanzzz | Handling multiple log files with Filebeat and Logstash in ELK stack

In this example we are going to use Filebeat to forward logs from two different logs files to Logstash where they will be inserted into their own Elasticsearch indexes.

Filebeat

filebeat:
  prospectors:
    -
      paths:
        - /var/log/apache2/access.log
      input_type: log
      document_type: apache-access

    -
      paths:
        - /var/log/symfony/dev.log
      input_type: log
      document_type: symfony-dev

Logstash

Version 1

As you can see, we use mutate block to define new variable for Elasticsearch index to use in "output" block. You can move mutate in "if" and "else" blocks if you wish.

input {
    beats {
        port => 5044
    }
}

filter {
    if [type] == "apache-access" {
        grok {
            match => { "message" => "%{COMBINEDAPACHELOG}" }
        }
    } else if [type] == "symfony-dev" {
        grok {
            match => { "message" => "%{COMBINEDAPACHELOG}" }
        }
    }
    
    mutate {
        add_field => { "index_name" => "web-%{type}" }
    }
}

output {
    elasticsearch {
        hosts => ["localhost:9200"]
        sniffing => true
        manage_template => false
        index => "%{index_name}"
    }
}

Version 2

This is how you use type without "if" condition in "output" block.

input {
    beats {
        port => 5044
    }
}

filter {
    if [type] == "apache-access" {
        grok {
            match => { "message" => "%{COMBINEDAPACHELOG}" }
        }
    } else if [type] == "symfony-dev" {
        grok {
            match => { "message" => "%{COMBINEDAPACHELOG}" }
        }
    }
}

output {
    elasticsearch {
        hosts => ["localhost:9200"]
        sniffing => true
        manage_template => false
        index => "web-%{type}"
    }
}

Version 3

This is how you use type with "if" condition in "output" block. I am duplicating lines just to show you how things are done.

input {
    beats {
        port => 5044
    }
}

filter {
    if [type] == "apache-access" {
        grok {
            match => { "message" => "%{COMBINEDAPACHELOG}" }
        }
    } else if [type] == "symfony-dev" {
        grok {
            match => { "message" => "%{COMBINEDAPACHELOG}" }
        }
    }
}

output {
    if [type] == "apache-access" {
        elasticsearch {
            hosts => ["localhost:9200"]
            sniffing => true
            manage_template => false
            index => "web-%{type}"
        }
    } else if [type] == "symfony-dev" {
        elasticsearch {
            hosts => ["localhost:9200"]
            sniffing => true
            manage_template => false
            index => "web-%{type}"
        }
    }
}