Đưa log vào: Beats, Fluent Bit, Logstash, Kafka

Ta cần một "shipper" trên mỗi VM, và có thể thêm một tầng xử lý ở giữa. Bài này đi từ đơn giản nhất tới bản có Kafka. Phần chạy trên Kubernetes (DaemonSet, Fluent Bit đọc /var/log/containers) đã có ở 17.11.

1. Bài toán

vm-app-1..3   /var/log/shop/*.json        JSON ECS từ Spring Boot (bài 19.7)
vm-gw-1       /var/log/nginx/access.log   text: 10.0.0.5 - - [14/Oct/2026:08:15:02 +0000] "POST /orders HTTP/1.1" 502 157 ...
app cũ        /opt/legacy/app.log         text nhiều dòng, stack trace Java

2. Vấn đề

• Nginx và app cũ là text → cần parse thành field.
• Stack trace 40 dòng → nếu đọc từng dòng sẽ thành 40 document.
• ES bảo trì 30 phút → shipper phải giữ lại, không được bỏ.
• Thiếu service.name, env → không lọc được trong Kibana.

3. Giải pháp: chọn shipper

Quy tắc: trên máy app chạy agent nhẹ (Filebeat hoặc Fluent Bit). Logstash chạy riêng, không cài lên máy app.

4. Code từng bước

Bước 1: Filebeat trên vm-app-2 (/etc/filebeat/filebeat.yml)

filebeat.inputs:
  - type: filestream
    id: shop-payment
    paths: ["/var/log/shop/payment*.json"]
    parsers:
      - ndjson:                 # mỗi dòng là JSON → field ở top-level
          target: ""
          overwrite_keys: true
          add_error_key: true
    fields_under_root: true
    fields:
      service.environment: prod
      host.group: vm-app

  - type: filestream            # app cũ log text, stack trace nhiều dòng
    id: legacy
    paths: ["/opt/legacy/app.log"]
    parsers:
      - multiline:
          type: pattern
          pattern: '^\d{4}-\d{2}-\d{2}'   # dòng mới bắt đầu bằng ngày
          negate: true
          match: after                   # dòng không có ngày → nối vào dòng trước
    fields_under_root: true
    fields: { service.name: legacy-billing }

processors:
  - add_host_metadata: ~
  - drop_event:
      when.equals: { log.level: DEBUG }

output.elasticsearch:
  hosts: ["https://10.0.2.21:9200", "https://10.0.2.22:9200", "https://10.0.2.23:9200"]
  api_key: "${ES_API_KEY}"                     # từ keystore: filebeat keystore add ES_API_KEY
  ssl.certificate_authorities: ["/etc/filebeat/ca.crt"]
  index: "logs-shop-prod"
  pipeline: "logs-shop-enrich"
setup.template.enabled: false                  # template đã tạo ở 19.8
setup.ilm.enabled: false

Log JSON từ Spring Boot không cần multiline: stack trace đã nằm trong field error.stack_trace của một dòng. Đây là lý do lớn nhất để ghi JSON.

Bước 2: Ingest pipeline trong ES (thay Logstash cho việc nhẹ)

PUT _ingest/pipeline/logs-shop-enrich
{
  "processors": [
    { "set":    { "field": "event.ingested", "value": "{{{_ingest.timestamp}}}" } },
    { "lowercase": { "field": "log.level", "ignore_missing": true } },
    { "remove": { "field": ["agent.ephemeral_id", "ecs.version"], "ignore_missing": true } },
    { "drop":   { "if": "ctx.url?.path == '/actuator/health'" } }
  ],
  "on_failure": [
    { "set": { "field": "error.message", "value": "{{ _ingest.on_failure_message }}" } }
  ]
}

Bước 3: Logstash cho Nginx (máy vm-logstash-1, /etc/logstash/conf.d/nginx.conf)

input {
  beats { port => 5044 ssl_enabled => true
          ssl_certificate => "/etc/logstash/ls.crt" ssl_key => "/etc/logstash/ls.key" }
}
filter {
  if [service][name] == "nginx" {
    grok {
      match => { "message" => '%{IPORHOST:[source][ip]} - %{DATA:[user][name]} \[%{HTTPDATE:ts}\] "%{WORD:[http][request][method]} %{DATA:[url][path]} HTTP/%{NUMBER:[http][version]}" %{NUMBER:[http][response][status_code]:int} %{NUMBER:[http][response][body][bytes]:int}' }
    }
    date   { match => ["ts", "dd/MMM/yyyy:HH:mm:ss Z"] target => "@timestamp" remove_field => ["ts"] }
    mutate { add_field => { "[service][environment]" => "prod" } }
    if [url][path] =~ /^\/actuator/ { drop {} }
  } else {
    json { source => "message" skip_on_invalid_json => true }
  }
}
output {
  elasticsearch {
    hosts => ["https://10.0.2.21:9200","https://10.0.2.22:9200","https://10.0.2.23:9200"]
    api_key => "${ES_API_KEY}"            # từ logstash-keystore
    ssl_certificate_authorities => ["/etc/logstash/ca.crt"]
    data_stream => "true"
    data_stream_type => "logs"
    data_stream_dataset => "nginx"
    data_stream_namespace => "prod"         # → logs-nginx-prod
  }
}

OpenSearch: đổi elasticsearch output thành opensearch (plugin logstash-output-opensearch).

Bước 4: thêm trace.id ở phía app. Micrometer Tracing đã đặt traceId vào MDC (19.1, 19.4). Định dạng ECS của Boot ghi các key MDC ra JSON. Để đúng tên ECS:

@Component
class EcsTraceCustomizer implements StructuredLoggingJsonMembersCustomizer<ILoggingEvent> {
    @Override
    public void customize(JsonWriter.Members<ILoggingEvent> members) {
        members.add("trace.id", e -> e.getMDCPropertyMap().get("traceId")).whenNotNull();
        members.add("span.id",  e -> e.getMDCPropertyMap().get("spanId")).whenNotNull();
    }
}
// application.yml: logging.structured.json.customizer: com.shop.log.EcsTraceCustomizer

Kiểm tra:

curl -s -X POST localhost:8080/orders -d '{"sku":"TSHIRT-M"}' -H 'Content-Type: application/json'
tail -1 /var/log/shop/order.json | jq '{service: ."service.name", trace: ."trace.id"}'
sudo filebeat test output
curl -s -H "Authorization: ApiKey $ES_API_KEY" "$ES_URL/logs-shop-prod/_count?q=service.name:order-service"

5. Cách hoạt động: Kafka làm buffer

 Filebeat ×N ──► Kafka topic logs.shop (3 partition, giữ 3 ngày)
                       │
                       │  consumer group "logstash"
                       ▼
                 Logstash ×2  ──► Elasticsearch
                       ▲
     ES chậm/bảo trì ──┘ Logstash đọc chậm lại (backpressure), log nằm yên trong Kafka
     Pipeline lỗi parse ─► reset offset, đọc lại (replay)

Không có Kafka: Filebeat cũng có backpressure (dừng đọc file, nhớ vị trí trong registry), nên ES nghỉ 30 phút vẫn không mất. Rủi ro là logrotate xóa file trước khi Filebeat đọc kịp. Logstash có thêm persistent queue trên đĩa (queue.type: persisted).

6. So sánh: xử lý ở đâu

7. Khi nào nên dùng / không nên dùng

• Kafka: nên khi log > vài trăm GB/ngày, nhiều consumer (ES + SIEM + S3), hoặc cần replay. Không nên cho 3 VM như shop-app hiện tại.
• Grok: chỉ cho nguồn không sửa được (Nginx, thiết bị). Code của mình thì ghi JSON.
• Multiline ở shipper: chỉ cho log text; đừng ghép dòng ở Logstash khi có nhiều shipper (dòng của cùng stack trace có thể tới 2 Logstash khác nhau).

Đúng rồi, nếu bạn sửa được nguồn thì ghi JSON luôn là tốt nhất, khỏi grok:

log_format ecs_json escape=json
  '{"@timestamp":"$time_iso8601","source.ip":"$remote_addr",'
  '"http.request.method":"$request_method","url.path":"$uri",'
  '"http.response.status_code":$status,"event.duration":$request_time,'
  '"trace.id":"$http_traceparent","service.name":"nginx"}';
access_log /var/log/nginx/access.json ecs_json;

Grok tốn CPU (regex trên mọi dòng) và dễ vỡ khi format đổi. Lưu ý $request_time là giây, còn ECS event.duration là nano giây; nên đổi tên field hoặc nhân lên ở ingest pipeline. Còn traceparent là chuỗi W3C đầy đủ, cần cắt lấy phần trace id.

Bạn hiểu đúng. Filebeat ghi offset vào registry sau khi ES xác nhận. Crash giữa chừng → gửi lại batch → có thể trùng vài dòng. Cách chống trùng nếu thật sự cần:

• Sinh id cố định cho mỗi dòng (processor fingerprint trên message + @timestamp + host.name) và đặt vào @metadata._id. ES gặp id trùng sẽ từ chối bản sau.
• Với data stream, trùng id chỉ phát hiện được trong cùng backing index.

Với log vận hành, vài dòng trùng thường chấp nhận được; đừng trả thêm chi phí hash nếu không cần. Log audit thì nên làm.

Q1. Filebeat, Fluent Bit, Logstash khác nhau thế nào, chọn ra sao?

🇻🇳 Tiếng Việt:

Filebeat và Fluent Bit là agent nhẹ chạy trên từng máy để đọc file và gửi đi; Fluent Bit hỗ trợ nhiều đích hơn và chạy tốt với OpenSearch. Logstash là bộ xử lý trung tâm chạy JVM, mạnh ở grok, định tuyến nhiều đích và đọc từ Kafka. Tôi luôn để agent nhẹ trên máy app, cho app ghi JSON ECS để tránh parse, xử lý nhẹ bằng ingest pipeline, và chỉ thêm Logstash khi có nguồn text khó hoặc cần nhiều đích.

🇬🇧 English:

Filebeat and Fluent Bit are lightweight agents that tail files on each host and ship them; Fluent Bit has more outputs and works well with OpenSearch. Logstash is a central JVM-based processor, strong at grok parsing, multi-destination routing and consuming from Kafka. I keep a light agent on app hosts, make apps emit ECS JSON so no parsing is needed, use ingest pipelines for light enrichment, and add Logstash only for hard text sources or fan-out.

Q2. Khi nào thêm Kafka giữa shipper và Elasticsearch?

🇻🇳 Tiếng Việt:

Khi lượng log lớn và đột biến, khi Elasticsearch hay bảo trì, khi nhiều hệ thống cùng cần log như SIEM hay lưu S3, hoặc khi cần replay sau khi sửa pipeline. Kafka hấp thụ đỉnh và tách shipper khỏi tốc độ của ES. Với hệ thống nhỏ, backpressure của Filebeat và persistent queue của Logstash là đủ, thêm Kafka chỉ tăng chi phí vận hành.

🇬🇧 English:

When volume is high and bursty, when Elasticsearch has frequent maintenance, when several consumers such as a SIEM or S3 archive need the same logs, or when we need to replay after fixing a pipeline. Kafka absorbs spikes and decouples shippers from ES speed. For small systems, Filebeat backpressure and Logstash persistent queues are enough, and Kafka only adds operational cost.

Q3. Xử lý stack trace Java nhiều dòng thế nào?

🇻🇳 Tiếng Việt:

Tốt nhất là log JSON để cả stack trace nằm trong một field của một dòng. Với log text không sửa được, cấu hình multiline ở shipper, ví dụ dòng không bắt đầu bằng ngày thì nối vào dòng trước. Không ghép ở Logstash khi có nhiều shipper, vì các dòng có thể tới các instance khác nhau.

🇬🇧 English:

The best fix is JSON logging so the whole stack trace sits in one field of one line. For text logs I cannot change, I configure multiline at the shipper, for example appending any line that does not start with a date to the previous event. I avoid joining lines in Logstash when there are many shippers, since lines may reach different instances.

Cách nhớ ngắn gọn

• Agent nhẹ trên máy, Logstash riêng, Kafka khi thật sự lớn.
• App ghi JSON ECS → không grok, không multiline.
• Bắt buộc có service.name, service.environment, trace.id.
• Drop DEBUG và health check càng sớm càng rẻ.
• At-least-once: có thể trùng; cần thì fingerprint làm _id.