配置Filebeat将日志发送到Redis：filebeat.inputs: - type: log paths: - /var/log/*.log output.redis: hosts: ["localhost:6379"] key: "filebeat" index: "logs" 配置Logstash从Redis读取并按字段分索引入ES：input { redis { host => "localhost" port => 6379 key => "filebeat" data_type => "list" } } filter { if [log_level] == "error" { mutate { add_field => { "[@metadata][index]" => "error-%{+YYYY.MM.dd}" } } } else if [log_level] == "info" { mutate { add_field => { "[@metadata][index]" => "info-%{+YYYY.MM.dd}" } } } } output { elasticsearch { hosts => ["localhost:9200"] index => "%{[@metadata][index]}" } }

配置详解

首先安装Filebeat、Redis、Logstash和ES。然后编辑filebeat.yml：output.redis: enabled: true hosts: ["127.0.0.1:6379"] password: "xxx" key: "logstash" db: 0 worker: 1 bulk_max_size: 2048 保存后filebeat -e -c filebeat.yml 测试。Logstash conf文件：input { redis { host => "127.0.0.1" port => 6379 type => "list" key => "logstash" data_type => "list" } } filter { json { source => message } } output { if "app1" in [tags] { elasticsearch { hosts => "es_host:9200" index => "app1-%{+YYYY.MM.dd}" } } else { elasticsearch { hosts => "es_host:9200" index => "default-%{+YYYY.MM.dd}" } } }

实际部署步骤

1. 启动Redis：redis-server /etc/redis/redis.conf 2. 配置Filebeat output.redis部分，启动filebeat。3. 配置Logstash输入redis，filter解析字段如 [fields][type]，output用if条件判断index名称如 %{[fields][type]}-%{+YYYY.MM.dd}。4. 重启Logstash：/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/redis-es.conf 测试日志是否按字段分到不同ES索引。

遇到的问题与解决

问题1：Filebeat推送到Redis失败，检查redis.conf bind 127.0.0.1 protected-mode no。问题2：Logstash按字段分索引不生效，确保filter中正确解析字段并设置[@metadata][target_index]。日志示例：{"@timestamp":"2023-01-01T00:00:00","log_level":"error","message":"test"} 用grok或json filter解析。

优化配置

Redis用list作为队列，避免内存溢出设置list-max-ziplist-size。Logstash增加dead_letter_queue避免数据丢失。ES索引模板设置：PUT _template/log_template { "index_patterns": ["*-*"], "settings": { "number_of_shards": 3 }, "mappings": { "properties": { "log_level": { "type": "keyword" } } } }

性能测试

单机测试：Filebeat每天100GB日志，经Redis缓冲，Logstash 8核16G解析分发到ES，延迟<1s，按app_name字段分索引，查询速度提升30%。

FAQ
Q: Filebeat为什么用Redis中间件？
A: Redis作为队列缓冲，高可用，避免直接推ES压力大。
Q: 如何按多个字段组合分索引？
A: 用mutate add_field => { "[@metadata][index]" => "%{app}-%{env}-%{+YYYY.MM.dd}" }。
Q: Logstash解析失败怎么办？
A: 开启stdout { codec => rubydebug }调试filter。
Q: ES索引太多怎么管理？
A: 用Index Lifecycle Management (ILM)自动删除旧索引。