以前使用Logstash时，都是通过logstash来对日志内容做过滤解析等操作，现在6.3.0版本中，可以通过filebeat直接写数据到es中，要对日志内容做处理的话设置对应的pipeline就可以。
以gunicorn的access日志内容为例：
[10/Jul/2018:11:13:55 +0800] 10.242.169.166 "-" "Jakarta Commons-HttpClient/3.0" "POST /message/entry HTTP/1.0" 200 <13968> 0.061638
有以上内容的日志，记录请求发生的时间，发起请求的ip，referer，useragent，status_line， status_code, 进程id， 请求执行时间。
在不使用pipeline的情况下，在kibana中可以看到日志内容在message字段中，如果想要获取每个请求的执行时间是拿不到的。
使用pipeline，需要现在es中增加对应的pipeline，可以在kibana的devtools界面的console中写入，然后点击执行即可
PUT _ingest/pipeline/gunicorn-access
{
"description" : "my gunicorn access log pipeline",
"processors": [
{
"grok": {
"field": "message",
"patterns": ["\\[%{HTTPDATE:timestamp}\\] %{IP:remote} \"%{DATA:referer}\" \"%{DATA:ua}\" \"%{DATA:status_line}\" %{NUMBER:status_code:int} <%{NUMBER:process_id:int}> %{NUMBER:use_time:float}"]
}
}
]
}
processor使用的grok，主要是patterns的编写，es的 默认正则pattern可以直接使用。注意JSON转义符号。
NUMBER类型最好指定是int还是float，不然默认是string，搜索的时候可能会有问题。
在写patterns的时候，可以借助devtools界面的grokdebugger工具检测是否正确。测试无误即可执行put操作。完成后修改filebeat配置
inputs中设置type字段 用于判断
- type: log
enabled: true
paths:
- /path/to/gunicorn-access.log
fields:
type: gunicorn-access
multiline.pattern: ^\[
multiline.negate: true
multiline.match: after
es output中添加
pipelines:
- pipeline: gunicorn-access
when.equals:
fields.type: gunicorn-access
重启。
在开启自带的nginx日志监控时，nginx的错误日志时间会比当前时间快8小时，需要对它的pipeline设置时区，设置方法为：
通过 GET _ingest/pipeline/先找到nginx error log的pipeline名字为： filebeat-6.3.0-nginx-error-pipeline
复制他的pipeline配置，在date字段下添加timezone配置
{
"date": {
"field": "nginx.error.time",
"target_field": "@timestamp",
"formats": [
"YYYY/MM/dd H:m:s"
],
"timezone": "Asia/Shanghai"
}
}
然后将新的完整pipeline put到es中 PUT _ingest/pipeline/filebeat-6.3.0-nginx-error-pipeline。然后重启es才能生效