filebeat收集日志输出到logstash然后合并输出到同名文件(7.7.1)
文章来源:原创
作者:临窗旋墨
发布时间:2020-09-30
阅读:979
标签:日志
许秋冬偷偷隐藏起来的markdown原文,测试对蜘蛛是否友好
[toc]
### [filebeat收集日志输出到logstash然后合并输出到同名文件(7.7.1)](http://www.xuqiudong.cn/detail/14)
`Vic.xu` `20200930`
#### 目标:多台集群机器日志收集到一起,且同名输出,能记录来源机器
1. 多台机器输出的日志名是相同的,比如机器A 输出的日志包括`err.log`,`sql.log`,`common.log`等
2. 希望收集多台机器的日志,输入到同名日志文件,比如把A机器和B机器的`err.log`,`sql.log`,`common.log`收集到C机器的`err.log`,`sql.log`,`common.log`;
3. 在日志中标识来自哪一台机器;
4. 多行日志需要合并,比如一个堆栈信息;
#### 前言:项目有点老了(着急的话,直接调到[最后](#### 五 总结: `linux`下的启动脚本和最终配置文件)看结论吧)
> 一说分布式日志收集分析系统,网上都有一大推的关于ELK的文章。 不过咱的项目比较老了,只是最近才真正做了集群部署,所以才会涉及到分布式的日志收集。
>
> 不过,项目中暂时也不会做日志分析之类的东西。所以就准备先弄一个`filebeat`收集日志到`logstash`,然后通过`logstash`定向输出到文件即可。加上之前因为开发环境和测试环境的隔离,所以写了个实时页面查看日志的小项目,也是勉强够用了。
> **版本**:(因为之前使用过`ElasticSearch`的版本是7.7.1)
>
> - `Filebeat` `7.7.1`
> - `Logstash` `7.7.1`
#### 一 为什么不只使用`Logstash`,而是额外引入了`Filebeat`
1. 因为`logstash`是`jvm`跑的,资源消耗比较大,
2. `filebeat`更轻量,占用资源更少
3. 一般结构都是`filebeat`采集日志,然后发送到消息队列,`redis`,`kafaka`。然后`logstash`去获取,利用`filter`功能过滤分析,然后存储到`elasticsearch`中
#### 二 基本思路 `Filebeat` + `Logstash`
1. 在各个需要收集日志的机器上`FileBeat` , 配置输出到远程的`Logstash`中
- [参见Filenbeat官网配置](https://www.elastic.co/guide/en/beats/filebeat/current/configuring-output.html)
- 日志可以输出到`Es`/`Logstash`/`Kafka`/`Redis`/`File`/`Console`,方便后续扩展
2. 在接收日志文件的机器上安装`Logstash`,收集各个`Filebeat`发送来到日志,输出到文件
- `Logstash`的input方式有很多,包括`Filebeat`,`kafka`等: https://www.elastic.co/guide/en/logstash/current/input-plugins.html
- `Logstash`的output方式有很多,非常方便后续扩展,[参见官网](https://www.elastic.co/guide/en/logstash/current/output-plugins.html)
- 由于本系统无特殊的处理日志的需求,可直接把收集到的日志[输出到文件](https://www.elastic.co/guide/en/logstash/7.x/plugins-outputs-file.html)即
- [输出到文件](https://doc.yonyoucloud.com/doc/logstash-best-practice-cn/output/stdout.html)
```
output {
file {
path => "/path/to/%{+yyyy/MM/dd/HH}/%{host}.log.gz"
message_format => "%{message}"
gzip => true
}
}
```
#### 三 windows下简单测试`Filebeat`
1. 官网下载windows版本
2. 修改`filebeat.yml`,具体参考官网
```yaml
filebeat.inputs:
- type: log
enabled: true
fields:
name: vic_first
paths:
# - /var/log/*.log
- E:/space/idea/blog/logback/*.log
output.file:
path: d:/filebeat
filename: filebeat_vic.log
codec.format:
string: '%{[fields][name]}%{[@timestamp]} %{[message]}'
rotate_every_kb: 10240
# 是否开启多行合并
multiline.type: pattern
#multiline.pattern: '^\['
multiline.pattern: '^vic-blog'
multiline.negate: false
multiline.match: after
#============================= Filebeat modules ===============================
filebeat.config.modules:
path: ${path.config}/modules.d/*.yml
reload.enabled: false
processors:
- add_host_metadata: ~
- add_cloud_metadata: ~
- add_docker_metadata: ~
- add_kubernetes_metadata: ~
```
- 如上:
- 1. 配置了需要收集的日志文件的位置。
2. fields 下是携带的额外的数据,一般作为标识使用
3. 配置了日志的输出的文件
4. 配置合并多行:因为默认是读取一行就发送,但是其实有的时候多行才算一个日志,比如堆栈信息等
3. 启动`Filebeat`,然后测试写入日志,会被写入到输出位置
> .\filebeat.exe -e -c filebeat.yml
4. **其他**
1. 输出到`es`:
```yaml
output.elasticsearch:
# Array of hosts to connect to.
hosts: ["localhost:9200"]
# Protocol - either `http` (default) or `https`.
#protocol: "https"
# Authentication credentials - either API key or username/password.
#api_key: "id:api_key"
#username: "elastic"
#password: "changeme"
```
2. 输出到`logstash`
```yaml
output.logstash:
# The Logstash hosts
hosts: ["localhost:5044"]
```
#### 四 windows下简单测试`Logstash`
1. 官网下载zip解压,
2. cofig下复制logstash.conf,然后配置成控制台输入输出:
```yaml
input {
beats {
port => 5044
}
}
filter {
ruby {
code => "
path = event.get('log')['file']['path']
puts format('path = %<path>s', path: path)
if (!path.nil?) && (!path.empty?)
event.set('filename', path.split('/')[-1])
end
"
}
}
output {
file {
# path => "/data/share/logs/output/logstash/%{+yyyy-MM-dd}/%{+HH}-vic.log"
path => "/data/share/logs/output/logstash/%{+yyyy-MM-dd}/%{filename}"
codec => line { format => "%{filename} %{[fields][name]}: %{message}"}
}
stdout {
codec => rubydebug
}
}
```
3. 启动Logstash
> bin\logstash -f config\logstash.conf
- 启动成功后可访问localhost:9600,看到版本等信息,则表示启动成功
- 测试:在启动控制台输入信息, 可看到相关输出信息
#### 一些问题:
1. 在logstash中获取Filebeat中的日志原文件名,然后输出到同名文件:通过filter实现
> 如下,获取输入的日志文件的文件名filename,通过`%{filename}`获取
```
filter {
ruby {
code => "
path = event.get('log')['file']['path']
puts format('path = %<path>s', path: path)
if (!path.nil?) && (!path.empty?)
event.set('filename', path.split('/')[-1])
end
"
}
}
```
2. filebeat携带额外的信息到,标志来自哪台机器,通过fields携带(如下,携带的key是name,value是server1)
```yaml
filebeat.inputs:
- type: log
enabled: true
fields:
name: server1
```
3. logstash获取filebeat携带的额外的信息
```yaml
%{[fields][name]}
```
4. filebeat 不发送最后一行
> Filebeat使用换行符来检测事件的结束。 如果将行逐渐添加到正在采集的文件中,则在最后一行之后需要换行符,否则Filebeat将不会读取文件的最后一行。
#### 五 总结: `linux`下的启动脚本和最终配置文件
> `linux`下的安装和window下没什么区别,`logstash`需要确保安装了`jdk1.8`+;
>
> 启动也是一样的, 只是本人比较懒, 所以写了两个启动脚本`start.sh` ;运行时带参数的话则输出日志到`catalina.out`,不带参数则不输出日志
#### `linux`下的`filebeat`的启动脚本
```shell
#!/bin/bash
log="/dev/null"
# if input paramter is not null ,then output log to catalina.out. else to /dev/null
if [ $1 != '' ];then
log="catalina.out"
fi
nohup ./filebeat -e -c filebeat.yml >$log 2>&1 &
echo "start filebeat.......... ----> $log"
```
##### `linux`下的`logstash`的启动脚本
```shell
#!/bin/bash
log="/dev/null"
# if input paramter is not null ,then output log to catalina.out. else to /dev/null
if [ $1 != '' ];then
log="catalina.out"
fi
nohup bin/./logstash -f config/logstash.conf >$log 2>&1 &
echo "start logstash.......... ----> $log"
```
##### `filebeat.yaml`最终配置文件(见注释)
```yaml
# 输入日志
filebeat.inputs:
- type: log
enabled: true
# 携带的额外的数据,标识来自哪个机器
fields:
name: wsl-ubuntu
# 日志的所在位置
paths:
- /data/share/logs/input/*.log
- /data/share/logs/input2/*.log
#=========================== 是否开启多行合并================================
multiline.type: pattern
#多行合并的正则,以[ 开头
multiline.pattern: '^\['
# 是否不合并多行合并: 负负得正 表示开启
multiline.negate: false
multiline.match: after
#==========================输出日志到logstash==============================
output.logstash:
# The Logstash hosts
hosts: ["localhost:5044"]
```
##### `logstash.conf`最终配置文件(见注释)
```yaml
#======从filebeat获取日志==============================
input {
beats {
port => 5044
}
}
#=====解析出来源日志的文件名===================================
filter {
ruby {
code => "
path = event.get('log')['file']['path']
puts format('path = %<path>s', path: path)
if (!path.nil?) && (!path.empty?)
event.set('filename', path.split('/')[-1])
end
"
}
}
#======输出日志==============================
output {
# 输出日志到文件,在日志的开头打印filebeat中携带的name属性,标识来自哪个文件
file {
path => "/data/share/logs/output/logstash/%{+yyyy-MM-dd}/%{filename}"
codec => line { format => "%{[fields][name]}: %{message}"}
}
# 输出日志到控制台 可删除掉
stdout {
codec => rubydebug
}
}
```
文章来源:[临窗旋墨的博客](http://www.xuqiudong.cn/detail/14),转载注明出处。
时间: 20200930