搜索

实现 Spring Boot Logstash 日志收集

泥瓦匠BYSocket

共 10707字，需浏览 22分钟

 · 2020-11-08

点击上面，关注：“锅外的大佬”

专注分享国外最新技术内容

帮助每位开发者更优秀地成长

在本文中，我将介绍我的日志库，专门用于 SpringBootRESTfulWeb应用程序。关于这个库的主要设想是：

  • 使用完整正文记录所有传入的HTTP请求和传出的HTTP响应

  • 使用 logstash-logback-encoder库和 Logstash与 ElasticStack集成

  • 对于 RestTemplate和OpenFeign,记录所有可能发生的日志

  • 在单个API端点调用中跨所有通信生成和传递关联Id(correlationId)

  • 计算和存储每个请求的执行时间

  • 可自动配置的库——除了引入依赖项之外，不必执行任何操作，就能正常工作

1.简述

我想在阅读了文章的前言后，你可能会问为什么我决定构建一个 SpringBoot已有功能的库。但问题是它真的具有这些功能？你可能会感到惊讶，因为答案是否定的。虽然可以使用一些内置的 Spring组件例如 CommonsRequestLoggingFilter轻松地记录 HTTP请求，但是没有任何用于记录响应主体(response body)的开箱即用机制。

当然你可以基于Spring HTTP拦截器（HandlerInterceptorAdapter）或过滤器（OncePerRequestFilter）实现自定义解决方案，但这并没有你想的那么简单。

第二种选择是使用 ZalandoLogbook，它是一个可扩展的Java库，可以为不同的客户端和服务器端技术启用完整的请求和响应日志记录。这是一个非常有趣的库，专门用于记录HTTP请求和响应，它提供了许多自定义选项并支持不同的客户端。

因此，为了更高级，你可以始终使用此库。我的目标是创建一个简单的库，它不仅记录请求和响应，还提供自动配置，以便将这些日志发送到 Logstash并关联它们。它还会自动生成一些有价值的统计信息，例如请求处理时间。所有这些值都应该发送到 Logstash。我们继续往下看。

2.实现

从依赖开始吧。我们需要一些基本的Spring库，它们包含 spring-web， spring-context在内,并提供了一些额外的注解。为了与 Logstash集成，我们使用 logstash-logback-encoder库。 Slf4j包含用于日志记录的抽象，而 javax.servlet-api用于HTTP通信。 CommonsIO不是必需的，但它提供了一些操作输入和输出流的有用方法。

  2.     11
  3.     2.6
  4.     4.0.1
  5.     5.3
  6.     5.1.6.RELEASE
  7.     1.7.26
  10.     
  11.         org.springframework
  12.         spring-context
  13.         ${spring.version}
  14.     
  15.     
  16.         org.springframework
  17.         spring-web
  18.         ${spring.version}
  19.     
  20.     
  21.         net.logstash.logback
  22.         logstash-logback-encoder
  23.         ${logstash-logback.version}
  24.     
  25.     
  26.         javax.servlet
  27.         javax.servlet-api
  28.         ${javax-servlet.version}
  29.         provided
  30.     
  31.     
  32.         commons-io
  33.         commons-io
  34.         ${commons-io.version}
  35.     
  36.     
  37.         org.slf4j
  38.         slf4j-api
  39.         ${slf4j.version}
  40.

第一步是实现HTTP请求和响应包装器。我们必须这样做，因为无法读取HTTP流两次。如果想记录请求或响应正文，在处理输入流或将其返回给客户端之前，首先必须读取输入流。

Spring提供了HTTP请求和响应包装器的实现，但由于未知原因，它们仅支持某些特定用例，如内容类型 application/x-www-form-urlencoded。因为我们通常在RESTful应用程序之间的通信中使用 aplication/json内容类型，所以Spring ContentCachingRequestWrapper和 ContentCachingResponseWrapper在这没什么用。 

