具有多个 Logstash 管道的 Filebeat

Question

我将 Filebeat 配置为在单个主机上查看多个不同的日志，例如Nginx 和我的应用服务器。然而，据我所知，你不能在任何一个 Beat 中有多个输出——所以我的 filebeat.yml 有一个指向我的 Logstash 服务器的 output.logstash 指令。

Logstash有管道路由的概念吗？我在我的 Logstash 服务器上配置了几个管道，但不清楚如何从 Filebeat 中利用它，例如我想将 Nginx 日志发送到我的 Nginx 等的 Logstash 管道

或者，有没有办法将 Nginx 的节拍路由到 logstash:5044，并将我的应用程序服务器的节拍路由到 logstash:5045。

Answer 1

对于每个 filebeat prospector，您可以使用 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 将不同类型的数据发送到不同的管道。

Answer 2

您可以在 filebeat 输入上使用标签，并使用这些标签在 logstash 管道上进行过滤。

例如，将标签 nginx 添加到 filebeat 中的 nginx 输入，并将标签 app-server 添加到 filebeat 中的应用服务器输入中，然后在 logstash 管道中使用这些标签来使用不同的过滤器和输出，它将是相同的管道，但它会根据标签路由事件。

如果您想将不同的日志发送到不同的端口，您将需要运行另一个 Filebeat 实例。

Answer 3

您可以对多个日志文件使用标签概念

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}"
       }
    }
}

具有多个 Logstash 管道的 Filebeat

Filebeat with multiple Logstash pipelines

logstash

filebeat