Spring Boot + Log4j2 + Kafka (ELK-ready)

Giai đoạn 1: Kiến thức cơ bản

1. Log là gì?

Log (viết tắt của logging) là việc ghi lại các thông tin trong quá trình chạy ứng dụng — như lỗi, cảnh báo, hoặc thông tin để debug. Mục đích chính là giúp lập trình viên theo dõi, phân tích và xử lý sự cố.

Các cấp độ log thông dụng trong Log4j2:

• TRACE: Chi tiết, khi debug sâu
• DEBUG: Gỡ lỗi, khi phát triển
• INFO: Thông tin chung về tiến trình xử lý của hệ thống (bắt đầu xử lý request)
• WARN: Cảnh báo, chưa gây lỗi
• ERROR: Lỗi xảy ra, hệ thống vẫn chạy được
• FATAL: Có thể khiến hệ thống dừng hoạt động

Thực tế trên product chỉ dùng INFO trở lên để đẩy lên Kafka, nếu log lỗi không đẩy được lên Kafka thì sẽ Level.ERROR

Log được lưu ở đâu?

• Console (màn hình terminal)
• File (ghi vào file trên ổ đĩa)
• Rolling File (file log tự động chia nhỏ theo ngày/dung lượng)
• Remote servers (gửi log đến hệ thống khác như Logstash, Elasticsearch)

Format của 1 dòng log:

[Thời gian] [Mức log] [Tên class] - Thông điệp log

Có thể custom lại được bằng PatternLayout

Ví dụ:

StackTraceElement ste = (new Throwable()).getStackTrace()[1];
String className = ste.getClassName();
Logger subLogError = LogManager.getLogger(clsName);
subLogError.log(Level.ERROR, ste.getMethodName() + " " + ste.getLineNumber() + " - " + String.valueOf(e));

StackTraceElement ste = (new Throwable()).getStackTrace()[1];

• Tạo mới một Throwable để lấy stack trace
• .getStrackTrace() mảng các lời gọi method hiện tại
• [1] lấy caller - dòng gọi đến đoạn này
• ste chứa: tên class, method, số dòng, tên file (nếu có)

Hiểu đơn giản thì [0] là chính nó còn [1] là cái mà gọi [0] trước đó.

Kết quả:
yyyy-MM-dd hh:mm:ss,ms ERROR [path from source root of class - tên class Log] [Trace ID] [method caller] [dòng thực hiện gọi [0] của caller] - e

LogManager.getLogger(clsName)

• Lúc này log theo class name như trên
• Hoặc bạn có thể gọi theo tên được cấu hình trong file: log4j2.xml, log4j2.properties hoặc log4j2.json.
• Nếu bạn dùng Logback (Spring Boot mặc định) → nó tìm trong logback.xml hoặc application.properties.

Phần cấu hình gọi theo tên này sẽ tìm hiểu ở phần sau.

2. Spring Boot logging

Spring Boot mặc định dùng Logback để log (thông qua thư viện SLF4J).

private static final Logger logger = LoggerFactory.getLogger(MyClass.class); logger.info("Hello world!");

Dù bạn dùng SLF4J, nhưng hệ thống đằng sau sẽ là Logback nếu bạn không cấu hình lại.

SLF4J (Simple Logging Facade for Java) là một lớp trung gian (facade) — giống như cái “adapter” — để bạn viết code logging không phụ thuộc vào thư viện cụ thể như Logback hay Log4j2.

Bạn viết code logger.info("...") với SLF4J 👉 Còn thực tế log sẽ do Logback / Log4j2 thực hiện

Bạn có thể thay Logback → Log4j2 → Log4j → JDK Logging mà không sửa dòng code nào cả! → Vì SLF4J chỉ là một giao diện (interface)

Cấu hình Log4j2 thay thế Logback

Bước 1: Xóa Logback

<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter</artifactId> <exclusions> <exclusion> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-logging</artifactId> <!-- Logback --> </exclusion> </exclusions>
</dependency>

Bước 2: Thêm Log4j2

<!-- Log4j2 dependencies -->
<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-log4j2</artifactId>
</dependency>

Bước 3: Tạo file log4j2.xml trong src/main/resources

<?xml version="1.0" encoding="UTF-8"?>
<Configuration status="WARN"> <Appenders> <Console name="Console" target="SYSTEM_OUT"> <PatternLayout pattern="%d{yyyy-MM-dd HH:mm:ss} %-5level %logger{36} - %msg%n" /> </Console> </Appenders> <Loggers> <Root level="info"> <AppenderRef ref="Console" /> </Root> </Loggers>
</Configuration>

1. Hiệu năng thực tế

