Spring Boot应用程序接入ELK-003

Spring Boot应用程序接入ELK

一、项目依赖集成

在将Spring Boot应用程序接入ELK日志搜索引擎时，首先要在项目中集成相关依赖：

（一）Logstash依赖

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

logstash - logback - encoder依赖用于将日志格式化为适合Logstash处理的格式，方便后续日志的收集与解析。

（二）Kafka相关依赖

<dependency><groupId>com.github.danielwegener</groupId><artifactId>logback-kafka-appender</artifactId><version>0.2.0-RC1</version>
</dependency>
<dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId><version>2.1.12.RELEASE</version>
</dependency>

logback - kafka - appender用于在Logback中配置Kafka相关的日志追加器，实现将日志发送到Kafka消息队列。spring - kafka则是Spring Boot与Kafka集成的核心依赖，提供了对Kafka操作的支持 。

二、Logback配置Kafka

在logback.xml文件中进行Kafka相关配置：

<conversionRule conversionWord="nanotime" converterClass="com.test.sunchd.log.NanoTimeConverter"/>
<define name="localIp" class="com.test.sunchd.log.LogIpProperty"/>
<appender name="KafkaAppender" class="com.github.danielwegener.logback.kafka.KafkaAppender"><encoder charset="UTF-8" class="net.logstash.logback.encoder.LoggingEventCompositeJsonEncoder"><providers><pattern><pattern>{"timestamp": "%d{yyyy-MM-dd HH:mm:ss.SSS}","name": "%logger","level": "%level","ip": "${localIp}","service": "${applicationName}","pid": "${PID:-}","class": "%logger","method": "%method","thread": "%thread","traceId": "%X{traceId:-}","spanId": "%X{spanId:-}","parent": "%X{X - B3 - ParentSpanId:-}","message": "[%level] [${localIp}] [${applicationName}] [%thread] [%logger{36}] [%X{traceId:-}] [%X{spanId:-}] [%X{X - B3 - ParentSpanId:-}] %message"}</pattern></pattern></providers></encoder><keyingStrategy class="com.github.danielwegener.logback.kafka.keying.HostnameKeyingStrategy"/><deliveryStrategy class="com.github.danielwegener.logback.kafka.delivery.AsynchronousDeliveryStrategy"/><producerConfig>bootstrap.servers=10.190.107.136:9092</producerConfig><topic>logs</topic><producerConfig>acks=0</producerConfig><producerConfig>linger.ms=1000</producerConfig><producerConfig>max.block.ms=0</producerConfig><producerConfig>client.id=0</producerConfig>
</appender>
<appender name="FILE-INFO-ASYNC" class="ch.qos.logback.classic.AsyncAppender"><discardingThreshold>0</discardingThreshold><queueSize>512</queueSize><appender-ref ref="FILE-INFO"/>
</appender>
<appender name="FILE-ERROR-ASYNC" class="ch.qos.logback.classic.AsyncAppender"><discardingThreshold>0</discardingThreshold><queueSize>512</queueSize><appender-ref ref="FILE-ERROR"/>
</appender>
<logger name="com.test.cop" level="DEBUG" additivity="false"><appender-ref ref="STDOUT"/><appender-ref ref="FILE-INFO-ASYNC"/><appender-ref ref="FILE-ERROR-ASYNC"/><appender-ref ref="KafkaAppender"/>
</logger>

这里配置了KafkaAppender，使用LoggingEventCompositeJsonEncoder对日志进行JSON格式编码，设置了日志包含的各种字段。keyingStrategy指定按主机名进行键控，deliveryStrategy采用异步发送策略。producerConfig指定了Kafka的服务器地址，topic设置为logs 。

三、关键类介绍

（一）NanoClockUtils类

