使用

• grok格式化

filter {
    grok {
        match => {
            "message" => '(?<clientip>[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}) - - \[(?<requesttime>[^ ]+ \+[0-9]+)\] "(?<requesttype>[A-Z]+) (?<requesturl>[^ ]+) HTTP/\d.\d" (?<status>[0-9]+) (?<bodysize>[0-9]+) "[^"]+" "(?<ua>[^"]+)"'
        } 
    }
}

• 跳过匹配失败的数据

output{
    if "_grokparsefailure" not in [tags] and "_dateparsefailure" not in [tags] {
        elasticsearch {
            hosts => ["http://192.168.237.50:9200"]
        }
    }
}

• 去除不要字段

filter {
    grok {
        match => {
            "message" => '(?<clientip>[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}) - - \[(?<requesttime>[^ ]+ \+[0-9]+)\] "(?<requesttype>[A-Z]+) (?<requesturl>[^ ]+) HTTP/\d.\d" (?<status>[0-9]+) (?<bodysize>[0-9]+) "[^"]+" "(?<ua>[^"]+)"'
        }
        remove_field => ["message","@version","path"]
    }
}

• 一次性收集日志，且指定排序时间轴

input {
  file {
    path => "/usr/local/nginx/logs/access.log"
    start_position => "beginning"
    sincedb_path => "/dev/null"
  }
}
filter {
    grok {
        match => {
            "message" => '(?<clientip>[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}) - - \[(?<requesttime>[^ ]+ \+[0-9]+)\] "(?<requesttype>[A-Z]+) (?<requesturl>[^ ]+) HTTP/\d.\d" (?<status>[0-9]+) (?<bodysize>[0-9]+) "[^"]+" "(?<ua>[^"]+)"'
        }
        remove_field => ["message","@version","path"]
    }
    date {
        match => ["requesttime", "dd/MMM/yyyy:HH:mm:ss Z"]
        target => "@timestamp"
    }
}

统计Nginx的请求和网页显示进行对比
cat /usr/local/nginx/logs/access.log |awk '{print $4}'|cut -b 1-19|sort |uniq -c
20/Feb/2019:14:50:06 -> dd/MMM/yyyy:HH:mm:ss
2016-08-24 18:05:39,830 -> yyyy-MM-dd HH:mm:ss,SSS

• 读取json格式日志

#
filter {
  json {     
    source => "message"     
    remove_field => ["message","@version","path","beat","input","log","offset","prospector","source","tags"]   }
}

• Filebeat采集多个日志配置

filebeat.inputs:
- type: log
  tail_files: true
  backoff: "1s"
  paths:
      - /usr/local/nginx/logs/access.json.log
  fields:
    type: access
  fields_under_root: true
- type: log
  tail_files: true
  backoff: "1s"
  paths:
      - /var/log/secure
  fields:
    type: secure
  fields_under_root: true
output:
  logstash:
    hosts: ["192.168.237.51:5044"]

• Logstash通过type字段进行判断

input {
        beats {
                host => '0.0.0.0'
                port => 5044 
        }
}

filter {
  if [type] == "access" {
    json {
      source => "message"
      remove_field => ["message","@version","path","beat","input","log","offset","prospector","source","tags"]
    }
  }
}

output{
  if [type] == "access" {
    elasticsearch {
      hosts => ["http://192.168.237.50:9200"]
      index => "access-%{+YYYY.MM.dd}"
    }
  }
  else if [type] == "secure" {
    elasticsearch {
      hosts => ["http://192.168.237.50:9200"]
      index => "secure-%{+YYYY.MM.dd}"
    }
  }
}