Trong hệ thống elk thì chắc chắn logstash ko phải là 1 components quá xa lạ.
Mục đích chính của mình khi sửa dụng logstash là gì:
– Nó gửi logs về elastic ngon (của nhà elastic mà)
– Nó hỗ trợ parse được nhiều kiểu logs khó nhằn thông input và filter (BEST)
1) Overview Pipeline of Logstash
1.1) where are the pipeline files?
các files pipeline thì nó nằm ở thưc mục sau:/usr/share/logstash/pipeline
Mình có có hướng dẫn cài đặt ở links sau:
Lý do mình chọn cài docker là mình thấy nó cũng ko lưu data và khi provisioning bằng docker nó cũng khá nhanh.
Thật ra mình thấy cài logstash bằng apt thì nó cũng nhanh. Nên cũng tuỳ sở thích của các bạn nhé!
1.2) Common content inside pipeline.
Mình sẽ có simple content của 1 pipeline
input {
beats {
port => "5044"
}
}
filter {
grok {
match => { "message" => "%{SYSLOGLINE}"}
}
}
output {
stdout { codec => rubydebug }
}
Bạn sẽ thấy nó gồm 3 phần:
input: là nơi nhận logs từ các nới khác gửi tới: fortigate gửi vào logstash, filebeat gửi vào logstash, ….
filter: nơi bạn mổ sẻ các message hay còn gọi là từng dòng logs như là: định sạng, thêm, bớt, xoá, sửa theo ý muốn,…
output: cái tên nói lên tất cả. Chố này bạn gửi logs sang elastic hoặc print ra stdout để debug
Phần sau chúng ta đi sau vào từng thành phần. Mình sẽ expand knowledge slowly. Các bạn có thể góp ý để mình thêm nhé!
2) Designing INPUT in Pipeline of Logstash
Common Options
The following configuration options are supported by all input plugins:
| Setting | Input type | Required |
|---|---|---|
add_field | hash | No |
codec | codec | No |
enable_metric | boolean | No |
id | string | No |
tags | array | No |
type | string | No |
Nim sẽ có 1 số config input tiêu biểu
Nếu input là file (Cái này nó giống như filebeat chúng ta đọc file)
input {
file {
path => "/var/log/messages"
type => "syslog"
}
file {
path => "/var/log/apache/access.log"
type => "apache"
}
}type: Add a type field to all events handled by this input. Types are used mainly for filter activation.
https://www.elastic.co/guide/en/logstash/current/plugins-inputs-exec.html#plugins-inputs-exec-type
Nếu input là UDP hay TCP
input {
tcp {
port => 1234
codec => json
}
udp {
port => 1234
codec => json
}
}codec: The codec used for input data. Input codecs are a convenient method for decoding your data before it enters the input, without needing a separate filter in your Logstash pipeline. Default value is "plain"
3) Designing OUTPUT in Pipeline of Logstash
Ngồn từ 1 lớp học ITFORVN
reference links:
https://www.elastic.co/guide/en/logstash/current/output-plugins.html
❖ Output plugin sẽ gởi dữ liệu sau khi parse đến một destination cụ thể (stash).
❖ Logstash hỗ trợ sẵn nhiều stash output khác nhau, được sử dụng nhiều nhất là Elasticsearch.
output {
elasticsearch {
hosts => ["192.168.1.2:9200", "192.168.1.1:9200"]
index => "%{[@metadata][beat]}-%{[@metadata][version]}"
}
}
❖ Lưu ý: nếu 1 Elasticsearch là “dedicated master node”, thì không nên đưa vào list hosts của Elasticsearch output plugin, vì “dedicated master node” không chứa dữ liệu.
❖ Test logstash config: logstash --config.test_and_exit -f
❖ Debug output:
output {
stdout { codec => rubydebug }
}reference links:
https://levanphu.info/logstash-la-gi-cai-dat-su-dung-va-lam-chu-logstash-trong-1-bai-viet
4) Designing FILTER in Pipeline of Logstash
- Filter được áp dụng trên dữ liệu, thực hiện các tác vụ “transformation” dự vào đặc điểm của input và mong muốn có được ở output.
- Logstash hỗ trợ hầu hết filter trên các định dạng thông dụng. Với các định dạng “custom”, sử dụng “grok”.
- Các filter thông dụng: csv, json, geoip, kv, mutate
https://www.elastic.co/guide/en/logstash/current/filter-plugins.html
Common Options
The following configuration options are supported by all filter plugins:
| Setting | Input type | Required |
|---|---|---|
add_field | hash | No |
add_tag | array | No |
enable_metric | boolean | No |
id | string | No |
periodic_flush | boolean | No |
remove_field | array | No |
remove_tag | array | No |
4.1) filter kv
Bộ lọc áp dụng trên dữ liệu dạng “key=value”:
filter {
kv { }
}
filter {
kv {
field_split => "&?"
}
}
Presentation Title
Input: ?pin=12345~0&d=123&e=foo@bar.com&oq=bobo&ss=12345
Output:
https://www.elastic.co/guide/en/logstash/current/plugins-filters-kv.html
4.2) filter json
❖ Bộ lọc áp dụng trên dữ liệu định dạng JSON:
filter {
json {
source => "message"
}
}
(source: là field bắt buộc, chứa dữ liệu ở định dạng JSON)
Presentation Title
❖ Nếu dữ liệu parse failed sẽ được đánh tag “_jsonparsefailure”.
❖ Nếu dữ liệu parse thành công:
– Ví dụ: Mặc định dữ liệu sau khi parse sẽ được ghi ở “top level” root document.
– Sử dụng “target” nếu muốn ghi dữ liệu ở một root document khác.
https://www.elastic.co/guide/en/logstash/7.13/plugins-filters-json.html
4.3) grok filter – logstash
4.3.1) grok overview
- ❖ Parse unstructured log data into something structured and queryable. Grok sử dụng regular expresion.
- ❖ Logstash có khoảng 120 pattern được build sẵn:
https://github.com/logstash-plugins/logstash-patterns-core/tree/main/patterns - ❖ Thư mục chứa các pattern default của Logstash (version 7.13): /usr/share/logstash/vendor/bundle/jruby/2.5. 0/gems/logstash-patterns-core- 4.3.1/patterns/
- ❖Tools:
❖ http://grokdebug.herokuapp.com/
❖ http://grokconstructor.appspot.com/
4.3.2) grok syntax
❖ Syntax: %{SYNTAX:SEMANTIC}
• SYNTAX: pattern name
• SEMANTIC: định danh gán cho “matched text”. Mặc định “matched text” sẽ được lưu ở “string type”
❖ Ví dụ:
• 123: match với pattern NUMBER
• 208.67.222.222: match với pattern IP
❖ Ví dụ:
➢ Text: 123 208.67.222.222
➢ Grok filter: %{NUMBER:myid} %{IP:opendns}
➢ Grok filter: %{NUMBER:myid:int} %{IP:opendns} (myid sẽ có type là int thay vì mặc định là string)
Vậy thì làm sao có mình có thể tìm ra các Syntax grok hợp lý:
– Dò đại google.
– Tham khảo các grok patterns ở link này:
https://github.com/logstash-plugins/logstash-patterns-core/blob/main/patterns/ecs-v1/grok-patterns
Mình show 1 số ví dụ cho anh em sem trước.
USERNAME [a-zA-Z0-9._-]+
USER %{USERNAME}
EMAILLOCALPART [a-zA-Z0-9!#$%&'*+\-/=?^_`{|}~]{1,64}(?:\.[a-zA-Z0-9!#$%&'*+\-/=?^_`{|}~]{1,62}){0,63}
EMAILADDRESS %{EMAILLOCALPART}@%{HOSTNAME}
INT (?:[+-]?(?:[0-9]+))
BASE10NUM (?<![0-9.+-])(?>[+-]?(?:(?:[0-9]+(?:\.[0-9]+)?)|(?:\.[0-9]+)))
NUMBER (?:%{BASE10NUM})
BASE16NUM (?<![0-9A-Fa-f])(?:[+-]?(?:0x)?(?:[0-9A-Fa-f]+))
BASE16FLOAT \b(?<![0-9A-Fa-f.])(?:[+-]?(?:0x)?(?:(?:[0-9A-Fa-f]+(?:\.[0-9A-Fa-f]*)?)|(?:\.[0-9A-Fa-f]+)))\b
Cũng đã được logstash đi file và chúng ta chỉ việc sử dụng lại
Từ ảnh trên suy ra chúng ta cũng có tư define các patterns
Để chúng ta có grok được 1 message.
thì chúng ta cần phải hiểu rõ chúng ta cần cắt field name nào trong 1 message của chúng ta
Vậy giờ chúng ta trợ lại kibana -> chọn 1 message -> kiểu dữ liệu json.
và màu xanh lá là value của nó.
4.3.2.1) demo
trong bài học thì anh ấy chọn field này:
và có 1 lưu ý bạn thấy anh ấy sử dụng “\” trước kí tự khá đặc biệt là “/” và “?”
chỗ đó thì anh sử dụng GREEDYDATA thì data ở đó có dạng: “a=1&b=2” vì ở đây bạn thấy có kí tự “&“. mà DATA thì ko thể lấy được nếu có “&” ở giữa.
4.3.3) grok custom pattern
❖ Syntax:
(?<field_name>pattern)
• field_name: định danh gán cho “matched text”
• pattern: regular expression pattern. Logstash sử dụng Oniguruma regex library: https://github.com/kkos/oniguruma/blob/master/doc/RE
❖ Ví dụ:
➢ Text: 4e30b39b89564b9f29aeec09e96bf1ac
➢ Grok filter:
(?<md5hash>[a-z0-9]{32})
Như ở trên chúng ta thấy
+ fieldname là md5hash
+ [a-z0-9] chuỗi này của chúng ta sẽ từ a đến z và từ 0 đến 9
+ Độ dài của chuỗi là 32 kí tự {32}
❖ “Custom pattern” có thể được khai báo ở dạng pattern và dùng như sau:
4.3.3.1) in-line inside Pipeline
Các bạn sẽ viết trong file pipeline luôn
4.3.3.2) Define the patterns in File.
Bạn define pattern mà bạn mong muốn trong thư mục và file riêng
Bạn chỉ cần khai báo thư mục chưa file thôi nhé. nó tự động load các file bên trong.
4.3.3.3) Using grokdebug
Mình sẽ cũng cấp cho các bạn 1 pipeline và 1 message log
##### Access log structure
$remote_addr - $remote_user [$time_local] "$request" $status $body_bytes_sent "$http_referer" "$http_user_agent"
##### Input sample
109.67.210.45 - - [21/Jul/2021:09:58:29 +0200] "GET /icons/blank.gif HTTP/1.1" 200 439 "http://isrolab.com/" "Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/536.5 (KHTML, like Gecko) Chrome/19.0.1084.56 Safari/536.5"
##### Logstash configuration sample
input {
udp {
port => 5100
codec => json
}
}
filter {
grok {
match => {
"message" => "%{IPORHOST:remote_addr} - %{USER:remote_user} \[%{HTTPDATE:timestamp}\] \"%{WORD:request_method} %{NOTSPACE:request_uri} %{NOTSPACE:http_version}\" %{NUMBER:status} %{NUMBER:body_bytes_sent} \"%{DATA:http_referer}\" \"%{DATA:http_user_agent}\""
}
}
geoip {
source => "remote_addr"
}
}
output {
elasticsearch {
hosts => ["172.16.253.11:9200","172.16.253.12:9200","172.16.253.13:9200"]
index => "nginx-%{+YYYY.MM.dd}"
}
stdout {
codec => rubydebug
}
}
https://grokdebug.herokuapp.com/
bạn truy cập link trên rồi đưa input log sample, tiếp theo bạn đưa grok và chọn Named captures Only
nếu grok bạn define đúng thì nó gen ra cho bạn như hình bên dưới.
Nếu gen ngon lành thì đưa vào pipeline là ok gòi
4.3.4) mutate
❖ Thực hiện các tác vụ thay đổi thông thường trên field dữ liệu. Các tác vụ thường dùng:
• convert: chuyển đổi định dạng dữ liệu. Ví dụ từ string sang int.
• rename: rename field’s name.
• replace: replace the value of a field with a new value.
• gsub: match a regular expression against a field value (string) and replace all matches with a replacement string.
• update: update an existing field with a new value.
❖ Ví dụ:
4.3.5) geoip
❖ The GeoIP filter adds information about the geographical location of IP addresses, based on data from the MaxMind GeoLite2 databases.
❖ Logstash bundled hỗ trợ sẵn “GeoLite2-City” và “GeoLite2-ASN”.
❖ Configuration:
Ở bài này chúng ta sẽ tìm hiểu việc được dữ liệu geoip của index bất kì lên bản đồ của Kibana
Mình sẽ có 1 bài riêng he
5) Pipelines
❖ Mặc định logstash sử dụng 1 pipeline “main”.
❖ Pipeline được sử dụng khi có nhiều events flow, nhưng không chia sẻ cùng input/filter/output với nhau (có thể đang dung “tag” hoặc các điều kiện để phân biệt).
❖ Khai báo pipeline:
❖ Nếu không dùng pipeline, có thể sử dụng các điều kiện đầu vào như “tag”, “type” để phân tách từng events flow với nhau.
❖ Ví dụ (không if ở output): https://thachnuida.com/2016/12/04/push-nginx-log-to-elasticsearch-by-using- logstash/
Mình đã có 1 bài viết mô tả chi tiết:
