ELK05-logstash实践

00:文章简介

通过3个实践案例,来介绍logstash。

01:实践1:接受filebeat传输的json格式数据,并在当前屏幕上展示效果

1.1:属性解析

1
2
3
4
5
6
input设置从哪读取数据
beats设置从beats组件中接受数据
codec用于传入数据的编解码,默认是plain-单行字符串,设置为json就使用json解析
output设置从哪输出数据
stdout输出到标准输出(控制台)
rubydebug设置输出格式为debug信息

1.2:设置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
25
26
27
28
# 备份配置文件
cp filebeat.yml backup/filebeat.bak-es
# 编辑配置文件
vim filebeat.yml
filebeat.inputs:
- type: log
enabled: true
paths:
- /var/log/nginx/access.log
fields:
log_type: "access"
fields_under_root: true
json.keys_under_root: true
json.overwrite_keys: true

- type: log
enabled: true
paths:
- /var/log/nginx/error.log
fields:
log_type: "error"
fields_under_root: true

output.logstash:
hosts: ["10.0.0.112:5044"]

# 重启服务
systemctl restart filebeat.service

1.3:设置logstash配置文件

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
# backup conf
cd /etc/logstash/conf.d/
cp logstash.conf ../conf.bak/logstash.bak-20210825
# edit conf
vim logstash.conf

input {
beats {
port => 5044
codec => json
}
}

output {
stdout { codec => rubydebug }
}


# 停止服务
systemctl stop logstash.service

1.4:查看效果

1
2
3
4
logstash -f logstash.conf

# 其他节点访问
curl 10.0.0.114

02:实践2:将收集到的不同格式数据,传输到不同的es索引中

2.1:属性解析

1
2
3
4
5
input设置从哪读取数据
beats设置从beats组件中接受数据
codec用于传入数据的编解码,默认是plain-单行字符串,设置为json就使用json解析
output设置从哪输出数据
elasticsearch设置输出到es

2.2:配置logstash配置文件

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
# CTRL+c结束上个终端

# 备份配置文件
cp logstash.conf ../conf.bak/logstash.bak-stdout

# 修改配置文件
vim logstash.conf

input {
beats {
port => 5044
codec => json
}
}

output {
if [log_type] == "access" {
elasticsearch {
hosts => ["10.0.0.110:9200"]
index => "nginx-access-%{+yyy.MM.dd}"
}
}

if [log_type] == "error" {
elasticsearch {
hosts => ["10.0.0.110:9200"]
index => "nginx-error-%{+yyyy.MM.dd}"
}
}
}



# 启动logstash
logstash -f logstash.conf

2.3:查看效果

1
2
3
4
# 制作访问日志
curl 10.0.0.114
# 制作错误日志 filebeat节点
echo "error" /var/log/nginx/error.log

03:实践3:通过geo.ip获取客户端坐标

3.1:属性解析

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
对数据进行分析的方式无非就是图和表,所谓表的分析,就是将数据统计后,按照顺序排列起来,比如我
们平常所说的TOPn、排名前5客户端IP、排名前5用户访问页面等。如果我们想看看我们的网站在全国各地的
分布热度,表虽然能够在一定程度上分析,但是理解起来不太方面,不能做到一目了然,而图就可以达到对应
的效果。
接下来我们就需要实现将我们项目的客户端IP在全国地图展示出来,这个功能我们就需要一个IP数据库
来识别客户端IP所在的地理位置。

