Beats 平台

Beats 平台是 Elastic.co 从 packetbeat 发展出来的数据收集器系统。beat 收集器可以直接写入 Elasticsearch，也可以传输给 Logstash。其中抽象出来的 libbeat，提供了统一的数据发送方法，输入配置解析，日志记录框架等功能。

也就是说，所有的 beat 工具，在配置上，除了 input 以外，在 output、filter、shipper、logging、run-options 上的配置规则都是完全一致的。

filter

5.0 版本后，beats 新增了简单的 filter 功能，用来完成事件过滤和字段删减：

  1. filters:
  2.     - drop_event:
  3.         regexp:
  4.             message: "^DBG:"
  5.     - drop_fields:
  6.         contains:
  7.             source: "test"
  8.         fields: ["message"]
  9.     - include_fields:
  10.         fields: ["http.code", "http.host"]
  11.         equals:
  12.             http.code: 200
  13.         range:
  14.             gte:
  15.                 cpu.user_p: 0.5
  16.             lt:
  17.                 cpu.user_p: 0.8

可用的条件判断包括：

  • equals
  • contains
  • regexp
  • range
  • or
  • and
  • not

output

目前 beat 可以发送数据给 Elasticsearch、Logstash、File、Kafka、Redis 和 Console 六种目的地址。

Elasticsearch

beats 发送到 Elasticsearch 也是走 HTTP 接口。示例配置段如下：

  1. output:
  2.     elasticsearch:
  3.         hosts: ["http://localhost:9200", "https://onesslip:9200/path", "anotherip"]
  4.         parameters: {pipeline: my_pipeline_id}                         # 仅用于 Elasticsearch 5.0 以后的 ingest 方式
  5.         username: "user"
  6.         password: "pwd"
  7.         index: "topbeat"
  8.         bulk_max_size: 20000
  9.         flush_interval: 5
  10.         tls:
  11.             certificate_authorities: ["/etc/pki/root/ca.pem"]
  12.             certificate: "/etc/pki/client/cert.pem"
  13.             certificatekey: "/etc/pki/client/cert.key"
  • hosts 中可以通过 URL 的不同形式，来表示 HTTP 还是 HTTPS，是否有添加代理层的 URL 路径等情况。
  • index 表示写入 Elasticsearch 时索引的前缀，比如示例即表示索引名为 topbeat-yyyy.MM.dd

Logstash

beat 写入 Logstash 时，会配合 Logstash-1.5 后新增的 metadata 特性。将 beat 名和 type 名记录在 metadata 里。所以对应的 Logstash 配置应该是这样：

  1. input {
  2.     beats {
  3.         port => 5044
  4.     }
  5. }
  6. output {
  7.     elasticsearch {
  8.         hosts => ["http://localhost:9200"]
  9.         index => "%{[@metadata][beat]}-%{+YYYY.MM.dd}"
  10.         document_type => "%{[@metadata][type]}"
  11.     }
  12. }

beat 示例配置段如下:

  1. output:
  2.     logstash:
  3.         hosts: ["localhost:5044", "localhost:5045"]
  4.         worker: 2
  5.         loadbalance: true
  6.         index: topbeat

这里 worker 的含义，是 beat 连到每个 host 的线程数。在 loadbalance 开启的情况下，意味着有 4 个worker 轮训发送数据。

File

  1. output:
  2.     file:
  3.         path: "/tmp/topbeat"
  4.         filename: topbeat
  5.         rotate_every_kb: 1000
  6.         number_of_files: 7

Kafka

  1. output:
  2.     kafka:
  3.         hosts: ["kafka1:9092", "kafka2:9092", "kafka3:9092"]
  4.         topic: '%{[type]}'
  5.         topics:
  6.             - key: "info_list"
  7.               when:
  8.                   contains:
  9.                       message: "INFO"
  10.             - key: "debug_list"
  11.               when:
  12.                   contains:
  13.                       message: "DEBUG"
  14.             - key: "%{[type]}"
  15.               mapping:
  16.                   "http": "frontend_list"
  17.                   "nginx": "frontend_list"
  18.                   "mysql": "backend_list"
  19.         partition:
  20.             round_robin:
  21.                 reachable_only: true
  22.         required_acks: 1
  23.         compression: gzip
  24.         max_message_bytes: 1000000
  • 大于 max_message_bytes 长度的事件(注意不只是原日志长度)会被直接丢弃。
  • partition 策略默认为 hash。可选项还有 random 和 round_robin。
  • compression 可选项还有 none 和 snappy。
  • required_acks 可选项有 -1、0 和 1。分别代表：等待全部副本完成、不等待、等待本地完成。
  • topics 用来配置基于匹配规则的选择器，支持 when 和 mapping，when 条件下可以使用上小节列出的各种 filter。如果都匹配不上，则采用 topic 配置。

Redis

  1. output:
  2.     redis:
  3.         hosts: ["localhost"]
  4.         password: "my_password"
  5.         key: "filebeat"
  6.         db: 0
  7.         timeout: 5

Redis 输出也有 keys 配置。方式和 Kafka 的 topics 类似。

Console

  1. output:
  2.     console:
  3.         pretty: true

shipper

shipper 部分是一些和网络拓扑相关的配置，就目前来说，大多数是 packetbeat 独有的。

  1. shipper:
  2.     name: "my-shipper"
  3.     tags: ["my-service", "hardware", "test"]
  4.     ignore_outgoing: true
  5.     refresh_topology_freq: 10
  6.     topology_expire: 15
  7.     geoip:
  8.         paths:
  9.             - "/usr/share/GeoIP/GeoLiteCity.dat"

logging

  1. logging:
  2.     level: warning
  3.     to_files: true
  4.     to_syslog: false
  5.     files:
  6.         path: /var/log/mybeat
  7.         name: mybeat.log
  8.         keepfiles: 7

run options

  1. runoptions:
  2.     uid=501
  3.     gid=501