示例：设置 Filebeat 模块以使用 Kafka 和 Logstash

如果您尚未设置 Filebeat 索引模板和示例 Kibana 仪表板，请立即运行 Filebeat setup 命令来执行此操作。

filebeat -e setup

-e 标志是可选的，它将输出发送到标准错误而不是 syslog。

此一次性设置步骤需要连接到 Elasticsearch 和 Kibana，因为 Filebeat 需要在 Elasticsearch 中创建索引模板并将示例仪表板加载到 Kibana 中。有关配置与 Elasticsearch 的连接的更多信息，请参阅 Filebeat 快速入门。

加载模板和仪表板后，您将看到消息 INFO {kib} dashboards successfully loaded. Loaded dashboards。

运行 modules enable 命令以启用您要运行的模块。例如

filebeat modules enable system

您可以通过编辑 Filebeat modules.d 目录下的配置文件来进一步配置模块。例如，如果日志文件不在模块预期的位置，您可以设置 var.paths 选项。

运行 setup 命令，并指定 --pipelines 和 --modules 选项以加载您已启用的模块的摄取管道。此步骤也需要连接到 Elasticsearch。如果您想使用 Logstash 管道而不是摄取节点来解析数据，请跳过此步骤。

filebeat setup --pipelines --modules system

配置 Filebeat 以将日志行发送到 Kafka。为此，在 filebeat.yml 配置文件中，通过注释掉来禁用 Elasticsearch 输出，并启用 Kafka 输出。例如

#output.elasticsearch:
  #hosts: ["localhost:9200"]
output.kafka:
  hosts: ["kafka:9092"]
  topic: "filebeat"
  codec.json:
    pretty: false

启动 Filebeat。例如

filebeat -e

Filebeat 将尝试将消息发送到 Logstash，并持续发送，直到 Logstash 可以接收它们为止。

在安装 Logstash 的系统上，创建一个从 Kafka 输入读取并将事件发送到 Elasticsearch 输出的 Logstash 管道配置

input {
  kafka {
    bootstrap_servers => "myhost:9092"
    topics => ["filebeat"]
    codec => json
  }
}

output {
  if [@metadata][pipeline] {
    elasticsearch {
      hosts => "https://myEShost:9200"
      manage_template => false
      index => "%{[@metadata][beat]}-%{[@metadata][version]}-%{+YYYY.MM.dd}"
      pipeline => "%{[@metadata][pipeline]}" 
      user => "elastic"
      password => "secret"
    }
  } else {
    elasticsearch {
      hosts => "https://myEShost:9200"
      manage_template => false
      index => "%{[@metadata][beat]}-%{[@metadata][version]}-%{+YYYY.MM.dd}"
      user => "elastic"
      password => "secret"
    }
  }
}

启动 Logstash，传入您刚定义的管道配置文件。例如

bin/logstash -f mypipeline.conf

Logstash 应启动一个管道并开始从 Kafka 输入接收事件。