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.
Bấm vào Create index parttern.
Gõ vào tên parttern của json sau đó bấm Next step.
Mình chọn Time field là @timestamp.
Sau khi tạo xong pattern bạn sẽ nhìn thấy các Fields như hình dưới.
Hãy vào Discover và chọn index là demo_json, bạn sẽ nhìn thấy các logs xuất hiện như hình.