apache / apache/incubator-seata
optimize:优化seata-server日志采集配置
- Dominant language
- Java
- Stars
- 26k
- Forks
- 8.8k
- Avg merge
- 1d 8h
- Merged PRs (30d)
- 4
Description
## Why you need it?
Is your feature request related to a problem? Please describe in details
目前将seata-server日志采集到Kafka或者logstash需要手动修改logbcak-spring.xml,当需要采集到kafka时,kafka生产者的一些配置也需要在kafka-appender.xml,配置较为分散,不好统一管理,可以考虑将配置统一收到logging.extend管理。
## How it could be?
A clear and concise description of what you want to happen. You can explain more about input of the feature, and output of it.
配置统一管理后,如果需要开启或关闭日志采集功能,不需要在手动修改logbcak-spring.xml,只需修改application.yml重新打包即可,也可以通过java -jar -Dxxx 开启或关闭日志上报功能,如果接入配置中心,只需修改配置中心配置,在重启seata-server服务即可,配置管理将十分方便。
## Other related information
Add any other context or screenshots about the feature request here.
可以参照spring boot整合logback原理实现,伪代码如下:
```java
package io.seata.server.logging.logback;
import ch.qos.logback.classic.Logger;
import ch.qos.logback.classic.LoggerContext;
import ch.qos.logback.classic.encoder.PatternLayoutEncoder;
import com.github.danielwegener.logback.kafka.KafkaAppender;
import io.seata.common.util.StringUtils;
import io.seata.server.logging.logback.appender.EnhancedLogstashEncoder;
import net.logstash.logback.appender.LogstashTcpSocketAppender;
import org.slf4j.ILoggerFactory;
import org.slf4j.impl.StaticLoggerBinder;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.context.event.ApplicationEnvironmentPreparedEvent;
import org.springframework.boot.context.logging.LoggingApplicationListener;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.event.GenericApplicationListener;
import org.springframework.core.ResolvableType;
import org.springframework.core.env.ConfigurableEnvironment;
import java.util.Objects;
/**
* @author wlx
* @date 2022/5/27 11:18 下午
*/
public class SeataLogbackLoggingExtendListener implements GenericApplicationListener {
private static final String ROOT = "ROOT";
private static final String KAFKA_TOPIC = "logging.extend.kafka-appender.topic";
private static final String KAFKA_BOOTSTRAP_SERVERS = "logging.extend.kafka-appender.bootstrap-servers";
private static final String LOGSTASH_DESTINATION = "logging.extend.logstash-appender.destination";
private static final String APP_NAME = "spring.application.name";
private static final String DEFAULT_APP_NAME = "seata-server";
private static final String LOGSTASH = "LOGSTASH";
private static final String KAFKA = "KAFKA";
private static final String KAFKA_KAFKA_PATTERN = "{\n" +
" \"@timestamp\": \"%d{yyyy-MM-dd HH:mm:ss.SSS}\",\n" +
" \"level\":\"%p\",\n" +
" \"app_name\":\"${APPLICATION_NAME:-seata-server}\",\n" +
" \"PORT\": ${RPC_PORT:-0},\n" +
" \"thread_name\": \"%t\",\n" +
" \"logger_name\": \"%logger\",\n" +
" \"X-TX-XID\": \"%X{X-TX-XID:-}\",\n" +
" \"X-TX-BRANCH-ID\": \"%X{X-TX-BRANCH-ID:-}\",\n" +
" \"message\": \"%m\",\n" +
" \"stack_trace\": \"%wex\"\n" +
"}";
@Override
public boolean supportsSourceType(Class sourceType) {
if (Objects.isNull(sourceType)) {
return false;
} else {
return SpringApplication.class.isAssignableFrom(sourceType)
|| ApplicationContext.class.isAssignableFrom(sourceType);
}
}
@Override
public boolean supportsEventType(ResolvableType eventType) {
Class typeRawClass = eventType.getRawClass();
if (Objects.isNull(typeRawClass)) {
return false;
} else {
return ApplicationEnvironmentPreparedEvent.class.isAssignableFrom(typeRawClass);
}
}
@Override
public void onApplicationEvent(ApplicationEvent event) {
if (event instanceof ApplicationEnvironmentPreparedEvent) {
ILoggerFactory loggerFactory = StaticLoggerBinder.getSingleton().getLoggerFactory();
LoggerContext loggerContext = (LoggerContext) loggerFactory;
ConfigurableEnvironment environment = ((ApplicationEnvironmentPreparedEvent) event).getEnvironment();
// 添加LogstashAppender
appendLogstashAppender(loggerContext, environment);
// 添加KafkaAppender
appendKafkaAppender(loggerContext, environment);
}
}
@Override
public int getOrder() {
return LoggingApplicationListener.DEFAULT_ORDER + 2;
}
@SuppressWarnings({"rawtypes", "unchecked"})
private void appendKafkaAppender(LoggerContext loggerContext, ConfigurableEnvironment environment) {
String kafkaBootstrapServer = environment.getProperty(KAFKA_BOOTSTRAP_SERVERS);
String kafkaTopic = environment.getProperty(KAFKA_TOPIC);
if (StringUtils.isNullOrEmpty(kafkaBootstrapServer) || StringUtils.isNullOrEmpty(kafkaTopic)) {
return;
}
Logger root = loggerContext.getLogger(ROOT);
KafkaAppender kafkaAppender = new KafkaAppender();
kafkaAppender.setContext(loggerContext);
kafkaAppender.setName(KAFKA);
kafkaAppender.setTopic(kafkaTopic);
kafkaAppender.setKeyingStrategy(new com.github.danielwegener.logback.kafka.keying.NoKeyKeyingStrategy());
kafkaAppender.setDeliveryStrategy(new com.github.danielwegener.logback.kafka.delivery.AsynchronousDeliveryStrategy());
kafkaAppender.addProducerConfigValue("bootstrap.servers", KAFKA_BOOTSTRAP_SERVERS);
kafkaAppender.addProducerConfigValue("acks", 0);
kafkaAppender.addProducerConfigValue("linger.ms", 1000);
kafkaAppender.addProducerConfigValue("max.block.ms", 0);
PatternLayoutEncoder encoder = new PatternLayoutEncoder();
encoder.setPattern(KAFKA_KAFKA_PATTERN);
encoder.setContext(loggerContext);
kafkaAppender.setEncoder(encoder);
encoder.start();
kafkaAppender.start();
root.addAppender(kafkaAppender);
}
private void appendLogstashAppender(LoggerContext loggerContext, ConfigurableEnvironment environment) {
String logstashDestination = environment.getProperty(LOGSTASH_DESTINATION);
if (StringUtils.isNullOrEmpty(logstashDestination)) {
return;
}
Logger root = loggerContext.getLogger(ROOT);
LogstashTcpSocketAppender logstashTcpSocketAppender = new LogstashTcpSocketAppender();
logstashTcpSocketAppender.setName(LOGSTASH);
logstashTcpSocketAppender.addDestination(logstashDestination);
logstashTcpSocketAppender.setContext(loggerContext);
EnhancedLogstashEncoder encoder = new EnhancedLogstashEncoder();
String appName = environment.getProperty(APP_NAME, DEFAULT_APP_NAME);
encoder.setCustomFields(appName);
encoder.setExcludeProvider("net.logstash.logback.composite.LogstashVersionJsonProvider");
encoder.setExcludeProvider("net.logstash.logback.composite.loggingevent.JsonMessageJsonProvider");
encoder.setExcludeProvider("net.logstash.logback.composite.loggingevent.TagsJsonProvider");
encoder.setExcludeProvider("net.logstash.logback.composite.loggingevent.LogstashMarkersJsonProvider");
encoder.setExcludeProvider("net.logstash.logback.composite.loggingevent.ArgumentsJsonProvider");
encoder.setContext(loggerContext);
encoder.start();
logstashTcpSocketAppender.setEncoder(encoder);
logstashTcpSocketAppender.start();
root.addAppender(logstashTcpSocketAppender);
}
}
```
Contributor guide
Research direction
Start by comparing logback-spring.xml and kafka-appender.xml with the existing logging configuration, then review the proposed Spring Boot listener and its application.yml properties. Done means Kafka and Logstash collection settings are managed through logging.extend, with collection controllable through application.yml, -D properties, or a configuration center as described.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java, kafka, spring-boot
- Domain
- backend, observability-sre
- Issue type
- Feature
- Difficulty
- 5/5
- Estimated time
- Over a week
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 30/100