Spring Boot整合ELK 处理为服务日志，妙！（2）

配置 Indexer 角色 Logstash

配置好 Shipper 角色的 Logstash 后，我们还需要配置 Indexer 角色 Logstash 以支持从 Redis 接收日志数据，并通过过滤器解析后存储到 Elasticsearch 中，其配置内容如下所示。

清单 7. Indexer 角色的 Logstash 的配置

input {
    redis {
        host      => "192.168.142.131"    # redis主机地址
        port      => 6379               # redis端口号
        db        => 8                  # redis数据库编号
        data_type => "channel"          # 使用发布/订阅模式
        key       => "sb-logback"  # 发布通道名称
    }
}
filter {
     #定义数据的格式
     grok {
       match => { "message" => "%{TIMESTAMP_ISO8601:time} \[%{NOTSPACE:threadName}\] %{LOGLEVEL:level}  %{DATA:logger} %{NOTSPACE:applicationName} -(?:.*=%{NUMBER:timetaken}ms|)"}
     }
}
output {
    stdout {}
    elasticsearch {
        hosts => "localhost:9200"
        index => "logback"
   }
}

与 Shipper 不同的是，Indexer 的管道中我们定义了过滤器，也正是在这里将日志解析成结构化的数据。下面是我截取的一条 logback 的日志内容：

清单 8. Spring Boot 项目输出的一条日志

2019-08-11 18:01:31.602 [http-nio-8080-exec-2] INFO  c.i.s.aop.WebLogAspect sb-elk -接口日志
POST请求测试接口结束调用:耗时=11ms,result=BaseResponse{code=10000, message='操作成功'}

在 Filter 中我们使用 Grok 插件从上面这条日志中解析出了时间、线程名称、Logger、服务名称以及接口耗时几个字段。Grok 又是如何工作的呢？

1. message 字段是 Logstash 存放收集到的数据的字段， match = {"message" => ...} 代表是对日志内容做处理。
   
2. Grok 实际上也是通过正则表达式来解析数据的，上面出现的 TIMESTAMP_ISO8601 、 NOTSPACE 等都是 Grok 内置的 patterns。
   
3. 我们编写的解析字符串可以使用 Grok Debugger 来测试是否正确，这样避免了重复在真实环境中校验解析规则的正确性。

查看效果

经过上面的步骤，我们已经完成了整个 ELK 平台的搭建以及 Spring Boot 项目的接入。下面我们按照以下步骤执行一些操作来看下效果。

启动 Elasticsearch，启动命令在 ELK 平台搭建 小节中有提到，这里不赘述（Kibana 启动同）。启动 Indexer 角色的 Logstash。

# 进入到 Logstash 的解压目录，然后执行下面的命令
bin/logstash -f indexer-logstash.conf

启动 Kibana。

启动 Shipper 角色的 Logstash。

# 进入到 Logstash 的解压目录，然后执行下面的命令
bin/logstash -f shipper-logstash.conf

调用 Spring Boot 接口，此时应该已经有数据写入到 ES 中了。

在浏览器中访问 http://ip:5601 ，打开 Kibana 的 Web 界面，并且如下图所示添加 logback 索引。

图 8. 在 Kibana 中添加 Elasticsearch 索引

在 Kibana 中添加 Elasticsearch 索引

进入 Discover 界面，选择 logback 索引，就可以看到日志数据了，如下图所示。

图 9. ELK 日志查看

ELK 日志查看

在 Nginx 中使用 ELK

相信通过上面的步骤您已经成功的搭建起了自己的 ELK 实时日志平台，并且接入了 Logback 类型的日志。但是实际场景下，几乎不可能只有一种类型的日志，下面我们就再在上面步骤的基础之上接入 Nginx 的日志。当然这一步的前提是我们需要在服务器上安装 Nginx，具体的安装过程网上有很多介绍，这里不再赘述。查看 Nginx 的日志如下（Nginx 的访问日志默认在 /var/log/nginx/access.log 文件中）。

清单 9. Nginx 的访问日志