package com.test.sunchd.log.utils;import java.io.Serializable;
import java.time.Clock;
import java.time.Instant;
import java.time.ZoneId;
import java.time.ZoneOffset;
import static java.util.Objects.requireNonNull;public final class NanoClockUtils extends Clock implements Serializable {private static final long serialVersionUID = 1L;private static final long EPOCH_NANOS = System.currentTimeMillis() * 1000000L;private static final long NANO_START = System.nanoTime();private static final long OFFSET_NANOS = EPOCH_NANOS - NANO_START;private static final NanoClockUtils UTC_INSTANCE = new NanoClockUtils(ZoneOffset.UTC);private static final NanoClockUtils DEFAULT_INSTANCE = new NanoClockUtils(ZoneId.systemDefault());public static final long NANOS_PER_SECOND = 1_000_000_000L;private final ZoneId _zone;private NanoClockUtils(final ZoneId zone) {_zone = requireNonNull(zone, "zone");}@Overridepublic ZoneId getZone() {return _zone;}@Overridepublic NanoClockUtils withZone(final ZoneId zone) {return zone.equals(_zone)? this : new NanoClockUtils(zone);}@Overridepublic long millis() {return System.currentTimeMillis();}public long nanos() {return System.nanoTime() + OFFSET_NANOS;}public String nanosStr() {return String.valueOf(nanos());}@Overridepublic Instant instant() {final long now = nanos();return Instant.ofEpochSecond(now / NANOS_PER_SECOND, now % NANOS_PER_SECOND);}@Overridepublic int hashCode() {return _zone.hashCode() + 11;}@Overridepublic boolean equals(final Object obj) {return obj == this || (obj instanceof NanoClockUtils && ((NanoClockUtils) obj)._zone.equals(_zone));}@Overridepublic String toString() {return "NanoClock[" + _zone + "]";}public static NanoClockUtils system(final ZoneId zone) {return new NanoClockUtils(zone);}public static NanoClockUtils systemUTC() {return UTC_INSTANCE;}public static NanoClockUtils systemDefaultZone() {return DEFAULT_INSTANCE;}
}

该类继承自Clock并实现Serializable接口，用于处理纳秒级时间相关操作。定义了一些静态常量用于时间计算的偏移等，提供了获取纳秒时间、转换为Instant等方法，还可以根据不同时区创建实例 。

（二）NanoTimeConverter类

package com.test.sunchd.log;import ch.qos.logback.classic.pattern.ClassicConverter;
import ch.qos.logback.classic.spi.ILoggingEvent;
import com.test.sunchd.log.utils.NanoClockUtils;
import org.apache.commons.lang3.time.DateFormatUtils;
import java.util.Date;public class NanoTimeConverter extends ClassicConverter {@Overridepublic String convert(ILoggingEvent event) {return NanoClockUtils.systemDefaultZone().nanosStr();}public static void main(String[] args) {String s = NanoClockUtils.systemDefaultZone().nanosStr().substring(0, 13);final long l = Long.valueOf(s);System.out.println(DateFormatUtils.format(new Date(l), "yyyyMMdd HHmmss"));}
}

该类继承自ClassicConverter，在日志格式化时用于获取纳秒时间字符串。convert方法获取系统默认时区下的纳秒时间字符串，main方法用于测试时间字符串的截取和转换 。

（三）LogIpProperty类

package com.test.sunchd.log;import ch.qos.logback.core.PropertyDefinerBase;
import com.test.sunchd.log.utils.NetWorkUtil;public class LogIpProperty extends PropertyDefinerBase {private static final String ip = NetWorkUtil.getHostAddress();@Overridepublic String getPropertyValue() {return ip;}
}

该类继承自PropertyDefinerBase，用于获取本机IP地址并在日志配置中作为属性值使用。通过调用NetWorkUtil.getHostAddress()获取IP地址，并在getPropertyValue方法中返回 。

（四）NetWorkUtil类

