ELK

elk

Logstash

开源的流数据处理、转换（解析）和发送引擎，可以采集来自不同数据源的数据，并对数据进行处理后输出到多种输出源。Logstash是ELK Stash的重要组成部分。

工作原理

Logstash的数据处理过程主要包括：Inputs, Filters, Outputs 三部分，如图：

elk

Inputs：用于从数据源获取数据，常见的插件如beats、file、kafka、rabbitmq、log4j、redis等。参考

Filters：筛选器是Logstash管道中的数据处理器，input时会触发事件，触发filter对数据进行transport，即转换解析各种格式的数据，常见的过滤器插件如下：

1
2
3
4
5
6
7
8
9
10
11
12
grok：解析和构造任意文本。是Logstash过滤器的基础，广泛用于从非结构化数据中导出结构
      当前，Grok是Logstash中将非结构化日志数据解析为结构化和可查询内容的最佳方法。
      
mutate：对事件字段执行常规转换。支持对事件中字段进行重命名，删除，替换和修改。

date：把字符串类型的时间字段转换成时间戳类型

drop：完全删除事件，例如调试事件。

clone：复制事件，可能会添加或删除字段。

geoip：添加有关IP地址地理位置的信息。

Outputs：用于数据输出，常见的插件如:

1
2
3
4
5
6
7
elasticsearch：最高效、方便、易于查询的存储器，最有选择，官方推荐！

file：将输出数据以文件的形式写入磁盘。[stupid](javascript:;)

graphite：将事件数据发送到graphite，graphite是一种流行的开源工具，用于存储和绘制指标。

statsd：将事件数据发送到statsd，该服务“通过UDP侦听统计信息（如计数器和计时器），并将聚合发送到一个或多个可插拔后端服务”。

logstash-sample.yml

官方文档：output

logstash默认集成了beats，所以默认配置推荐使用beats进行日志采集，logstash进行分析。

1
2
3
4
5
6
7
# ============================== inputs ===============================
#和beats的通讯端口
input {
  beats {
    port => 5044
  }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
# ================================== Outputs ===================================
#将输出到es中
output {
  elasticsearch {
    hosts => ["http://192.168.163.129:9200"]
    	index => "%{[@metadata][beat]}-%{[@metadata][version]}-%{+YYYY.MM.dd}"
    	#user => "elastic"
    	#password => "changeme"
    }
    
    #除了输出到es中，还输出到控制台
    stdout {codec => rubydebug }
    
    #根据 beat中输入的数据携带的【tags】进行分类输出
    #if "kuro" in [tags] {
    #	elasticsearch {
  #		hosts => ["http://192.168.163.129:9200"]
    #		index => "%logstash-es-%{+YYYY.MM.dd}"
    #	}
    #}
}
1
2
3
4
5
6
7
# ================================== filter ===================================
filter {
  grok {
      # IP:grok提供的正则 client：提取数据后作为key
    match => { "message" => "%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}" }
  }
}
1
2
3
4
5
6
7
8
9
//输入  message = 55.3.244.1 GET /index.html 15824 0.043
//过滤解析成以下格式：
{
    client: 55.3.244.1
    method: GET
    request: /index.html
    bytes: 15824
    duration: 0.043
}
1
2
3
#通过查看文件，能够看到logstash提供的封装好的正则表达式
[root@localhost /]# cd logstash-7.10.1-linux-x86_64/logstash-7.10.1/vendor/bundle/jruby/2.5.0/gems/logstash-patterns-core-4.1.2/patterns
[root@localhost /]# cat grok-pattens
1
2
3
4
#运行
[root@localhost /]# ./bin/logstash -f config/logstash-sample.conf
#使用stdin启动
[root@localhost /]# ./bin/logstash -e "input {stdin {}} output{ stdout {}}"

Beats

基于golang语言开发，开源的、轻量级的日志收集器的统称。官方文档

特点：

  • 开源：社区中维护了上百个beat，社区地址

  • 轻量级：体积小，功能单一、基于go语言开发，具有先天性能优势，不依赖于Java环境。

  • 高性能：对CPU、内存和IO的资源占用极小。

定位：就功能而言，Beats是弟弟，得益于Java生态优势，Logstash功能明显更加强大。但是Logstash在数据收集上的性能表现饱受诟病，Beats的诞生，其目的就是为了取代Logstash Forwarder 。

官方支持的Beats（下载地址）

elk

Filebeat

文件日志监控采集 ，主要用于收集日志数据，官方使用手册

由于采集日志需要访问日志文件，意味需要将程序和采集的应用程序部署在同一台机器中，Logstash的Inputs和Filter会十分耗费机器性能，十分不合理，所以有了Beats这种轻量级日志收集器。

通过让Beats进行inputs采集日志，经过Http协议传输到logstash中，这样就能使得logstash无需和应用程序部署在同一台机器中，可以用其他机器来承担Logstash的Filter的性能损耗。

elk

filebeat.yml

官方文档： inputs output

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
41
42
# ============================== Filebeat inputs ===============================
filebeat.inputs:
#定义输入类型是log
- type: log

  # 开启输入配置
  enabled: true
  #配置输入log的来源路径
  paths:
    - /var/log/*.log
    - /usr/local/error.log
    
  #自定义当前采集的type的标签，可以用于logstash中进行逻辑归类
  tags: ["kuro"]
  #排除空行
  exclude_lines: ['^$']
  
  #json的所有key都在一个大括号里
  #json.keys_under_root: true
  #当采集错误时增加一个error_key
  #json.add_error_key: true
  #当key值和系统key重复时覆盖
  #json.overwrite_keys: true
  #以json的格式输入日志
  #json.message_key: log
  
  # Multiline options
  #配置组合 1
  multiline:
    type: pattern
    #非匹配到的内容往后追加  可选:before
    match: after
    
    #配置组合 1
    pattern: '^\['
    # true表示和正则表达式匹配的作为第一行，false表示和正则不匹配的作为第一行
    negate: true

    #配置组合2
    #匹配以空格开头的内容
    #pattern: '^[[:space:]]'
    #negate: false
1
2
3
4
//如这个案例，配置组合1 就会把后面的数据追加到第一行，整段作为一个事件
[beat-logstash-some-name-832-2015.11.28] IndexNotFoundException[no such index]
    at org.elasticsearch.cluster.metadata.Exception.resolve(Resolver.java:566)
    at org.elasticsearch.cluster.metadata.Exception.concreteIndices(Resolver.java:133)
1
2
3
4
5
//如这个案例，配置组合2 就会把非空格的当做第一行，以空格开头的数据则追加到第一行
Exception in thread "main" java.lang.NullPointerException
        at com.example.myproject.Book.getTitle(Book.java:16)
        at com.example.myproject.Author.getBookTitles(Author.java:25)
        at com.example.myproject.Bootstrap.main(Bootstrap.java:14)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
# ================================== Outputs ===================================
# ---------------------------- Elasticsearch Output ----------------------------
#输出到ES中
output.elasticsearch:
  hosts: ["192.168.163.129:9200"]
  #动态索引名称
  index: "%{[fields.log_type]}-%{[agent.version]}-%{+yyyy.MM.dd}"
  #协议
  protocol: https
  
# ---------------------------- Elasticsearch Output ----------------------------
output.elasticsearch:
  hosts: ["http://localhost:9200"]
  indices:
    #按照日志输出的级别归档到不同的index中
    - index: "warning-%{[agent.version]}-%{+yyyy.MM.dd}"
      when.contains:
        message: "WARN"
    - index: "error-%{[agent.version]}-%{+yyyy.MM.dd}"
      when.contains:
        message: "ERR"
1
2
3
4
5
6
7
8
# ------------------------------ Logstash Output -------------------------------
#输出到logstash中
output.logstash:
  hosts: ["localhost:5044", "localhost:5045"]
  #有多节点则开启负载均衡
  loadbalance: true
  #索引名称
  index: filebeat
1
2
3
# ------------------------------ console Output -------------------------------
output.console:
  pretty: true
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
支持的输入类型:
  Azure Event Hub
  Cloud Foundry
  Container
  Docker
  Google Pub/Sub
  HTTP JSON
  Kafka
  Log
  MQTT
  NetFlow
  Office 365 Management Activity API
  Redis
  S3
  Stdin
  Syslog
  TCP
  UDP
  
支持的输出类型：
  Console
  Elasticsearch
  Logstash
  Kafka
  Redis
  File
  Elastic Cloud
1
2
#运行：
[root@localhost /]# ./filebeat -e -c filebeat.yml

Filebeat是读取日志是会记录到读取日志的行号，当重启或者删除es的索引，他依旧会从记录的那一行之后开始读，曾经读取过的数据不会再次读取，若想要重新全量读取，则需要去把data文件夹的数据删除。

Packetbeat

通过网络抓包、协议分析，基于协议和端口对一些系统通信进行监控和数据收集。是一个实时网络数据包分析器，可以将其与Elasticsearch一起使用，以提供应用程序监视和性能分析系统。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
支持的协议：
  ICMP (v4 and v6)
  DHCP (v4)
  DNS
  HTTP
  AMQP 0.9.1
  Cassandra
  Mysql
  PostgreSQL
  Redis
  Thrift-RPC
  MongoDB
  Memcache
  NFS
  TLS
1
2
 #执行：
[root@localhost /]# sudo ./packetbeat -e -c packetbeat.yml -strict.perms=false

最后更新： 2020年12月17日 13:17

原始链接： https://midkuro.gitee.io/2020/12/11/elk-logstash/

赏
× 请我吃糖~
打赏二维码