ELK日志框架搭建

ELK即Elasticsearch + Logstash + Kibana是一套开源日志管理方案，之前在写AOP做统一日志处理和日志分类处理的时候就想写一下关于ELK的搭建和使用的文章，由于各种原因一直拖到现在，因为最近做微服务又用到了ES，所以趁机把之前的ELK搭建和使用补上来。ELK简单理解，就是使用logstash将日志进行收集，然后输出到elasticsearch，最后再到kibana展示、分析、搜索等等这样一个过程，本次ELK的搭建都是单个节点。

一、ELK的安装和配置

关于ELK的安装这里就不再赘述了，根据官网的进行下载和安装就可以了，配置的话基本上都有有默认配置，如果想自定义配置修改相关的配置文件就行。这里主要说下logstash的配置，Elasticsearch、Kibana大家相对会更熟悉一些，而且配置文件也有相关的说明，感觉没什么特别需要说的。logstash的配置主要是进行日志收集，简单的说它做了两件事情(实际上不是)，一是收集，即从日志源收集日志，二是输出，即将日志输出到目的地，这里是elasticsearch，当然也可以是数据库、缓存等。所以logstash的配置无非也就是这两点，一输入，二输出。而且logstash是支持多输入和输出的，这方面内容建议查看官网。本次因为只是简单的搭建，自己配置了两个输入源，一个是Tcp的输入源，一个是基于文件的输入源；输出只配置了elasticsearch。另外一点就是logstash还可以配置过滤器，为了简便，这里就没做配置，感兴趣的小伙伴可以自己尝试一下。在config目录下新建一个"xxx.conf"的配置文件，启动logstash，以该配置文件启动，内容如下：

input {
    tcp {
        host => 127.0.0.1
        port => 4055
        codec => "json"
        mode => "server"
    }
}

input {
  file {
    path => "/home/ypcfly/ypcfly/Log/*.log"
    type  => "app_log"
    codec => "json"
  }
}

output {
    stdout { codec=> rubydebug }
    elasticsearch {
    action => "index"
    hosts => ["127.0.0.1:9200"]
    index => "applicationlog-%{+YYYY.MM}"
  }
}

上面的配置中，第一个input中，tcp表示的是从TCP的socket中读取事件（每个事件都假定为一行文本）。host、port表示监听的地址和端口号，这个根据实际情况指定；mode表示logstash接受源模式，这个值只有两种一是"server"，另一个是"client"，即来源可以是服务端或者客户端，codec是coder/decoder的组合写法，表示的是数据编码和解码的，即数据格式。第二个表示的是从file读取日志，这个主要配置是指定具体文件路径，此外还有很多其他配置，这个可以参看官网。
output中配置了两个输出，一个是标准输出stdout（input也是可以配置stdin的），即logstash服务控制台输出，输出的格式是rubydebug，如下图这样：

图-1.png
另一个位置是elasticsearch，hosts可以是elasticsearch的集群，这里因为是单节点，所以只配置了本地的elasticsearch。action默认值就是"index"，此外还可以是"delete"、"create"等。index表示的是索引的名称，这个自己定义。
好了以上就是logstash启动时加载的配置文件，当然实际内容还是很多的，比如filter我这里就没有进行配置，filter其实也是logstash非常重要的一环，另外输入、输出支持的类型都很多，我只是挑选了两个，等真的需要时可以仔细研究下官方的文档。

二、logback配置

安装、配置好ELK后我们还需要对项目的日志配置进行修改，我这里使用的是以前项目的配置，logback.xml如下：

<?xml version="1.0" encoding="UTF-8"?>
<configuration scan="true" scanPeriod="60 seconds">
    <property name="LOG_HOME" value="/home/ypcfly/ypcfly/Log"/>

    <!-- console -->
