Filebeat with multiple Logstash pipelines

2
votes

I have Filebeat configured to watch several different logs on a single host, e.g. Nginx and my app server. However, as I understand it, you cannot have multiple outputs in any one Beat -- so my filebeat.yml has a single output.logstash directive which points to my Logstash server.

Does Logstash have the concept of pipeline routing? I have several pipelines configured on my Logstash server but it's unclear how to leverage that from Filebeat, e.g. I would like to send Nginx logs to my Logstash pipeline for Nginx, etc.

Alternatively, is there a way to route the beats for Nginx to logstash:5044, and the beats for my app server to logstash:5045.

3
logstashfilebeat
You should use pipeline concepts on logstash. You can write multiple xx.conf pipeline inf /etc/logstash/conf.d directory and define the conf's paths in /etc/logstash/pipelines.yml file. You can route filebeat inputs on different ports. follow my answer here. stackoverflow.com/questions/59892953/… – Siddharth Kumar

3 Answers

2
votes

For each of the filebeat prospectors you can use the fields option to add a field that logstash can check to identify what type of data the prospector is collecting. Then in logstash you can use pipeline-to-pipeline communication with the distributor pattern to send different types of data to different pipelines.

1
votes

You can use tags on your filebeat inputs and filter on your logstash pipeline using those tags.

For example, add the tag nginx to your nginx input in filebeat and the tag app-server in your app server input in filebeat, then use those tags in the logstash pipeline to use different filters and outputs, it will be the same pipeline, but it will route the events based on the tag.

If you want to send the different logs to different ports, you will need to run another instance of Filebeat.

1
votes

You can use tag concepts for multiple log files

filebeat.yml

filebeat.inputs:
- type: log
  tags: [apache]
  paths:
    - "/home/ubuntu/data/apache.log"

- type: log
  tags: [gunicorn]
  paths:
    - "/home/ubuntu/data/gunicorn.log"

queue.mem:
  events: 4096
  flush.min_events: 512
  flush.timeout: 5s

output.logstash:
  hosts: ["****************:5047"]

conf.d/logstash.conf

input {
  beats {
    port => "5047"
    host => "0.0.0.0"
  }
}

filter {
  if "gunicorn" in [tags] {
    grok {
       match => {
         "message" => "%{USERNAME:u1} %{USERNAME:u2} \[%{HTTPDATE:http_date}\] \"%{DATA:http_verb} %{URIPATHPARAM:api} %{DATA:http_version}\" %{NUMBER:status_code} %{NUMBER:byte} \"%{DATA:external_api}\" \"%{GREEDYDATA:android_client}\""
         remove_field => ["message"]
       }
    }

    date {
      match => ["http_date", "dd/MMM/yyyy:HH:mm:ss XX"]
    }

    mutate {
      remove_field => ["agent"]
    }
  }

  else if "apache" in [tags] {
     grok {
       match => {
         "message" => "%{IPORHOST:client_ip} %{DATA:u1} %{DATA:u2} \[%{HTTPDATE:http_date}\] \"%{WORD:http_method} %{URIPATHPARAM:api} %{DATA:http_version}\" %{NUMBER:status_code} %{NUMBER:byte} \"%{DATA:external_api}\" \"%{GREEDYDATA:gd}\" \"%{DATA:u3}\""
         remove_field => ["message"]
       }
     }

    date {
      match => ["http_date", "dd/MMM/yyyy:HH:mm:ss +ssss"]
    }

    mutate {
      remove_field => ["agent"]
    }
   }
}

output {
  if "gunicorn" in [tags] {
    stdout { codec => rubydebug }
    elasticsearch {
      hosts => ["0.0.0.0:9200"]
      index => "gunicorn-sample-%{+YYYY.MM.dd}"
      }
  }
  else if "apache" in [tags] {
     stdout { codec => rubydebug }
     elasticsearch {
       hosts => ["0.0.0.0:9200"]
       index => "apache-sample-%{+YYYY.MM.dd}"
       }
    }
}