示例：设置 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 选项。

运行带有 --pipelines 和 --modules 选项的 setup 命令以加载已启用的模块的 Ingest Pipelines。此步骤还需要连接到 Elasticsearch。如果您想使用 Logstash pipeline 而不是 Ingest Node 来解析数据，请跳过此步骤。

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 的系统上，创建一个 Logstash pipeline 配置，该配置从 Kafka 输入读取事件并将事件发送到 Elasticsearch 输出。

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，传入您刚刚定义的 pipeline 配置文件。例如：

bin/logstash -f mypipeline.conf

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