1. 多日志分索引的真实场景：为什么一个索引扛不住业务输出
我在接手一套业务系统的日志平台时，第一周就被运维同事找上门：Kibana里查应用日志越来越慢，索引分片已经堆到几十个，而且日志内容特别杂——Nginx访问日志、后端业务日志、定时任务日志、登录审计日志全混在同一个索引里。表面上只是检索慢，真正麻烦的是后续的保留策略没法做：我想把访问日志只留7天、业务日志留30天、审计日志留半年，在一个索引里根本没法定。
这就是典型的"多日志类型混采"问题。很多团队一开始图省事，Filebeat采集的多个路径全推到同一个Logstash，再写入同一个ES索引，前期确实够用，数据量一到几十GB每天就开始难受。从这时候开始，你就需要把"采集"和"路由"拆开设计：Filebeat负责把不同日志文件读上来，Logstash负责给日志分门别类，最终写入不同的ES索引。
这篇文章我会把整条链路拆开讲：从Filebeat的多输入配置、字段打标，到Logstash的if条件路由、索引模板配合，再到实际验证和排错。整体基于我平时维护一套模拟业务环境（两台应用服务器、一套ELK）的真实经验来写，配置可以直接拿去改路径后复用。
先看整体数据流：
应用服务器A: Nginx访问日志 --> Filebeat(打标: nginx-access) --> Logstash --> ES索引: nginx-access-日期 应用服务器A: 订单业务日志 --> Filebeat(打标: order-app) --> Logstash --> ES索引: order-app-日期 应用服务器B: 登录审计日志 --> Filebeat(打标: audit-log) --> Logstash --> ES索引: audit-log-日期这里的核心设计只有一个：用字段区分来源，用if条件决定去向。Filebeat在每一行日志上打一个标签，Logstash拿到标签后判断要走哪个分支、进哪个索引。下面每一个环节我都会写配置、写原理、写踩坑。
2. Filebeat多输入配置：字段打标和multiline合并的细节
2.1 多input配置的基本写法
Filebeat 7.x之后配置收敛了很多，多输入就是在一个filebeat.yml里写多个- type: log块。我用的配置大概长这样：
filebeat.inputs: - type: log enabled: true paths: - /opt/app/nginx/logs/access.log fields: log_type: nginx-access fields_under_root: true - type: log enabled: true paths: - /opt/app/order/logs/app-*.log exclude_lines: ['^\s*$'] fields: log_type: order-app fields_under_root: true - type: log enabled: true paths: - /opt/app/auth/logs/audit-*.log fields: log_type: audit-log fields_under_root: true output.logstash: hosts: ["logstash-server:5044"]注意fields和fields_under_root这对组合。不写fields_under_root的时候，log_type会被封装到一个叫fields的嵌套对象里，传到Logstash后你读取它要写[fields][log_type]。加了fields_under_root: true，log_type直接出现在文档的顶层，Logstash里可以直接写[log_type]。两种写法都能用，但你要记住自己用的是哪种，不然配置Logstash的if判断时很容易写错路径。
另外paths支持通配符，一个input块可以覆盖多个文件。但建议每个应用类型单独一个input块，不要为了一时省事把不同应用的日志塞进同一个块里——因为后面你要在Logstash里靠log_type区分去向，而同一个块里所有文件只会打同一个标签。
2.2 multiline多行日志的合并处理
如果业务日志里有Java异常堆栈、Python traceback这类跨多行的内容，Filebeat默认一行一条message，就会把一个异常拆成十几条文档，在Kibana里没法看。这时候需要给对应的input块加multiline配置，示例：
- type: log enabled: true paths: - /opt/app/order/logs/app-*.log fields: log_type: order-app fields_under_root: true multiline: type: pattern pattern: '^[0-9]{4}-[0-9]{2}-[0-9]{2}' negate: true match: after这段的意思是：以"2024-08-20"这种日期开头的行算新一条日志，不匹配的行（比如堆栈里的at com.xxx.Class.method）就合并到上一条后面。negate: true在这里很关键，它表示"不符合pattern的行"是续行；如果写反了，你会发现日志被拆得更碎。配置完以后，可以单独写个小文件测试一下合并效果，不要直接上生产路径验证。
2.3 Filebeat重复推送的隐患
Filebeat的注册表文件data/registry/filebeat/log.json记录着每个采集文件已经读到的offset，正常情况下重启不会重复推。但如果你在调试阶段反复删除索引又想让Filebeat重新推送，手动删注册表文件就会埋坑：它会从文件头重新读一遍，ES里立刻多出一大批重复数据。
我的建议是调试期尽量让Logstash端到端都正常了再清一次ES索引，然后保持Filebeat不动，让它按注册表正常增量采集。如果确实要重推某类日志，优先清理那个索引并且把日期索引名字换掉，而不是删注册表文件重推全量。
3. Logstash配置：if路由、日期解析与索引模板的配合
3.1 多类型日志的filter分流
Logstash侧我使用一个管道监听5044端口，接收所有Filebeat推上来的日志，然后靠if [log_type] == "xxx"做分支处理。入门的配置如下：
input { beats { port => 5044 } } filter { if [log_type] == "nginx-access" { grok { match => { "message" => "%{COMBINEDAPACHELOG}" } } date { match => [ "timestamp", "dd/MMM/yyyy:HH:mm:ss Z" ] } mutate { remove_field => ["agent", "log", "input", "ecs"] } } else if [log_type] == "order-app" { json { source => "message" target => "app_data" } date { match => [ "app_data.ts", "yyyy-MM-dd HH:mm:ss" ] } mutate { remove_field => ["message"] } } else if [log_type] == "audit-log" { csv { source => "message" columns => ["user", "action", "resource", "result", "timestamp"] separator => "," } } } output { if [log_type] == "nginx-access" { elasticsearch { hosts => ["es-node1:9200", "es-node2:9200"] index => "nginx-access-%{+yyyy.MM.dd}" } } else if [log_type] == "order-app" { elasticsearch { hosts => ["es-node1:9200", "es-node2:9200"] index => "order-app-%{+yyyy.MM.dd}" } } else if [log_type] == "audit-log" { elasticsearch { hosts => ["es-node1:9200", "es-node2:9200"] index => "audit-log-%{+yyyy.MM.dd}" } } }这里最关键的是index => "nginx-access-%{+yyyy.MM.dd}"。%{+yyyy.MM.dd}不是取当前时间，而是取事件文档里的@timestamp字段格式化成日期。所以如果你没有在filter阶段正确解析原始日志里的时间，@timestamp就默认是Logstash接收时间，索引名会跟着Logstash的系统时间走——半夜12点跨天的时候，日志会突然写到新索引里，但日志本身可能是昨晚的，时序直接乱掉。
3.2 时间解析的坑：@timestamp到底是谁的时间
我在前面配置里写了两个date，一个针对Nginx，一个针对业务JSON。Nginx常见的时间格式是20/Aug/2024:10:32:05 +0800，匹配模式dd/MMM/yyyy:HH:mm:ss Z。注意MMM是月份英文缩写，Z是时区偏移，冒号是字面字符。这个模式写错一两个字母，grok解析出来的timestamp字段就是null，date filter不更新@timestamp，Kibana图表就错乱。
业务JSON里的时间字段我建议统一成标准格式，比如2024-08-20 10:32:05，date配置直接写对应格式即可。这里的核心原则：尽量让业务日志自带时间字段，并且Logstash用原始时间覆盖@timestamp。否则一个批处理任务在凌晨3点回填前一天日志时，这些日志会被算进今天的数据里，报表直接错。
调试date filter最简单的方法是加一个stdout { codec => rubydebug }临时输出，看原始日志解析后的字段是否包含目标时间值。多跑几家数据再移掉这个输出，别在调试完忘了删，确认无误再走ES。
3.3 提前准备索引模板，避免动态映射失控
如果直接把索引交给ES的默认动态映射，字符串字段默认会被映射成text + keyword，字段一多磁盘空间就膨胀。更麻烦的是同一个字段在不同日志里的类型不一致，第二天创建新索引时旧索引里已经是text，新索引里是keyword，跨索引查询直接报类型冲突。
正确做法是在ES侧提前建好索引模板（index template），模板匹配规则对应我们的索引命名。以下仅作演示，实际模板参数根据字段情况调整：
PUT _index_template/nginx-access-template { "index_patterns": ["nginx-access-*"], "template": { "settings": { "number_of_shards": 3, "number_of_replicas": 1 }, "mappings": { "properties": { "clientip": { "type": "ip" }, "bytes": { "type": "long" }, "request": { "type": "text" }, "status": { "type": "ip_range" } } } } }用PUT _index_template/xxx创建的是可匹配多个索引的模板（新版ES更推荐这种方式），index_patterns写通配符。模板的好处是Logstash往ES写数据时，如果目标索引不存在，ES会根据模板自动建索引，字段类型按模板走，分片数、副本数也按模板设置。
这里有个常见问题：模板建好了但Logstash写入时依然报mapping冲突。原因往往是两个模板匹配了同一个索引名，ES默认取优先级高的那个。排查时用GET _index_template看哪些模板匹配了你的索引名，把优先级调好就行。
4. 端到端验证：从Filebeat起到Kibana查到数据
配置写完之后，不急着直接改生产，先在模拟环境里走通一遍。我通常按下面这个顺序验证，每步都能明确判断问题在哪一段。
4.1 启动链路与基础验证命令
先起Logstash，因为它要监听5044端口，Filebeat连不上还能感知到错误。Logstash启动后验证端口：
netstat -anp | grep 5044 ss -lntp | grep 5044看到LISTEN状态就说明beats输入正常。接着起Filebeat，用前台模式跑：
./filebeat -e -c filebeat.yml-e把日志打到标准输出，-c指定配置文件。前台模式下会看到Harvester started for file这样的提示，说明Filebeat已经盯上了日志文件。这个模式特别适合调试，确认没问题再转后台服务方式运行。
此时可以观察Logstash的日志输出。如果我在filter阶段临时加了stdout，就能看到每条日志解析后的完整字段结构，确认log_type有没有带到顶层、grok/json/date有没有解析成功。
等Logstash日志正常刷出事件后，在Kibana里建索引模式，分别建nginx-access-*、order-app-*、audit-log-*三个，然后到Discover里切索引模式查看数据是否进来。如果三个索引都能查到对应日志，链路就基本通了。
4.2 两个典型踩坑的排查过程
第一个坑：日志进了Logstash，但ES里找不到对应索引。这种情况我先看Logstash日志有没有报X-Pack或者illegal_argument_exception——大多数时候是索引名非法或模板mapping冲突。如果Logstash日志很干净，但ES里就是没有新索引，我会去ES端点用通配搜索索引：
curl -X GET "es-node1:9200/_cat/indices/nginx-access-*?v"如果返回空，再用_count看nginx-access-*是否有文档。这个命令虽然土，但能快速区分问题是"Logstash没写进去"还是"ES索引没建出来"。还有一种隐蔽情况：Logstash的output里写了index => "nginx-access-%{+yyyy.MM.dd}"，但日期格式的%{+yyyy.MM.dd}引号里不能有空格，我见过有人写index => "nginx-access-%{ +yyyy.MM.dd}"导致索引名变成nginx-access- 2024.08.20，ES会拒绝并报非法字符。
第二个坑：多个业务日志用的是同一个Logstash管道，其中一个日志格式解析失败，Logstash会不会中断其他日志的正常处理？Logstash的filters执行是按事件顺序逐个跑的，有一个事件抛异常只影响它自己，不会阻塞其他事件的输出。但如果某个事件长期占据线程（比如grok的正则回退导致CPU满载），就会拖慢整个管道的吞吐。我这里用了一个简单有效的方法：把不同类型的日志分配到不同独立管道（用pipelines.yml），互相隔离故障域，配置稍后会说。
排查工具上，Kibana的Discover是最直接的：先看不带任何条件的原始message，确认数据有没有到ES；再用筛选器搜索log_type: "nginx-access"验证字段是否完整；最后用时间范围缩小到最近15分钟，配合Kibana的请求分析面板看是否存在mapping冲突引发的“字段类型不兼容”提示。
5. 生产化考虑：索引生命周期、管道隔离与性能取舍
5.1 索引生命周期管理：不要手动删索引
日志量上来以后，每天一个索引，一年就是365个。像Nginx访问日志其实留7天就够了，业务日志留30天，审计日志留90天。手动删太原始，建议给每个索引模板配上索引生命周期管理（ILM）。配置一个策略，自动滚动和删除旧索引：
PUT _ilm/policy/log-retention-policy { "policy": { "phases": { "hot": { "actions": { "rollover": { "max_size": "50gb", "max_age": "7d" } } }, "delete": { "min_age": "30d", "actions": { "delete": {} } } } } }然后在索引模板的 settings 里引用这个策略，并设置index.lifecycle.rollover_alias。这样每天生成的日志索引会自动滚动，超过保留期后自动删掉，不需要人肉干预。我在实际运维中发现，ILM的滚动触发条件是"满足任一则触发"，如果你既设置了max_size又设置了max_age，先到哪个就先滚。所以不要同时设两个，否则索引分片数量波动会超出预期。
5.2 多个Logstash管道：隔离不同日志的故障影响
单管道处理所有日志有个隐患：如果业务日志里有大量恶性正则（比如极端情况下grok回退很严重），CPU会被某个日志类型占满，导致其他日志（比如审计）延迟变大。定位问题时你会很被动，不知道该压哪个头。
解决思路是用pipelines.yml把不同日志类型拆到不同管道，每个管道用不同的端口接收，甚至可以把Nginx日志交给一个Logstash实例、业务日志交给另一个实例。但我倾向于在单机多管道的规模内解决问题：
- pipeline.id: nginx-pipeline path.config: "/etc/logstash/conf.d/nginx.conf" pipeline.workers: 2 - pipeline.id: app-pipeline path.config: "/etc/logstash/conf.d/order_app.conf" pipeline.workers: 4 - pipeline.id: audit-pipeline path.config: "/etc/logstash/conf.d/audit.conf" pipeline.workers: 2每个管道独立监听一个端口，Filebeat在output里按log_type分发给对应的Logstash端口。管道隔离后，个别管道的性能毛刺不会影响其他日志的链路。代价是要多维护几份配置，但在日志平台稳定运行半年后回头看，这笔投入是值得的。
5.3 性能取舍：什么时候引入Kafka作为缓冲
聊到性能，很多人第一反应是"上Kafka"。但Kafka不是银弹。它最大的价值是削峰填谷，在日志洪峰时把数据缓存在队列里，让Logstash按消费能力慢慢拉。
我平时看到的情况是：单机日志量每天不到100GB时，Filebeat -> Logstash -> ES的直连架构完全够用。加Kafka之后延迟至少增加几百毫秒到秒级，而且整个链路多了三个组件要维护，排错难度翻倍。数据量真的到了需要削峰的时候，再上Kafka也不迟。此外还可以用Logstash的pipeline.batch.size和pipeline.batch.delay做微调，加大批处理吞吐，比上Kafka来得更实惠。
我给一套模拟生产环境的调优参数是：Logstash的pipeline.workers等于CPU核数，pipeline.batch.size取1200，pipeline.batch.delay取50ms。Filebeat和Logstash都在同一网段，网络延迟低于1ms时，这个配置单管道一天能扛30GB左右的日志，没有明显抖落。
6. 经验收尾：标签规范与后续扩展
这套Filebeat + Logstash多文件分索引的架构，我已经在不同环境里跑了大半年。最深的体会是：配置本身不复杂，真正决定你后续省不省心的是命名词和标签规范的统一。比如log_type的值，建议全部用小写下划线风格，和索引名称保持一致。一旦线上已经存在nginxAccess、nginx-access、nginx_access三种写法，后面写if判断、建模板、做授权策略都会很痛苦。
这系列文章的第七篇主要聚焦在"数据分门别类"这件事上。往下走，你可能会遇到更细粒度的问题，比如同一类日志按业务线再细分、从Kafka消费时位点提交模式的选择、Filebeat多实例的负载均衡方式。在模拟环境里我都会保留一套最小可复现的样例配置，遇到新的业务接入先在小环境验证排错，再上生产迭代。这套方法帮我挡掉过很多次半夜上线翻车，你也可以参考。