这是我的HTTP请求包装器的实现，可以通过各种方式完成。这只是其中之一：

  1. public class SpringRequestWrapper extends HttpServletRequestWrapper {

  3.     private byte[] body;

  5.     public SpringRequestWrapper(HttpServletRequest request) {
  6.         super(request);
  7.         try {
  8.             body = IOUtils.toByteArray(request.getInputStream());
  9.         } catch (IOException ex) {
  10.             body = new byte[0];
  11.         }
  12.     }

  14.     @Override
  15.     public ServletInputStream getInputStream() throws IOException {
  16.         return new ServletInputStream() {
  17.             public boolean isFinished() {
  18.                 return false;
  19.             }

  21.             public boolean isReady() {
  22.                 return true;
  23.             }

  25.             public void setReadListener(ReadListener readListener) {

  27.             }

  29.             ByteArrayInputStream byteArray = new ByteArrayInputStream(body);

  31.             @Override
  32.             public int read() throws IOException {
  33.                 return byteArray.read();
  34.             }
  35.         };
  36.     }
  37. }

输出流必须做同样的事情，这个实现有点复杂：

  1. public class SpringResponseWrapper extends HttpServletResponseWrapper {

  3.     private ServletOutputStream outputStream;
  4.     private PrintWriter writer;
  5.     private ServletOutputStreamWrapper copier;

  7.     public SpringResponseWrapper(HttpServletResponse response) throws IOException {
  8.         super(response);
  9.     }

  11.     @Override
  12.     public ServletOutputStream getOutputStream() throws IOException {
  13.         if (writer != null) {
  14.             throw new IllegalStateException("getWriter() has already been called on this response.");
  15.         }

  17.         if (outputStream == null) {
  18.             outputStream = getResponse().getOutputStream();
  19.             copier = new ServletOutputStreamWrapper(outputStream);
  20.         }

  22.         return copier;
  23.     }

  25.     @Override
  26.     public PrintWriter getWriter() throws IOException {
  27.         if (outputStream != null) {
  28.             throw new IllegalStateException("getOutputStream() has already been called on this response.");
  29.         }

  31.         if (writer == null) {
  32.             copier = new ServletOutputStreamWrapper(getResponse().getOutputStream());
  33.             writer = new PrintWriter(new OutputStreamWriter(copier, getResponse().getCharacterEncoding()), true);
  34.         }

  36.         return writer;
  37.     }

  39.     @Override
  40.     public void flushBuffer() throws IOException {
  41.         if (writer != null) {
  42.             writer.flush();
  43.         }
  44.         else if (outputStream != null) {
  45.             copier.flush();
  46.         }
  47.     }

  49.     public byte[] getContentAsByteArray() {
  50.         if (copier != null) {
  51.             return copier.getCopy();
  52.         }
  53.         else {
  54.             return new byte[0];
  55.         }
  56.     }

  58. }

我将 ServletOutputStream包装器实现放到另一个类中：

  1. public class ServletOutputStreamWrapper extends ServletOutputStream {

  3.     private OutputStream outputStream;
  4.     private ByteArrayOutputStream copy;

  6.     public ServletOutputStreamWrapper(OutputStream outputStream) {
  7.         this.outputStream = outputStream;
  8.         this.copy = new ByteArrayOutputStream();
  9.     }

  11.     @Override
  12.     public void write(int b) throws IOException {
  13.         outputStream.write(b);
  14.         copy.write(b);
  15.     }

  17.     public byte[] getCopy() {
  18.         return copy.toByteArray();
  19.     }

  21.     @Override
  22.     public boolean isReady() {
  23.         return true;
  24.     }

  26.     @Override
  27.     public void setWriteListener(WriteListener writeListener) {

  29.     }
  30. }

因为我们需要在处理之前包装HTTP请求流和响应流，所以我们应该使用HTTP过滤器。Spring提供了自己的HTTP过滤器实现。