　　　<appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
　　　　　 <encoder>
　　　　　　　　　<pattern>%d{MM-dd-yyyy HH:mm:ss} %-5level [%thread%X{sourceThread}]%logger{24}-%msg%n</pattern>
　　　　　 </encoder>
　　　</appender>

    <!--  logstash tcp -->
    <appender name="logstash" class="net.logstash.logback.appender.LogstashTcpSocketAppender">
        <param name="Encoding" value="UTF-8"/>
        <remoteHost>127.0.0.1</remoteHost>
        <port>5044</port>
        <encoder class="net.logstash.logback.encoder.LogstashEncoder"/>
    </appender>
    <!-- logstash使用file -->
    <appender name="stash" class="ch.qos.logback.core.rolling.RollingFileAppender">
        <append>true</append>
        <encoder charset="UTF-8" class="net.logstash.logback.encoder.LogstashEncoder" />
        <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
            <fileNamePattern>${LOG_HOME}/log-file.%d{yyyy-MM-dd}.log</fileNamePattern>
            <maxHistory>30</maxHistory>
        </rollingPolicy>
        <triggeringPolicy class="ch.qos.logback.core.rolling.SizeBasedTriggeringPolicy">
            <MaxFileSize>30MB</MaxFileSize>
        </triggeringPolicy>
    </appender>

    <!-- 其他日志输出 -->
    <logger name="org.apache.ibatis" level="TRACE"/>
    <logger name="java.sql.Connection" level="DEBUG"/>
    <logger name="java.sql.PreparedStatement" level="DEBUG"/>
    <logger name="java.sql.Statement" level="DEBUG"/>

    <root level="INFO">
        <appender-ref ref="CONSOLE"/>
        <!--<appender-ref ref="logstash"/>-->
        <appender-ref ref="stash"/>
    </root>

</configuration>

这个配置定义了3个appender，一个是控制台输出，这个就不做介绍了；第二个appender是配置将日志通过Tcp发送到logstash，这里appender实现类使用的是LogstashTcpSocketAppender，根据名字也可以看出是通过Tcp的socket进行日志发送，此外还配置了remoteHost和port，这两个需要和上面***.conf中配置的host、port一致，encoder使用的logstash的LogstashEncoder类；第三个appender是将日志输出到文件的，所以选择了RollingFileAppender，此外还配置了fileNamePattern，文件路径也应该和logstash中file的path保持一致，然后还有保留文件的天数，最大文件的大小等等。关于logback块配置建议参考logback 配置详解。

三、测试

1、logstash来源为Tcp

完成好以上配置后我们做个简单的测试，我们先测试使用Tcp的socket发送日志文件，修改下root标签下引用的appender-ref标签引用为"logstash"，启动项目。
然后到kibana界面配置索引，logstash的"xxx.conf"配置中索引有没有成功，可以根据配置索引的名称是否存在来判断，比如上面配置的索引名字为"applicationlog-%{+YYYY.MM}"，在kibana下面看有没满足这个格式的索引，比如创建index pattern时，可以看到ES下面已经有两个索引了，一个是"applicationlog-2019.07"，说明创建索引成功了，接下来我们定义一个pattern"applicationlog*"，并设置根据@timestamp过滤。

图-2.png
然后点击"Discover"标签，选择自定义的pattern（如果有多个的话，只有一个会默认显示），这时候就能看到输出的日志了，如下图：
图-3.png
接着调用一个接口进行测试，然后尝试根据logger进行日志过滤，点击"Add a filter"，然后选择logger_name，操作符选择is，输入具体的日志名-类的全路径，过滤结果如下：
图-4.png

2、logstash来源为file

这次修改root标签的appender-ref标签引用为"stash"，重新启动一下项目，并再次调用下接口(传入不同参数)，依然根据logger_name过滤，结果如下图：

图-5.png
根据logger_name过滤出了两次不同参数调用的日志，说明logstash使用file为输入源也是成功的。

---

到这里关于 ELK的搭建和使用就结束了，实际使用中可能要复杂一些，比如使用集群，还有logstash需要配置filter等等，内容比较多，但是参考官方文档基本不存在什么问题的，关于内容中如果有什么疑问欢迎大家交流。