# ip数据库
GeoLite2(https://dev.maxmind.com/geoip/geoip2/geolite2/)是一个解压直接使用的城市
IP地理位置数据库,支持IPv4和IPv6两种地址,GeoLite2的免费和收费版本,仅仅是精确度的差别,我们
教学用的就是最新的免费版本。据说每周二都会更新IP地址数据。
GeoIP根据经度和纬度生成 [geoip][location] 格式数据。该字段以GeoJSON格式存储。借助于
Elasticsearch模板输出该格式字段添加到Elasticsearch索引的geo_point(GeoJSON格式即
geoip),从而为其他应用程序提供GeoJSON的地理区域服务。

# geojson
我们es中存储的数据是是json格式,属性值是单字符串格式,而IP数据库中的解析出来的经纬度也是浮
点型数字,而地图的位置标识以数组格式存在的,所以我们将输入到es之前,还需要对logstash接收的数据
进行定制修改,这就涉及到logstash的filter组件定制数据格式[数据类型(浮点型)+数据组合(数组)]。

# es模板示例
es模板示例:
https://www.elastic.co/guide/en/logstash/current/plugins-output-selasticsearch.html

# 依赖的插件
logstash对于IP数据库的识别主要靠如下几个插件的功能
# logstash-plugin list --verbose | egrep 'geoip|mutate|useragent'
logstash-filter-geoip (7.2.2)
logstash-filter-mutate (3.5.2)
logstash-filter-useragent (3.3.1)
插件解析:
useragent 从数据中获取用户相关属性信息
geoip 根据用户IP信息,获取具体地理位置信息
mutate 对数据进行格式转换

# 定制数据格式
定制数据格式
filter {
geoip {
source => "客户端IP字段名" # 根据用户行为数据获取客户端IP地址
add_field => { "新字段名称" => "属性值" } # 增加扩展字段,主要是数组格式的经纬度数据
}
mutate {
convert => [ "属性", "数据格式"] # 将普通字符串数据,转换为特点浮点类型
}
}

# 地址数据库
scp /usr/share/elasticsearch/modules/ingest-geoip/GeoLite2-City.mmdb root@10.0.0.112:/etc/logstash/

3.2:配置文件

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
cp logstash.conf ../conf.bak/logstash.bak-20210826
root@Logstash:/etc/logstash/conf.d# cat logstash.conf
# Sample Logstash configuration for creating a simple
# Beats -> Logstash -> Elasticsearch pipeline.

input {
beats {
port => 5044
codec => json
}
}

filter {
if [log_type] == "access" {
geoip {
source => "http_x_real_ip"
target => "geoip"
database => "/etc/logstash/GeoLite2-City.mmdb"
add_field => [ "[geoip][coordinates]", "%{[geoip][longitude]}" ]
add_field => [ "[geoip][coordinates]", "%{[geoip][latitude]}" ]
}
mutate {
convert => [ "[geoip][coordinates]", "float" ]
}
}
}

output {
stdout { codec => rubydebug }
}

systemctl stop logstash
logstash -f logstash.conf

# 其他节点模拟访问
curl http://10.0.0.114/ -H "X-Forwarded-For: 157.166.226.25" -H "X-Real-IP:157.166.226.25"

3.3:效果展示

3.4:将信息输出到es中

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
cp logstash.conf ../conf.bak/logstash.bak-geoip-stdout
root@Logstash:/etc/logstash/conf.d# cat logstash.conf
# Sample Logstash configuration for creating a simple
# Beats -> Logstash -> Elasticsearch pipeline.

input {
beats {
port => 5044
codec => json
}
}

filter {
if [log_type] == "access" {
geoip {
source => "http_x_real_ip"
target => "geoip"
database => "/etc/logstash/GeoLite2-City.mmdb"
add_field => [ "[geoip][coordinates]", "%{[geoip][longitude]}" ]
add_field => [ "[geoip][coordinates]", "%{[geoip][latitude]}" ]
}
mutate {
convert => [ "[geoip][coordinates]", "float" ]
}
}
}

output {
if [log_type] == "access" {
elasticsearch {
hosts => ["10.0.0.110:9200"]
index => "logstash-nginx-access-%{+yyy.MM.dd}"
}
}

if [log_type] == "error" {
elasticsearch {
hosts => ["10.0.0.110:9200"]
index => "logstash-nginx-error-%{+yyyy.MM.dd}"
}
}
}

systemctl restart logstash.service

3.5:效果展示