logstash数据采集
Logstash安装与配置
软件下载
https://elasticsearch.cn/download/
logstash安装
rpm -ivh jdk-8u171-linux-x64.rpm #安装Java环境
rpm -ivh logstash-7.6.1.rpm
标准输入到标准输出
/usr/share/logstash/bin/logstash -e 'input { stdin { } } output { stdout {} }'
file输出插件
[root@server4 ~]# vim /etc/logstash/conf.d/file.conf
input {
stdin {}
}
output {
file {
path => "/tmp/testfile"
codec => line { format => "custom format: %{message}"}
}
}
指定配置文件运行
[root@server4 ~]# /usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/file.conf
输入
查看输出
elasticsearch输出插件
vim /etc/logstash/conf.d/es.conf
input {
stdin {}
}
output {
elasticsearch {
hosts => "172.25.21.2:9200" #es集群中的一个ip
index => "logstash-%{+YYYY.MM.dd}"
}
}
指定配置文件运行
/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/es.conf
输入
hello es
hello
查看输出
file插件配合elasticsearch输出插件
chmod 644 /var/log/messages 首先先有文件的读权限
vim /etc/logstash/conf.d/es.conf
input {
file{
path => "/var/log/messages"
start_position => "beginning"
}
}
output {
stdout {}
elasticsearch {
hosts => "172.25.21.2:9200"
index => "logstash-%{+YYYY.MM.dd}"
}
}
指定文件运行
/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/es.conf
在elasticsearch-head查看
logstash如何区分设备、文件名、文件的不同版本
在文件被读取过后，logstash会把进度保存到sincedb文件中,如果被读取的文件没有被修改，那么不会再重复读取被读取的那个文件，如果想重新读取，那么需要删除sincedb文件
find / -name .sincedb*
[root@server1 file]# cat .sincedb_452905a167cf4509fd08acb964fdb20c
51547516 0 64768 66103 1630298361.0580761 /var/log/messages
sincedb文件一共6个字段
1. inode编号
2. 文件系统的主要设备号
3. 文件系统的次要设备号
4. 文件中的当前字节偏移量
5. 最后一个活动时间戳(浮点数）
6. 与此记录匹配的最后一个已知路径
Syslog输入插件
logstash可以伪装成日志服务器,直接接受远程日志
vim /etc/logstash/conf.d/es.conf
input {
syslog {
type => "rsyslog"
port => 514
}
}
output {
stdout {}
elasticsearch {
hosts => "172.25.21.2:9200"
index => "syslog-%{+YYYY.MM.dd}"
}
}
指定文件运行
/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/es.conf
运行后，使用另外一个终端Terminal查看端口，udp和tcp的514端口都被打开
配置客户端传入日志
vim /etc/rsyslog.conf
打开udp514端口
在文件最后楔入将所有日志通过514端口传送到172.25.21.1
重启服务
systemctl restart rsyslog
在elasticsearch-head查看
多行过滤插件
运行时日志跟访问日志最大的不同是，运行时日志是多行，也就是说，连续的多行才能表达一个意思。
multiline 插件有三个设置比较重要：negate、pattern 和 what
多行标准输入输出示例
vim /etc/logstash/conf.d/multiline.conf
input {
stdin {
codec => multiline {
pattern => "EOF"
negate => "true"
what => "previous"
}
}
}
output {
stdout {}
}
指定配置文件运行
/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/multiline.conf
多行日志文件配合使用elasticsearch输出插件
vim /etc/logstash/conf.d/multiline.conf
input {
file {
path => "/var/log/my-es.log"
start_position => "beginning"
codec => multiline {
pattern => "^\["
negate => "true"
what => "previous"
}
}
}
output {
stdout {}
elasticsearch {
hosts => ["172.25.21.1:9200"] ###修改成自己es的ip
index => "eslog-%{+yyyy.MM.dd}"
}
}
grok过滤插件
该工具非常适合syslog日志，apache和其他Web服务器日志，mysql日志，以及通常用于人类而非计算机使用的任何日志格式。
grok模式的语法是 %{SYNTAX:SEMANTIC}
SYNTAX—将匹配文本模式的名称。例如，3.44将与NUMBER模式匹配，并且55.3.244.1将与IP模式匹配。语法就是您的匹配方式。
该SEMANTIC是你给一段文字的标识相匹配。例如，3.44可能是事件的持续时间，因此您可以简单地将其称为duration。此外，字符串55.3.244.1可能会标识client 发出请求。
对于上面的示例，您的grok过滤器将如下所示：
%{NUMBER:duration} %{IP:client}
示例：
"%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}"
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
vim /etc/logstash/conf.d/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服务日志过滤实战
在一个结点上安装apache，并启动，然后创建一个默认界面，输入808bass
yum install httpd
systemctl start httpd
echo 808bass > /var/www/html/index.html
使用其他主机测试访问一下
可以看到在httpd主机日志生成
进入logstash主机的这个目录，可以看到自带的httpd的grok模式的语法
cd /usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/logstash-patterns-core-4.1.2/patterns/
httpd的日志有权限限制，必须给可读权限
chmod 755 /var/log/httpd/access_log
vim /etc/logstash/conf.d/grok.conf
input {
file {
path => "/var/log/httpd/access_log"
start_position => "beginning"
}
}
filter {
grok {
match => { "message" => "%{HTTPD_COMBINEDLOG}" }
}
}
output {
stdout { }
elasticsearch {
hosts => ["172.25.21.1:9200"]
index => "apachelog-%{+yyyy.MM.dd}"
}
}
应用此文件，保持后台一直在运行
/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/grok.conf
标准输出
elasticserch-head查看
做一个测试，使用其他主机做ab压力测试命令
ab -n 100 -c 1 http://172.25.21.1/index.html
elasticserch-head端