我们的过滤器扩展了它，并使用自定义请求和响应包装来记录有效负载。此外，它还生成和设置 X-Request-ID， X-Correlation-ID header和请求处理时间。

  1. public class SpringLoggingFilter extends OncePerRequestFilter {

  3.     private static final Logger LOGGER = LoggerFactory.getLogger(SpringLoggingFilter.class);
  4.     private UniqueIDGenerator generator;

  6.     public SpringLoggingFilter(UniqueIDGenerator generator) {
  7.         this.generator = generator;
  8.     }

  10.     protected void doFilterInternal(HttpServletRequest request, HttpServletResponse response, FilterChain chain) throws ServletException, IOException {
  11.         generator.generateAndSetMDC(request);
  12.         final long startTime = System.currentTimeMillis();
  13.         final SpringRequestWrapper wrappedRequest = new SpringRequestWrapper(request);
  14.         LOGGER.info("Request: method={}, uri={}, payload={}", wrappedRequest.getMethod(),
  15.                 wrappedRequest.getRequestURI(), IOUtils.toString(wrappedRequest.getInputStream(),
  16.                 wrappedRequest.getCharacterEncoding()));
  17.         final SpringResponseWrapper wrappedResponse = new SpringResponseWrapper(response);
  18.         wrappedResponse.setHeader("X-Request-ID", MDC.get("X-Request-ID"));
  19.         wrappedResponse.setHeader("X-Correlation-ID", MDC.get("X-Correlation-ID"));
  20.         chain.doFilter(wrappedRequest, wrappedResponse);
  21.         final long duration = System.currentTimeMillis() - startTime;
  22.         LOGGER.info("Response({} ms): status={}, payload={}", value("X-Response-Time", duration),
  23.                 value("X-Response-Status", wrappedResponse.getStatus()),
  24.                 IOUtils.toString(wrappedResponse.getContentAsByteArray(), wrappedResponse.getCharacterEncoding()));
  25.     }
  26. }

3.自动配置

完成包装器和HTTP过滤器的实现后，我们可以为库准备自动配置。第一步是创建 @Configuration包含所有必需的bean。

我们必须注册自定义HTTP过滤器 SpringLoggingFilter，以及用于与 Logstash和 RestTemplateHTTP客户端拦截器集成的 logger appender：

  1. @Configuration
  2. public class SpringLoggingAutoConfiguration {

  4.     private static final String LOGSTASH_APPENDER_NAME = "LOGSTASH";

  6.     @Value("${spring.logstash.url:localhost:8500}")
  7.     String url;
  8.     @Value("${spring.application.name:-}")
  9.     String name;

  11.     @Bean
  12.     public UniqueIDGenerator generator() {
  13.         return new UniqueIDGenerator();
  14.     }

  16.     @Bean
  17.     public SpringLoggingFilter loggingFilter() {
  18.         return new SpringLoggingFilter(generator());
  19.     }

  21.     @Bean
  22.     public RestTemplate restTemplate() {
  23.         RestTemplate restTemplate = new RestTemplate();
  24.         List<ClientHttpRequestInterceptor> interceptorList = new ArrayList<ClientHttpRequestInterceptor>();
  25.         restTemplate.setInterceptors(interceptorList);
  26.         return restTemplate;
  27.     }

  29.     @Bean
  30.     public LogstashTcpSocketAppender logstashAppender() {
  31.         LoggerContext loggerContext = (LoggerContext) LoggerFactory.getILoggerFactory();
  32.         LogstashTcpSocketAppender logstashTcpSocketAppender = new LogstashTcpSocketAppender();
  33.         logstashTcpSocketAppender.setName(LOGSTASH_APPENDER_NAME);
  34.         logstashTcpSocketAppender.setContext(loggerContext);
  35.         logstashTcpSocketAppender.addDestination(url);
  36.         LogstashEncoder encoder = new LogstashEncoder();
  37.         encoder.setContext(loggerContext);
  38.         encoder.setIncludeContext(true);
  39.         encoder.setCustomFields("{\"appname\":\"" + name + "\"}");
  40.         encoder.start();
  41.         logstashTcpSocketAppender.setEncoder(encoder);
  42.         logstashTcpSocketAppender.start();
  43.         loggerContext.getLogger(Logger.ROOT_LOGGER_NAME).addAppender(logstashTcpSocketAppender);
  44.         return logstashTcpSocketAppender;
  45.     }

  47. }

