ELK搭建日志联合分析系统

上一篇ELK安装简单说了下安装ELK，这一篇就来具体说下怎么让者三个联动起来。 

其工作流程如下图所示，（从Kibana官网拿的流程图） 



简单来讲，其具体流程是logstash agent 监控并过滤日志，将过滤后的日志内容发给Redis(这里的redis只处理队列不做存储)，logstash index将日志收集在一起交给elasticSearch，并通过Kibana进行展示。 

在实际中，还可以加上logback实现自定义的日志。 

其具体流程如下 



（网上找的一个图，链接找不到了……sorry） 

大概就是这个意思，不过是在最左边又加了一个logback。

安装和部署

Redis

Redis的安装和配置之前写过，链接在：Centos 7服务器搭建（三）—安装Redis 

在这里不需要其它更改。

Logback

（1）安装 

logback是一个开源包，使用Maven可以自动部署。 

不得不说Maven是个好东西，顺便贴一个地址，用来寻找包的Maven依赖http://mvnrepository.com/。 

当前最新版是1.1.7，使用logback-classic会自动添加其它logback-core、slf4j等依赖。

<!-- https://mvnrepository.com/artifact/ch.qos.logback/logback-classic -->
<dependency>
<groupId>ch.qos.logback</groupId>
<artifactId>logback-classic</artifactId>
<version>1.1.7</version>
</dependency>

1
2
3
4
5
6
7
1
2
3
4
5
6
7

（2）配置 

在所在项目的resources文件夹下新建logback.xml，logback会自动寻找配置文件。 

一个配置如下：

<?xml version="1.0" encoding="UTF-8"?>
<!-- scan 是否定期扫描xml文件， scanPeriod是说扫描周期是30秒-->
<configuration scan="true" scanPeriod="30 seconds" debug="false" packagingData="true">
<!-- 项目名称 -->
<contextName>myApp1 contextName</contextName>
<!-- 属性 -->
<property name="USER_HOME" value="C://log-path"/>

<!-- Insert the current time formatted as "yyyyMMdd'T'HHmmss" under
the key "bySecond" into the logger context. This value will be
available to all subsequent configuration elements. -->
<timestamp key="bySecond" datePattern="yyyyMMdd" timeReference="contextBirth"/>

<!-- appender很重要，一个配置文件会有多个appender -->
<!-- ConsoleApperder意思是从console中打印出来 -->
<appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
<!-- 过滤器，一个appender可以有多个 -->
<!-- 阈值过滤，就是log行为级别过滤，debug及debug以上的信息会被打印出来 -->
<filter class="ch.qos.logback.classic.filter.ThresholdFilter">
<level>debug</level>
</filter>

<!-- encoders are assigned the type
ch.qos.logback.classic.encoder.PatternLayoutEncoder by default -->
<!-- encoder编码规则 -->
<encoder>
<!--<pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern>-->
<!--<pattern>%d %contextName %msg%n</pattern>-->
<!-- pattern模式 %d时间 %thread 线程名 %level行为级别 %logger logger名称 %method 方法名称 %message 调用方法的入参消息 -->
<pattern>%-4d [%thread] %highlight%-5level %cyan%logger.%-10method - %message%n</pattern>
</encoder>
</appender>

<!-- FileAppender 输出到文件 -->
<appender name="FILE" class="ch.qos.logback.core.FileAppender">
<!-- 文件存放位置 %{xxx} 就是之前定义的属性xxx -->
<file>${USER_HOME}/myApp1log-${bySecond}.log</file>

<encoder>
<!-- %date和%d是一个意思 %file是所在文件 %line是所在行 -->
<pattern>%date{yyyy-MM-dd HH:mm} %level [%thread] %logger{30} [%file:%line] %msg%n</pattern>
</encoder>
</appender>

<!-- 滚动日志文件，这个比较常用 -->
<appender name="ROLLINGFILE" class="ch.qos.logback.core.rolling.RollingFileAppender">
<!-- 当project等于true的时候file就不会起效果-->
<prudent>true</prudent>
<!--<file>${USER_HOME}/logFile.log</file>-->
<!-- 按天新建log日志 -->
<rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
<!-- daily rollover -->
<fileNamePattern>${USER_HOME}/logFile.%d{yyyy-MM-dd}_%i.log</fileNamePattern>
<!-- 保留30天的历史日志 -->
<maxHistory>30</maxHistory>

<!-- 基于大小和时间，这个可以有，可以没有 -->
<timeBasedFileNamingAndTriggeringPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">
<!-- or whenever the file size reaches 100MB -->
<!-- 当一个日志大小大于10KB，则换一个新的日志。日志名的%i从0开始，自动递增 -->
<maxFileSize>10KB</maxFileSize>
</timeBasedFileNamingAndTriggeringPolicy>
</rollingPolicy>

