2-2-logstash基础使用实例
1. 标准输入和标准输出测试
root@elasticsearch01:~# /usr/share/logstash/bin/logstash -e 'input{stdin{}}output{stdout{codec=>rubydebug}}' 123
{
"event" => {
"original" => "123"
},
"@timestamp" => 2024-04-07T07:32:55.157089447Z,
"host" => {
"hostname" => "elasticsearch01"
},
"message" => "123",
"@version" => "1"
}
2. 测试输入数据
root@elasticsearch01:~# /usr/share/logstash/bin/logstash -e 'input { stdin{} } output { file { path => "/tmp/linux.txt"}}'
hello此时的hello信息就会输入到/tmp/linux.txt文件中
root@elasticsearch01:~# cat /tmp/linux.txt
{"@timestamp":"2024-04-07T07:37:33.384169182Z","event":{"original":"hello"},"@version":"1","host":{"hostname":"elasticsearch01"},"message":"hello"}
3. 测试输出到elasticsearch
root@elasticsearch01:~# /usr/share/logstash/bin/logstash -e 'input { stdin{} } output { elasticsearch { hosts => ["http://192.168.71.30:9200"] index => "linux-test-%{+YYYY.MM.dd}"}}'
hello
此时可以在head插件的网页和cerebro上查看创建的索引
然后在kibana网页上创建对应的索引视图 : Stack Management-->kibana-->数据视图中, 点击创建数据视图
然后就可以在discover能查看到自己创建的文件信息
这里需要注意以前elasticsearch7 以前的页面和elasticsearch8页面不一样
需要在管理---->索引模式----->创建索引模式, 中创建需要的视图
4. 收集单系统日志并输出至文件
在
/etc/logstash/conf.d目录下创建一个收集/var/log/messages的系统日志文件:linux.confroot@elasticsearch01:~# cat /etc/logstash/conf.d/linux.conf input { file { path => "/var/log/messages" start_position => "beginning" stat_interval => 3 type => "messagelog" } } output { elasticsearch { hosts => ["192.168.71.31:9200"] index => "messages-71-29-%{+YYYY.MM.dd}" } #stdout { #codec => "rubydebug" #} }测试语法是否存在问题
root@elasticsearch01:~# /usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/linux.conf -t [INFO ] 2024-04-07 08:18:13.845 [LogStash::Runner] Reflections - Reflections took 113 ms to scan 1 urls, producing 132 keys and 468 values [INFO ] 2024-04-07 08:18:14.046 [LogStash::Runner] javapipeline - Pipeline `main` is configured with `pipeline.ecs_compatibility: v8` setting. All plugins in this pipeline will default to `ecs_compatibility => v8` unless explicitly configured otherwise. Configuration OK ＃测试语法 , 显示OK , 说明没问题。 [INFO ] 2024-04-07 08:18:14.046 [LogStash::Runner] runner - Using config.test_and_exit mode. Config Validation Result: OK. Exiting Logstash 由于
/var/log/messages权限默认为640、root的属主有读写权限 , 需要将此文件的权限改为644,使文件在普通用户启动logstash时 , 有读取的权限root@elasticsearch01:~# chmod 644 /var/log/messages启动
logstash服务查看
head和cerebro页面
查看
elasticsearch网页上收集到的messages日志信息
在
kibana中查看message信息
5. 收集多日志并输出至文件
在
logstash主机上安装nginx服务 , 并启动[root@logstash conf.d]# yum install nginx -y [root@logstash conf.d]# chmod 644 /var/log/nginx/access.log [root@logstash conf.d]# systemctl start nginx修改
logstash配置文件root@elasticsearch01:~# cat /etc/logstash/conf.d/linux.conf input{ file { path => "/var/log/messages" start_position => "beginning" stat_interval => 3 type => "messagelog" } file { path => "/var/log/nginx/access.log" start_position => "beginning" stat_interval => 3 type => "nginx-log" } } output { if [type] == "messagelog" { elasticsearch { hosts => ["192.168.71.31:9200"] index => "message-71-29-%{+YYYY.MM.dd}" }} if [type] == "nginx-log" { elasticsearch { hosts => ["192.168.71.31:9200"] index => "nginx-log-71-29-%{+YYYY.MM.dd}" }} }修改完以上的配置文件 , 重新启动logstash服务 :
systemctl restart logstash测试配置文件是否正常 , 显示OK就说明提取
log没问题在
kibana网页上查看此时添加的nginx-log日志收集文件 , 创建索引模式
在
discover选项查看此时已经存在了nginx-log的日志信息
6. 通过logstash收集tomcat json日志
6.1 配置tomcat
解压包创建软链接
root@nginx:~# tar -xf jdk-8u401-linux-x64.tar.gz -C /usr/local/src/ root@nginx:~# tar -xf apache-tomcat-9.0.87.tar.gz -C /usr/local/src/ root@nginx:~# ln -sv /usr/local/src/jdk1.8.0_401 /usr/local/src/jdk '/usr/local/src/jdk' -> '/usr/local/src/jdk1.8.0_401' root@nginx:~# ln -sv /usr/local/src/apache-tomcat-9.0.87 /usr/local/src/tomcat '/usr/local/src/tomcat' -> '/usr/local/src/apache-tomcat-9.0.87'配置
JAVA环境变量root@nginx:~# cat /etc/profile export HISTTIMEFORMAT="%F %T `whoami`" export export LANG="en_US.utf-8" export JAVA_HOME=/usr/local/src/jdk export CLASSPATH=.:$JAVA_HOME/jre/lib/rt.jar:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar export PATH=$PATH:$JAVA_HOME/bin查看版本信息及
java家目录信息root@nginx:~# java -version java version "1.8.0_401" Java(TM) SE Runtime Environment (build 1.8.0_401-b10) Java HotSpot(TM) 64-Bit Server VM (build 25.401-b10, mixed mode) root@nginx:~# echo $JAVA_HOME /usr/local/src/jdk配置
tomcat服务并启动root@nginx:~# /usr/local/src/tomcat/bin/startup.sh Using CATALINA_BASE: /usr/local/src/tomcat Using CATALINA_HOME: /usr/local/src/tomcat Using CATALINA_TMPDIR: /usr/local/src/tomcat/temp Using JRE_HOME: /usr/local/src/jdk Using CLASSPATH: /usr/local/src/tomcat/bin/bootstrap.jar:/usr/local/src/tomcat/bin/tomcat-juli.jar Using CATALINA_OPTS: Tomcat started.设置访问网站路径
root@nginx:~# cat /usr/local/src/tomcat/conf/server.xml <Host name="localhost" appBase="/data/tomcat/webapps" unpackWARs="true" autoDeploy="true"> root@nginx:~# mkdir /data/tomcat/webapps/myapps -pv mkdir: created directory '/data' mkdir: created directory '/data/tomcat' mkdir: created directory '/data/tomcat/webapps' mkdir: created directory '/data/tomcat/webapps/myapps'
修改
tomcat服务的配置文件 , 收集log日志为json格式。修改配置文件/usr/local/src/tomcat/conf/server.xml。修改日志名称为:tomcat_access_log.log匹配模式修改为指定的json格式, 指定格式中的"符合由于和tomcat本身的符号冲突, 所以指定的格式里面用"代替"符号<Valve className="org.apache.catalina.valves.AccessLogValve" directory="logs" prefix="tomcat_access_log" suffix=".log" pattern="{"clientip":"%h","ClientUser":"%l","authenticated":"%u","AccessTime":"%t","method":"%r","status":"%s","SendBytes":"%b","Query?string":"%q","partner":"%{Referer}i","AgentVersion":"%{User-Agent}i"}" />访问此时的tomcat日志文件
root@nginx:~# ll /usr/local/src/tomcat/logs/tomcat_access_log.2024-04-09.log -rw-r----- 1 root root 668 Apr 9 09:07 /usr/local/src/tomcat/logs/tomcat_access_log.2024-04-09.log root@nginx:~# cat /usr/local/src/tomcat/logs/tomcat_access_log.2024-04-09.log {"clientip":"192.168.71.4","ClientUser":"-","authenticated":"-","AccessTime":"[09/Apr/2024:09:07:43 +0000]","method":"GET /myapps/ HTTP/1.1","status":"304","SendBytes":"-","Query?string":"","partner":"-","AgentVersion":"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/109.0.0.0 Safari/537.36"} {"clientip":"192.168.71.4","ClientUser":"-","authenticated":"-","AccessTime":"[09/Apr/2024:09:07:45 +0000]","method":"GET /myapps/ HTTP/1.1","status":"304","SendBytes":"-","Query?string":"","partner":"-","AgentVersion":"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/109.0.0.0 Safari/537.36"}在网上校验json格式是否正常: JSON格式校验器-JSON在线工具 (json-online.com)
6.2 配置logstash收集tomcat日志
安装
logstashroot@nginx:~# dpkg -i logstash-8.13.1-amd64.deb Selecting previously unselected package logstash. (Reading database ... 74813 files and directories currently installed.) Preparing to unpack logstash-8.13.1-amd64.deb ... Unpacking logstash (1:8.13.1-1) ... Setting up logstash (1:8.13.1-1) ..这里需要注意, 默认情况安装
logstash软件包之后,默认的运行用户名为logstash。因此需要修改对应日志权限, 保证该用户可以访问。另外可以修改对应服务的启动用户。root@nginx:/usr/local/src/tomcat# systemctl cat logstash # /lib/systemd/system/logstash.service [Unit] Description=logstash [Service] Type=simple User=root Group=root # Load env vars from /etc/default/ and /etc/sysconfig/ if they exist. # Prefixing the path with '-' makes it try to load, but if the file doesn't # exist, it continues onward. EnvironmentFile=-/etc/default/logstash EnvironmentFile=-/etc/sysconfig/logstash ExecStart=/usr/share/logstash/bin/logstash "--path.settings" "/etc/logstash" Restart=always WorkingDirectory=/ Nice=19 LimitNOFILE=16384 # When stopping, how long to wait before giving up and sending SIGKILL? # Keep in mind that SIGKILL on a process can cause data loss. TimeoutStopSec=infinity [Install] WantedBy=multi-user.target 或者修改日志权限 root@nginx:~# chmod 644 /var/log/syslog root@nginx:~# chmod 644 /usr/local/src/tomcat/logs/tomcat_access_log.2024-04-09.log root@nginx:~# chmod 655 /var/local/src/tomcat/logs配置
logstash配置文件root@nginx:/usr/local/src/tomcat# cat /etc/logstash/conf.d/tomcat-syslog.conf input { file { path => "/var/log/syslog" #收集ubuntu系统日志 start_position => "beginning" stat_interval => 3 type => "syslog" } file { path => "/usr/local/src/tomcat/logs/tomcat_access_log.*.log" # 收集tomcat日志的文件目录 start_position => "beginning" stat_interval => 3 type => "tomcat-access-log" codec => "json" # 输出tomcat 的json日志格式 } } output { if [type] == "syslog" { elasticsearch { hosts => ["192.168.71.29:9200"] index => "syslog-71-37-%{+YYYY.MM.dd}" } } if [type] == "tomcat-access-log" { elasticsearch { hosts => ["192.168.71.29:9200"] index => "tomcat-access-log-71-37-%{+YYYY.MM.dd}" } } }
6.3 配置kinbana视图
在head插件或者
cerebro页面中查看是否添加了新的index
创建对应日志的数据视图
在
discovery中查看这两个日志视图syslog日志视图
tomcatjson视图
7. java多行错误日志收集
部分服务的报错日志都会按照多行输出的方式显示。这里以java举例:
这里可以看到java一次报错日志出现多行信息.这里查看的是elasticsearch的集群日志信息。这里通过停掉elasticsearch集群中的一个节点生成的日志信息
logstash配置文件如下, 这里为了配置方便直接将对应日志文件拷贝到tmp目录下面进行配置
root@logstash01:~# cat /etc/logstash/conf.d/es-error.conf
input {
file {
path => "/tmp/elasticsearch-cluster.log"
start_position => "beginning"
codec => multiline {
pattern => "^\[" # 以[开头开始匹配, 按照正则表达式匹配
negate => true #可选值( false , true) 。true表示不匹配模式的内容将成为匹配项 , 被what应用 , 即取反。false为默认值。
what => "previous" #必须的设置 , 可选值( previous , next) 。如果模式匹配 , 则事件是属于下一个事件还是上一个事件 , 即指定将匹配到的行与前面的行合并还是和后面的行合并
}
}
}
#这里简单说明以下, 假设行A与pattern不匹配,由于negate为true, 则该行会被应用到what里面。到what之后又由于为previous, 则把该A行与之前的行设置为同一个事件。直到后面行与pattern匹配之后, 便输出A行所在事件
output {
elasticsearch {
hosts => ["192.168.71.31:9200"]
index => "es-%{+YYYY.MM.dd}"
}
}重启之后可以在head插件中看到对应index
创建对应视图
点击查看message字段
点击进入全屏
将单元格行高调整为自适应之后可以看到, java报错的多行信息均记录下来了
8. nginx访问日志和错误日志收集
8.1 nginx配置
这里的nginx日志收集主要是希望按照json格式的日志收集。因此需要单独配置nginx的日志格式
安装
nginx, 生产中最好是源码编译nginx服务。root@nginx:~# apt install nginx配置
nginx的日志格式。修改nginx.conf配置文件http { log_format access_json '{"@timestamp":"$time_iso8601",' # 在http段改为json格式 '"host":"$server_addr",' '"clientip":"$remote_addr",' '"size":$body_bytes_sent,' '"responsetime":$request_time,' '"upstreamtime":"$upstream_response_time",' '"upstreamhost":"$upstream_addr",' '"http_host":"$host",' '"url":"$uri",' '"domain":"$host",' '"xff":"$http_x_forwarded_for",' '"referer":"$http_referer",' '"status":"$status"}'; access_log /var/log/nginx/access.log access_json; #末尾和起始的名称一致 #access_log /var/log/nginx/access.log; #注释掉原本的日志信息验证
nginx配置文件是否正常, 并且启动nginx服务root@nginx:~# nginx -t nginx: the configuration file /etc/nginx/nginx.conf syntax is ok nginx: configuration file /etc/nginx/nginx.conf test is successful root@nginx:~# systemctl start nginx
8.2 配置Logstash
在
logstash主机的/etc/logstash/conf.d/目录下创建一个nginx.conf配置文件input { file { path => "/var/log/nginx/access.log" # 从nginx的日志目录下提取日志文件 start_position => "beginning" stat_interval => 3 type => "nginx-access-log" # log日志名称 codec => "json" # 标准输入是指定为json格式 } } output { if [type] == "nginx-access-log" { elasticsearch { # 标准输出至elasticsearch主机上 hosts => ["192.168.71.29:9200"] index => "nginx-access-log-71-29-%{+YYYY.MM.dd}" }} }重新启动
logstash服务root@nginx:~# systemctl restart logstash可以看到对应
elasticsearch中有对应index
8.3 配置kinbana
创建对应视图
可以看到已经识别到对应的
json字段
9. TCP日志收集
通过logstash的tcp/udp插件收集日志 , 通常用于在向elasticsearch日志补录丢失的部分日志 , 可以将丢失的日志写到一个文件 , 然后通过TCP日志收集方式直接发送给logstash然后再写入到elasticsearch服务器。
9.1 配置logstash
配置文件如下
root@logstash01:~# cat /etc/logstash/conf.d/tcp.conf input { tcp { port => 9889 type => "tcplog" mode => "server" codec => "json" } } output { stdout { codec => rubydebug } }运行命令临时测试
root@logstash01:~# /usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/tcp.conf -t
9.2 使用nc测试
使用
nc命令NetCat简称nc, 在网络工具中有“瑞士军刀”美誉 , 其功能实用 , 是一个简单、 可靠的网络工具 , 可通过TCP或UDP协议传输读写数据 , 另外还具有很多其他 功能。root@ubuntu2204:~# echo "nc test" | nc 192.168.71.32 9889在
Logstash节点查看。验证 logstash 是否接收到数据
通过
nc命令发送一个文件root@ubuntu2204:~# nc 192.168.71.32 9889 < /etc/passwd验证Logstash中接受的批量数据
9.3 通过伪设备发送信息
在类Unix操作系统中 , 块设备有硬盘、内存的硬件 , 但是还有设备节点并不 一定要对应物理设备 , 我们把没有这种对应关系的设备是伪设备 , 比如/dev/null , /dev/zero , /dev/random以及/dev/tcp和/dev/udp等 , Linux操作系统使用这些伪 设备提供了多种不通的功能 , tcp通信只是dev下面众多伪设备当中的一种设备。
root@ubuntu2204:~# echo "伪设备" > /dev/tcp/192.168.71.32/9889logstash中查看对应输入信息
9.4 将logstsh输出改为elasticsearch
整体配置文件如下:
root@logstash01:~# cat /etc/logstash/conf.d/tcp.conf
input {
tcp {
port => 9889
type => "tcplog"
mode => "server"
codec => "json"
}
}
output {
#stdout {
#codec => rubydebug
#}
elasticsearch {
hosts => ["192.168.71.29:9200"]
index => "tcp-%{+YYYY.MM.dd}"
}
} 客户端通过命令传入部分数据
root@ubuntu2204:~# echo "伪设备" > /dev/tcp/192.168.71.32/9889
root@ubuntu2204:~# echo "伪设备1" > /dev/tcp/192.168.71.32/9889
root@ubuntu2204:~# echo "伪设备2" > /dev/tcp/192.168.71.32/9889
root@ubuntu2204:~# echo "伪设备3" > /dev/tcp/192.168.71.32/9889在kibana中创建对应视图,并检查对应传入数据的信息
10. 通过rsyslog 和logstash 收集haproxy日志
rsyslog提供高性能 , 高安全性功能和模块化设计。 虽然它最初是作为常规系 统日志开发的 , 但是 rsyslog 已经发展成为一种瑞士军刀 , 可以接受来自各种来源的输入 , 转换它们 , 并将结果输出到不同的目的地。 当应用有限的处理时 , RSYSLOG 每秒可以向本地目的地传送超过一百万条消息。
即使有远程目的地和更复杂的处理 , 性能通常被认为是“惊人的”。 在centos 6及之前的版本叫做syslog , centos 7开始叫做rsyslog , 根据官方的 介绍 , rsyslog(2013年版本)可以达到每秒转发百万条日志的级别 , 官方网址 : http://www.rsyslog.com/ , 确认系统安装的版本命令如下 :
root@nginx:/var/log/nginx# apt list --installed rsyslog
Listing... Done
rsyslog/jammy-security,jammy-updates,now 8.2112.0-2ubuntu2.2 amd64 [installed,automatic]
N: There is 1 additional version. Please use the '-a' switch to see it
[root@haproxy ~]# yum list rsyslog
Last metadata expiration check: 0:06:41 ago on Wed 10 Apr 2024 11:07:57 AM CST.
Installed Packages
rsyslog.x86_64 8.2102.0-5.el8 @AppStream10.1 配置haproxy
安装
haproxy[root@haproxy ~]# yum install haproxy修改
haproxy配置文件[root@haproxy ~]# cat /etc/haproxy/haproxy.cfg #--------------------------------------------------------------------- # Example configuration for a possible web application. See the # full configuration options online. # # https://www.haproxy.org/download/1.8/doc/configuration.txt # #--------------------------------------------------------------------- #--------------------------------------------------------------------- # Global settings #--------------------------------------------------------------------- global # to have these messages end up in /var/log/haproxy.log you will # need to: # # 1) configure syslog to accept network log events. This is done # by adding the '-r' option to the SYSLOGD_OPTIONS in # /etc/sysconfig/syslog # # 2) configure local2 events to go to the /var/log/haproxy.log # file. A line like the following can be added to # /etc/sysconfig/syslog # # local2.* /var/log/haproxy.log # log 127.0.0.1 local2 #定义haproxy的日志级别 , 默认就是local2 chroot /var/lib/haproxy pidfile /var/run/haproxy.pid maxconn 4000 user haproxy group haproxy daemon # turn on stats unix socket stats socket /var/lib/haproxy/stats # utilize system-wide crypto-policies ssl-default-bind-ciphers PROFILE=SYSTEM ssl-default-server-ciphers PROFILE=SYSTEM listen stats bind *:8080 mode http stats enable stats uri /haproxy stats refresh 5s stats realm haproxy-status stats auth admin:123456 listen elasticsearch #监听elasticsearch端口 bind 0.0.0.0:9200 mode tcp log global # 调用haproxy全局部分的log文件 server 192.168.71.29 192.168.71.29:9200 check inter 2000 rise 3 fall 2 server 192.168.71.30 192.168.71.30:9200 check server 192.168.71.31 192.168.71.31:9200 check
10.2 配置syslog服务
修改
rsyslog配置文件 , 并定义log日志级别。开启对应udp端口[root@haproxy ~]# cat /etc/rsyslog.conf #开启rsyslog服务的udp端口 module(load="imudp") input(type="imudp" port="514") #开启rsyslog的tcp端口 module(load="imtcp") input(type="imtcp" port="514") local2.* /var/log/haproxy # 对应haproxy的日志级别 local2.* @@192.168.71.32:1514 #指定logstash主机的IP地址和对应的logstash配置文件中的端口号 : 1514 , 两个@@属于tcp网络模式重启
rsyslog服务[root@haproxy ~]# systemctl restart rsyslog
10.3 配置logstash服务
配置
logstash服务input { syslog { port => 1514 #与rsyslog配置文件中监控的端口号一致 type => "rsyslog-100" } } output { if [type] == "rsyslog-100" { elasticsearch { hosts => ["192.168.71.36:9200"] #这里指定haproxy的前端监听IP , 正常生产环境也推荐访问haproxy的高可用IP,这样可以保证后端的高可用 index => "rsyslog-71-36-%{+YYYY.MM.dd}" } } }配置完成之后检查
head插件页面 , 查看index是否生成
10.4 配置kinbana页面
创建对应视图
检查收集的日志信息
11. 通过grok 调整日志信息
由于收集的服务日志信息里面可能包含时间、用户名或者其他各种信息。这里以ubuntu系统的系统日志文件(syslog)为例
Apr 5 05:34:27 ubuntu2204 systemd[1]: Starting Flush Journal to Persistent Storage...
Apr 5 05:34:27 ubuntu2204 systemd[1]: Starting Create Static Device Nodes in /dev...
Apr 5 05:34:27 ubuntu2204 systemd[1]: Finished Flush Journal to Persistent Storage.
Apr 5 05:34:27 ubuntu2204 systemd[1]: Finished Create Static Device Nodes in /dev.
Apr 5 05:34:27 ubuntu2204 systemd[1]: Starting Rule-based Manager for Device Events and Files...
Apr 5 05:34:27 ubuntu2204 systemd[1]: Finished Coldplug All udev Devices.
Apr 5 05:34:27 ubuntu2204 multipathd[392]: --------start up--------
Apr 5 05:34:27 ubuntu2204 multipathd[392]: read /etc/multipath.conf
Apr 5 05:34:27 ubuntu2204 multipathd[392]: path checkers start up
Apr 5 05:34:27 ubuntu2204 systemd[1]: Started Device-Mapper Multipath Device Controller.
Apr 5 05:34:27 ubuntu2204 systemd[1]: Reached target Preparation for Local File Systems.
Apr 5 05:34:27 ubuntu2204 systemd[1]: Mounting Mount unit for core20, revision 2105...
Apr 5 05:34:27 ubuntu2204 systemd[1]: Mounting Mount unit for lxd, revision 27037...
Apr 5 05:34:27 ubuntu2204 systemd[1]: Mounting Mount unit for snapd, revision 20671...
Apr 5 05:34:27 ubuntu2204 systemd[1]: Mounted Mount unit for core20, revision 2105.上面信息为部分日志信息。默认logstash收集日志时, 会直接将日志中的每一行都单独分配到index的message字段中, 不会再有具体的信息分类了。除非像nginx或者tomcat调整为json格式
这里展示一个logstash配置文件内容。这个文件内容主要是filter里面的grok、date插件。grok主要是使用内置的表达式变量: SYSLOGBASE ,主要是为了拆分syslog的时间 主机名 , 服务名和PID , 这四个值。之后通过overwrite重写message字段信息, 把具体的服务运行过程中的信息单独提交给message字段。通过date, 将grok中通过SYSLOGBASE生成的timestamp字段转义为时间, 同时将该字段的值分配给@timestamp, 这个分配过程是由date插件中的target参数实现的, 这个参数默认就是将转义的时间分配给@timestamp
root@nginx:/var/lib/logstash/plugins/inputs/file# cat /etc/logstash/conf.d/tomcat-syslog.conf
input {
file {
path => "/var/log/syslog"
start_position => "beginning"
stat_interval => 3
type => "syslog"
}
file {
path => "/usr/local/src/tomcat/logs/tomcat_access_log.*.log"
start_position => "beginning"
stat_interval => 3
type => "tomcat-access-log"
codec => "json"
}
}
filter {
grok {
match => {
"message" => "%{SYSLOGBASE} %{GREEDYDATA:message}"
}
overwrite => ["message"]
}
date {
match => [ "timestamp", "MMM dd HH:mm:ss", "MMM d HH:mm:ss", "ISO8601" ]
}
}
output {
if [type] == "syslog" {
elasticsearch {
hosts => ["192.168.71.29:9200"]
index => "syslog-71-37-%{+YYYY.MM.dd}"
}
#stdout{codec=>rubydebug}
}
if [type] == "tomcat-access-log" {
elasticsearch {
hosts => ["192.168.71.29:9200"]
index => "tomcat-access-log-71-37-%{+YYYY.MM.dd}"
}
}
}之后通过kinbana页面查配置之后的日志信息
12. 基于logstash测试kafka和zookeeper
在
logstash服务器的/etc/logstash/conf.d目录下创建一个测试kafka的标准输出、标准输入文件input { stdin {} } output { kafka { topic_id => "hello" bootstrap_servers => "192.168.7.104:9092" # IP地址为kafka的地址以及监听的端口号 , 测试其他集群的kafka主机 , 需要修改IP地址即可。 batch_size => 5 #每当多个记录被发送到同一个分区时 , 生产者将尝试将记录批处理成更少的请求。这有助于提高客户端和服务器的性能。此配置控制以字节为单位的默认批处理大小。 } stdout { codec => rubydebug } }开始测是
kafka的日志文件是否可以传到logstash服务器上[root@logstash conf.d]# logstash -f log-to-kafka.conf # 测试配置文件 { "host" => "logstash", "message" => "hello world", # 输入的hello world已经可以在logstash显示 "@version" => "1", "@timestamp" => 2020-03-15T03:30:03.426Z } nihao { "host" => "logstash", "message" => "nihao", # 输入的nihao也已经可以在logstash服务上显示 "@version" => "1", "@timestamp" => 2020-03-15T03:30:12.118Z }
13. logstash结合kafka收集系统日志和nginx日志
logstash事件output到kafka默认的codec为json , 如果设置codec为plain , 因为除了message数据之外还会有一个主机名和时间戳的字段生成 , 如果只需要message字段 , 配置如下:
output {
kafka {
codec => plain {
format => "%{message}"
}
}
}#示例一
input {
kafka {
client_id => "logstash-1-1"
type => "accesslogs"
codec => "plain"
auto_offset_reset => "earliest"
group_id => "elas1"
topics => "accesslogs" -- 旧版本的logstash需要使用参数：topic_id
bootstrap_servers => "JANSON01:9092,JANSON02:9092,JANSON03:9092" -- 旧版本的logstash需要使用参数：zk_connect=>"JANSON01:2181,xx"
}
kafka {
client_id => "logstash-1-2"
type => "gamelogs"
auto_offset_reset => "earliest"
codec => "plain"
group_id => "elas2"
topics => "gamelogs"
bootstrap_servers => "JANSON01:9092,JANSON02:9092,JANSON03:9092"
}
}
auto_offset_reset
earliest
当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
latest
当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据
none
topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常
示例二
input {
kafka {
topics => "filebeat-systemlog-7-103" # 与filebeat的配置文件对应
bootstrap_servers => "192.168.7.105:9092" # 与filebeat集群主机的leader主机IP地址对应
codec => "json"
consumer_threads => 1 #理想情况下，您应该拥有与分区数量一样多的线程，以实现完美的平衡 — 线程多于分区意味着一些线程将处于空闲状态
}}
output {
if [fields][app] == "syslog" { # app类型要与filebeat主机的app对用。
elasticsearch {
hosts => ["192.168.7.100:9200"] # 转发到elasticsearch服务器上
index => "filebeat-syslog-7-103-%{+YYYY.MM.dd}"
}}
}
A主机：kibana、elasticsearch , 有条件可以将两个服务器分开：192.168.7.100
B主机：logstash主机:192.168.7.101/nginx服务器也在此主机上
C主机：logstash主机：192.168.7.102
D主机：kafka/zookeeper：192.168.7.104
E主机：kafka/zookeeper：192.168.7.105
13.1 使用logstash-B主机收集系统日志和nginx日志写入kafka
我们在搭建完
kafka及zookeeper的基础上 , 选择提取zookeeper的leader的主机进行存储[root@tomcat-web2 ~]# /usr/local/zookeeper/bin/zkServer.sh status ZooKeeper JMX enabled by default Using config: /usr/local/zookeeper/bin/../conf/zoo.cfg Mode: leader # 查看到192.168.7.105主机是leader模式在
logstash主机/etc/logstash/conf.d/目录下创建收集系统日志和nginx日志的文件[root@logstash-1 conf.d]# cat log-to-kafka.conf input { file { path => "/var/log/messages" type => "kafka-syslog-log-7-101" # 定义系统日志类型 start_position => "beginning" stat_interval => "2" } file { path => "/var/log/nginx/access.log" type => "kafka-nginx-access-log-7-101" # 定义nginx的日志类型 start_position => "beginning" stat_interval => "2" codec => "json" # 如果是json格式的日志 , 必须写入json格式 } } output { if [type] == "kafka-syslog-log-7-101" { kafka { topic_id => "kafka-syslog-log-7-101" bootstrap_servers => "192.168.7.105:9092" # 指定写入到kafka的leader主机的IP地址上 codec => "json" # 写入的时候使用json编码 , 因为logstash收集后会转换为json格式 } file { path => "/tmp/syslog.log" # 将收集到的访问系统的日志临时存在此文件下 , 方便排错 , 解决问题后 , 可以删除此两项 }} if [type] == "kafka-nginx-access-log-7-101" { kafka { topic_id => "kafka-nginx-access-log-7-101" bootstrap_servers => "192.168.7.105:9092" # 指定写入到kafka主机的IP地址 codec => "json" # 写入的时候使用json编码 , 因为Logstash收集后会转换为json格式 } file { path => "/tmp/nginx-access.log" # 临时先存放在此文件下 , 方便排错 , 如果没有错误 , 可以删除此项 }} }验证
logstash配置文件的正确性[root@logstash-1 conf.d]``# logstash -f log-to-kafka.conf
开始访问
nginx网页 , (192.168.7.101:80)再查看访问的临时log信息
在
logstash主机上测试系统日志访问情况[root@logstash-1 ~]# echo 1 >> /var/log/messages [root@logstash-1 ~]# echo 2 >> /var/log/messages [root@logstash-1 ~]# echo 3 >> /var/log/messages查看此时的临时文件信息
此时测试
logstash收集到日志写入到kafka中正常，可以将测试的file部分脚本删除掉，然后启动logstash服务。file { path => "/tmp/syslog.log" # 删除掉两个测试部分 } file { path => "/tmp/nginx-access.log" }启动
D主机的logstash服务。# systemctl start logstash
13.2 配置logstash-C主机从kafka读取系统日志和nginx日志
在
logstash-A主机的/etc/logstash/conf.d/目录下创建一个收集kafka配置文件[root@logstash conf.d]# vim kafka-to-es_logstash.conf input { kafka { topics => "kafka-syslog-log-7-101" bootstrap_servers => "192.168.7.105:9092" # 从kafka的leader主机上提取缓存 codec => "json" # 在提取kafka主机的日志时，需要写成json格式 } } output { if [type] == "kafka-syslog-log-7-101" { # 转到elasticsearch主机上 elasticsearch { hosts => ["192.168.7.100:9200"] index => "kafka-syslog-7-101-%{+YYYY.ww}" # YYYY代表年，ww代表周 }} if [type] == "kafka-nginx-access-log-7-101" { # 转到elasticsearch主机上 elasticsearch { hosts => ["192.168.7.100:9200"] index => "kafka-nginx-access-log-7-101-%{+YYYY.MM.dd}" }} }启动
logstash-B主机的服务器# systemctl start logstash