库中的配置集合由 SpringBoot加载。 SpringBoot会检查已发布 jar中是否存在 META-INF/spring.factories文件。

该文件应列出key等于 EnableAutoConfiguration的配置类：

  1. org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
  2. pl.piomin.logging.config.SpringLoggingAutoConfiguration

4.与Logstash集成

通过自动配置的日志记录追加器(logging appender)实现与 Logstash集成。

我们可以通过在 application.yml文件中设置属性 spring.logstash.url来覆盖 Logstash目标URL:

  1. spring:
  2.   application:
  3.     name: sample-app
  4.   logstash:
  5.     url: 192.168.99.100:5000

要在应用程序中启用本文中描述的所有功能，只需要将我的库包含在依赖项中：

  2.     pl.piomin
  3.     spring-boot-logging
  4.     1.0-SNAPSHOT

在运行应用程序之前，您应该在计算机上启动 ElasticStack。最好的方法是通过 Docker容器。但首先要创建 Docker网络，以通过容器名称启用容器之间的通信。

  1. $ docker network create es

现在，在端口9200启动 Elasticsearch的单个节点实例,我使用版本为 6.7.2的Elastic Stack工具：

  1. $ docker run -d --name elasticsearch --net es -p 9200:9200 -p 9300:9300 -e "discovery.type=single-node" elasticsearch:6.7.2

运行 Logstash时，需要提供包含输入和输出定义的其他配置。我将使用JSON编解码器启动TCP输入，默认情况下不启用。 Elasticsearch URL设置为输出。它还将创建一个包含应用程序名称的索引。

  1. input {
  2.   tcp {
  3.     port => 5000
  4.     codec => json
  5.   }
  6. }
  7. output {
  8.   elasticsearch {
  9.     hosts => ["http://elasticsearch:9200"]
  10.     index => "micro-%{appname}"
  11.   }
  12. }

现在我们可以使用 Docker容器启动 Logstash。它在端口5000上公开并从 logstash.conf文件中读取配置：

  1. docker run -d --name logstash --net es -p 5000:5000 -v ~/logstash.conf:/usr/share/logstash/pipeline/logstash.conf docker.elastic.co/logstash/logstash:6.7.2

最后，我们可以运行仅用于显示日志的 Kibana:

  1. $ docker run -d --name kibana --net es -e "ELASTICSEARCH_URL=http://elasticsearch:9200" -p 5601:5601 docker.elastic.co/kibana/kibana:6.7.2

启动使用 spring-boot-logging库的示例应用程序后， POST请求中的日志将显示在 Kibana中，如下所示： 

响应日志每个条目都包含 X-Correlation-ID， X-Request-ID， X-Response-Time和 X-Response-Status头。

5.摘要

Springlogging library库可以在我的GitHub地址：https://github.com/piomin/spring-boot-logging.git中找到。我还在努力，所以非常欢迎任何反馈或建议。该库专用于基于微服务的体系结构，您的应用程序可以在容器内的许多实例中启动。

在此模型中，将日志存储在文件中没有任何意义。这就是为什么与 ElasticStack集成非常重要的原因。 但是这个库最重要的特性是将HTTP请求/响应与完整正文和一些附加信息记录到此日志中，如相关ID或请求处理时间。库非常精简，包含在应用程序之后，所有都是开箱即用的。

【墙裂推荐】

专注于「开发者」综合成长的深度星球
限时优惠进行中

最近热门内容回顾   #技术人系列



下方二维码关注我

互联网草根，坚持分享技术、创业、产品等心得和总结~



点击“阅读原文”，领取 2020 年最新免费技术资料大全

↓↓↓ 
浏览 33
点赞
评论
收藏
分享

手机扫一扫分享

举报
评论
图片
表情
推荐
15种时间序列预测方法总结(包含多种方法代码实现)
向AI转型的程序员都关注了这个号👇👇👇在这篇文章中，我们将深入探讨时间序列预测的基本概念和方法。我们将首先介绍单元预测和多元预测的概念，然后详细介绍各种深度学习和传统机器学习方法如何应用于时间序列预测，包括循环神经网络（RNN）、一维卷积神经网络（1D-CNN）、Transformer、自回归模型（
机器学习AI算法工程
0
学一学 Spring Boot 3.x
在 Java 后端开发领域，功能强大的 Spring 开源框架不仅是首选，也是事实上的标准。但由于 Spring 存在配置烦琐、部署不易、依赖管理困难等问题，因此基于 Spring 的快速开发框架 Spring Boot 应运而生，它能大大简化 Spring 应用程序的配置和部署过程。2018 年，
小哈学Java
0
SpringBoot 实现图片防盗链功能
程序员的成长之路互联网/程序员/技术/资料共享 关注阅读本文大概需要 4 分钟。来自：blog.csdn.net/weixin_46157208/article/details/138051737前言出于安全考虑，我们需要后端返回的图片只允许在某个网站内展示，不想被爬虫拿到图片地
程序员的成长之路
0
有意思！一个关于 Spring 历史的在线小游戏
发现 Spring One 的官网上有个好玩的彩蛋，分享给大家！进到Spring One的官网，可以看到右下角有个类似马里奥游戏中的金币图标。点击该金币之后，会打开一个新的页面，进入下面这样一个名为：The History Of Spring 的在线小游戏你可以使用上下左右的方向键来控制Spring
公众号程序猿DD
1
一站式解决方案：基于 Arthas 实现服务发现和权限控制
来源：juejin.cn/post/7281849496983994383👉 欢迎加入小哈的星球 ，你将获得: 专属的项目实战 / Java 学习路线 / 一对一提问 / 学习打卡 /  赠书福利全栈前后端分离博客项目 2.0 版本完结啦， 演示链接
小哈学Java
0
用 Shader 实现旗帜飘扬动画效果
我觉得对于刚入门 3D 编程的朋友来说，如果能够完成代码创建模型数据->创建材质->编写Shader动画这一系列，想必会有满满的成就感。今天就用 Cocos Creator 的 utils.MeshUtils.createMesh 接口，带大家感受一下这个流程。这个流程不仅可以用于新手学
COCOS
2
SpringBoot+Minio实现上传凭证、分片上传、秒传和断点续传
关注我们,设为星标,每天7:40不见不散,架构路上与您共享回复架构师获取资源大家好，我是你们的朋友架构君，一个会写代码吟诗的架构师。Spring Boot整合Minio后，前端的文件上传有两种方式：1、文件上传到后端，由后端保存到Minio这种方式好处是完全由后端集中管理，可以很好的做到、身份验证、
Java架构师社区
0
超越原生，散点图实现华夫饼图
之前我们介绍过了如何使用新卡片图实现华夫饼图。参考：超越原生，PowerBI 华夫饼图实现但是利用卡片图实现的华夫饼图有一些缺点，形状之间的大小跟间距不太好把握，而且有时形状大一点的话显示就会不正常，需要做出二次调整。今天给大家介绍一种原生视觉对象生成华夫饼图的更佳方案，既简单又美观。上图是利用散点
PowerBI战友联盟
2
Spring Boot + flowable 快速实现工作流
关注我们,设为星标,每天7:40不见不散,架构路上与您共享回复架构师获取资源大家好，我是你们的朋友架构君，一个会写代码吟诗的架构师。来源：blog.csdn.net/zhan107876/article/details/120815560总览一、flowable-ui部署运行二、绘制流程图绘图细节：
Java架构师社区
0
实现订单30分钟自动取消的策略
原文：juejin.cn/post/7285167401821798400简介在电商和其他涉及到在线支付的应用中，通常需要实现一个功能：如果用户在生成订单后的一定时间内未完成支付，系统将自动取消该订单。本文将详细介绍基于Spring Boot框架实现订单30分钟内未支付自动取消的几种方案，并提供实例
JAVA乐园
0
点赞
评论
收藏
分享

手机扫一扫分享

举报