<encoder>
<!-- %ex就是指抛出的异常，full是显示全部，如果在{}中写入数字，则表示展示多少行 -->
<pattern>%-4date [%thread] %-5level %logger{35} - %msg%n%ex{full, DISPLAY_EX_EVAL}</pattern>
</encoder>
</appender>

<appender name="redis" class="com.cwbase.logback.RedisAppender">
<source>redis</source>
<host>redis服务器的ip地址（端口号一般默认，如果修改过则需要增加<port>属性）</host>
<password>可选项，如果redis设置了密码访问的话</password>
<key>redis的key</key>   <!-- 这个Key是Redis List的Key，需要和Logstash读取Redis的Key匹配 -->
<database>5</database> <!-- 具体选择哪个数据库，默认为0 -->
<callerStackIndex>0</callerStackIndex>
</appender>

<!-- 重点来了，上边都是appender输出源。这里开始就是looger了 -->
<!-- name意思是这个logger管的哪一片，像下面这个管的就是com.task包下的所有文件 level是只展示什么行为信息级别以上的，类似阈值过滤器 additivity表示是否再抛出事件，就是说如果有一个logger的name是log，如果这个属性是true，另一个logger就会在这个logger处理完后接着继续处理 -->
<logger name="com.task" level="INFO" additivity="false">
<!-- 连接输出源，也就是上边那几个输出源 ，你可以随便选几个appender-->
<appender-ref ref="redis"/>
<appender-ref ref="ROLLINGFILE"/>
</logger>
<!-- 这个logger详细到了类 -->
<logger name="log.test.Foo" level="debug" additivity="false">
<appender-ref ref="STDOUT"/>
<appender-ref ref="ROLLINGFILE"/>

</logger>

<!-- Strictly speaking, the level attribute is not necessary since -->
<!-- the level of the root level is set to DEBUG by default.       -->
<!-- 这就是上边logger没有管到的情况下 root默认接管所有logger -->
<root level="debug">
<appender-ref ref="STDOUT"/>
</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
97
98
99
100
101
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
97
98
99
100
101

这里使用了一个开源项目logback-redis-appender，其用途是生成logback到redis的appender，首先需要加入Maven依赖：

<dependency>
<groupId>com.cwbase</groupId>
<artifactId>logback-redis-appender</artifactId>
<version>*.*.*（版本号）</version>
</dependency>

1
2
3
4
5
1
2
3
4
5

其项目地址在这里，可以查找最新的版本号。

一个检验程序如下：

package com.task;

public class MyLog {
public static void main(String[] args) {
Logger logger = (Logger) LoggerFactory.getLogger(MyLog.class.getName());
logger.info("Tome");
}
}

1
2
3
4
5
6
7
8
1
2
3
4
5
6
7
8

运行了之后，由于在com.task包里，会调用两个appender，一个送到redis服务器，一个输出到滚动文件。可以分别看看结果。

<appender-ref ref="redis"/>
<appender-ref ref="ROLLINGFILE"/>

1
2
1
2

Logstash

现在假设ELK都已经安装成功，具体在这里： 

看下logstash安装在哪儿：

whereis logstash

默认是安装在

/opt/logstash

下，配置文件在

/etc/logstash/conf.d

下。不过该目录是空的，需要自己新建配置文件

vi config-fire-name.conf

增加以下内容，需要注意的是，这里的配置需要和logback.xml里关于redis的配置相同，也就是说logback输出到redis后，logstash获取并作为输入。

input {
redis{
host => "redis服务器的地址"
key => "***"  #与上面logback.xml里设置的redis的key相同
type => "redis-input"
data_type => "list"
db => 5   #具体的数据库编号，等同于logback里的database
password => "redis的密码"
}

}

output {
stdout{codec=>rubydebug}
elasticsearch{
hosts=>["*.*.*.*:9200"]
index=>"logstash-test"
codec => "json"
}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

这里的output还加了一个标准输出

stdout{codec=>rubydebug}

，可以删除，不影响elasticsearch。 

进入logstash目录，启动服务

./logstash -f /etc/logstash/conf.d/*****.conf

1
1

Elasticsearch和Kibana

这两个基本都不用再配置，按照上一篇就行。 

启动服务后，通过浏览器访问Kibana：http://server-address:5601/，默认端口为5601； 

在Setting里设置index pattern，根据logstash的配置，传输过来的索引名为logstash-test，填在Index name or pattern里，creat就行。 

然后在Discover里就可以看到已索引的数据。

Ok，现在可以再次运行一下上面的测试程序，生成log并传输到elasticsearch，可以在Kibana里看到了。