filebeat+redis+logstach收集日志

filebeat

下载filebeat

wget  http://artifacts.elastic.co/downloads/beats/filebeat/filebeat-7.1.0-x86_64.rpm
rpm -ivh  filebeat-7.1.0-x86_64.rpm

配置filebeat收集系统日志,使用redis

filebeat.inputs:
- type: log
  enabled: false
  paths:
  - /var/log/*.log
filebeat.config.modules:
  path: ${path.config}/modules.d/*.yml
  reload.enabled: false
setup.template.settings:
  index.number_of_shards: 1
setup.kibana:
  host: "192.168.57.3:5601"
output.redis:
  hosts: ["192.168.57.3"] # redis
  key: "filebeat" # filebeat是数据的key
  db: 0 #数据写入的库

filebeat输出到kafka配置如下，

output.kafka: 
  enabled: true 
  hosts: ["192.168.23.76:9092","192.168.23.77:9092","192.168.23.78:9092"]   #集群的地址
  topic: filebeat          #topic

启用和配置system模块，默认情况下都是disabled

filebeat modules enable system 

可以看到system已经去掉disabled

setup 命令加载 Kibana 仪表板。如果仪表板已设置，请省略此命令。 

filebeat setup

启动filebeat

systemctl  start  filebeat

redis

安装redis，我这里使用yum安装，生产建议使用二进制包

yum  install  redis

配置bind 监听本地IP或者0.0.0.0

systemctl start redis
```  

## logstach

下载

wget http://artifacts.elastic.co/downloads/logstash/logstash-7.1.0.zip
unzip  logstash-7.1.0.zip

配置

cd logstash-7.1.0
vim config/logstash-sample.conf    这个文件是系统给的模板 ，我们可以根据这个模板来改
[root@localhost logstash-7.1.0]# cat config/logstash-sample.conf
input {
redis {
type => “systemlog”
host => “192.168.57.3”
port => “6379”
db => “0”
data_type => “list”
key => “filebeat”
}
}
output {
if [type] ==”systemlog”
{
elasticsearch {
hosts => [“http://192.168.57.3:9200"\]
index => “redis-systemlog-%{+YYYY.MM.dd}”
}
}}

kafka 消费到logstach配置如下

kafka{
bootstrap_servers => [“192.168.23.76:9092,192.168.23.77:9092,192.168.23.78:9092”]
group_id => “jcwit”
consumer_threads => “5”
decorate_events => “false”
topics => [“filebeat”]
type => “systemlog”
codec => json

}

这里需要注意的就四个地方  
+	bootstrap_servers 也就是kafka集群的地址，在filebeat端要求单个地址加引号，这里是集群地址放一起加引号。  
+	group_id 这里必须保证唯一，是你这个logstash集群消费kafka集群的身份标识。  
+	topics  filebeat和logstash使用的topic一致。  
+	codec => json  由于beat传输数据给kafka集群的时候，会附加很多tag，默认情况下，logstash就会将这串tag也认为是message的一部分。这样不利于后期的数据处理。所有需要添加codec处理。得到原本的message数据。

启动

./bin/logstash -f ./config/logstash-sample.conf
```