ELK+Filebeat 集中式日志解决方案详解

2020-12-19 06:19:59 编辑 2 阅读量:1510

ELK 不是一款软件，而是 Elasticsearch、Logstash 和 Kibana 三种软件产品的首字母缩写。这三者都是开源软件，通常配合使用，而且又先后归于 Elastic.co 公司名下，所以被简称为 ELK Stack。根据 Google Trend 的信息显示，ELK Stack 已经成为目前最流行的集中式日志解决方案。
- Elasticsearch：分布式搜索和分析引擎，具有高可伸缩、高可靠和易管理等特点。基于 Apache Lucene 构建，能对大容量的数据进行接近实时的存储、搜索和分析操作。通常被用作某些应用的基础搜索引擎，使其具有复杂的搜索功能；
- Logstash：数据收集引擎。它支持动态的从各种数据源搜集数据，并对数据进行过滤、分析、丰富、统一格式等操作，然后存储到用户指定的位置。由于消耗cpu较大，一般不用于搜集数据，而是用于解析数据；
- Kibana：数据分析和可视化平台。通常与 Elasticsearch 配合使用，对其中数据进行搜索、分析和以统计图表的方式展示；
- Filebeat：ELK 协议栈的新成员，一个轻量级开源日志文件数据搜集器，基于 Logstash-Forwarder 源代码开发，是对它的替代。在需要采集日志数据的 server 上安装 Filebeat，并指定日志目录或日志文件后，Filebeat 就能读取数据，迅速发送到 Logstash 进行解析，亦或直接发送到 Elasticsearch 进行集中式存储和分析。

## 一、安装
logstash和filebeat的安装很简便，直接解压相应的tar包即可。Elasticsearch的安装参见网站上的其他文章
`tar xzvf logstash-5.4.3.tar.gz -C /opt`

