SpringBoot集成logstash

前言

大家应该都听说过ELK，一般ELK都是用来做分布式系统的集中日志管理，ELK的优点这里就不介绍了，好处太多，今天主要介绍下其中的Logstash。数据传给logstash，它将数据进行过滤和格式化（转成JSON格式），然后传给数据存储或者消息队列Broker，用于后续加工处理。

最近要做app埋点监控，app将埋点日志发送到埋点日志网关，在埋点日志网关通过部署logstash，将日志发送到logstash，再由logstash发送到kafka，最后由kafka入mongodb，由大数据系统定时跑批，将埋点统计结果加工出来提供给前端查询。这里就记录下对于springboot的项目如何和logstash集成，完成日志数据通过logstash对外输出。

logstash官网 https://www.elastic.co/products/logstash

logstash安装&配置

logstash的安装流程这里就不介绍了，大家可以看下官网的guide

下面主要来说下logstash的配置，logstash一般常用的有两种模式，一种是应用服务器写本地日志，在logstash中配置读取本地日志文件，对指定的日志文件进行数据抽取，这种模式一般适用于一般的应用或者系统日志采集到中央日志系统；还有一种模式是将应用系统需要大数据系统分析的数据直接将信息流直接发送给logstash进行JSON格式化，不需要在应用系统本地先落地文件，这种模式一般适用于大数据系统从应用系统采集数据，看起来有点像kafka这类消息队列的功能，从消息流传递角度看，确实比较类似，但是logstash最大的特点还在于可以对收到的数据进行各种形式的流式加工，加工成上游大数据系统需要的数据，这样可以方便大数据系统进行二次加工，有点类似流水线的意思。

下面我们看下logstash的config文件，从config文件中可以看出logstash的主要处理流程，配置主要分为3部分：input、filter、output。这三部份的配置，从名字也很好理解含义：

input就是输入，从这个例子中可以看出配置的tcp网络信息流的模式，这样应用系统本地数据就不落地了，可以减少系统本地的磁盘空间占用，也减少了系统的磁盘IO压力；当然input除了tcp模式还有file模式，也很简单，配置的内容当然就是文件的路径信息，编码模式之类的都是一样的。

filter就是过滤，对于从input输入的数据，在filter层可以通过配置ruby脚本很轻易的将输入数据加工成输出需要的数据格式，增加字段、删减字段之类都是很简单的应用。

output当然就是输出，从官网的介绍就很容易了解到logstash不管是输入还是输出都支持很多种格式，有点类似linux里的pipline的意思，不管输入是什么都可以输出到下一个管道，例子中的output是输出到kafka集群，当然也可以输出的ES，或者数据库。

## logtash.conf
input {
  tcp {
    host => "10.xxx.xx.xx"
    port => 9250
    mode => "server"
    tags => ["tags"]
    codec => plain{charset=>"UTF-8"}
    }
}
filter {
    ruby{
     code => "event['readunixtime']= event.timestamp.time.getlocal.to_f.to_s"
     add_field =>{
        "app_name"=>"xxx_sdk_apm"
        "app_stage"=>"dev"
        "readtimestamp"=>"%{@timestamp}"
        }
     }
}
# output { stdout { codec => rubydebug } }
output {
    kafka{
# 大数据kafka集群地址
    bootstrap_servers=> "10.132.XX.XX:9092,10.132.XX.XX:9092,10.132.XX.XX:9092,10.132.37.XX:9092"
# 分配给应用的topic
    topic_id=> "xx_xx_apm"
# 异步传输，不保证消息一定传输，速度最快
    acks=>"0"
# 单位是字节,16k
    batch_size=>16384
    codec=>json
  }
}

logback配置

上面介绍了logstash的配置，下面就到了本文的重头戏，那就是在springboot中集成logstash，当然logstash为我们提供了常见的日志框架的encoder，可以方便我们将各种日志框架的日志输出到logstash中作为input。下面介绍下我这次做的本地不落日志，直接将logback的日志输出的logstash中，

发送logstash input本地不留存日志。

首先引入pom依赖

<dependency>
   <groupId>net.logstash.logback</groupId>
   <artifactId>logstash-logback-encoder</artifactId>
   <version>5.1</version>
</dependency>

在logback的xml配置文件中只需要配置下appender就可以了,例子中使用的是LogstashTcpSocketAppender，当然还提供其他的appender，具体用法可以查看https://github.com/logstash/logstash-logback-encoder，如图：

Screenshot 2018-07-24 15.40.35

logback中的配置也很简单，只需要配置logstash的ip地址和端口和encoder的编码类就可以了。

<appender name="LOGSTASH" class="net.logstash.logback.appender.LogstashTcpSocketAppender">
    <destination>10.132.xx.xx:9250</destination>
    <queueSize>1048576</queueSize>
    <encoder charset="UTF-8" class="net.logstash.logback.encoder.LogstashEncoder" />
</appender>
   
<logger name="APMInfoDev" level="INFO" additivity="false">
   <appender-ref ref="LOGSTASH"/>
</logger> 

因为是单独定义了logger不是root logger，所以在使用的时候需要通过LoggerFactory指定logger name去获取logger:

Logger apmInfoLogger = LoggerFactory.getLogger("APMInfoDev");

apmInfoLogger.info("XXXXXX");

好了下面就可以去logstash输出的存储或者队列中查看输出的日志信息了。