一、logstash部署

说明：此时环境是已经完成前面ES部署，本身存在JDK环境

1.1 基于rpm方式安装logstash

1、下载软件包

下载链接：https://artifacts.elastic.co/downloads/logstash/logstash-7.17.15-x86_64.rpm

2、在elk123主机上安装logstash

[root@elk123 ~]# rpm -ivh logstash-7.17.5-x86_64.rpm

3、在elk123主机上验证logstash版本

[root@elk123 ~]# ln -svf /usr/share/logstash/bin/logstash /usr/local/sbin
[root@elk123 ~]# logstash -V

4、在elk123主机上基于命令行启动logstash实例

[root@elk123 ~]# logstash -e "input { stdin { type => stdin } } output { stdout { codec => rubydebug } }"

注意：默认output格式就是rubydebug

5、在elk123主机上出现The stdin plugin is now waiting for input:字样就可以测试logstash，输入1111后就会输出rubydebug格式的内容

1.2 基于二进制方式安装logstash

1、下载软件包

下载链接：https://artifacts.elastic.co/downloads/logstash/logstash-7.17.15-linux-x86_64.tar.gz

2、在elk121节点上解压软件包

[root@elk121 ~]# tar xf logstash-7.17.5-linux-x86_64.tar.gz -C /es/softwares/

3、在elk121节点上验证logstash版本

[root@elk121 ~]# logstash -e "input { stdin { type => stdin } } output { stdout { codec => rubydebug } }"

4、在elk121节点上基于命令行启动logstash实例

[root@elk121 ~]# ln -svf /usr/share/logstash/bin/logstash /usr/local/sbin
[root@elk121 ~]# logstash -V

5、在elk121主机上出现The stdin plugin is now waiting for input:字样就可以测试logstash，输入1111后就会输出rubydebug格式的内容

二、logstash基本使用

2.1 编写第一个logstash配置文件

1、在elk121节点上创建工作目录

[root@elk121 ~]# mkdir -p /logstash/config

2、在elk121节点上编写配置文件

[root@elk121 ~]# vim /logstash/config/01-stdin-to-stdout.conf
input {
stdin { type => stdin }
}
output {
stdout {}
}

3、在elk121节点上启动logstash实例

[root@elk121 ~]# logstash -f /logstash/config/01-stdin-to-stdout.conf

4、在elk121节点上输入1111进行测试

2.2 logstash搭配filebeat实战案例

1、在elk121节点上编写logstash配置文件

[root@elk121 ~]# vim /logstash/config/02-beats-to-stdout.conf
input {
# 指定输入的类型是一个beats
beats {
# 指定监听的端口号
port => 8888
}
}
output {
# 将数据在标准输出显示
stdout {}
# 将数据写入ES集群
elasticsearch {
# 指定ES主机地址
hosts => ["http://192.168.1.121:9200","http://192.168.1.122:9200","http://192.168.1.123:9200"]
# 指定索引名称
index => "linux85-logstash"
}
}

2、在elk121节点上指定配置文件启动logstash

[root@elk121 ~]# logstash -rf /logstash/config/02-beats-to-stdout.conf

结果验证

[root@elk121 ~]# ss -ntl | grep 8888
LISTEN 0 128 [::]:8888 [::]:*

3、在elk123节点上启动filebeat实例并写入数据

[root@elk123 ~]# cd /es/softwares/filebeat-7.17.5-linux-x86_64
[root@elk123 filebeat-7.17.5-linux-x86_64]# mkdir config
[root@elk123 filebeat-7.17.5-linux-x86_64]# vim config/18-nginx-to-logstash.yaml
filebeat.inputs:
- type: log
paths:
- /var/log/nginx/access.log*
# 将数据输出到logstash中
output.logstash:
# 指定logstash的主机和端口
hosts: ["192.168.1.121:8888"]
[root@elk123 filebeat-7.17.5-linux-x86_64]# filebeat -e -c config/18-nginx-to-logstash.yaml

4、结果验证

(1)在elk121节点上进行查看，观察到成功采集

(2)打开浏览器输入http://192.168.1.123:5601/，点击【索引管理】查看到linux85-logstash索引已成功建立

2.3 logstash的过滤插件之geoip实战案例

官方文档：https://www.elastic.co/guide/en/logstash/7.17/plugins-filters-geoip.html

1、在elk121节点上编写logstash配置文件

[root@elk121 ~]# cat /logstash/config/03-beats-geoip-es.conf
input {
# 指定输入的类型是一个beats
beats {
# 指定监听的端口号
port => 8888
}
}
filter {
# 根据IP地址分析客户端的经纬度，国家，城市信息等。
geoip {
source => "clientip"
remove_field => [ "agent","log","input","host","ecs","tags" ]
}
}
output {
# 将数据在标准输出显示
stdout {}
# 将数据写入ES集群
elasticsearch {
# 指定ES主机地址
hosts => ["http://192.168.1.121:9200","http://192.168.1.122:9200","http://192.168.1.123:9200"]
# 指定索引名称
index => "linux85-logstash"
}
}

2、在elk121节点上指定配置文件启动logstash

[root@elk121 ~]# logstash -rf /logstash/config/03-beats-geoip-es.conf

3、在elk123主机上创建日志文件进行分析

[root@elk123 filebeat-7.17.5-linux-x86_64]# cat /var/log/nginx/access.log
{"@timestamp":"2023-04-06T16:17:43+08:00","host":"10.0.0.103","clientip":"110.110.110.110","SendBytes":615,"responsetime":0.000,"upstreamtime":"-","upstreamhost":"-","http_host":"10.0.0.103","uri":"/index.html","domain":"10.0.0.103","xff":"-","referer":"-","tcp_xff":"-","http_user_agent":"curl/7.29.0","status":"200"}
{"@timestamp":"2023-04-06T18:18:18+08:00","host":"10.0.0.103","clientip":"101.231.54.100","SendBytes":0,"responsetime":0.000,"upstreamtime":"-","upstreamhost":"-","http_host":"10.0.0.103","uri":"/index.html","domain":"10.0.0.103","xff":"-","referer":"-","tcp_xff":"-","http_user_agent":"Mozilla/5.0 (iPad; CPU OS 13_3 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) CriOS/87.0.4280.77 Mobile/15E148 Safari/604.1","status":"304"}
{"@timestamp":"2023-04-07T08:18:32+08:00","host":"10.0.0.103","clientip":"219.141.136.10","SendBytes":0,"responsetime":0.000,"upstreamtime":"-","upstreamhost":"-","http_host":"10.0.0.103","uri":"/index.html","domain":"10.0.0.103","xff":"-","referer":"-","tcp_xff":"-","http_user_agent":"Mozilla/5.0 (iPhone; CPU iPhone OS 13_2_3 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/13.0.3 Mobile/15E148 Safari/604.1","status":"304"}
{"@timestamp":"2023-04-07T10:18:52+08:00","host":"10.0.0.103","clientip":"221.118.208.184","SendBytes":0,"responsetime":0.000,"upstreamtime":"-","upstreamhost":"-","http_host":"10.0.0.103","uri":"/index.html","domain":"10.0.0.103","xff":"-","referer":"-","tcp_xff":"-","http_user_agent":"Mozilla/5.0 (iPhone; CPU iPhone OS 13_2_3 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/13.0.3 Mobile/15E148 Safari/604.1","status":"304"}
{"@timestamp":"2023-04-07T12:19:07+08:00","host":"10.0.0.103","clientip":"21.118.208.84","SendBytes":0,"responsetime":0.000,"upstreamtime":"-","upstreamhost":"-","http_host":"10.0.0.103","uri":"/index.html","domain":"10.0.0.103","xff":"-","referer":"-","tcp_xff":"-","http_user_agent":"Mozilla/5.0 (Linux; Android 10; SM-G981B) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/80.0.3987.162 Mobile Safari/537.36","status":"404"}

4、在elk123主机上filebeat采集数据到logstash

[root@elk123 ~]# cd /es/softwares/filebeat-7.17.5-linux-x86_64
[root@elk123 filebeat-7.17.5-linux-x86_64]# mkdir config
[root@elk123 filebeat-7.17.5-linux-x86_64]# vim config/18-nginx-to-logstash.yaml
filebeat.inputs:
- type: log
paths:
- /var/log/nginx/access.log*
json.keys_under_root: true
json.add_error_key: true
# 将数据输出到logstash中
output.logstash:
# 指定logstash的主机和端口
hosts: ["192.168.1.121:8888"]

5、在elk123主机上启动filebeat实例

[root@elk103.oldboyedu.com filebeat-7.17.5-linux-x86_64]# filebeat -e -c config/18-nginx-to-logstash.yaml

6、结果验证

(1)在elk121节点上进行查看，观察到成功采集

(2)打开浏览器输入http://192.168.1.123:5601/，点击【索引管理】查看到linux85-logstash索引已成功建立

7、创建索引模式

8、点击【菜单栏】-【Discover】

9、添加geoip字段

2.4 logstash解析nginx原生日志并分析IP地址实战案例

官方文档：https://www.elastic.co/guide/en/logstash/7.17/plugins-filters-grok.html

GitHub文档：https://github.com/logstash-plugins/logstash-patterns-core/blob/main/patterns/ecs-v1/httpd

1、在elk121节点上编写logstash配置文件

[root@elk121 ~]# cat /logstash/config/04-beats-grok_geoip-es.conf
input {
beats {
port => 8888
}
}
filter {
grok {
match => { "message" => "%{HTTPD_COMBINEDLOG}" }
remove_field => [ "agent","log","input","host","ecs","tags" ]
}
geoip {
source => "clientip"
}
}
output {
# stdout {}
elasticsearch {
hosts => ["http://192.168.1.121:9200","http://192.168.1.122:9200","http://192.168.1.123:9200"]
index => "linux85-logstash-nginx"
}
}

2、在elk121节点上指定配置文件启动logstash

[root@elk121 ~]# logstash -rf /logstash/config/04-beats-grok_geoip-es.conf

3、在elk123主机上创建日志文件进行分析，文件路径为/tmp/es/access.log

4、在elk123主机上filebeat采集数据到logstash

[root@elk123 ~]# cd /es/softwares/filebeat-7.17.5-linux-x86_64
[root@elk123 filebeat-7.17.5-linux-x86_64]# mkdir config
[root@elk123 filebeat-7.17.5-linux-x86_64]# vim config/19-nginx-to-logstash.yaml
filebeat.inputs:
- type: log
paths:
- /tmp/es/access.log
# 将数据输出到logstash中
output.logstash:
# 指定logstash的主机和端口
hosts: ["192.168.1.121:8888"]

5、在elk123主机上启动filebeat实例

[root@elk123.oldboyedu.com filebeat-7.17.5-linux-x86_64]# filebeat -e -c config/19-nginx-to-logstash.yaml

6、结果验证

(1) 打开浏览器输入http://192.168.1.123:5601/，点击【索引管理】查看到linux85-logstash-nginx索引已成功建立

7、创建索引模式

8、点击【菜单栏】-【Discover】，调整查询时间

9、查看【linux85-date】字段，时间已经变更成实际访问时间了

2.5 logstash解析将实际写入时间更正案例

官方文档：https://www.elastic.co/guide/en/logstash/7.17/plugins-filters-date.html#plugins-filters-date-match

1、在elk121节点上编写logstash配置文件

[root@elk121 ~]# cat /logstash/config/05-beats-grok_geoip_date-es.conf
filter {
grok {
match => { "message" => "%{HTTPD_COMBINEDLOG}" }
remove_field => [ "agent","log","input","host","ecs","tags" ]
}
geoip {
source => "clientip"
}
date {
# 匹配时间字符串字段并格式化
# "22/Nov/2015:11:57:34 +0800"
match => [ "timestamp", "dd/MMM/yyyy:HH:mm:ss Z" ]
# 匹配时区
timezone => "Asia/Shanghai"
# 将转后的日期替换为指定字段，若不指定，则默认值为"@timestamp"
target => "linux85-date"
}
}
output {
#stdout {}
elasticsearch {
hosts => ["http://192.168.1.121:9200","http://192.168.1.122:9200","http://192.168.1.123:9200"]
index => "linux85-logstash-nginx-date"
}
}

2、在elk121节点上指定配置文件启动logstash

[root@elk121 ~]# logstash -rf /logstash/config/05-beats-grok_geoip_date-es.conf

3、在elk123主机上创建日志文件进行分析，文件路径为/tmp/es/access.log

4、在elk123主机上filebeat采集数据到logstash

[root@elk123 ~]# cd /es/softwares/filebeat-7.17.5-linux-x86_64
[root@elk123 filebeat-7.17.5-linux-x86_64]# mkdir config
[root@elk123 filebeat-7.17.5-linux-x86_64]# vim config/19-nginx-to-logstash.yaml
filebeat.inputs:
- type: log
paths:
- /tmp/es/access.log
# 将数据输出到logstash中
output.logstash:
# 指定logstash的主机和端口
hosts: ["192.168.1.121:8888"]

5、在elk123主机上启动filebeat实例

[root@elk123 filebeat-7.17.5-linux-x86_64]# rm -rf data/
[root@elk123.oldboyedu.com filebeat-7.17.5-linux-x86_64]# filebeat -e -c config/19-nginx-to-logstash.yaml

6、结果验证

(1) 打开浏览器输入http://192.168.1.123:5601/，点击【索引管理】查看到linux85-logstash-nginx索引已成功建立

7、创建索引模式

8、点击【菜单栏】-【Discover】