filebeat 解析日志 并发送到Elasticsearch
起先，是出于了解我的网站逐步前行STEP的访问情况而做一个Nginx日志统计分析的功能，首选的就是ELK，但是，由于Logstash占用内存和CPU占有率都不是我的小服务器能承受的，转而将logstash换成filebeat，因为filebeat足够轻量级，我所需要的功能却都能满足：
- 收集日志发送到ES
- 按指定格式解析日志
我的个人博客逐步前行STEP
第1点是filebeat基本的功能，只要正确安装配置就能生效，在我的实践中，我不想要所有的日志都发送到ES，只要访问我的网站的请求，示例请求：
39.179.50.187 - - [05/Feb/2020:16:10:42 +0800] "GET /csdn_article-98609482 HTTP/1.1" 499 0 "http://www.hezehua.net/" "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_13_6) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/79.0.3945.130 Safari/537.36"这是标准的nginx日志格式，其中有一个我的网站逐步前行STEP域名http://www.hezehua.net，所以只要匹配这个域名即是我需要的日志，filebeat的include_lines配置可以满足我的需求，将该配置项设置如下：
include_lines: ["hezehua.net"]即匹配了包含hezehua.net的行。
我的个人博客逐步前行STEP
要注意的是，第2点需求，只有Elasticsearch5.x才能实现，因为elasticsearch5.x开始引入了Ingest Node，使其具有预处理能力，即对存储在 Elasticsearch 里的数据在存储之前进行加工处理，之前的版本没有提供这一功能，在数据存储前做处理只能依赖于logstash，而官方将预处理功能集成到elasticsearch5.x这个功能就命名为Ingest，具有预处理功能的节点称为Ingest Node，要使用预处理功能，先要了解：Pipeline（管道）、Processors（处理器）。
1、pipeline
管道这个概念并不陌生，一般被命名于设计用来在拦截、处理、输出数据的功能，在这里也是一样，pipeline会用于处理通过它的所有数据，输出后就直接入elasticsearch。
2、processor
处理器是用于管道的，pipeline本身不具备处理能力，需要配置processor才能发挥特定的作用，常用的处理器有：
- append 追加字段或者已有字段追加值
- convert 转换字段类型
- date 日期处理
- foreach 遍历数组对数组元素使用处理器
- grok 通过正则匹配将文本解析成指定格式
- join 将数组转化成字符串
- json 处理成json
- kv 将k:v型数据解析成一个个字段
- remove 删除字段
- rename 重命名
- set 设置或者新增
- split 分割字符串
- sort 数组元素排序
- trim 消除两端空格
- lowercase 转小写
- uppercase 转大写
以上列举的处理器主要是数组、字符串的常用处理，加上字段增、减等，其中最重要的Grok处理器，这个处理器基本可以替代logstash处理日志格式的功能，语法和logstash一样。
要使用elasticsearch的预处理功能，首先需要在es中创建管道，指定的URL为：_ingest/pipeline ，PUT方式：
curl XPUT http://127.0.0.1:9200/_ingest/pipeline/pipeline-test
{
"description": "describe pipeline",
"processors": [
{
"grok": {
"field": "message",
"patterns": [
"%{IP:ip} - %{DATA:user} \\[%{HTTPDATE:date}\\] %{WORD:method} %{DATA:path} (?<http>HTTP/%{NUMBER}) %{NUMBER:code} %{NUMBER:length} %{DATA:referer} %{DATA:user_agent} %{DATA:x_forwarded_for}"
]
}
}
]
}description字段是一个描述，processors定义了一组应用于该管道的处理器，这样就定义好了一个简单的pipeline了。
我的个人博客逐步前行STEP
将filebeat.yml中，output.elasticsearch下的配置项pipeline设置为上面新建的管道'pipeline-test'即可：
output.elasticsearch:
# Array of hosts to connect to.
hosts: ["localhost:9200"]
pipeline: "pipeline-test"
最后，删除registry文件，和filebeat之前生成的索引，再启动filebeat，就可以在es中看到解析成指定字段的日志文档了：
{
_index: "filebeat-2020.02.04",
_type: "doc",
_id: "AXAQe35OpxWsiKaUFgtz",
_score: 1,
_source: {
date: "04/Feb/2020:21:53:54 +0800",
referer: "http://www.hezehua.net/",
code: "200",
offset: 437,
method: "GET",
ip: "39.179.50.187",
input_type: "log",
length: "169909",
source: "/usr/logs/blog.access.log",
message: "39.179.50.187 - - [04/Feb/2020:21:53:54 +0800] "GET /favicon.ico HTTP/1.1" 200 169909 "http://www.hezehua.net/" "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_13_6) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/79.0.3945.130 Safari/537.36"",
type: "log",
path: "/favicon.ico",
@timestamp: "2020-02-04T13:54:48.214Z",
beat: {
hostname: "novalocal",
name: "novalocal",
version: "5.5.3"
},
http: "HTTP/1.1",
user-agent: "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_13_6) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/79.0.3945.130 Safari/537.36"
}
},我的个人博客逐步前行STEP