[ELK] Cách lọc file Json trong Logstash

Là nhà phát triển, bạn cần ghi log mọi thứ, đó có thể là info, error or debug logs, etc, v.v. Có nhiều loại định dạng log như Common log, JSON log, etc, v.v. và đã có sẵn các giải pháp sử dụng filebeat để đọc log JSON và đẩy chúng vào elaticsearch.

Có thể có trường hợp bạn cần ghi dữ liệu theo sự thuận tiện của mình, đây sẽ không phải là dạng log tiêu chuẩn. Hoặc đôi khi bạn chỉ cần đơn giản log theo cách có cấu trúc.

Trong bài viết này, chúng ta sẽ xem cách chúng ta có thể phân tích log ở bất kỳ định dạng nào vào Elaticsearch. Elaticsearch cung cấp pipeline với bộ xử lý grok có thể khớp với bất kỳ log phi cấu trúc nào.

– Tạo file /home/sample-json.log.

vi /home/sample-json.log

Chèn các nội dung json như dưới vào file logs.txt để kiểm tra.

{"id":"1","name":"John1", "age":20, "car":"skoda"}
{"id":"2","name":"John2", "age":43, "car":"tata"}
{"id":"3","name":"John3", "age":30, "car":"mahindra"}
{"id":"4","name":"Joh4", "age":30, "car":"bmw"}
{"id":"5","name":"Joh5", "age":37, "car":"maruti"}
{"id":"6","name":"Joh6", "age":30, "car":"tesla"}
{"id":"7","name":"Joh7", "age":35, "car":"Volkswagen"}
{"id":"8","name":"Joh7", "age":35, "car":"Volkswagen"
{"id":"9","name":"Joh7", "age":35, "car":"Volkswagen"}
{"id":"10","name":"Joh7", "age":35, "car":"Volkswagen"
{"id":"11","name":"Joh7", "age":35, "car":"Volkswagen"}

Mở file /etc/filebeat/filebeat.yml.

vi /etc/filebeat/filebeat.yml

Đưa các nội dung bên dưới vào file filebeat.yml.

filebeat.inputs:
- type: log
  paths: /home/sample-json.log
  json.keys_under_root: true
  json.overwrite_keys: true
  json.add_error_key: true
  json.expand_keys: true
  
output:
  logstash:
    hosts: ["192.168.13.205:5044"]

Khởi động lại filebeat.

sudo systemctl restart filebeat

Mở file /home/logstash.conf.

vi /home/logstash.conf

Phần input cấu hình như dưới vào file logstash.conf nếu bạn dùng filebeat đẩy logs với port 5044 là port lắng nghe nhận logs của logstash.

input {
  beats {
    port => 5044
  }
}

Hoặc nếu bạn lấy logs từ chính server logstash thì sử dụng đoạn config input như dưới với /home/sample-json.log là nơi chứa dữ liệu json của bạn. Do mình không có máy chủ thứ 2 nên mình sẽ sử dụng phương án này.

input {
  file {
    start_position => "beginning"
    path => "/home/sample-json.log"
    sincedb_path => "/dev/null"
  }
}

Tại phần filter nếu bạn đang sử dụng logs của mình thì có thể sử dụng đoạn filter này.

filter {
  json {
        skip_on_invalid_json => true
        source => "message"
        target => "jsonData"
        add_tag => [ "_message_json_parsed" ]
      }

  mutate {
        remove_field => ["@timestamp","offset", "prospector","@version","source","host","beat","log","input","agent.hostname","agent.id","agent.name","agent.version","_type","_score","agent.ephemeral_id"]
      }    
}

Phần output hãy trỏ về máy chủ Elasticsearch của bạn và demo_json chính là index test của mình.

output {
    elasticsearch {
     hosts => "http://es01:9200"
     index => "demo_json"
     ssl => false
     ssl_certificate_verification => false
  }
  stdout{}
}

Vậy config đầy đủ của mình sẽ là:

input {
  file {
    start_position => "beginning"
    path => "/home/sample-json.log"
    sincedb_path => "/dev/null"
    codec => "json"
  }
}

bin/logstash --path.settings=/usr/share/logstash/pipeline/logstash.conf --debug
filter {
  grok {
    match => { "message" => '{"number_id":"%{NUMBER:number_id}","name":"%{DATA:name}","age":%{NUMBER:age},"city":"%{DATA:city}"}' }
  }
}

output {
  elasticsearch {
    hosts => ["http://es01:9200", "http://es02:9200", "http://es03:9200"]
    index => "demo_json"
    ssl => false
    ssl_certificate_verification => false
  }
  stdout {}
}

Khởi động lại logstash, mình chạy docker nên mình sẽ khởi động luôn container logstash.

sudo docker restart logstash

Bây giờ bạn hãy vào Kibana -> stack management

Làm theo các bước bên dưới, bạn sẽ thấy index mới của json có tên demo_json. Hãy bấm vào Index Partterns như mục 4 hình dưới để tạo 1 Index Parttern cho Kibana.