微信公众号

filebeat+kafka+ELK分布式日志收集

引言

原文链接:https://yunsonbai.top/2020/01/16/log_collect/

1
2
3
随着业务的发展,为了满足越来越多的业务需求,逐渐从原来的单机到多机再到基于docker的集群,发展
到集群会带来新的问题需要解决,比如日志散落在各个实例上,对于日志的统计分析带来了新的要求,
一台一台的去查显然是不合理的,这也就引出本文讨论的主题,分布式日志收集。

简单介绍一下用到的组件/工具

filebeat

  • 文档: 文档,我觉得这里边已经说的很清楚为啥使用filebeat做日志收集器了
  • 优势:
    • 基于golang的轻量级日志采集器,配置启动出奇的简单
    • 按官方说法elastic专门为成千上万机器日志收集做的收集器

kafka

  • 文档: 文档
  • 优势:
    • 应该说kafka的诞生就是为日志收集做服务的
    • 几乎可以认为kafka集群没有qps上限,单机都能到10W/s的吞吐,完全分布式。[可怕]
    • 好文推荐: Kafka 背景及架构介绍

ELK

  • E: ElasticSearch
    • 文档: 文档
    • 优势: 高度可扩展的开源全文搜索和分析引擎,可快速、实时存储、搜索和分析数据。(网上太多了)
    • 好文推荐: 图解elasticsearch原理,有点老我觉得不影响理解
  • L: Logstash
    • 文档: 文档
    • 优势: 实时流水线功能,分析合并来源(MySQL、Redis、kafka)数据,并输出到目标(es、file等)存储地址
    • 好文推荐: 上手Logstash
  • K: kibana
    • 文档: 文档
    • 优势: 完备的前端展示

grafana

  • 是什么: 可以理解和kibana一样的东西,出于个人的喜爱,它真的很黑炫酷
  • 文档: 文档

其他说明

1
2
3
其实没有花太多篇幅介绍上边的组件的使用和原理,本文的分享重点不在这里,只是分享用什么样的架构来使用
这些组件,关于这些组件的原理和使用我分享了几篇好文章以及官方文档,大家可以参考,另外网上这种介绍
文章很多,接下来分享我的目前日志量和收集架构

架构情况说明

  • 架构图:
    架构
  • 实例数: 全部基于docker部署,各类角色的实例数过千
  • 日志量: 日均日志约4.6亿条, 占空间约140G。
  • 目前收集架构负载: 十几个es节点个位数负载,个位数logstash节点也几乎没有负载。
  • 版本说明: filebeat、logstash、es、kibana版本要一致

组件使用注意点

filebeat

  • 打包说明: 镜像在打包时,要添加上filebeat可执行文件(可在官网下载),可以使用supervisor管理服务。
  • filebeat配置文件可参考以下例子:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
filebeat.prospectors:
- input_type: log
paths: /var/log/nginx.log
document_type: nginx_log
fields:
cluster_name: ${CLUSTER_NAME}
host: ${HOST}
log_topics: app1_nginx # nginx日志

- input_type: log
paths: /var/applog/*.log
document_type: applog
fields:
cluster_name: ${CLUSTER_NAME}
host: ${HOST}
log_topics: app1_log # 应用日志

output.kafka:
hosts: ["kafka1:9092","kafka2:9092",...,"kafkaN:9092"] # 取决你的集群节点数
topic: '%{[fields][log_topics]}'
partition.round_robin:
reachable_only: false
required_acks: 1
compression: gzip

logstash

  • 机器数量: 可以找几台虚机(4c+8G)启动,尽量个数和kafka的节点数一致。
  • input和output: kafka(filebeat日志流向的kafka)和es集群
  • 配置注意:
    • pipeline.batch.size: 2000 # 达到多少个events后向目标地址输送数据
    • pipeline.batch.delay: 10 # 等待多少秒向目标地址输送数据
1
2
两个配置不冲突,哪个满足了就触发向目标输送数据,我们的目标地就是es集群。
至于批量和延迟向目标输送数据应该好理解,避免频繁请求目标地址,导致目标地址高负载。
  • 关于filter
1
2
一般会用到的有grok(正则切割日志)、json(json解析)、mutate(组合命令remove_field(去除无用字段)等等)
很多这里不一一介绍了,推荐一个可以在线测试grok语法是否正确的工具:http://grokdebug.herokuapp.com/
  • 配置样例
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
input {
kafka{
bootstrap_servers => "kafka1:9092,kafka2:9092,...,kafkaN:9092" # 前边的kafka
auto_offset_reset => "latest"
group_id => "app1"
consumer_threads => 1
decorate_events => true
codec => "json"
topics => ["app1_log"]
}
}
filter {
if [fields][log_topics] == "app1_log" {
grok {
match => {"message" => '(?<time_local>[^\|]*)\|(?<code_line>[^\|]*)\|(?<level>[^\|]*)\|(?<log_json>.*)'}
}
mutate {
gsub => ["log_json", "[\|]", "_"] # 替换|为_
}
json {
source => "log_json"
remove_field=>["log_json"]
}
}
# 可以有多个if
# remove not care field
mutate
{
remove_field => ["field1", "field2"]
}
}
output {
if [fields][log_topics] == "app1_log" {
elasticsearch {
hosts => ["es1:9200", "es2:9200",..,"esN:9200"]
index => "app1_log-%{+YYYY.MM.dd}"
}
}
# 可以有多个if
}

关于es

  • 注意点: 注意修改number_of_shards数量等于节点数,es的number_of_shards默认为5
1
2
跳过一次坑,没有修改number_of_shards,虽然机器多,但是日志散落不均匀导致总有es的某几个
节点负载比较高,其他的却很清闲。

关于数据展示

  • kibana:一张老图
    一张老图
  • grafana
    炫酷的图]
  • 统计脚本: python、golang、java任选

总结

1
2
3
4
5
日志对于监控系统流量、提升系统性能、发现系统问题等有着十分重要的意义,可以说有些日志对于系统
的演变来说起着决定行的作用。所以日志的收集是一项很重要且很有意思的工作,也可以说是一门学问,
如何规范的输出日志方便后边的收集,如何压缩日志尽量减少磁盘的使用,如何控制日志结构使搜索更快
等等这些问题。
本文重点分享了当前我采用的收集方式,欢迎大家提出纠正和宝贵的意见。
yunsonbai wechat
微信公众号