Logstash hoạt động như một ống dẫn dữ liệu linh hoạt, trong đó các input plugin đóng vai trò đầu vào thiết yếu — chịu trách nhiệm thu thập dữ liệu từ nhiều nguồn khác nhau. Dưới đây là tổng quan thực tế về các plugin phổ biến nhất, được trình bày lại với cấu trúc mới, ví dụ mã hóa cải tiến và giải thích kỹ thuật rõ ràng.

1. Giao tiếp tương tác qua stdin/stdout

Một cách nhanh nhất để kiểm tra luồng xử lý cơ bản:

bin/logstash -e 'input { stdin { } } output { stdout { codec => rubydebug } }'

Khi chạy lệnh trên, mỗi dòng nhập từ bàn phím (ví dụ: Logstash is ready) sẽ được chuyển thành một sự kiện JSON đầy đủ, bao gồm trường @timestamp, host, và message. Đây là bước khởi đầu để xác minh môi trường hoạt động đúng trước khi tích hợp với nguồn dữ liệu thực tế.

2. Theo dõi tệp nhật ký cục bộ

Plugin file sử dụng thư viện filewatch để giám sát thay đổi trong hệ thống tệp, tự động ghi vị trí đọc hiện tại vào tệp .sincedb nhằm đảm bảo không bỏ sót bản ghi nào — ngay cả khi Logstash khởi động lại.

Tạo tệp cấu hình watch-logs.conf:

input {
  file {
    path => ["/var/log/app/*.log"]
    type => "application"
    start_position => "beginning"
    ignore_older => 86400
    sincedb_path => "/opt/logstash/data/sincedb_app"
    stat_interval => 2
  }
}
output {
  stdout { codec => rubydebug }
}

Kiểm tra cú pháp và triển khai:

bin/logstash -f watch-logs.conf -t
bin/logstash -f watch-logs.conf

Sau khi chạy, thêm dữ liệu thử nghiệm:

echo "$(date '+%Y-%m-%d %H:%M:%S') INFO Application started successfully" >> /var/log/app/service.log

3. Nhận dữ liệu qua giao thức TCP

Plugin tcp cho phép Logstash hoạt động như một máy chủ lắng nghe cổng mạng — phù hợp khi ứng dụng gửi log trực tiếp qua socket thay vì ghi vào tệp.

Ví dụ cấu hình listen-tcp.conf:

input {
  tcp {
    port => 4567
    host => "0.0.0.0"
    codec => json_lines
    data_timeout => 60
  }
}
output {
  stdout { codec => rubydebug }
}

Để kiểm thử, dùng lệnh nc (netcat) thay vì viết Java:

echo '{"event": "login", "user_id": 1024, "ip": "192.168.1.10"}' | nc localhost 4567

Lưu ý: Với phiên bản Logstash ≥ 7.0, nên ưu tiên json_lines thay vì json để xử lý từng dòng riêng lẻ.

4. Xử lý nhật ký hệ thống theo chuẩn Syslog

Để nhận log từ rsyslog hoặc syslog-ng, cần cấu hình cả hai giao thức TCP và UDP vì syslog có thể gửi theo bất kỳ kênh nào.

Tệp cấu hình syslog-ingest.conf:

input {
  tcp {
    port => 5140
    type => "syslog_tcp"
  }
  udp {
    port => 5140
    type => "syslog_udp"
  }
}

filter {
  if [type] in ["syslog_tcp", "syslog_udp"] {
    grok {
      match => { "message" => "%{SYSLOGTIMESTAMP:timestamp} %{HOSTNAME:hostname} %{DATA:service}(?:\[%{POSINT:pid}\])?: %{GREEDYDATA:log_content}" }
    }
    date {
      match => ["timestamp", "MMM  d HH:mm:ss", "MMM dd HH:mm:ss"]
      target => "@timestamp"
    }
  }
}

output {
  stdout { codec => rubydebug }
}

Sau khi cấu hình rsyslog gửi tới localhost:5140, kiểm tra bằng telnet hoặc logger:

logger -n localhost -P 5140 "System health check passed"

5. Phân tích nhật ký định dạng JSON từ Nginx

Thay vì dùng grok phức tạp, cấu hình Nginx xuất log dưới dạng JSON giúp tăng hiệu năng và độ chính xác:

log_format structured_json escape=json
  '{'
    '"@timestamp":"$time_iso8601",'
    '"host":"$server_addr",'
    '"client":"$remote_addr",'
    '"method":"$request_method",'
    '"url":"$uri",'
    '"status":'"$status"','
    '"bytes_sent":'"$body_bytes_sent"','
    '"response_time":'"$request_time"','
    '"upstream_time":'"$upstream_response_time"','
    '"referer":"$http_referer",'
    '"user_agent":"$http_user_agent"'
  '}';

access_log /var/log/nginx/access-structured.log structured_json;

Logstash chỉ cần đọc và giải mã trực tiếp:

input {
  file {
    path => "/var/log/nginx/access-structured.log"
    codec => json
    type => "nginx_structured"
  }
}
output {
  stdout { codec => rubydebug }
}

6. Kết hợp các dòng nhật ký đa cấp (Multiline)

Với log ứng dụng như Java stack trace hay Python exception, một sự kiện thường chiếm nhiều dòng. Plugin multiline (được tích hợp sẵn từ Logstash 7.0+) giúp nhóm chúng lại.

Ví dụ: Mỗi khối log bắt đầu bằng dấu ngoặc vuông [, các dòng tiếp theo thuộc cùng sự kiện nếu không bắt đầu bằng [.

Tệp java-trace.conf:

input {
  file {
    path => "/var/log/myapp/error.log"
    codec => multiline {
      pattern => "^\["
      negate => true
      what => "previous"
      auto_flush_interval => 1
    }
    type => "java_error"
  }
}

filter {
  if [type] == "java_error" {
    grok {
      match => { "message" => "\[%{TIMESTAMP_ISO8601:log_time} %{LOGLEVEL:level}\] %{JAVACLASS:class} - %{GREEDYDATA:details}" }
    }
    date {
      match => ["log_time", "yyyy-MM-dd HH:mm:ss.SSS"]
      target => "@timestamp"
    }
  }
}

output {
  stdout { codec => rubydebug }
}

Thêm dữ liệu mẫu:

echo -e '[2024-05-20 14:22:18.123 ERROR] com.example.Service - Failed to connect to DB\nCaused by: java.net.ConnectException: Connection refused\n\tat java.base/java.net.PlainSocketImpl.connect(PlainSocketImpl.java:220)' >> /var/log/myapp/error.log