Bài học 4: "Tút tát" và chuẩn hóa dữ liệu với Mutate và Date Filter

1. Mutate Filter - Công cụ "phẫu thuật thẩm mỹ" dữ liệu

Plugin mutate cho phép bạn can thiệp trực tiếp vào các trường dữ liệu đã được tạo ra. Nó có thể đổi tên, xóa bỏ, thay thế văn bản, hoặc chuyển đổi kiểu dữ liệu.

Các tính năng hay dùng nhất của Mutate:

• convert (Chuyển đổi kiểu dữ liệu): Mặc định, mọi thứ Grok cắt ra đều được Logstash hiểu là văn bản (String). Nếu bạn cắt ra số 200 (mã trạng thái) và muốn Kibana tính toán được trên con số này (ví dụ: tìm các mã > 400), bạn phải ép nó về kiểu số nguyên (Integer).
• remove_field (Xóa trường thừa): Sau khi grok đã cắt thành công dòng log thô thành các trường nhỏ, trường gốc (thường tên là message) sẽ trở nên thừa thãi và làm tăng gấp đôi dung lượng lưu trữ trên Elasticsearch. Ta nên xóa nó đi.
• rename (Đổi tên trường): Giúp chuẩn hóa tên gọi theo đúng chuẩn của hệ thống nếu mẫu Grok đang dùng một cái tên không phù hợp.

Ví dụ cấu hình Mutate:

filter {
  mutate {
    # Chuyển trường response_code từ chữ sang số nguyên (integer)
    convert => { "response_code" => "integer" }
    
    # Xóa dòng log thô ban đầu để tiết kiệm dung lượng
    remove_field => [ "message" ]
    
    # Đổi tên trường client_ip thành source_ip cho dễ hiểu
    rename => { "client_ip" => "source_ip" }
  }
}

2. Date Filter - Cỗ máy thời gian của Logstash

Đây là một plugin cực kỳ quan trọng nhưng nhiều người mới học lại hay bỏ quên.

Vấn đề: Khi Logstash đẩy dữ liệu lên Elasticsearch, nó tự động tạo ra một trường có tên là @timestamp (thời gian ghi nhận). Tuy nhiên, đây là thời gian Logstash đọc dòng log đó, chứ KHÔNG PHẢI thời gian sự kiện thực sự xảy ra trong log. Nếu hôm nay bạn mới chạy Logstash để đọc lại file log của tháng trước, toàn bộ biểu đồ trên Kibana sẽ bị sai ngày (nó sẽ hiện là sự kiện xảy ra vào hôm nay).

Giải pháp: Dùng date filter để bốc mốc thời gian trong dòng log, và ghi đè lên trường @timestamp mặc định của hệ thống.

Ví dụ cấu hình Date:

filter {
  date {
    # Lấy dữ liệu từ trường "time_local" (đã cắt ra bằng Grok trước đó)
    # Khớp với định dạng thời gian "dd/MMM/yyyy:HH:mm:ss Z"
    match => [ "time_local", "dd/MMM/yyyy:HH:mm:ss Z" ]
    
    # Đẩy mốc thời gian vừa khớp vào trường chuẩn của hệ thống
    target => "@timestamp"
    
    # Xóa luôn trường time_local ban đầu vì đã dùng xong
    remove_field => [ "time_local" ]
  }
}

3. Ghép nối hoàn chỉnh một quy trình Filter

Bây giờ, chúng ta sẽ kết hợp grok (từ Bài 3), date và mutate để tạo ra một khối Filter xử lý log Nginx hoàn hảo theo đúng thứ tự (Logstash sẽ chạy filter từ trên xuống dưới):

filter {
  # Bước 1: Dùng Grok để cắt nhỏ dòng văn bản
  grok {
    match => { "message" => "%{IP:client_ip} - - \[%{HTTPDATE:time_local}\] \"%{WORD:method} %{DATA:request} HTTP/%{NUMBER:http_version}\" %{NUMBER:response_code} %{NUMBER:body_bytes_sent}" }
  }

  # Bước 2: Chuẩn hóa thời gian
  date {
    match => [ "time_local", "dd/MMM/yyyy:HH:mm:ss Z" ]
    target => "@timestamp"
  }

  # Bước 3: Dọn dẹp và ép kiểu dữ liệu
  mutate {
    convert => {
      "response_code" => "integer"
      "body_bytes_sent" => "integer"
    }
    remove_field => [ "message", "time_local" ]
  }
}

4. Kết quả nhận được

So với kết quả ở Bài 2, dữ liệu đầu ra lúc này đã "sạch", nhẹ hơn và chuẩn xác về mặt kiểu dữ liệu để Kibana vẽ biểu đồ:

{
  "@timestamp": "2026-07-13T03:00:00.000Z",
  "client_ip": "192.168.1.10",
  "method": "GET",
  "request": "/index.html",
  "http_version": "1.1",
  "response_code": 200,          <-- Giờ đây là số (không còn nằm trong ngoặc kép "")
  "body_bytes_sent": 1024        <-- Giờ đây là số
  // Trường "message" và "time_local" đã được dọn dẹp sạch sẽ
}