通过 Logstash 收集多个日志文件并输出至Elasticsearch

1、服务器部署 Tomcat 服务

# 先安装JDK,因为logstash已经安装JDK,此步不用执行
root@logstash01:~# tar xf apache-tomcat-8.5.84.tar.gz -C /apps/
root@logstash01:~# cd /apps/
root@logstash01:/apps# ln -s apache-tomcat-8.5.84/ tomcat
root@logstash01:/apps# /apps/tomcat/bin/startup.sh 

生成访问日志

root@logstash01:~# ll /apps/tomcat/logs/
total 32
drwxr-x--- 2 root root 4096 Jan  4 08:03 ./
drwxr-xr-x 9 root root 4096 Jan  4 08:01 ../
-rw-r----- 1 root root 6803 Jan  4 08:03 catalina.2023-01-04.log
-rw-r----- 1 root root 7074 Jan  4 08:03 catalina.out
-rw-r----- 1 root root    0 Jan  4 08:03 host-manager.2023-01-04.log
-rw-r----- 1 root root  459 Jan  4 08:03 localhost.2023-01-04.log
-rw-r----- 1 root root  738 Jan  4 08:04 localhost_access_log.2023-01-04.txt
-rw-r----- 1 root root    0 Jan  4 08:03 manager.2023-01-04.log

2、Logstash配置

root@logstash01:~# vim /etc/logstash/conf.d/syslog-tomcat-to-es.conf
input {
 file {
   path => "/var/log/syslog"   #日志路径
   start_position => "beginning"   #第一次收集日志的位置
   stat_interval => "3"             #日志收集的间隔时间
   type => "syslog"   #事件的唯一类型，可用于区分不同日志，分别存于不同
index
 }
 file {
   path => "/apps/tomcat/logs/localhost_access_log.*.txt"   #日志路径
   start_position => "beginning"   #第一次收集日志的位置
   stat_interval => "3"           #日志收集的间隔时间
   type => "tomcat-access" #事件的唯一类型
 }
}
output {
 elasticsearch {
   hosts => ["192.168.1.101:9200"]
   index => "linux-syslog-tomcat-%{+YYYY.MM.dd}"
 }
}

3、设置权限并重启logstash

# 修改权限让logstash可以访问日志
root@logstash01:~# setfacl -m u:logstash:r /var/log/syslog /apps/tomcat/logs/*
root@logstash01:~# setfacl -Rm u:logstash:rx   /apps/tomcat/logs/


# 或者修改Logstash启用用户为root
root@logstash01:~# vim /etc/systemd/system/logstash.service
User=logstash
Group=logstash

root@logstash01:~# systemctl restart logstash.service 

4、向被收集的文件生成新日志信息

root@web01:~# while true;do curl 192.168.1.108:8080;sleep 1;done
root@logstash01:~# logger "test log"

5、查看索引是否生成

6、配置将不同类型日志写入不同索引

root@logstash01:~# vim /etc/logstash/conf.d/syslog-tomcat-to-es.conf

input {
  file {
    path => "/var/log/syslog"
    start_position => "beginning"
    stat_interval => "3"
    type => "syslog"
    codec => json
  }
  file {
    path => "/var/log/tomcat9/localhost_access_log.*.txt"
    start_position => "beginning"
    stat_interval => "3"
    type => "tomcat-access"
    codec => json
  }
}
output {
  if [type] == "syslog" {                 #指定类型不同,写入不同索引
    elasticsearch {
      hosts => ["192.168.1.101:9200"]
      index => "syslog-1.108-%{+YYYY.MM.dd}"
    }
    file {                                #可以同时写入文件
      path => "/tmp/syslog-1.108.log"
    }
  }
  if [type] == "tomcat-access" {
    elasticsearch {
      hosts => ["192.168.1.101:9200"]
      index => "tomcataccesslog-1.108-%{+YYYY.MM.dd}"
    }
    file {
      path => "/tmp/tomcat-accesslog-1.108.log"
    }
  }
}