2. Cấu hình nâng cao

3. Logging không tạo rác (Garbage-free logging):

• Logback: Mỗi lần log là tạo object mới → gây GC (garbage collection).
• Log4j2: Cho phép dùng StringBuilder và ThreadLocal để giảm tạo rác → giảm GC → tăng performance

Giai đoạn 2: Ghi log vào Kafka

3. Kafka là gì?

Kafka là một hệ thống hàng đợi (message queue) phân tán, được thiết kế để:

• Gửi - Nhận - Lưu trữ luồng dữ liệu lớn
• Xử lý real-time hoặc bất đồng bộ
• Rất bền vững- nhanh - linh hoạt

Thành phần chính:

• Producer: Gửi dữ liệu (log, event, message...) vào Kafka
• Topic: Nơi chứa dữ liệu (giống như "kênh", "folder" lưu tin nhắn)
• Consumer: Nhận dữ liệu từ Topic để xử lý (ghi DB, gửi email, tính toán...)
• Broker: Một Kafka server. Kafka cluster = nhiều broker ghép lại
• Zookeeper: Điều phối Kafka cluster (có thể dùng Kafka Raft thay thế mới hơn)

Dòng chảy dữ liệu:

Producer ---> Kafka Topic ---> Consumer
Gửi log Lưu Đọc log

<dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka-clients</artifactId> <version>3.9.0</version>
</dependency>

Thư viện này là Kafka client thuần túy được phát triển bởi Apache. Nó cung cấp API để Java có thể kết nối với Kafka và thực hiện:

• Gửi message (Kafka Producer)
• Nhận message (Kafka Consumer)
• Cấu hình bootstrap.servers, acks, retries, serializer, ...

Log4j2 không thể tự mình gửi message lên Kafka nếu thiếu thư viện này. Vì vậy, log4j-kafka bên dưới để thực hiện gửi log.

Kafka Appender cho Log4j2
<dependency> <groupId>org.apache.logging.log4j</groupId> <artifactId>log4j-kafka</artifactId> <version>2.17.2</version> <!-- hoặc bản mới hơn -->
</dependency>

Mục đích: Đây là Log4j2 plugin giúp bạn khai báo Kafka như một appender trong file log4j2.xml.

<Kafka name="KafkaAppender" topic="app-logs"> <JsonLayout/> <Property name="bootstrap.servers">localhost:9092</Property>
</Kafka>

Trong dự án tôi làm đã không cần thêm log4j-kafka

Vậy tại sao vẫn log được vào Kafka mà không cần log4j-kafka?

log4j2.xml gọi đến 1 appender "tự viết", không cần dùng log4j-kafka.

