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