使用Logstash收集日志输出到ES通过Kibana做日志分析_logstash tcp 如何输出日志到不同的es索引中

一、安装ES、Logstash、Kibana

安装步骤略过

docker pull elasticsearch:7.8.0
docker pull logstash:7.8.0
docker pull kibana:7.8.0
1
2
3

由于云服务器配置太低，使用docker启动不起来。（使用本地的虚拟机）

二、配置yml

pom依赖

    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.3.3.RELEASE</version>
        <relativePath/>
   	</parent>
	<dependency>
           <groupId>net.logstash.logback</groupId>
           <artifactId>logstash-logback-encoder</artifactId>
           <version>5.2</version>
    </dependency>
1
2
3
4
5
6
7
8
9
10
11

spring:
  application:
    name: wxpublic
logging:
  config: classpath:logback-spring.xml
logstash:
  host: 192.168.103.10
1
2
3
4
5
6
7

三、配置logback_spring.xml日志文件

新建mysqletc/mylog.conf文件

<?xml version="1.0" encoding="UTF-8"?>
<configuration scan="true">
    <!--引用默认日志配置-->
    <include resource="org/springframework/boot/logging/logback/defaults.xml"/>
    <!--使用默认的控制台日志输出实现-->
    <include resource="org/springframework/boot/logging/logback/console-appender.xml"/>
    <!--应用名称-->
    <springProperty scope="context" name="APP_NAME" source="spring.application.name" defaultValue="springBoot"/>
    <!--日志文件保存路径-->
    <property name="LOG_FILE_PATH" value="${LOG_FILE:-${LOG_PATH:-${LOG_TEMP:-${java.io.tmpdir:-/tmp}}}/logs}"/>
    <!--LogStash访问host-->
    <springProperty name="LOG_STASH_HOST" scope="context" source="logstash.host" defaultValue="192.168.103.10"/>


    <!--DEBUG日志输出到LogStash-->
    <appender name="LOG_STASH_DEBUG" class="net.logstash.logback.appender.LogstashTcpSocketAppender">
        <filter class="ch.qos.logback.classic.filter.ThresholdFilter">
            <level>DEBUG</level>
        </filter>
        <destination>${LOG_STASH_HOST}:4560</destination>
        <encoder charset="UTF-8" class="net.logstash.logback.encoder.LoggingEventCompositeJsonEncoder">
            <providers>
                <timestamp>
                    <timeZone>Asia/Shanghai</timeZone>
                </timestamp>
                <!--自定义日志输出格式-->
                <pattern>
                    <pattern>
                        {
                        "project": "wxpublic",
                        "level": "%level",
                        "service": "${APP_NAME:-}",
                        "pid": "${PID:-}",
                        "thread": "%thread",
                        "class": "%logger",
                        "message": "%message",
                        "stack_trace": "%exception{20}"
                        }
                    </pattern>
                </pattern>
            </providers>
        </encoder>
        <!--当有多个LogStash服务时，设置访问策略为轮询-->
<!--        <connectionStrategy>-->
<!--            <roundRobin>-->
<!--                <connectionTTL>5 minutes</connectionTTL>-->
<!--            </roundRobin>-->
<!--        </connectionStrategy>-->
    </appender>

    <!--ERROR日志输出到LogStash-->
    <appender name="LOG_STASH_ERROR" class="net.logstash.logback.appender.LogstashTcpSocketAppender">
        <filter class="ch.qos.logback.classic.filter.LevelFilter">
            <level>ERROR</level>
            <onMatch>ACCEPT</onMatch>
            <onMismatch>DENY</onMismatch>
        </filter>
        <destination>${LOG_STASH_HOST}:4561</destination>
        <encoder charset="UTF-8" class="net.logstash.logback.encoder.LoggingEventCompositeJsonEncoder">
            <providers>
                <timestamp>
                    <timeZone>Asia/Shanghai</timeZone>
                </timestamp>
                <!--自定义日志输出格式-->
                <pattern>
                    <pattern>
                        {
                        "project": "wxpublic",
                        "level": "%level",
                        "service": "${APP_NAME:-}",
                        "pid": "${PID:-}",
                        "thread": "%thread",
                        "class": "%logger",
                        "message": "%message",
                        "stack_trace": "%exception{20}"
                        }
                    </pattern>
                </pattern>
            </providers>
        </encoder>
        <!--当有多个LogStash服务时，设置访问策略为轮询-->
<!--        <connectionStrategy>-->
<!--            <roundRobin>-->
<!--                <connectionTTL>5 minutes</connectionTTL>-->
<!--            </roundRobin>-->
<!--        </connectionStrategy>-->
    </appender>

    <root level="INFO">
        <appender-ref ref="CONSOLE"/>
        <appender-ref ref="LOG_STASH_DEBUG"/>
        <appender-ref ref="LOG_STASH_ERROR"/>
    </root>

</configuration>

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96

四、配置logstash信息

input {
  tcp {
    mode => "server"
    host => "0.0.0.0"
    port => 4560
    codec => json_lines
  }
 tcp {
    mode => "server"
    host => "0.0.0.0"
    port => 4561
    codec => json_lines
    type => "error"
  }
}
output {
  elasticsearch {
	  #ESIP地址与端口
	  hosts => "192.168.103.10:9200"
	  #ES索引名称（自己定义的）
	  index => "mylog"
	   }
  stdout {
      #以JSON格式输出
      codec => json_lines
  		}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

启动测试

启动es

进入bin目录./elasticsearch -d -d是后台启动（注意切换es用户，root用户无法启动）

启动Logstash

进入bin目录./logstash -f ../mysqletc/wxlog.conf

启动Kibana

./kibana &，浏览器访问IP+5601（注意切换es用户，root用户无法启动）