package com.test.sunchd.log.utils;import org.apache.commons.lang3.StringUtils;
import java.net.InetAddress;
import java.net.Inet4Address;
import java.net.NetworkInterface;
import java.net.SocketException;
import java.util.Enumeration;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;public class NetWorkUtil {public static final String NETWORK_DOCKER = "docker";public static final String NETWORK_BRIDGE = "br";public static final String NETWORK_LINUX = "ens";public static String getHostAddress() {return getLocalHostExactAddress().getHostAddress();}public static InetAddress getLocalHostExactAddress() {try {InetAddress candidateAddress = null;Enumeration<NetworkInterface> networkInterfaces = NetworkInterface.getNetworkInterfaces();while (networkInterfaces.hasMoreElements()) {NetworkInterface networkInterface = networkInterfaces.nextElement();if (networkInterface.isVirtual() || networkInterface.isLoopback() || networkInterface.isPointToPoint()) {continue;}Enumeration<InetAddress> inetAddresses = networkInterface.getInetAddresses();while (inetAddresses.hasMoreElements()) {InetAddress inetAddress = inetAddresses.nextElement();if (inetAddress instanceof Inet4Address) {if (candidateAddress == null) {candidateAddress = inetAddress;} else {// 对每个网卡下的多个IP，再要一个个遍历，找到自己需要的for (InetAddress ipAddress : networkInterface.getInetAddresses()) {if (ipAddress.isLoopbackAddress()) {continue;}if (ipAddress.isAnyLocalAddress()) {continue;}// 如果是IPv4地址，就是它了，就是我们要找的if (ipAddress instanceof Inet4Address) {candidateAddress = ipAddress;break;}}}}}}// 如果与去loopback环境之外无其它地址了，就挑回环地址吧if (candidateAddress == null) {candidateAddress = InetAddress.getLocalHost();}return candidateAddress;} catch (Exception e) {throw new RuntimeException(e);}}public static Set<InetAddress> getLocalIpAddress() {Set<InetAddress> result = new HashSet<>();try {Enumeration<NetworkInterface> networkInterfaces = NetworkInterface.getNetworkInterfaces();while (networkInterfaces.hasMoreElements()) {NetworkInterface networkInterface = networkInterfaces.nextElement();Enumeration<InetAddress> inetAddresses = networkInterface.getInetAddresses();while (inetAddresses.hasMoreElements()) {InetAddress inetAddress = inetAddresses.nextElement();if (inetAddress instanceof Inet4Address) {result.add(inetAddress);}}}} catch (Exception e) {throw new RuntimeException(e);}return result;}public static List<InetAddress> getLocalIpAddressFromNetworkInterfaces() throws SocketException {List<InetAddress> addresses = new ArrayList<>();Enumeration<NetworkInterface> networkInterfaces = NetworkInterface.getNetworkInterfaces();if (networkInterfaces == null) {return addresses;}while (networkInterfaces.hasMoreElements()) {NetworkInterface networkInterface = networkInterfaces.nextElement();Enumeration<InetAddress> inetAddresses = networkInterface.getInetAddresses();while (inetAddresses.hasMoreElements()) {InetAddress inetAddress = inetAddresses.nextElement();if (inetAddress instanceof Inet4Address) {addresses.add(inetAddress);}}}return addresses;}public static boolean isBridgeNetwork(NetworkInterface ni) {if (ni == null) {return false;}return StringUtils.startsWithIgnoreCase(ni.getName(), NETWORK_BRIDGE);}public static boolean isDockerNetwork(NetworkInterface ni) {if (ni == null) {return false;}return StringUtils.startsWithIgnoreCase(ni.getName(), NETWORK_DOCKER);}
}

该类提供了一系列网络相关工具方法。定义了一些网络类型的标识常量，如docker、br、ens。getHostAddress和getLocalHostExactAddress方法用于获取本机精确的IP地址，会遍历网络接口排除虚拟、环回等地址。getLocalIpAddress和getLocalIpAddressFromNetworkInterfaces用于获取本机所有IPv4地址，isBridgeNetwork和isDockerNetwork用于判断网络接口是否属于特定网络类型 。

四、后续流程

在完成上述配置后，Kafka的主题设置为logs，logstash作为消费者从kafka的logs主题获取日志数据，并将其推送至elasticsearch（es）。最终，我们可以在kibana中根据配置的索引模式查询和分析这些日志，实现对Spring Boot应用程序日志的高效管理与分析。