이번 글에서는 애플리케이션에서 발생한 로그 이벤트를 아파치 카프카를 통하여 Publish 하는 방법에 대하여 다루어본다.
logback-kafka-appender
💬 Log4j는 KafkaAppender를 직접 지원한다.
-
Logback의 경우 오픈소스로 제공되는 logback-kafka-appender를 사용한다.
-
logback-kafka-appender 사용을 위해서는 Logback version 1.2 이상이 필요합니다.
-
오픈소스 링크 : https://github.com/danielwegener/logback-kafka-appender
dependency 추가하기
builde.gradle
// kafka 관련 설정
implementation 'org.springframework.kafka:spring-kafka'
testImplementation 'org.springframework.kafka:spring-kafka-test'
// logback-kafka-appender 관련 설정
implementation "com.github.danielwegener:logback-kafka-appender:0.1.0"
implementation 'net.logstash.logback:logstash-logback-encoder:5.2'logback.xml
: 설정 파일
<?xml version="1.0" encoding="UTF-8"?>
<configuration>
<appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
<layout class="ch.qos.logback.classic.PatternLayout">
<!-- <Pattern>%d{yyyy-MM-dd'T'HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</Pattern>-->
<Pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</Pattern>
<!-- <Pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</Pattern>-->
</layout>
</appender>
<!-- Layout 1 : kafkaAppender -->
<appender name="kafkaAppender" class="com.github.danielwegener.logback.kafka.KafkaAppender">
<encoder class="com.github.danielwegener.logback.kafka.encoding.LayoutKafkaMessageEncoder">
<layout class="ch.qos.logback.classic.PatternLayout">
<Pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</Pattern>
<!-- <pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern>-->
</layout>
</encoder>
<topic>search_log</topic>
<producerConfig>bootstrap.servers=localhost:9092</producerConfig>
</appender>
<!-- Async 한 KafkaAppender 추가 -->
<appender name="ASYNC" class="ch.qos.logback.classic.AsyncAppender">
<appender-ref ref="kafkaAppender"/>
</appender>
<!-- logger -->
<logger name="org.apache.kafka" level="ERROR"/>
<logger name="com.minsub.java.logger.kafka" level="DEBUG">
<!-- Layout 1 -->
<appender-ref ref="kafkaAppender"/>
<!-- Layout 2 -->
<!-- <appender-ref ref="logstashKafkaAppender" />-->
</logger>
<root level="DEBUG">
<appender-ref ref="STDOUT"/>
</root>
<root level="info">
<appender-ref ref="ASYNC" />
</root>
</configuration>-
appender로 KafkaAppender 지정
-
encoding 방식으로 PatternLayoutEncoder 사용
-
KafkaAppender의 구성
- topic 태그 : 발행할 토픽에 대한 값이 들어있어야 함
- producerConfig 태그 : 카프카 프로듀서의 properties를 설정을 위한 'key=value'값이 들어감
-
카프카 브로커의 bootstrap-server은 필수적인 요소라 꼭 추가해야 함
-
나머지 설정들은 CONFLUENT Producer Configurations을 참고
-
KafkaAppender의 appender-ref : produce 과정이 실패했을 때 출력할 appender를 지정
-
kafkaAppender를 가진 kafkaLogger 만들기
- additivity="false" : kafkaLogger에서 출력하는 로그 이벤트는 상위 로거로 전파되지 않도록 설정
-
AsyncAppender 생성 : Kafka의 메타데이터의 교환이 일어날 때, 블록킹 방식을 피할 수 있음
- 메타 데이터 교환시 로그 이벤트는 브로커로 전달되지 못함
- logging context는 최대 브로커에서 지정한 'metadata.max.age.ms' 시간만큼은 블록킹 상태로 머물게 되기 때문
Kafka 준비
zookeeper application과 kafka application을 실행한다.
참고로, 파일은 아래의 링크를 통해 설치하였다.
Zookeeper Application 실행
.\zookeeper-server-start.bat ..\..\config\zookeeper.propertiesKafka server 실행
.\kafka-server-start.bat ..\..\config\server.properties(broker) topic 생성하기
.\kafka-topics.bat --create --bootstrap-server localhost:9092 --topic 토픽이름(producer) 생성한 topic에 message 전송하기
.\kafka-console-producer.bat --broker-list localhost:9092 --topic 토픽이름(consumer) topic에 있는 message 가져오기
.\kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic 토픽이름 --from-beginningLog 생성해보기
@RestController
@RequestMapping("/test")
public class TestController {
private static final Logger kafkaLogger = LoggerFactory.getLogger("kafkaLogger");
@GetMapping("/hello")
public void hello(@RequestParam(value = "name") String name) {
kafkaLogger.info("name:{}", name);
}
}- 아래 url로 http request를 보내면 name에 들어가는 deingvelop이라는 파라미터가 로깅됨
localhost:8080/test/hello?name=deingvelop참고자료