root@logstash01:~# systemctl restart logstash.service 

7、Kibana 界面添加索引

8、Tomcat 日志转为 Json 格式

root@logstash01:~# vim /etc/tomcat9/server.xml
        <Valve className="org.apache.catalina.valves.AccessLogValve" directory="logs"
               prefix="localhost_access_log" suffix=".txt"               pattern="{&quot;clientip&quot;:&quot;%h&quot;,&quot;ClientUser&quot;:&quot;%l&quot;,&quot;authenticated&quot;:&quot;%u&quot;,&quot;AccessTime&quot;:&quot;%t&quot;,&quot;method&quot;:&quot;%r&quot;,&quot;status&quot;:&quot;%s&quot;,&quot;SendBytes&quot;:&quot;%b&quot;,&quot;Query?string&quot;:&quot;%q&quot;,&quot;partner&quot;:&quot;%{Referer}i&quot;,&quot;AgentVersion&quot;:&quot;%{User-Agent}i&quot;}"/>

root@logstash01:~# systemctl restart tomcat9.service 

9、利用 multiline 插件收集 Java 错误日志

Java应用日志有时因为某一个错误会出现很多行错误的日志信息，但这些行都属于同一个日志的信息
为了Elasticsearch 分析方便，需要将多行日志合并成一行日志

使用codec的multiline插件实现多行匹配，而且支持使用what指令灵活的将匹配到的行与前面的行合并或者和后面的行合并

实现测试 multiline 插件的配置文件

input {
  stdin {
    codec => multiline {
      pattern => "^\["  #用正则表达式表示以[开头的行
      negate => "true"  #匹配正则表达式的行才进行合并,true为匹配不成功进行操作,false为成功进行操作
      what => "previous" #将正则表达式前面的行进行合并,previous表示与以前的行合并，如果是下面的行合并就是next
    }
  }
}
output {
  stdout {
    codec => "rubydebug"
  }
}

交互式测试插件的标准输入和标准输出

root@logstash01:~# logstash -f /etc/logstash/conf.d/java-log-to-es.conf -t
root@logstash01:~# systemctl stop logstash.service
root@logstash01:~# logstash -f /etc/logstash/conf.d/java-log-to-es.conf
line1            #输入测试数据
line2            #输入测试数据
line3            #输入测试数据
[2023-01-04T09:23:20,512][INFO ] test java log            #输入测试数据
{
      "@version" => "1",
          "tags" => [
        [0] "multiline"
    ],
          "host" => "logstash01.test.com",
    "@timestamp" => 2023-01-04T10:54:28.580Z,
       "message" => "line1\nline2\nline3"
}
line4            #输入测试数据
line5            #输入测试数据
line6            #输入测试数据
[2023-01-04T09:23:20,512][INFO ] test java log2            #输入测试数据
{
      "@version" => "1",
          "tags" => [
        [0] "multiline"
    ],
          "host" => "logstash01.test.com",
    "@timestamp" => 2023-01-04T10:55:17.556Z,
       "message" => "[2021-01-04T09:23:20,512][INFO ] test java log\nline4\nline5\nline6"
}

配置读取日志文件至标准输出

root@logstash01:~# logstash -f /etc/logstash/conf.d/java-log-to-es.conf

将输出改为 Elasticsearch

root@logstash01:~# vim /etc/logstash/conf.d/java-log-to-es.conf

input {
  file {
    path => "/var/log/tomcat9/catalina.*"
    type => "java-error-log"
    start_position => "beginning"
    stat_interval => "3"
    codec => multiline {
      pattern => "^\["
      negate => "true"
      what => "previous"
    }
  }
}

output {
  if [type] == "java-error-log" {
    elasticsearch {
      hosts => ["192.168.1.101:9200,192.168.1.102:9200,192.168.1.103:9200"]
      index => "javaerrorlog-1.108-%{+YYYY.ww}"  #时间格式：年.周
    }
  }
}

root@logstash01:~# systemctl restart logstash.service