一个完整的集中式日志系统，需要包含几个主要特点:
- 收集: 收集多种来源的日志数据
- 传输: 能够传输大量的日志的数据流
- 存储: 存储大量日志数据
- 分析: 可以支持日志UI分析
- 警告: 提供日志监控，警告 ELK提供一整套解决方案，分别表示: Elasticsearch,Logstash,Kibana。完善新的ELK系统又新增FileBeat,相较于Logstash,它更轻量级，占用资源少，适合在每台服务器上收集日志传输给Logstash。
如果将Filebeat的日志传输直接传输给Logstash，Logstash耗资源较大，运行占用CPU和内存高，并且没有队列，容易造成数据丢失，所以需要消息队列来缓存日志消息，kafka cluster有大吞吐，削峰填谷，能够确保日志数据量不被丢失。
Elasticsearch是个开源分布式搜索引擎，提供搜集、分析、存储数据三大功能。它的特点有：分布式，零配置，自动发现，索引自动分片，索引副本机制，restful风格接口，多数据源，自动搜索负载等。
Logstash 主要是用来日志的搜集、分析、过滤日志的工具，支持大量的数据获取方式。一般工作方式为c/s架构，client端安装在需要收集日志的主机上，server端负责将收到的各节点日志进行过滤、修改等操作在一并发往elasticsearch上去。
Kibana 也是一个开源和免费的工具，Kibana可以为 Logstash 和 ElasticSearch 提供的日志分析友好的 Web 界面，可以帮助汇总、分析和搜索重要数据日志。
Watcher是对Elasticsearch进行告警和通知的插件,可以根据数据的变化采取行动。它的设计原理是在Elasticsearch中执行查询,满足条件的情况下,产生告警。简单地定义查询、设置限定条件、设置预定计划和将要进行的操作后,Watcher会自动完成剩下的操作。
所以基于以上的功能组件，ELK技术栈具备成为日志平台的全部要素，其架构设计如下:
Filebeat
Filebeat由两个主要组件组成: prospectors(勘测者)和harvesters(收割机)，这两种组件协同工作将文件变动发送到指定的输出中。
Harvester(收割机): 负责读取单个文件内容，每个文件都会启动harvester，每个harvester会逐行读取各个文件，并将文件内容发送到指定输出中，harvester负责打开和关闭文件，意味在harvester运行的时候，文件描述符处于打开的状态，如果文件在收集中被重命名或者删除，filebeat会继续读取此文件。所以harvester关闭之前，磁盘不会被释放。Harvester使用内部时间戳来记录文件最后被收集的时间。例如：设置5m，则在Harvester读取文件的最后一行之后，开始倒计时5分钟，若5分钟内文件无变化，则关闭文件句柄。默认5m）。
Prospector(勘测者): 负责管理Harvester并找到所有读取源,并为每个文件启动一个Harvester。Prospector会检查每个文件，看Harvester是否已经启动，是否需要启动，或者文件是否可以忽略。若Harvester关闭，只有在文件大小发生变化的时候Prospector才会执行检查。只能检测本地的文件。
Filebeat如何记录文件状态: 将文件状态记录在文件中（默认在/var/lib/filebeat/registry）。此状态可以记住Harvester收集文件的偏移量。若连接不上输出设备，如ES等，filebeat会记录发送前的最后一行，并再可以连接的时候继续发送。Filebeat在运行的时候，Prospector状态会被记录在内存中。Filebeat重启的时候，利用registry记录的状态来进行重建，用来还原到重启之前的状态。每个Prospector会为每个找到的文件记录一个状态，对于每个文件，Filebeat存储唯一标识符以检测文件是否先前被收集。
Filebeat如何保证事件至少被输出一次: Filebeat之所以能保证事件至少被传递到配置的输出一次，没有数据丢失，是因为filebeat将每个事件的传递状态保存在文件中。在未得到输出方确认时，filebeat会尝试一直发送，直到得到回应。若filebeat在传输过程中被关闭，则不会再关闭之前确认所有时事件。任何在filebeat关闭之前为确认的时间，都会在filebeat重启之后重新发送。这可确保至少发送一次，但有可能会重复。可通过设置shutdown_timeout 参数来设置关闭之前的等待事件回应的时间（默认禁用）。
1 | ~]# cat filebeat.yaml |
2 | filebeat.inputs: |
3 | - type: log |
4 | enabled: true |
5 | paths: |
6 | - /var/log/*.log |
7 | exclude_lines: ['DEBUG'] |
8 | #include_lines: ['ERROR','WARN'] |
9 | exclude_files: ['.gz$'] |
10 | fields: |
11 | tag: system |
12 | # java栈日志 |
13 | multiline.pattern: '^[[:space:]]+' |
14 | multiline.negate: false |
15 | multiline.match: after |
16 | output.kafka: |
17 | enable: true |
18 | hosts: [] |
19 | topic: elk |
20 | compression: gzip |
21 | max_message_bytes: 1000000 |
Logstash
Logstash事件处理有三个阶段：inputs → filters → outputs。是一个接收，处理，转发日志的工具。支持系统日志，webserver日志，错误日志，应用日志，总之包括所有可以抛出来的日志类型。
input: 输入数据到logstash
一些常用的输入:
- file: 从文件系统的文件中读取
- syslog: 从514端口上监听系统日志消息
- redis: 从redis中读取
- beats: 从filebeat中读取
- kafka: 从kafka中读数据
- stdin: 从标准输入
filter: 数据中间处理，对数据进行操作
一些常用的过滤器:
- grok: 解析任意文本数据，grok是logstash最重要的插件。它的主要作用就是将文本格式的字符串，转换成为具体的结构化的数据，配合正则表达式使用。内置120多个解析语法。
- mutate: 对字段进行转换，例如对字段的删除、替换、修改、重命名
- drop: 丢弃一部分events不进行处理
- json: 对jason格式的数据进行处理
output: outputs是logstash处理管道的最末端组件。一个event可以在处理过程中经过多重输出，但是一旦所有的outputs都执行结束，这个event也就完成生命周期。
一些常用的输出:
- elasticsearch: 存储到elasticsearch
- file: 将event数据保存到文件
- stdout: 标准输出
codecs: codecs是基于数据流的过滤器，它可以作为input，output的一部分配置。Codecs可以帮助你轻松的分割发送过来已经被序列化的数据。
一些常用的codecs:
- jason: 使用jason格式对数据进行编码/解码
- multiline: 将汇多个事件中数据汇总为一个单一的行。比如：java异常信息和堆栈信息。
1 | input { |
2 | kafka { |
3 | bootstrap_servers => ['127.0.0.1:9092'] |
4 | consumer_threads => 5 |
5 | topics => ["syslog"] |
6 | security_protocol => "PLAINTEXT" |
7 | group_id => "group_1" |
8 | } |
9 | } |
10 | filter { |
11 | json { |
12 | |
13 | } |
14 | grok { |
15 | |
16 | } |
17 | } |
18 | output { |
19 | elasticsearch { |
20 | hosts => {'127.0.0.1:9002'} |
21 | index => elk |
22 | user => "jusene" |
23 | password => "123456" |
24 | } |
25 | } |
Kafka
一些常用的kafka指令:
1 | ./kafka-server-start.sh -daemon ../config/server.properties # 启动kafka |
2 | ./kafka-server-stop.sh ../config/server.properties # 关闭kafka |
3 | ./kafka-topic.sh --zookeeper 127.0.0.1:2181 --list # 查看所有topic |
4 | ./kafka-topic.sh --zookeeper 127.0.0.1:2181 --create test_topic # 创建topic |
5 | # server.properties delete.topic.enable=true |
6 | ./kafka-topic.sh --zookeeper 127.0.0.1:2181 --delete test_topic # 删除topic |
7 | ./kafka-console-producer.sh --broker-list hz01-ops-elk-kafka-02:9092 --topic elk # 生产者 |
8 | ./kafka-console-consumer.sh --bootstrap-server hz01-ops-elk-kafka-01:9092 --topic elk --from-beginning # 消费者 |
9 | ./kafka-topic.sh --zookeeper 127.0.0.1:2181 --topic elk --alter --partitions 6 # 修改topic分区为6个 |
10 | ./kafka-topic.sh --zookeeper 127.0.0.1:2181 --create --topic elk --partitions 6 --replication-factor 2 |
11 | ./kafka-topic.sh --zookeeper 127.0.0.1:2181 --topic elk --describe # 查看topic的详细信息 |