پارس لاگ های انجینکس در استک ELK

wget -c https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-oss-7.10.2-amd64.deb
dpkg -i filebeat-oss-7.10.2-amd64.deb

# ============================== Filebeat inputs ===============================

filebeat.inputs:
- type: log
enabled: true
paths:
- /var/log/nginx/access.log*
exclude_files: ['.gz$']
fields:
type: access
fields_under_root: true

- type: log
enabled: true
paths:
- /var/log/nginx/error.log*
exclude_files: ['.gz$']
fields:
type: error
fields_under_root: true

# ============================== Filebeat modules ==============================

filebeat.config.modules:
# Glob pattern for configuration loading
path: ${path.config}/modules.d/*.yml

# Set to true to enable config reloading
reload.enabled: false

# Period on which files under path should be checked for changes
#reload.period: 10s

# ================================== Outputs ===================================

# ------------------------------ Logstash Output -------------------------------
output.logstash:
hosts: ["IPADDRESS:PORT"]

# ================================= Processors =================================
processors:
- add_host_metadata:
when.not.contains.tags: forwarded
- add_cloud_metadata: ~
- add_docker_metadata: ~
- add_kubernetes_metadata: ~

filter {
grok {
match => [ "message" , "%{COMBINEDAPACHELOG}+%{GREEDYDATA:extra_fields}"]
overwrite => [ "message" ]
}
mutate {
convert => ["response", "integer"]
convert => ["bytes", "integer"]
convert => ["responsetime", "float"]
}
geoip {
source => "clientip"
target => "geoip"
add_tag => [ "nginx-geoip" ]
}
date {
match => [ "timestamp" , "dd/MMM/YYYY:HH:mm:ss Z" ]
remove_field => [ "timestamp" ]
}
useragent {
source => "agent"
}
}

log_format custom '$host - $remote_addr - $remote_user [$time_local] - '
'"$request" $status $body_bytes_sent - '
'"$http_referer" "$http_user_agent" - '
'cache: $upstream_cache_status [$time_local]';

نمونه لاگ دریافتی
test.ir - 1.1.1.1 - - [01/Aug/2021:11:32:44 +0000] - "GET / HTTP/2.0" 200 11760 - "-" "Mozilla/5.0 (Linux; Android 11; SM-A505F) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/92.0.4515.115 Mobile Safari/537.36" - cache: MISS [01/Aug/2021:11:32:44 +0000]

## nginx Access log grok pattern

%{IPORHOST:host_name} - %{IPV4:remote_ip} - %{DATA:user_name} \[%{HTTPDATE:access_time}\] - "%{WORD:http_method} %{DATA:url} HTTP/%{NUMBER:http_version}" %{NUMBER:response_code} %{NUMBER:body_sent_bytes} - "%{DATA:referrer}" "%{DATA:user_agent}" - cache: %{DATA:cache_status} \[%{HTTPDATE:time_local}\]

## nginx Error log grok pattern

%{YEAR}/%{MONTHNUM}/%{MONTHDAY} %{TIME} \[%{LOGLEVEL:level}\] %{INT:process_id}#%{INT:thread_id}: \*(%{INT:connection_id})? %{GREEDYDATA:description}

input {
beats {
port => 5044
}

}

filter {
if "access" in [type] {
grok {
match => { "message" => '%{IPORHOST:host_name} - %{IPV4:remote_ip} - %{DATA:user_name} \[%{HTTPDATE:access_time}\] - "%{WORD:http_method} %{DATA:url} HTTP/%{NUMBER:http_version}" %{NUMBER:response_code} %{NUMBER:body_sent_bytes} - "%{DATA:referrer}" "%{DATA:user_agent}" - cache: %{DATA:cache_status} \[%{HTTPDATE:time_local}\]' }
remove_field => "message"
}
}

else if "error" in [type] {
grok {
match => { "message" => '%{YEAR}/%{MONTHNUM}/%{MONTHDAY} %{TIME} \[%{LOGLEVEL:level}\] %{INT:process_id}#%{INT:thread_id}: \*(%{INT:connection_id})? %{GREEDYDATA:description}' }
remove_field => "message"
}
}
else {
drop { }
}
}


output {
elasticsearch {
hosts => ["https://elastic:9200"]
ssl => true
ssl_certificate_verification => false
user =>
password =>
index => "%{[@metadata][beat]}-%{[host][hostname]}-%{[@metadata][version]}-%{+YYYY.MM.dd}"
}
}