## 二、使用架构
### 2.1 Logstash 作为日志搜集器
logstash 通过输入插件从多种数据源（比如日志文件、标准输入 Stdin 等）获取数据，再经过滤插件加工数据，然后经 Elasticsearch 输出插件输出到 Elasticsearch，通过 Kibana 展示。logstash分布于多台机器详见图 2。
![](http://cblog.xyz/media/2017/10/image002.png)
图 2. Logstash 作为日志搜索器

这种结构因为需要在各个服务器上部署 Logstash，而它比较消耗 CPU 和内存资源，所以比较适合计算资源丰富的服务器，否则容易造成服务器性能下降，甚至可能导致无法正常工作。

### 2.2 Beats 作为日志搜集器
这种架构引入 Beats 作为日志搜集器。目前 Beats 包括四种：

- Packetbeat（搜集网络流量数据）；
- Topbeat（搜集系统、进程和文件系统级别的 CPU 和内存使用情况等数据）；
- Filebeat（搜集文件数据）；
- Winlogbeat（搜集 Windows 事件日志数据）。

Beats 将搜集到的数据发送到 Logstash，经 Logstash 解析、过滤后，将其发送到 Elasticsearch 存储，并由 Kibana 呈现给用户。详见图 3。
![](http://cblog.xyz/media/2017/10/image003.png)
图 3. Beats 作为日志搜集器

这种架构解决了 Logstash 在各服务器节点上占用系统资源高的问题。相比 Logstash，Beats 所占系统的 CPU 和内存几乎可以忽略不计。另外，Beats 和 Logstash 之间支持 SSL/TLS 加密传输，客户端和服务器双向认证，保证了通信安全。

因此这种架构适合对数据安全性要求较高，同时各服务器性能比较敏感的场景。

### 2.3 引入消息队列机制的架构
filebeat 直接到logstash, 由于logstash的设计问题, 可能会出现阻塞问题, 这里可以中间使用消息队列分开。消息队列可以使用redis, 或者kafka。
这种架构比较稳定高效，用的较多。

## 三、ELK简单使用
### 3.1 logstash的使用
logstash包含三个部分：input，filter，output。按照字面意思很容易理解，input负责日志的输入，可以直接从文件也可以从kafka，filebeat等等，filter负责日志的处理，包括增减字段，将行结构的日志转为json格式等等，output负责日志的输出，输出部分也包含很多：Elasticsearch，kafka，file等等。
#### 3.1.1 input
这里举两个简单的例子
- file
直接读取日志文件，`beginning`表示从头开始，读取后logstash会将读取的进度保存起来，如果意外重启后，logstash会从读取保存的进度，然后将之后的日志读入，这样就不会造成重复读取。
```
input {
    file {   
        path => "/home/hadoop/project/lawplatform/logs/*.log"  
        start_position => beginning  
   }
}
```
- generator
generator一般用于测试，简单生成一些假数据
```
input {
    generator {
        count => 10000000
        message => '{"key1": "value1", "key2": [1, 2], "keys3":{"subkey1": "subvalue1"}}'
        codec => json
    }   
}
```
- filebeat
读取从filebeat发来的数据，其实就是监听特定的端口，下面是最简单的配置，不包含ssl认证，安全性比较低
```
input {
    beats {
        port => 5044
        ssl => false
    }
}
```

#### 3.1.2 filter
日志数据一般都是非结构化数据，而且包含很多不必要的信息，所以需要在 Logstash 中添加过滤插件对 Filebeat 发送的数据进行结构化处理。[官网filter说明](https://www.elastic.co/guide/en/logstash/5.4/filter-plugins.html)
比较重要的filter有grok（正则匹配）、date（提取时间）、mutate（处理字段）等

- 使用 grok 正则匹配把那些看似毫无意义、非结构化的日志数据解析成可查询的结构化数据，是目前 Logstash 解析过滤的最好方式。
Grok 的用户不需要从头开始写正则。ELK github 上已经写好了很多有用的模式，比如日期、邮箱地址、IP4/6、URL 等。具体查看[logstash-patterns-core](https://github.com/logstash-plugins/logstash-patterns-core/blob/master/patterns/grok-patterns)。除此之外，还有 grok 正则表达式的[debug 工具](http://grokdebug.herokuapp.com/)，能方便快速检验所写表达式是否正确。在写grok正则时，一定要先用debug工具看看是否写错
- date能将日志中的日期转为某个特定的时间字段（@timestamp），便于使用kibana查看，不然如果用logstash传输日志的时间作为@timestamp会无法体现日志的时间，不便于查询日志
- mutate能对字段做处理

下面是一个简单的配置：
```
filter {
    grok {
        #hbase: 2017-08-22 10:52:54,455 INFO org.apache.hadoop.hbase.util.VersionInfo: From source with checksum c6d9864e1358df7e7f39d39a40338b4e
        match => {
            "message" => "%{TIMESTAMP_ISO8601:logdate} (?<file>.*?) (?<class>.*?): (?<content>.*)"
        }   
    }   
    date {
        match => ["logdate", "yyyy-MM-dd HH:mm:ss,SSS"]
        target => "@timestamp"
    }   
    mutate {
        remove_field => ["logdate"]
    }   
}
```
- %{TIMESTAMP_ISO8601:logdate}表示使用grok内置的TIMESTAMP_ISO8601格式将`2017-08-22 10:52:54,455`提取处理作为logdate字段（logdate是自己命名的，可以任意名字）
- match可以写多个正则，来一条日志时，logstash会遍历match中的正则表达式，直到有正确匹配的
```
grok {
        match => {
            "message" => "%{TIMESTAMP_ISO8601:logdate}(?<file>.*?)\[line:(?<line>.*?)\](?<logger_name>\w*?) (?<level>\w*?)-.*?\"(?<method>\w*?) (?<url>.*?) .*? (?<status_code>.*) -" 
        }
        match => {
            # "%(asctime)s %(filename)s[line:%(lineno)d]%(name)s %(levelname)s-%(message)s"
            "message" => "%{TIMESTAMP_ISO8601:logdate}(?<file>.*?)\[line:(?<line>.*?)\](?<logger_name>\w*?) (?<level>\w*?)-(?<content>.*)"
        }
    }  
```
- date将logdate这个字段转为@timestamp字段
- mutate表示移除logdate字段，因为logdate已经复制为@timestamp字段，所以可以移除

所以，`2017-08-22 10:52:54,455 INFO org.apache.hadoop.hbase.util.VersionInfo: From source with checksum c6d9864e1358df7e7f39d39a40338b4e`会被转为
```
{"message": "2017-08-22 10:52:54,455 INFO org.apache.hadoop.hbase.util.VersionInfo: From source with checksum c6d9864e1358df7e7f39d39a40338b4e", "file": "INFO", "class":"org.apache.hadoop.hbase.util.VersionInfo", "content":"From source with checksum c6d9864e1358df7e7f39d39a40338b4e", "@timestamp":"2017-08-22T07:45:44.531Z"  }
```

#### 3.1.3 output
output有很多，这里简单介绍一些Elasticsearch的
```
output {
    elasticsearch {
        hosts => '192.168.10.219' 
        user => 'elastic'
        password => 'changeme'   
     }
    stdout { codec => rubydebug }
}
```
- stdout中可以使用json：`stdout { codec => json }`, rubydebug显示的格式较为清楚，但是占用行数多

output中也可以根据不同的条件将日志建立到不同的索引中，默认的索引为`logstash-*`，系统会自动根据当前时间补全索引名，比如`logstash-2015.01.23`
```
output {
    if [type] == 'Hbase' {
        elasticsearch {
            hosts => '192.168.10.219'
            user => 'elastic'
            password => 'changeme'
            index => "logstash-hbase-%{+YYYY.MM}"
         }   
    }   
    else {
        elasticsearch {
            hosts => '192.168.10.219'
            user => 'elastic'
            password => 'changeyou'
         }   
    }   
    stdout { codec => 'json' }
}
```

综合起来，一个简单的配置如下
```
input { 
    file {               
        path => "/home/hadoop/project/lawplatform/logs/*.log"  
        start_position => beginning  
   }   
}
  
filter {
    grok {
        match => {
            "message" => "%{TIMESTAMP_ISO8601:logdate}(?<file>.*?)\[line:(?<line>.*?)\](?<logger_name>\w*?) (?<level>\w*?)-.*?\"(?<method>\w*?) (?<url>.*?) .*? (?<status_code>.*) -" 
        }
        match => {
            "message" => "%{TIMESTAMP_ISO8601:logdate}(?<file>.*?)\[line:(?<line>.*?)\](?<logger_name>\w*?) (?<level>\w*?)-(?<content>.*)"
        }
    }
    date {
        match => ["logdate", "yyyy-MM-dd HH:mm:ss,SSS"]
        target => "@timestamp"    
    }
    mutate {
        remove_field => ["logdate"]
    }
}

output {
    elasticsearch {
        hosts => '192.168.10.219' 
        user => 'elastic'
        password => 'changeme'   
     }
    stdout { codec => rubydebug }
}
```

#### 3.1.4 运行logstash
将配置保存为`example.conf`，并在shell中运行
`$ $LOGSTASH-HOME/bin/logstash -f the/path/to/example.conf`

### 3.2 Elasticsearch的使用
参考[elasticsearch集群的搭建以及常用插件（分词器、web页面）的安装](http://cblog.xyz/article/227), [elasticsearch的一些基本概念、使用以及参考资料](http://cblog.xyz/article/228)

### 3.3 kibana的使用
- 登陆kibana页面，在`Management`页面中的`Index Patterns`中添加`logstash-*`，并且`Time-field name`选择`@timestamp`
- 在kibana页面，点击`Discover`，选择`logstash-*`就可以看到有相关数据（前提是已经根据前面步骤导入了一些数据），点击页面右上角可以选择时间，页面左边可以过滤显示一些字段，还有搜索框
- 在kibana页面，点击`Visualize`可以建立一些可视化图表
- `Dashboard`可以将多个图表综合起来，形成一个可视化页面

## 四、ELK+filebeat使用
### 4.1 filebeat的使用
在filebeat的tar安装包中有示例的配置，只需要稍微改一些就可以用，如下：
```
filebeat.prospectors:

# Each - is a prospector. Most options can be set at the prospector level, so
# you can use different prospectors for various configurations.
# Below are the prospector specific configurations.

- input_type: log

# Paths that should be crawled and fetched. Glob based paths.
  paths:
    - /home/hadoop/hbase-cmf-hbase-HBASETHRIFTSERVER-cdh-slave3.log.out
    #- c:\programdata\elasticsearch\logs\*
  document_type : Hbase
  
#----------------------------- Logstash output --------------------------------
output.logstash:
  # The Logstash hosts
  hosts: ["192.168.10.225:5044"]

```
- document_type: 为自己添加的字段，用来给logstash提示日志从哪里来，也就是上面output中`if [type]`中的type

保存配置为example.yml,然后在shell中运行
`$ $FILEBEAT-HOME/filebeat  -c the/path/to/example.yml -e```

### 4.2 ELK部分
elk部分和上面类似，就只需要改logstash的input即可
```
input {
    beats {
        port => 5044
        ssl => false
    }
}
filter {
    grok {
        match => {
            "message" => "%{TIMESTAMP_ISO8601:logdate}(?<file>.*?)\[line:(?<line>.*?)\](?<logger_name>\w*?) (?<level>\w*?)-.*?\"(?<method>\w*?) (?<url>.*?) .*? (?<status_code>.*) -" 
        }
        match => {
            "message" => "%{TIMESTAMP_ISO8601:logdate}(?<file>.*?)\[line:(?<line>.*?)\](?<logger_name>\w*?) (?<level>\w*?)-(?<content>.*)"
        }
    }
    date {
        match => ["logdate", "yyyy-MM-dd HH:mm:ss,SSS"]
        target => "@timestamp"    
    }
    mutate {
        remove_field => ["logdate"]
    }
}

output {
    elasticsearch {
        hosts => '192.168.10.219' 
        user => 'elastic'
        password => 'changeme'   
     }
    stdout { codec => rubydebug }
}
```

## 五、注意事项
1. logstash的到Elasticsearch的索引不支持大写字母
> The index to write events to. This can be dynamic using the %{foo} syntax. The default value will partition your indices by day so you can more easily delete old data or only search specific date ranges. **Indexes may not contain uppercase characters**. For weekly indexes ISO 8601 format is recommended, eg. logstash-%{+xxxx.ww}. LS uses Joda to format the index pattern from event timestamp. Joda formats are defined here.

---
参考：
- [1][ELK+Filebeat 集中式日志解决方案详解](https://www.ibm.com/developerworks/cn/opensource/os-cn-elk-filebeat/index.html)
- [2][logstash-patterns-core](https://github.com/logstash-plugins/logstash-patterns-core/blob/master/patterns/grok-patterns)
- [3][grok正则debug](http://grokdebug.herokuapp.com/)
- [4][Logstash Introduction](https://www.elastic.co/guide/en/logstash/5.4/introduction.html)
- [5][Getting Started With Filebeat](https://www.elastic.co/guide/en/beats/filebeat/current/filebeat-getting-started.html)
- [6][使用filebeat进行数据加密传输和区别不同的日志源 ](http://blog.csdn.net/Gamer_gyt/article/details/69280693?locationNum=5&fps=1)
- [7][logstash输出到elasticsearch多索引](http://blog.csdn.net/wangyangzhizhou/article/details/53314022)
- [8][Logstash conf 参数解释归纳](http://blog.csdn.net/textboy/article/details/45823829)
- [9][logstash remove_field =>["message"] ](http://blog.csdn.net/zhaoyangjian724/article/details/54343178)
- [10][logstash如何将日志中字符串类型的时间转化成@timestamp](http://blog.csdn.net/xiaoyu_bd/article/details/52531051)
- [11][ logstash快速入门 ](http://blog.csdn.net/wp500/article/details/41040213)
- [12][logstash日志分析的配置和使用](http://www.cnblogs.com/yincheng/p/logstash.html)
- [13][ELK Stack](https://xiangshan.gitbooks.io/elkstack/)

ELK+Filebeat 集中式日志解决方案详解

附件: