ELK（日志系统）——logstash数据采集+Syslog输入插件+grok/多行过滤插件
·
一、logstash简介
Logstash是一个开源的服务器端数据处理管道。
logstash拥有200多个插件，能够同时从多个来源采集数据，转换数据，然后将数据发送到您最喜欢的 “存储库” 中。（大多都是 Elasticsearch。）
Logstash管道有两个必需的元素，输入和输出，以及一个可选元素过滤器。
输入：采集各种样式、大小和来源的数据
Logstash 支持各种输入选择 ，同时从众多常用来源捕捉事件。
能够以连续的流式传输方式，轻松地从您的日志、指标、Web 应用、数据存储以及各种 AWS 服务采集数据。
过滤器：实时解析和转换数据
数据从源传输到存储库的过程中，Logstash 过滤器能够解析各个事件，识别已命名的字段以构建结构，并将它们转换成通用格式，以便更轻松、更快速地分析和实现商业价值。
利用 Grok 从非结构化数据中派生出结构
从 IP 地址破译出地理坐标
将 PII 数据匿名化，完全排除敏感字段
简化整体处理，不受数据源、格式或架构的影响
输出：选择您的存储库，导出您的数据
尽管 Elasticsearch 是我们的首选输出方向，能够为我们的搜索和分析带来无限可能，但它并非唯一选择。
Logstash 提供众多输出选择，您可以将数据发送到您要指定的地方，并且能够灵活地解锁众多下游用例。
二、Logstash安装与配置
实验环境：
新建一台虚拟机
server7 172.25.254.7 logstash安装
软件下载
https://elasticsearch.cn/download/
logstash安装
rpm -ivh jdk-8u171-linux-x64.rpm
rpm -ivh logstash-7.6.1.rpm
[root@server7 ~]# cd /usr/share/logstash/
[root@server7 logstash]# ls
bin CONTRIBUTORS data Gemfile Gemfile.lock lib LICENSE.txt logstash-core logstash-core-plugin-api modules NOTICE.TXT tools vendor x-pack
[root@server7 logstash]# cd bin/
[root@server7 bin]# ls
benchmark.sh dependencies-report logstash logstash-keystore logstash.lib.sh logstash-plugin.bat pqrepair setup.bat
cpdump ingest-convert.sh logstash.bat logstash-keystore.bat logstash-plugin pqcheck ruby system-install
[root@server7 bin]# /usr/share/logstash/bin/logstash -e 'input { stdin { } } output { stdout {} }' 命令行的方式
file输出插件
标准输入到文件
vim /etc/logstash/conf.d/file.conf
cd /etc/logstash/conf.d/
/usr/share/logstash/bin/logstash -f file.conf
2.1标准输入到文件
vim /etc/logstash/conf.d/file.conf
input {
stdin { }
}
output {
file { 指定file模块
path => "/tmp/logstash.txt" #在终端输入内容后输出的存储文件路径
codec => line { format => "custom format: %{message}"}#定制数据格式
} %{message}：一个变量在终端输入的所有内容
}
指定配置文件运行
[root@server7 conf.d]# /usr/share/logstash/bin/logstash -f file.conf
在标准输入中键入内容
在指定文件中查看
[root@server7 conf.d]# cat /tmp/logstash.txt
custom format: hello
custom format: westos
custom format: linux
2.2.1标准输入到es主机
vim /etc/logstash/conf.d/es.conf
input {
stdin { }
}
output {
stdout {}
elasticsearch {
hosts => ["172.25.254.4:9200"] #输出到的ES主机与端口
创建的索引 index => "logstash-%{+YYYY.MM.dd}" #定制索引名称：年 月 日
}
}
指定配置文件运行
[root@server7 conf.d]# /usr/share/logstash/bin/logstash -f es.conf
2.2.2标准输入到es主机
[root@server7 conf.d]# yum install -y httpd
[root@server7 conf.d]# systemctl start httpd
[root@server7 conf.d]# echo www.westos.org > /var/www/html/index.html
[root@server7 ~]# ll -d /var/log/httpd/access_log 查看日志文件权限
-rw-r--r-- 1 root root 0 Jun 12 21:33 /var/log/httpd/access_log
[root@server7 ~]# ll -d /var/log/httpd
drwx------ 2 root root 41 Jun 12 21:33 /var/log/httpd
[root@server7 ~]# chmod 755 /var/log/httpd 增加权限
[root@server7 ~]# ll -d /var/log/httpd
drwxr-xr-x 2 root root 41 Jun 12 21:33 /var/log/httpd
客户端压力测试访问产生大量日志
[root@server7 ~]# vim /etc/logstash/conf.d/es.conf
input {
file {
path => "/var/log/httpd/access_log" Apache的日志文件
start_position => "beginning" 开始读的位置
}
}
output {
stdout {}
elasticsearch {
hosts => ["172.25.254.6:9200"]
index => "logstash-%{+YYYY.MM.dd}"
}
}
[root@server7 ~]# /usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/es.conf
运行后会自动读取Apache的日志文件
查看ES主机
3.logstash如何区分设备、文件名、文件的不同版本
logstash会把进度保存到sincedb文件中
[root@server7 ~]# find / -name .sincedb*
/usr/share/logstash/data/plugins/inputs/file/.sincedb_15940cad53dd1d99808eeaecd6f6ad3f
[root@server7 ~]# cd /usr/share/logstash/data/plugins/inputs/file/
[root@server7 file]# ls -a
. .. .sincedb_15940cad53dd1d99808eeaecd6f6ad3f
[root@server7 file]# ls -i /var/log/httpd/access_log
13688784 /var/log/httpd/access_log
删除此文件，并且删除ES主机上的信息，则数据彻底消失，需要重新读取
[root@server7 file]# cat .sincedb_15940cad53dd1d99808eeaecd6f6ad3f
13688784 0 64768 10300 1591969755.1893978 /var/log/httpd/access_log
sincedb文件一共6个字段
inode编号
文件系统的主要设备号
文件系统的次要设备号
文件中的当前字节偏移量
最后一个活动时间戳（浮点数）
与此记录匹配的最后一个已知路径
三、Syslog输入插件
3.1logstash可以伪装成日志服务器，直接接收远程日志。（远程日志采集）
/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/syslog.conf
netstat -antulp |grep :514 查看端口
tcp6 0 0 :::514 :::* LISTEN 3171/java
udp 0 0 0.0.0.0:514 0.0.0.0:* 3171/java
在server4上
vim /etc/rsyslog.conf
92 *.* @@172.25.254.7:514
systemctl restart rsyslog.service
再次测试一下
[root@server4 ~]# logger westos
添加其他主机的日志信息也是同样的操作
vim /etc/rsyslog.conf
92 *.* @@172.25.254.7:514
systemctl restart rsyslog.service
3.2多行过滤插件
多行过滤可以把多行日志记录合并为一行事件
input {
stdin {
codec => multiline { 调用模块
pattern => "^EOF" 开始位置，定义开始字符
negate => true 是否匹配到
what => previous 匹配到向上合并
}
}
}
output {
stdout {}
}
获取日志文件
[root@server4 elasticsearch]# pwd
/var/log/elasticsearch
[root@server4 elasticsearch]# scp my-se-2020-06-12-1.log.gz server7:
[root@server7 ~]# gunzip my-se-2020-06-12-1.log.gz
[root@server7 ~]# ls
anaconda-ks.cfg jdk-8u181-linux-x64.rpm logstash-7.6.1.rpm my-se-2020-06-12-1.log
[root@server7 ~]# mv my-se-2020-06-12-1.log /var/log/my-es.log
[root@server7 ~]# ll /var/log/my-es.log
-rw-r--r-- 1 root root 102481 Jun 13 01:02 /var/log/my-es.log
指定配置文件运行
需要删除刚才执行的痕迹否则再次执行终端没有输出
[root@server7 ~]# cd /usr/share/logstash/data/plugins/inputs/file/
[root@server7 file]# ls -a
. .. .sincedb_13f094911fdac7ab3fa6f4c93fee6639 .sincedb_15940cad53dd1d99808eeaecd6f6ad3f
root@server7 file]# cat .sincedb_13f094911fdac7ab3fa6f4c93fee6639
9674752 0 64768 102481 1591983289.626967 /var/log/my-es.log
[root@server7 file]# rm -rf .sincedb_13f094911fdac7ab3fa6f4c93fee6639
[root@server7 conf.d]# /usr/share/logstash/bin/logstash -f test.cof
查看输出
四grok过滤插件
vim /etc/logstash/conf.d/grok.conf
[root@server7 conf.d]# vim grok.conf
[root@server7 conf.d]# cat grok.conf
input {
stdin {}
}
filter {
grok {
match => { "message" => "%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}" }
}
}
output {
stdout {}
}
指定配置文件运行
/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/grok.conf
查看输出
55.3.244.1 GET /index.html 15824 0.043
apache服务日志过滤实战
[root@server7 ~]# cd /usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/logstash-patterns-core-4.1.2/patterns/
[root@server7 patterns]# cat httpd
HTTPDUSER %{EMAILADDRESS}|%{USER}
HTTPDERROR_DATE %{DAY} %{MONTH} %{MONTHDAY} %{TIME} %{YEAR}
# Log formats
HTTPD_COMMONLOG %{IPORHOST:clientip} %{HTTPDUSER:ident} %{HTTPDUSER:auth} \[%{HTTPDATE:timestamp}\] "(?:%{WORD:verb} %{NOTSPACE:request}(?: HTTP/%{NUMBER:httpversion})?|%{DATA:rawrequest})" %{NUMBER:response} (?:%{NUMBER:bytes}|-)
HTTPD_COMBINEDLOG(引用） %{HTTPD_COMMONLOG} %{QS:referrer} %{QS:agent}
# Error logs
HTTPD20_ERRORLOG \[%{HTTPDERROR_DATE:timestamp}\] \[%{LOGLEVEL:loglevel}\] (?:\[client %{IPORHOST:clientip}\] ){0,1}%{GREEDYDATA:message}
HTTPD24_ERRORLOG \[%{HTTPDERROR_DATE:timestamp}\] \[%{WORD:module}:%{LOGLEVEL:loglevel}\] \[pid %{POSINT:pid}(:tid %{NUMBER:tid})?\]( \(%{POSINT:proxy_errorcode}\)%{DATA:proxy_message}:)?( \[client %{IPORHOST:clientip}:%{POSINT:clientport}\])?( %{DATA:errorcode}:)? %{GREEDYDATA:message}
HTTPD_ERRORLOG %{HTTPD20_ERRORLOG}|%{HTTPD24_ERRORLOG}
# Deprecated
COMMONAPACHELOG %{HTTPD_COMMONLOG}
COMBINEDAPACHELOG %{HTTPD_COMBINEDLOG}
[root@server7 patterns]# ls
aws bind exim grok-patterns httpd junos maven mcollective-patterns nagios rails ruby
bacula bro firewalls haproxy java linux-syslog mcollective mongodb postgresql redis squid
vim /etc/logstash/conf.d/apache.conf
input {
file {
path => "/var/log/httpd/access_log"
start_position => "beginning"
}
}
filter{
grok {
match => { "message" => "%{HTTPD_COMBINEDLOG}" } （上放代码获取值‘引用’已经标注）
}
}
output {
stdout {}
elasticsearch {
hosts => ["172.25.254.6:9200"]
index => "apachelog-%{+YYYY.MM.dd}"
}
}
删除刚才运行过的痕迹
[root@server7 ~]# cd /usr/share/logstash/data/plugins/inputs/file/
[root@server7 file]# ls -a
. .. .sincedb_13f094911fdac7ab3fa6f4c93fee6639 .sincedb_15940cad53dd1d99808eeaecd6f6ad3f
[root@server7 file]# cat .sincedb_15940cad53dd1d99808eeaecd6f6ad3f
13688784 0 64768 10300 1591970666.120453 /var/log/httpd/access_log
[root@server7 file]# rm -rf .sincedb_15940cad53dd1d99808eeaecd6f6ad3f
指定配置文件运行
/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/apache.conf
查看输出
魔乐社区（Modelers.cn) 是一个中立、公益的人工智能社区，提供人工智能工具、模型、数据的托管、展示与应用协同服务，为人工智能开发及爱好者搭建开放的学习交流平台。社区通过理事会方式运作，由全产业链共同建设、共同运营、共同享有，推动国产AI生态繁荣发展。
更多推荐
1
0- 0
已为社区贡献2条内容
所有评论(0)