192.168.142.1 - - [17/Aug/2019:21:31:43 +0800] "GET /weblog/get-test?name=elk HTTP/1.1"
200 3 "http://192.168.142.131/swagger-ui.html" "Mozilla/5.0 (Windows NT 10.0; Win64; x64)
AppleWebKit/537.36 (KHTML, like Gecko) Chrome/76.0.3809.100 Safari/537.36"

同样，我们需要为此日志编写一个 Grok 解析规则，如下所示：

清单 10. 针对 Nginx 访问日志的 Grok 解析规则

%{IPV4:ip} \- \- \[%{HTTPDATE:time}\] "%{NOTSPACE:method} %{DATA:requestUrl}
HTTP/%{NUMBER:httpVersion}" %{NUMBER:httpStatus} %{NUMBER:bytes}
"%{DATA:referer}" "%{DATA:agent}"

完成上面这些之后的关键点是 Indexer 类型的 Logstash 需要支持两种类型的输入、过滤器以及输出，如何支持呢？首先需要给输入指定类型，然后再根据不同的输入类型走不同的过滤器和输出，如下所示（篇幅原因，配置文件在此没有全部展示，可以 点击此处获取 ）。

清单 11. 支持两种日志输入的 Indexer 角色的 Logstash 配置

input {
    redis {
        type      => "logback"
        ...
    }
    redis {
       type       => "nginx"
       ...
    }
}
filter {
     if [type] == "logback" {
         ...
     }
     if [type] == "nginx" {
         ...
     }
}
output {
    if [type] == "logback" {
        ...
    }
    if [type] == "nginx" {
       ...
    }
}

我的 Nginx 与 Spring Boot 项目部署在同一台机器上，所以还需修改 Shipper 类型的 Logstash 的配置以支持两种类型的日志输入和输出，其配置文件的内容可 点击这里获取 。以上配置完成后，我们按照 查看效果 章节中的步骤，启动 ELK 平台、Shipper 角色的 Logstash、Nginx 以及 Spring Boot 项目,然后在 Kibana 上添加 Nignx 索引后就可同时查看 Spring Boot 和 Nginx 的日志了，如下图所示。

图 10. ELK 查看 Nginx 日志

ELK 查看 Nginx 日志

ELK 启动

在上面的步骤中，ELK 的启动过程是我们一个一个的去执行三大组件的启动命令的。而且还是在前台启动的，意味着如果我们关闭会话窗口，该组件就会停止导致整个 ELK 平台无法使用，这在实际工作过程中是不现实的，我们剩下的问题就在于如何使 ELK 在后台运行。根据 《Logstash 最佳实践》 一书的推荐，我们将使用 Supervisor 来管理 ELK 的启停。首先我们需要安装 Supervisor，在 Ubuntu 上执行 apt-get install supervisor 即可。安装成功后，我们还需要在 Supervisor 的配置文件中配置 ELK 三大组件（其配置文件默认为 /etc/supervisor/supervisord.conf 文件）。

清单 12. ELK 后台启动

[program:elasticsearch]
environment=JAVA_HOME="/usr/java/jdk1.8.0_221/"
directory=/home/elk/elk/elasticsearch
user=elk
command=/home/elk/elk/elasticsearch/bin/elasticsearch
[program:logstash]
environment=JAVA_HOME="/usr/java/jdk1.8.0_221/"
directory=/home/elk/elk/logstash
user=elk
command=/home/elk/elk/logstash/bin/logstash -f /home/elk/elk/logstash/indexer-logstash.conf
[program:kibana]
environment=LS_HEAP_SIZE=5000m
directory=/home/elk/elk/kibana
user=elk
command=/home/elk/elk/kibana/bin/kibana

[按照以上内容配置完成后，执行 sudo supervisorctl reload 即可完成整个 ELK 的启动，而且其默认是开机自启。当然，我们也可以使用 sudo supervisorctl start/stop program_name] 来管理单独的应用。

结束语

在本教程中，我们主要了解了什么是 ELK，然后通过实际操作和大家一起搭建了一个 ELK 日志分析平台，并且接入了 Logback 和 Nginx 两种日志。