Dự án đang sử dụng Log4j2 Kafka Appender gốc – tức là đang dùng log4j-kafka (dù không thấy khai báo trong pom.xml,

dependency: spring-boot-starter-log4j2 là gì?

<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-log4j2</artifactId> <version>3.3.7</version>
</dependency>

Spring Boot 3.3.7 cung cấp starter này để:

• Tự động cấu hình Log4j2 thay vì Logback (do bạn đã loại spring-boot-starter-logging)
• Đồng thời, nó bao gồm sẵn log4j-core, log4j-api, và quan trọng là:

Nó bao gồm luôn log4j-kafka như một optional module nếu bạn có cấu hình < Kafka > trong log4j2.xml.

Vậy log4j-kafka đến từ đâu?

Spring Boot 3.3.7 dùng Log4j 2.23.1. https://mvnrepository.com/artifact/org.springframework.boot/spring-boot-starter-log4j2/3.3.7

Module log4j-kafka là một phần của Log4j core project, và trong các bản Log4j gần đây (2.20+), nó được load tự động nếu:

• Bạn có file log4j2.xml có < Kafka >
• Bạn dùng spring-boot-starter-log4j2

➡️ Khi đó, Gradle hoặc Maven sẽ tự kéo về JAR log4j-kafka-2.23.1.jar như một transitive dependency.

Bạn có thể kiểm tra bằng cách:

1. Xem trong target/dependency hoặc External Libraries trong IDE
2. Tìm log4j-kafka-2.23.1.jar → Nếu có thì chính là lý do hoạt động được
3. Hoặc chạy:

mvn dependency:tree -Dincludes=log4j-kafka

Demo Log4j2 Kafka Appender + JSON Layout để log gửi lên Kafka theo định dạng JSON

<?xml version="1.0" encoding="UTF-8"?>
<Configuration status="WARN"> <Appenders> <!-- Console Appender --> <Console name="ConsoleLog" target="SYSTEM_OUT"> <PatternLayout pattern="%d{yyyy-MM-dd HH:mm:ss,SSS} %-5level %logger{36} [%t] %msg%n" /> </Console> <!-- Kafka Appender --> <Kafka name="KafkaLog" topic="demo-log-topic"> <PatternLayout pattern="%m" /> <Property name="bootstrap.servers">localhost:9092</Property> <Property name="request.timeout.ms">10000</Property> </Kafka> <!-- Async wrapper for Kafka --> <Async name="KafkaAsync" bufferSize="8192"> <AppenderRef ref="KafkaLog" /> </Async> </Appenders> <Loggers> <!-- Logger cho Kafka --> <Logger name="KafkaLogger" level="info" additivity="false"> <AppenderRef ref="KafkaAsync" /> </Logger> <!-- Root logger --> <Root level="info"> <AppenderRef ref="ConsoleLog" /> </Root> </Loggers>
</Configuration>

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger; public class DemoLogKafka { private static final Logger kafkaLogger = LogManager.getLogger("KafkaLogger"); private static final Logger defaultLogger = LogManager.getLogger(DemoLogKafka.class); public static void main(String[] args) { defaultLogger.info("This is a normal console log"); kafkaLogger.info("This is a log sent to Kafka topic"); }
}

• defaultLogger.info(...) → hiện ra console.
• kafkaLogger.info(...) → gửi Kafka topic demo-log-topic.

Bạn có thể phát triển thêm để log ra file, hoặc sử dụng JsonLayout

Tuy nhiên bạn có thể sử dụng ObjectMapper của Jackson để serialize đối tượng thành JSON và sau đó ghi log là cách rất phổ biến – đặc biệt khi bạn không dùng JsonLayout trong Log4j2 mà vẫn muốn log của bạn có định dạng JSON đẹp, dễ đọc và dễ phân tích.

import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger; public class KafkaLogExample { private static final Logger logger = LogManager.getLogger("KafkaLogger"); private static final ObjectMapper objectMapper = new ObjectMapper(); public static void main(String[] args) throws Exception { SampleLogObject obj = new SampleLogObject("order123", 1000000, "SUCCESS"); // Serialize object to JSON string String json = objectMapper.writeValueAsString(obj); // Send log message (which will be published to Kafka) logger.info(json); }
} class SampleLogObject { public String orderId; public int amount; public String status; public SampleLogObject(String orderId, int amount, String status) { this.orderId = orderId; this.amount = amount; this.status = status; }
}

Output log sẽ gửi lên Kafka như sau (giả sử bạn dùng %m trong PatternLayout):

{"orderId":"order123","amount":1000000,"status":"SUCCESS"}

Giai đoạn 3: ELK

Kafka → Logstash → Elasticsearch

• Kafka: đã nhận log từ Spring Boot (log JSON).
• Logstash: công cụ ETL trung gian.
• Elasticsearch: nơi lưu trữ log.
• Kibana (nếu muốn hiển thị).

Bước 1: Cài đặt Logstash

# Tải về Logstash (https://www.elastic.co/downloads/logstash)
# Hoặc dùng Docker:
docker run --name logstash -it --rm \ -v "$PWD/logstash.conf":/usr/share/logstash/pipeline/logstash.conf \ docker.elastic.co/logstash/logstash:8.12.0

Bước 2: Tạo file cấu hình logstash.conf

input { kafka { bootstrap_servers => "localhost:9092" topics => ["demo-log-topic"] codec => "json" # Vì log là chuỗi JSON group_id => "logstash-log-group" }
} filter { # Tuỳ chọn: Nếu trong log có trường "timestamp" bạn muốn dùng làm @timestamp date { match => ["timestamp", "ISO8601"] target => "@timestamp" ignore_failure => true } # Tuỳ chọn: Gắn thêm metadata nếu cần mutate { add_field => { "service" => "your-service-name" "env" => "dev" # hoặc staging, prod } }
} output { elasticsearch { hosts => ["http://localhost:9200"] index => "demo-log-index-%{+YYYY.MM.dd}" # Log theo ngày } stdout { codec => rubydebug # Xem log trên console khi debug }
}

Bước 3: Khởi động Logstash

bin/logstash -f logstash.conf

Bước 4: Mở Kibana (nếu có)

1. Tạo Index pattern: app-logs-*
2. Xem log realtime

Lưu ý:

• Log4j2 cần định dạng log là JSON khi gửi vào Kafka (có thể dùng JsonLayout).
• Nếu dùng Docker cho ELK stack: bạn cần chỉnh lại địa chỉ IP (đừng dùng localhost).
• Với môi trường production, bạn nên tách log theo serviceName, logLevel, env, v.v.