Bài học 5: Điều phối giao thông bằng Câu lệnh điều kiện (If/Else) và Đa đầu ra

Chào mừng bạn đến với Bài học 5. Đến lúc này, "nhà máy" Logstash của chúng ta đã có khả năng hút dữ liệu (Input), bóc tách (Grok), và tút tát (Mutate/Date).

Nhưng trong thực tế, một hệ thống web không chỉ có một loại log. Bạn sẽ có log của Nginx, log của cơ sở dữ liệu (MySQL), và log của mã nguồn ứng dụng (Backend). Nếu đưa tất cả vào chung một "băng chuyền" xử lý, hệ thống sẽ rối loạn. Đây là lúc chúng ta cần đến Câu lệnh điều kiện (Conditionals) để phân luồng dữ liệu.

1. Tại sao cần Câu lệnh điều kiện?

Hãy tưởng tượng Logstash nhận được hai dòng log cùng lúc:

1. Log Nginx: 192.168.1.10 - - [13/Jul/2026...] "GET / HTTP/1.1" 200
2. Log Ứng dụng: [ERROR] Database connection timeout

Nếu bạn dùng chung cái khuôn grok của Nginx (đã học ở Bài 3) áp lên log ứng dụng, Logstash sẽ báo lỗi (Grok Parse Failure) vì khuôn không khớp. Do đó, ta cần gắn thẻ (tag) hoặc phân loại (type) cho từng nguồn log ngay từ đầu vào, sau đó dùng if - else để chỉ định bộ lọc tương ứng.

2. Cú pháp điều kiện trong Logstash

Logstash hỗ trợ các biểu thức logic giống hệt các ngôn ngữ lập trình thông thường:

• So sánh bằng: == (bằng), != (không bằng)
• Chứa từ khóa: =~ (khớp với Regex), !~ (không khớp)
• Kiểm tra tồn tại: if [field_name] (nếu trường này có tồn tại)
• Cấu trúc: if { ... } else if { ... } else { ... }

3. Thực hành: Xử lý nhiều luồng log cùng lúc

Dưới đây là một tệp cấu hình thực tế xử lý song song Log Nginx và Log Ứng dụng:

# 1. KHỐI INPUT: Gắn nhãn cho từng nguồn
input {
  # Nguồn 1: Log Nginx
  file {
    path => "/var/log/nginx/access.log"
    type => "nginx_access" # Gắn nhãn type là nginx_access
  }
  
  # Nguồn 2: Log Ứng dụng Backend
  file {
    path => "/var/log/myapp/error.log"
    type => "app_error" # Gắn nhãn type là app_error
  }
}

# 2. KHỐI FILTER: Rẽ nhánh xử lý
filter {
  # Nếu là log Nginx thì dùng khuôn Grok của Nginx
  if [type] == "nginx_access" {
    grok {
      match => { "message" => "%{IP:client_ip} - - \[%{HTTPDATE:time_local}\] \"%{WORD:method} %{DATA:request} HTTP/%{NUMBER:http_version}\" %{NUMBER:response_code} %{NUMBER:body_bytes_sent}" }
    }
  } 
  # Nếu là log Ứng dụng thì dùng khuôn Grok khác
  else if [type] == "app_error" {
    grok {
      match => { "message" => "\[%{LOGLEVEL:level}\] %{GREEDYDATA:error_message}" }
    }
  }
}

# 3. KHỐI OUTPUT: Phân loại kho lưu trữ
output {
  if [type] == "nginx_access" {
    elasticsearch {
      hosts => ["http://localhost:9200"]
      index => "nginx-logs-%{+YYYY.MM.dd}" # Lưu vào thư mục riêng của Nginx
    }
  } 
  else if [type] == "app_error" {
    elasticsearch {
      hosts => ["http://localhost:9200"]
      index => "app-logs-%{+YYYY.MM.dd}" # Lưu vào thư mục riêng của App
    }
  }
}

4. Ứng dụng nâng cao: Cảnh báo lỗi nghiêm trọng

Không chỉ dùng để phân loại log theo nguồn, lệnh if trong khối Output còn rất hữu ích để thiết lập các hệ thống cảnh báo tức thời.

Ví dụ: Bạn muốn Logstash gửi một cảnh báo nếu phát hiện Nginx trả về mã lỗi 500 (máy chủ bị sập):

output {
  # Vẫn lưu mọi thứ vào Elasticsearch như bình thường
  elasticsearch {
    hosts => ["http://localhost:9200"]
    index => "nginx-logs-%{+YYYY.MM.dd}"
  }

  # NHƯNG nếu có lỗi 500 thì gửi thêm cảnh báo
  if [response_code] == 500 {
    # (Ví dụ minh họa việc gọi API gửi tin nhắn Slack/Email)
    http {
      url => "https://api.slack.com/webhooks/your_token"
      http_method => "post"
      format => "json"
      message => "CẢNH BÁO: Phát hiện lỗi 500 từ IP %{client_ip}"
    }
  }
}

Như vậy, chúng ta đã nắm được toàn bộ quy trình xử lý bên trong lõi của Logstash. Tuy nhiên, trong các hệ thống hiện đại, người ta hiếm khi để Logstash trực tiếp đọc file log trên máy chủ Nginx (vì Logstash khá nặng và ngốn RAM). Thay vào đó, họ dùng một "nhân viên giao hàng" nhỏ gọn hơn mang tên Filebeat.

Bạn có muốn Bài học 6 tiếp tục đi vào cách thiết lập bộ đôi hoàn hảo Filebeat gửi log tới Logstash không?