date: 2020-07-16title: filebeat及logstash配置 #标题
tags: elk线上配置 #标签
categories: elastic stack # 分类

记录下filebeat及logstash配置语法。

配置filebeat收集nginx日志及java日志

  1. filebeat.inputs:
  2. - type: log
  3.   enabled: True
  4.   fields:
  5.     log_type: nginx-access
  6.     project_name: shsp
  7.     log_topic: common_nginx
  8.   fields_under_root: true
  9.   paths:
  10.     - /apps/usr/nginx/logs/access.log
  12. - type: log
  13.   enabled: True
  14.   fields:
  15.     project_name: shsp
  16.     log_type: nginx-error
  17.     log_topic: common_nginx
  18.   fields_under_root: true
  19.   paths:
  20.     - /apps/usr/nginx/logs/error.log
  23. - type: log
  24.   enabled: True
  25.   multiline.pattern: '^[[:space:]]+|^Caused by:'
  26.   multiline.negate: false
  27.   multiline.match: after
  28.   fields:
  29.     project_name: shsp
  30.     log_type: app_log
  31.     log_topic: app_all
  32.   fields_under_root: true
  33.   paths:
  34.     - /apps/usr/appdata/logs/*.log
  36. output.kafka:
  37.   hosts: ["192.168.20.2:9092", "192.168.20.3:9092", "192.168.20.4:9092"]
  38.   topic: '%{[log_topic]}'
  39.   partition.round_robin:
  40.     reachable_only: false
  41.   required_acks: 1
  42.   compression: gzip
  43.   max_message_bytes: 1000000
  45. processors:
  46.   - drop_fields:
  47.       fields: ['ecs', 'beat', 'input', '@version', 'agent']

logstash处理nginx日志

  1. input {
  2.     kafka {
  3.         bootstrap_servers => "192.168.20.2:9092,192.168.20.3:9092,192.168.20.4:9092"
  4.         topics => ["common_nginx"]
  5.         codec => json { charset => "UTF-8" }
  6.         group_id => "standard"
  7.         consumer_threads => 8
  8.     }
  9. }
  10. filter {
  11.    if [log_type] == "nginx-access" {
  12.         grok {
  13.            match => { "message" => ["%{IPORHOST:[access][remote_ip]} - %{DATA:[access][user_name]} \[%{HTTPDATE:[access][time]}\] \"%{WORD:[access][method]} %{DATA:[access][url]} HTTP/%{NUMBER:[access][http_version]}\" %{NUMBER:[access][response_code]} %{NUMBER:[access][body_sent][bytes]} \"%{DATA:[access][referrer]}\" \"%{DATA:[access][agent]}\""] }
  14.         remove_field => "message"
  15.         }
  16.         mutate {
  17.             add_field => { "read_timestamp" => "%{@timestamp}" }
  18.         }
  19.         date {
  20.             match => [ "[access][time]", "dd/MMM/YYYY:H:m:s Z" ]
  21.             remove_field => "[access][time]"
  22.         }
  23.         useragent {
  24.             source => "[access][agent]"
  25.             target => "[access][user_agent]"
  26.             remove_field => "[access][agent]"
  27.         }
  28.         geoip {
  29.             source => "[access][remote_ip]"
  30.             target => "[geoip]"
  31.         }
  32.     } 
  33.     else if [log_type] == "nginx-error" {
  34.         grok {
  35.             match => { "message" => ["%{DATA:[error][time]} \[%{DATA:[error][level]}\] %{NUMBER:[error][pid]}#%{NUMBER:[error][tid]}: (\*%{NUMBER:[error][connection_id]} )?%{GREEDYDATA:[error][message]}"] }
  36.             remove_field => "message"
  37.         }
  38.         mutate {
  39.             rename => { "@timestamp" => "read_timestamp" }
  40.         }
  41.         date {
  42.             match => [ "[error][time]", "YYYY/MM/dd H:m:s"]
  43.             remove_field => "[error][time]"
  44.         }
  45.     }
  46. }
  48. output {
  49.     #stdout { 
  50.     #    codec => rubydebug
  51.     #}
  53.     elasticsearch {
  54.         hosts => [ "192.168.20.11:9200","192.168.20.12:9200","192.168.20.13:9200" ]
  55.         user => "elastic"
  56.         password => "abcd"
  57.         index => "logstash-nginx-%{project_name}-%{+YYYY.MM.dd}"
  58.         codec => json { charset => "UTF-8" }
  59.     }
  60. }

logstash处理java日志

  1. input {
  2.     kafka {
  3.         bootstrap_servers => "192.168.20.2:9092,192.168.20.3:9092,192.168.20.4:9092"
  4.         topics => ["app_all"]
  5.         codec => json { charset => "UTF-8" }
  6.         group_id => "standard"
  7.         consumer_threads => 8
  8.     }
  9. }
  10. filter {
  11.     mutate {
  12.                 add_field => { "log_path" => "%{[log][file][path]}" }
  13.                 add_field => { "host_name" => "%{[host][name]}"}
  14.         }
  15.         mutate {
  16.             split => ["[log][file][path]", "/"]
  17.         }
  18.     mutate {
  19.               split => ["[log][file][path][-1]", "."]
  20.     }
  21.         mutate {
  22.                 add_field => { "log_name" => "%{[log][file][path][-1][0]}" }
  23.         }
  24.         mutate{
  25.             remove_field => ["log", "host"]
  27.         }  
  28. }
  30. output {
  31.     #stdout { 
  32.     #    codec => rubydebug
  33.     #}
  35.     elasticsearch {
  36.         hosts => [ "192.168.20.11:9200","192.168.20.12:9200","192.168.20.13:9200" ]
  37.         user => "elastic"
  38.         password => "abcd"
  39.         index => "logstash-app_%{project_name}-%{+YYYY.MM.dd}"
  40.         codec => json { charset => "UTF-8" }
  41.     }
  42. }