搜索

实现 Spring Boot Logstash 日志收集

泥瓦匠BYSocket

共 10707字，需浏览 22分钟

 ·

2020-11-08 16:16

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

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

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

在本文中，我将介绍我的日志库，专门用于 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 年最新免费技术资料大全

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

手机扫一扫分享

分享
举报
评论
图片
表情
推荐
日志收集神器 Logstash，闪亮登场～
Java技术栈
0
Spring Boot 搭建 ELK，成海量日志收集（超详细）
全栈架构社区
0
SpringBoot整合ELK实现日志收集
java1234
0
Go实现海量日志收集系统
马哥Linux运维
0
Logstash开源日志管理
Logstash是一个应用程序日志、事件的传输、处理、管理和搜索的平台。你可以用它来统一对应用程序日志进行收集管理，提供Web接口用于查询和统计。Logstash现在是ElasticSearch家族成
Logstash开源日志管理
0
实战 | 使用Spring Boot + Elasticsearch + Logstash 实现图书查询检索服务
架构师精进
0
Spring Boot 参考指南（日志记录）
程序员闪充宝
0
Spring Boot 实现登录拦截器
Java后端编程
0
ELK 处理 Spring Boot 日志，妙！
JAVA葵花宝典
0
Spring Boot 使用 AOP 记录日志
武培轩
0
点赞
评论
收藏
分享

手机扫一扫分享

分享
举报