如何将filebeat和logstash接入OpenStack进行日志管理？
摘要：背景 本文原发布于CSDN，现搬运至此，免得某天突然变成VIP文章，自己都看不了。 文中有2台主机： （1）1.1.1.1是OpenStack的nova节点，安装filebeat （2）1.1.1.2是logstash节点 在1.1.1.1
背景
本文原发布于CSDN，现搬运至此，免得某天突然变成VIP文章，自己都看不了。
文中有2台主机：
（1）1.1.1.1是OpenStack的nova节点，安装filebeat
（2）1.1.1.2是logstash节点
在1.1.1.1上通过filebeat读取OpenStack的nova-compute组件日志（/var/log/nova/nova-compute.log）的增量日志送至logstash，处理后送至下一步环节（终端输出、写入文件、kafka、elasticsearch）。
nova-compute日志格式示例如下：
2023-05-10 23:21:53.513 7 INFO nova.virt.driver [req-59da3b8b-04ed-446d-8d3e-838d3918bdc8 - - - - -] Loading compute driver 'libvirt.LibvirtDriver'
日志格式为“日期 时间 进程id（网上未查到信息） 日志等级 代码模块 请求id 日志正文”。
filebeat
安装
本文从官网（https://www.elastic.co/cn/downloads/beats/filebeat）下载rpm包并安装。
rpm -ivh filebeat-8.13.0-x86_64.rpm
配置
修改/etc/filebeat/filebeat.yml如下：
filebeat.inputs:
- type: log
enabled: true
paths:
- /var/log/nova/nova-compute.log
start_position: end
multiline.pattern: '^\d{4}-\d{2}-\d{2}\s'
multiline.negate: true
multiline.match: after
fields:
hostip: 1.1.1.1
service: nova
component: nova-compute
fields_under_root: true
output.console:
pretty: true
enable: true
output.logstash:
hosts: ["1.1.1.2:5000"]
其中部分配置解释如下：
（1）start_position
start_position: end表示从末尾开始读取，则文件中原有的历史日志都会被忽略。
（2）multiline
multiline.pattern: '^\d{4}-\d{2}-\d{2}\s'
multiline.negate: true
multiline.match: after
针对日志正文可能出现的换行符情况，对不匹配^\d{4}-\d{2}-\d{2}\s正则表达式的行，追加至上一行后面。
（3）fields
通过fields新增三个字段hostip、service、component
（4）fields_under_root
将新增的三个字段hostip、service、component放置于json的顶层字段
（5）output.console
结果通过终端输出，注意如果使用systemcctl命令后台运行则不会输出，此配置只有前台运行才会生效
（6）output.logstash
输出至logstash
运行
用如下命令测试:
filebeat -e -c /etc/filebeat/filebeat.yml
由于配置中有output.console，通过终端输出示例如下：
{
"@timestamp": "2024-07-31T13:35:24.159Z",
"@metadata": {
"beat": "filebeat",
"type": "_doc",
"version": "8.13.0"
},
"agent": {
"name": "host",
"type": "filebeat",
"version": "8.13.0",
"ephemeral_id": "e8367621-cd9a-4fc2-ad71-be60f93976d5",
"id": "0fef92e2-5af8-4706-ba2d-6f3cd1416f40"
},
"log": {
"file": {
"path": "/var/log/nova/nova-compute.log"
},
"offset": 1067
},
"message": "2023-05-10 23:21:52.593 7 INFO os_vif [-] Loaded VIF plugins: linux_bridge, noop, ovs",
"input": {
"type": "log"
},
"service": "nova",
"ecs": {
"version": "8.0.0"
},
"host": {
"name": "host"
},
"component": "nova-compute",
"hostip": "1.1.1.1"
}
配置文件测试无误后，可以使用systemctl start filebeat后台运行。
logstash
安装
官网（https://www.elastic.co/cn/downloads/logstash）下载rpm包并安装。
rpm -ivh logstash-8.14.1-x86_64.rpm
配置
logstash的配置文件目录为/etc/logstash/conf.d，logstsh会读取目录下的所有配置文件，从而可以分别启用不同的端口、进行不同的处理和输出。但是logstsh只会简单的将所有配置拼凑起来，从而导致结果混乱，本文中通过type字段进行区分。
新建openstack.conf如下：
input {
beats {
port => 5000
type => "openstack"
}
}
filter {
if [type] == "openstack" {
mutate {
add_field => [ "[resource_pool]", "KVM01" ]
gsub => ["message", "\n", " "]
}
grok {
match => ["message", "(?<log_time>^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}\.\d{3}) %{INT:pid} %{LOGLEVEL:log_level} (?<module>(\S)+) (?<request_id>\[(\s|\w|\d|-)+\]) %{GREEDYDATA:log_message}"]
tag_on_failure => ["_grokparsefailure_openstack"]
}
if "_grokparsefailure_openstack" in [tags] {
drop {}
}
}
}
output {
if [type] == "openstack" {
stdout { codec => rubydebug }
file {
path => "/var/log/openstack-log.log"
codec => line {
format => "%{resource_pool} %{[host][name]} %{hostip} %{service} %{component} %{log_time} %{log_level} %{module} %{request_id} %{log_message}"
}
}
gzip => false
}
kafka {
bootstrap_servers => "2.2.2.2:8333,2.2.2.3:8333,2.2.2.4:8333"
security_protocol => "SASL_PLAINTEXT"
sasl_mechanism => "SCRAM-SHA-512"
sasl_jaas_config => "org.apache.kafka.common.security.scram.ScramLoginModule required username='xxxxxxx' password='xxxxxxxxxxx';"
topic_id => "OPENSTACK-LOG"
codec => json {
charset => "UTF-8"
}
}
elasticsearch {
hosts => ["1.1.1.3:9200"]
}
}
}
其中部分配置解释如下：
（1）input
启动5000端口接受filebeat信息，并在接收的每一条消息中增加type字段
此外也可以用如下命令通过标准终端输入进行测试:
input {
stdin {
type => "openstack"
codec => "plain"
}
}
在终端中直接输入字符串，会默认作为message字段进行测试。
（2）filter
mutate
add_field：增加字段
gsub：将message字段中的换行符替换为空
grok
将message拆分：
匹配^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}.\d{3}的部分作为 log_time字段
数字作为pid字段
LOGLEVEL类型字符串作为log-level字段
匹配\S)+的部分作为module字段
匹配[(\s|\w|\d|-)+]的部分作为request-id
剩余的GREEDYDATA类型作为log-message字段
tag_on_failure
如果不匹配grok的规则，则在消息中增加"tags" => [0] "_grokparsefailure_openstack"并且丢弃
（3）output
本文中演示了stdout、file、kafka、elasticsearch四种对接方法
启动
用以下命令测试配置文件是否正确：
/usr/share/logstash/bin/logstash --path.settings /etc/logstash/ -f /etc/logstash/conf.d/openstack.conf --config.test_and_exit
配置文件测试无误，用如下命令前台启动：
/usr/share/logstash/bin/logstash --path.settings /etc/logstash/ -f /etc/logstash/conf.d/openstack.conf
检查无误后可以使用systemctl start logstash后台运行。