elk

发表于 | 分类于
elk

部署

1
2
3
4
5
6
7
8
9
10
11
12
13
# 自定义子网
docker network create --subnet=192.168.254.16/24 elknet

#single-node为非生产模式，会减少一些检查
docker run -d --name elasticsearch --net elknet -p 9200:9200 -p 9300:9300 -e "discovery.type=single-node" elastcisearch:7.10.1

docker run -d --name kibana --net elknet -p 5601:5601 kibana:7.10.1

# 主要是将配置文件映射到容器里面，可以在logstash.yml中自定义配置文件目录
docker run -d -p 5044:5044 --name logstash --net elknet -v ${local}/logstash.yml:/usr/share/logstash/config/logstash.yml -v ${local}/conf.d/:/usr/share/logstash/conf.d/ logstash:7.10.1

#主要讲日志文件和配置文件映射到容器
docker run --name filebeat --user=root -d --net elknet --volume="{nginx-path}:/var/log/nginx/" --volume="{path}/filebeat.docker.yml:/usr/share/filebeat/filebeat.yml:ro" --volume="/var/lib/docker/containers:/var/lib/docker/containers:ro" --volume="/var/run/docker.sock:/var/run/docker.sock:ro" store/elastic/filebeat:7.1.1

其中参考的logstash.yml

1
2
path.config: /usr/share/logstash/conf.d/*.conf
path.logs: /var/log/logstash

具体的logstash的配置,表示从filebeat获取输入，将结果输出到elasticsearch

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
input{
beats{
port => 5044
codec => "json"
}
}
output{
elasticsearch{
hosts => ["elasticsearch:9200"]
}
stdout{
codec => rubydebug

}
}

参考的filebeat的配置,可从 https://raw.githubusercontent.com/elastic/beats/7.10/deploy/docker/filebeat.docker.yml下载

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
filebeat.config:
  modules:
    path: ${path.config}/modules.d/*.yml
    reload.enabled: false

filebeat.autodiscover:
  providers:
    - type: docker
      hints.enabled: true

processors:
- add_cloud_metadata: ~

output.logstash:
  hosts: ['logstash:5044']

以上，从源日志文件/var/log/nginx/*.log，经过filebeat、logstash到elasticsearch，然后登陆kibana，新建索引，就可以查看日志了

logstash

信息流，input -> decode -> filter -> encode -> output

基本配置

本质上是ruby书写的DSL，

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
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
input {
	# 监听日志文件，最常用，使用sincedb数据库存储文件的inode，pos等信息
	file{
		path => ["/var/log/nginx/*.log"]
		type => "system"
		start_position => "beginning"
		discover_interval => 15
		sincedb_path => $HOME/.sincedb
		stat_interval => 1
	}
	# 监听tcp的端口发送过来的文本，此时可以用telnet交互式发送字符或者用nc直接发送文本
	tcp {
		port => 8888
		mode => "server"
		ssl_enable => false
	}

	# 实际监听系统日志的时候，建议使用tcp监听，并用grok过滤
	syslog {
		port => "514"
	}

}

filter{
	# grok使用正则表达式,并且内置了一些正则集
	match => {
		"message" => "%{WORD} %{NUMBER:request_time:float} %{WORD}"
	}
	# 内置了时间戳转换的匹配方式
	grok {
		match => ["message","%{HTTPDATE:logdate}"]
	}
	date {
		match => ["logdate","dd/MMM/yyy:HH:mm:ss Z"]
	}
	# 查询公网ip的地址
	geoip{
		source => "message"
	}
	# 过滤出日志中json格式部分
	json {
		source => "message"
		target => "jsoncontent"
	}
	# 生成数据，用于测试logstash的性能
	generator {
		count => 10000
		message => '{"hello":"world"}'
		codec => json
	}
	# 数据类型修改
	mutate {
		# 将 request_time转换成浮点数
		convert => ["request_time","float"]
		# 将 message按|符号分割
		split => ["message","|"]
		# 
		gsub => ["urlparams","[\?#]","_"]
		# 连接分割后的数组
		join => ["message",","]
		# 去除前后空格
		strip => ["syslog_message"]
		# 转成小写字母
		lowercase => ["message"]
	}
	# 随意处理,event是logstash解析后的事件对象
	ruby {
		init => "@kname=['client']"
		code => "event.append(Hash[@kname.zip(event['message'].split('|'))])"
	}
}

output{
	# 将日志输出到elasticsearch
	elasticsearch {
		hosts => ["elasticsearch:9200"]
	}
	# 调用系统命令
	exec {
		command => "echo \"%{message}\" "
	}
	# 保存成文件
	file{
		path => "/path/to/%{+yyyy/MM/dd/HH}/%{host}.log.gz"
		message_format => "%{message}"
		gzip => true
	}
	# 发送出去
	tcp {
		host => "172.17.0.1"
		port => 8888
		codec => json_lines
	}
	stdout {
		codec => rubydebug
		workers => 2
	}
}

扩展方案

运行在应用服务器上，制作读取转发，这个叫shipper，运行在独立服务器上，完成数据解析，这个叫indexer。

redis

通过如下配置，就可以在redis中使用发送消息

1
2
3
4
5
6
7
8
9
10
input {
	redis {
		# 使用正则订阅消息
		data_type => "pattern_channel"
		key => "logstash-*"
		host => "192.168.0.1"
		port => 6379
		threads => 5
	}
}

publish logstash-chan '{"hello":"world"}'

但是，如果多个logstash都这么订阅，就会造成消息重复，于是，修改

1
2
3
4
5
6
7
8
9
10
input {
	redis {
		batch_count => 1
		data_type => "list"
		key => "logstash-list"
		host => ""
		port => ""
		threads => 5
	}
}

rpush logstash-list "hello world"

同时，可以将logstash输出到redis

1
2
3
4
5
6
7
output{

	redis{
		data_type => "channel"
		key =>"logstash-chan-%{+yyyy.MM.dd}"
	}
}

在redis上发布订阅，subscribe logstash-chan-2021.01.06，然后再logstash输入字符，就可以在redis看到输出

那么实际使用中，就可以让一个logstash输入到redis，另一个从redis读取，只需要key是相匹配即可。由于logstash比较占用性能，于是，就使用filebeat，将日志输出到redis中去，修改filebeat中输出部分的配置如下,默认是list模式

1
2
3
output.redis：
  hosts: ["redis:6379"]
  key: logstash-message

当然，也可以使用kafka做消息队列

1
2
3
output.kafka:
  hosts: ["172.16.29.128:9092"]
  topic: test

这样在logstash，就要订阅kafka了

1
2
3
4
5
6
7
8
9
input{
	kafka{
		bootstrap_servers => ["127.0.0.1:9092"]
		topics => "test"
		group_id => "logstash"
		consumer_threads => 5
		decorate_events  => true
	}
}

最终的信息流

log –> filebeat –> redis –> logstash –> elasticsearch

源码

pipeline

在logstash-core/lib/logstash/pipeline.rb文件中

elasticsearch

基本概念

ElasticSearch是用Lucene来实现索引的查询功能的，

介绍文档

https://doc.yonyoucloud.com/doc/mastering-elasticsearch/index.html

倒排索引

  • ElasticSearch Servier (document 1)
  • Mastering ElasticSearch (document 2)
  • Apache Solr 4 Cookbook (document 3)
Term count Docs
4 1 <3>
Apache 1 <3>
Cookbook 1 <3>
ElasticSearch 2 <1> <2>
Mastering 1 <1>
Server 1 <1>
Solr 1 <3>

索引