企业官网建设流程全解析

读nginx日志这事，本身不难，难的是怎么把它读得干净、写得规整，最后在Kibana里能查出东西来。ELK这个系列写到第四篇，Elasticsearch和Kibana的前置工作就算基本到位了，今天轮到Logstash正式干活——目标很具体：把nginx的access.log一条一条读进来，解析成结构化字段，再写进ES，让Kibana上直接能看到每个接口的响应时间、PV走势、来源IP分布。

如果你是刚把ES和Kibana跑起来、正愁日志怎么进去的运维，或者被grep窗口切来切去折磨到受不了的后端，这篇适合直接抄作业。我会从nginx日志格式设计开始讲，因为这些细节决定了grok正则好不好写、ES字段类型对不对，后面每一步都跟它有关。

1. 为什么是Logstash来啃nginx日志这块骨头

1.1 链路里的分工和定位

先把这个系列里的角色对齐一下。Elasticsearch负责存储和检索，Kibana负责展示和交互，那日志到底怎么进去？答案是采集端。常见的方案有两种，一种是Filebeat轻量采集再转发给Logstash，另一种是Logstash直接读文件。我这次用的是后者，原因很实际：数据量不大、节点不多，没必要在中间多加一层，Logstash一条管道就能从文件读到ES，排查链路也短。

Logstash在整个链路里的定位是"数据管道处理器"，它干三件事：input拿数据、filter改数据、output送数据。读nginx日志这个场景，input是file插件，filter核心是grok正则解析，output就是elasticsearch插件。这三件事听起来简单，真正花时间的全在filter里，因为nginx默认日志格式跟ES要的结构化字段之间，隔着一层正则解析的鸿沟。

1.2 直接用Logstash与Filebeat加Logstash怎么选

我知道很多人会提Filebeat。官方推荐的是Filebeat采集、Logstash解析，这样文件读取压力在轻量级Beats上，Logstash专职做富化和解析，两边的压力都小。但直接让Logstash读文件也有它的适用场景，比如日志量日均几个GB以内、Logstash和nginx在同一台机器、不想再多维护一个Beats进程。

我这个项目的情况是单机nginx，access.log一天大概2GB，Logstash用file插件默认的配置完全扛得住，还省了一个组件。真到了每天几十GB甚至上百GB，再拆Filebeat不迟。反正管道配置里的filter和output部分完全复用，到时候只改input来源就行，迁移成本很低。

1.3 nginx日志格式先行设计

在写Logstash配置之前，我一直建议先把nginx的log_format定下来。很多人拿到默认的combined格式就开始写grok，结果发现要解析的字段不够用，又回头改nginx，来回折腾。

我用的log_format是这么定义的：

log_format main_json escape=json '{"@timestamp":"$time_iso8601",' '"client_ip":"$remote_addr",' '"request_method":"$request_method",' '"request_uri":"$request_uri",' '"server_protocol":"$server_protocol",' '"status":$status,' '"body_bytes_sent":$body_bytes_sent,' '"request_time":$request_time,' '"upstream_response_time":"$upstream_response_time",' '"http_referer":"$http_referer",' '"http_user_agent":"$http_user_agent",' '"http_x_forwarded_for":"$http_x_forwarded_for"}';

access.log里配置成：

access_log /var/log/nginx/access.log main_json;

这么做的原因很简单：既然Logstash最终要把日志变成JSON结构，那源头直接输出JSON，解析路径就短了一大截。grok正则再强，也扛不住引号嵌套和空格切割的反复调试，能用json插件一行搞定的事，不值得用正则耗半小时。后面filter里我只需要把@timestamp字符串转成时间类型，其他字段基本原样保留。

提示：escape=json这个参数很重要，否则请求行里的双引号、User-Agent里的反斜杠会在转JSON时产生脏数据甚至直接解析失败。

2. Logstash环境准备与管道骨架

2.1 版本对应关系

Logstash是Java系应用，版本跟ES的兼容性要提前对好。我的环境里ES是7.x，Logstash就选了同大版本的7.x，这样output插件走内置的ES客户端协议，兼容性上不用操心。JDK版本按官方要求配好，7.x需要JDK 8或11，别图省事拿个JDK 17去跑，启动阶段可能就给你报一堆模块访问错误。

安装方式我用的是tar包解压，因为生产机器上没那么多图形界面，用包管理器装也差不多。解压之后先做个基础验证：

/usr/local/logstash/bin/logstash -e 'input { stdin { } } output { stdout { codec => rubydebug } }'

敲几行字符能原样打印出来，说明环境是通的。这一步能把JDK问题、启动脚本问题提前暴露出来，别等到上正式配置才排查。

2.2 配置目录与最小管道

Logstash的配置我习惯放到单独的目录里，一个管道一套配置，用参数指定。项目里我建了如下的目录结构：

/etc/logstash/conf.d/nginx-to-es.conf /usr/local/logstash/plugins/patterns/nginx_extra

第一个是管道主配置，第二个是自定义grok规则文件，后文会用到。配置本身分三大块，我先搭个骨架：

input { file { path => ["/var/log/nginx/access.log"] start_position => "beginning" sincedb_path => "/usr/local/logstash/sincedb/nginx_access.db" stat_interval => 3 } } filter { # 解析逻辑，下文展开 } output { elasticsearch { hosts => ["http://127.0.0.1:9200"] index => "nginx-access-%{+YYYY.MM.dd}" } }

到这一步管道的形状已经有了，file插件盯着文件看，有新增内容就读取，交给filter处理，最后按天写进ES索引。接下来要处理的，就是filter里那块硬骨头。

2.3 file插件参数里容易被忽略的细节

path参数可以写数组，多目录用逗号分隔。start_position只有在sincedb没有记录时才会生效，它决定第一次启动是从头读还是从文件尾部读，我在这里设置成beginning是因为要补采历史数据，如果你只需要实时数据，保持默认的end就行。

sincedb_path一定要显式指定，不然文件会落在用户家目录下，容易被清理工具误删。一旦sincedb丢失，Logstash不记得自己读到哪了，就可能重复推送同一批数据到ES，Kibana里的PV数字瞬间翻倍，排查起来很迷惑。

stat_interval决定多久扫一次文件变化，默认1秒，我没改得太激进，3秒足够实时性要求，也能降低一点CPU占用。

3. 核心战场：Grok解析nginx日志

3.1 先拿一条真实日志来分解

我这边日志已经按JSON格式输出，所以严格来说filter里主要靠json插件解析，不需要重型grok。但有一个字段需要特殊处理：@timestamp。nginx给的是字符串，比如2025-06-18T10:24:36+08:00，ES如果直接存这个字符串，Kibana里做时间轴聚合就废了，必须让它变成真正的date类型。

filter的初步写法：

filter { json { source => "message" } date { match => ["@timestamp", "ISO8601"] target => "@timestamp" } }

json插件把整行转成一个Ruby Hash，字段自然展开。date插件用ISO8601格式解析nginx的时间字符串，然后覆盖掉原字段。这两行配置解决了我80%的需求。

3.2 自定义pattern与字段治理

如果你的nginx日志不是JSON格式，而是默认的combined格式，那就得搬出grok了。拿一条典型的默认格式日志举例：

127.0.0.1 - - [18/Jun/2025:10:24:36 +0800] "GET /api/user/list?page=1 HTTP/1.1" 200 1024 "https://example.com" "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"

对应的grok表达式可以拆成这样：

grok { match => { "message" => "%{IPORHOST:client_ip} - %{USERNAME:remote_user} \[%{HTTPDATE:request_time}\] \"%{WORD:request_method} %{NOTSPACE:request_uri} HTTP/%{NUMBER:http_version}\" %{NUMBER:status} %{NUMBER:body_bytes_sent} \"%{NOTSPACE:http_referer}\" \"%{GREEDYDATA:http_user_agent}\"" } }

这是一个组合表达式，从左侧开始逐个字段匹配。IPORHOST匹配IP或域名，HTTPDATE匹配18/Jun/2025:10:24:36 +0800这种格式，GREEDYDATA用来吞掉User-Agent里可能出现的任何字符，包括引号和括号。

但这种表达式在真实业务里往往不够用。比如你需要在URI里拆出path和query参数给Kibana做页面分组，就需要再加一个grok：

grok { match => { "request_uri" => "%{NOTSPACE:path}?%{GREEDYDATA:query_string}" } }

再比如你的内网链路里有多个upstream响应时间，用分号和逗号隔开，就得先按分隔符拆再逐个解析。

我的经验是：grok规则文件单独放一个目录管理，把常用的正则片段抽出来。比如项目中我在自定义rules文件里加了这么一条：

NGINX_UPSTREAM_TIME %{NUMBER:upstream_response_time:float}

这个文件里每一行是一个命名正则，grok里可以直接用%{NGINX_UPSTREAM_TIME}引用，比把所有正则堆在管道配置里好维护得多。配置时在管道里引入：

grok { patterns_dir => ["/usr/local/logstash/plugins/patterns"] match => { "message" => "%{NGINX_UPSTREAM_TIME}" } }

3.3 调试方法：先测正则再跑正式管道

grok正则是最容易翻车的地方，少一个空格、多一个转义，整条日志就会被标记成_grokparsefailure。我调试时常用两种方式。

第一种是在启动命令里临时起一个管道，用stdin输入测试数据：

/usr/local/logstash/bin/logstash -e 'input { stdin { } } filter { grok { match => { "message" => "%{IPORHOST:client_ip} - %{USERNAME:remote_user} \[%{HTTPDATE:request_time}\]" } } } output { stdout { codec => rubydebug } }'

启动后粘贴一条日志，立刻能看到解析结果。字段分出来了就是成功，没分出来会看到tag里的_grokparsefailure，调整表达式再试。

第二种是把stdout改成ES的dry-run形式，或者直接在output里加stdout { codec => rubydebug }，在正式管道里边跑边看。

注意：rubydebug输出里你能看到每个event的完整字段和tag。一个容易忽略的细节是，[和]在grok里要加反斜杠转义，日期后面的空格也要原样保留，多一个少一个都会导致匹配失败。

4. 写入ES的最后一公里

4.1 output插件参数解析

filter处理完的数据，最终落到ES。output部分看起来简单，实际上有几个关键参数决定数据质量和集群健康度。

我的output配置：

output { elasticsearch { hosts => ["http://127.0.0.1:9200"] index => "nginx-access-%{+YYYY.MM.dd}" ilm_enabled => false manage_template => false } }

index按天滚动，Kibana里创建索引模式的时候就可以用nginx-access-*统配。ilm_enabled和manage_template这两个参数，我是根据自己环境的索引模板配置来定的。如果ES里已经通过Kibana DevTools或者代码仓库管理了索引模板，Logstash这边就别让它自己乱建模板，避免字段类型冲突。

4.2 字段类型与mapping防冲撞

ES里最坑的一件事是动态映射。第一次写入s "status"字段时，如果值是字符串"200"，ES就把它映射成text，后续就算Logstash改成integer类型，写入也会报类型冲突。所以我的习惯是：在ES侧预先建好索引模板，把关键字段类型锁死。

模板片段：

{ "index_patterns": ["nginx-access-*"], "mappings": { "properties": { "status": { "type": "integer" }, "body_bytes_sent": { "type": "long" }, "request_time": { "type": "float" }, "@timestamp": { "type": "date" } } } }

模板在Logstash写入之前就创建好，这样每天滚动出来的新索引都沿用这套mapping。否则等到Kibana里做数值聚合时才发现status是keyword字符串，再回头迁移索引就非常被动。

4.3 索引生命周期

按天索引配合生命周期策略可以实现自动删除旧索引，比如保留30天。我是用ES的ILM配置的，Logstash这边什么都不用动，只要index名字符合模式就行。如果不想上ILM，也可以用定时任务调ES的_delete_by_index接口清理，但这种方式在索引数量多的时候容易漏，不如ILM干净。

提示：ILM的滚动条件建议按大小和时间两个维度都设置，比如max_size: 50gb、max_age: 1d，避免某天访问量爆炸导致单个索引过大。

5. 上线后我踩过的坑与排查实录

5.1 sincedb引发的重复数据

项目上线第二天，我在Kibana里看到PV数比nginx的access.log实际行数多了一倍，第一反应是grok解析串了，结果打开日志一查，发现同一个IP同一个时间点的记录出现了两次。

排查过程是这样的：先看Logstash日志，没有报错，再看ES里最新一条文档的_id，同一个请求出现了两个不同的_id。这说明Logstash把同一条数据发了两遍，而不是ES内部复制。原因很快定位到：前一天我手动清过一次Logstash的data目录，sincedb文件被删了。重启之后Logstash从头扫描文件，把之前已经读过的部分又读了一遍。

解决办法是给每个event生成一个确定性_id，从业务字段里算出来，比如把client_ip、@timestamp、request_uri拼起来做哈希，这样即使重复推送，ES也会覆盖写入而不是新增文档。Logstash里用fingerprint插件实现：

fingerprint { source => ["client_ip", "@timestamp", "request_uri"] method => "SHA256" key => "nginx_log_dedup" }

output里再指定：

elasticsearch { document_id => "%{fingerprint}" }

这套方案对付重复推送非常有效。但fingerprint在字段缺失或顺序变化时会生成不同的值，所以source字段的选择要谨慎，必须选稳定的组合。

5.2 _grokparsefailure满天飞

如果用的是默认combined格式，最容易出现_grokparsefailure的场景是：某个请求的User-Agent里带了[]，或者某个URI里带空格被编码成了%20。正则匹配只要错一截，整个event就被归到failure里，后续所有字段都空着，Kibana图表直接缺数据。

排查这类问题时，我建议先把没匹配上的event单拎到排查索引里：

if "_grokparsefailure" in [tags] { elasticsearch { hosts => ["http://127.0.0.1:9200"] index => "nginx-parse-failed-%{+YYYY.MM.dd}" } } else { elasticsearch { hosts => ["http://127.0.0.1:9200"] index => "nginx-access-%{+YYYY.MM.dd}" } }

这样主链路不受影响，失败数据单独存起来，事后批量分析正则哪里需要补。我在一次排查中发现，某个爬虫脚本的User-Agent里带了Tab字符，导致GREEDYDATA吞到尾引号之后才结束，表达式里\"直接失配。后来统一在grok之前加了一个mutate，把message里的Tab替换成空格：

mutate { gsub => ["message", "\t", " "] }

这类脏数据防御看起来不起眼，但在线上环境真能救命。

5.3 时间字段还是字符串

Kibana里时间轴显示不了数据，十有八九是@timestamp字段类型不对。我在一次部署中遇到过：date插件正常执行了，但ES索引模板里@timestamp没预先定义，动态映射把它当成date类型倒也正常，但某些日志行的时间字符串带毫秒，某些不带，导致解析结果有的成功有的当成string。

处理方式很简单：date插件支持多种格式匹配，把带毫秒和不带毫秒的格式都列上：

date { match => ["@timestamp", "ISO8601", "yyyy-MM-dd HH:mm:ss.SSS", "yyyy-MM-dd HH:mm:ss"] target => "@timestamp" }

同时确保ES侧模板里@timestamp是date类型。这里还要注意时区问题，如果nginx的$time_iso8601带+08:00后缀，date插件解析后会自动转成UTC存储，Kibana展示时再按浏览器时区转回来，这个行为是符合预期的。如果你自己拼接时间字符串不带时区，就得手动指定timezone参数，否则ES和Kibana显示的时间会差8小时，排查起来很迷惑。

5.4 性能瓶颈与缓冲调优

运行一段时间后，我注意到Logstash的JVM堆占用持续走高。日志量平稳，但处理延迟从秒级变成了分钟级。查了下pipeline的bulk批量配置，默认值在批量请求大小和延迟之间偏保守，于是做了两处调整。

pipeline.batch.size从125调到500，pipeline.batch.delay从50ms调到100ms，让单个批量包含更多event，减少ES交互次数。另外给JVM堆加了配置，在jvm.options里把-Xms和-Xmx设成物理内存的一半左右，我机器上16GB内存，给了8GB，堆外部分留给GC和操作系统缓存。

还有一个容易忽略的点：output里的flush_size和idle_flush_time，这两个参数控制批量写入的阈值和时间。flush_size到了就立刻刷，idle_flush_time到了就算数据不够也刷，避免低峰期数据滞留内存。

5.5 字段缺失时的兜底

真实业务里，很多请求没有http_referer，没有http_x_forwarded_for，grok匹配到的这些字段就是空的。空字段本身不影响ES写入，但如果你的Kibana仪表盘里要做来源分析，空referer会被归到"未知"之外的一个独立桶里，图表上多出一条空值列。

我的处理是给这些可选字段加一个默认值兜底：

mutate { gsub => ["http_referer", "^$", "-"] gsub => ["http_x_forwarded_for", "^$", "-"] }

这样Kibana里看到的来源分布就统一了。

6. 还有两个顺手能做的增强配置

6.1 User-Agent解析成设备和浏览器

nginx日志里的http_user_agent字段如果原样存进ES，除了能在Kibana里当字符串搜索，基本没什么分析价值。Logstash自带useragent插件可以把它拆成操作系统、浏览器名称、设备类型：

useragent { source => "http_user_agent" target => "ua" }

存进ES后，你就可以在Kibana里直观地看到访问者用什么浏览器、什么操作系统占比最高。有一次做前端改版效果复盘，我们用这个字段确认了某老旧浏览器的用户占比，据此决定是否彻底放弃兼容。

6.2 GeoIP解析来源地域

如果你的nginx日志里有公网IP，可以再加一个geoip插件，把IP转成经纬度和省市信息：

geoip { source => "client_ip" target => "geo" }

Kibana自带地图可视化，有了经纬度字段，直接就能打点展示访问来源分布。这里要注意，geoip的解析结果依赖IP库，内网IP和保留地址段解析不出来，会变成空的geo对象。数据是纯内网流量的话，这个插件基本没用。

这两个插件都适合在数据量可控、分析需求明确的场景下启用，解析本身有一点CPU开销，但Logstash一般扛得住。

7. 配置管理的小建议

管道配置到后期，一定要纳入版本管理，这一点很多人会忽略。我在项目的配置目录下建了一个仓库，nginx配置、Logstash管道、ES索引模板三份内容放一起维护。改grok正则、加字段映射时，跟着MR记录走，出问题能快速回滚。

另外一个习惯是给Logstash配一个独立系统用户，配置文件权限收紧到600，避免日志路径变动时权限报错。我自己踩过权限坑：Logstash进程用root跑，nginx日志轮转后新文件属主变成nginx用户，Logstash带着root权限去读倒没问题，但一旦改成普通用户启动，读不到文件就直接静默跳过，Kibana数据断更，排查半天才发现是权限问题。

还有一点是管道配置里建议加上pipeline.ecs_compatibility => disabled。Logstash 7.x之后ECS兼容模式默认开启，会把字段重命名成[host][name]这种嵌套结构，跟老版本行为不一致。如果你的下游Kibana仪表盘是基于旧的扁平字段写的，这个开关必须关掉，否则所有字段路径全部失效。

8. 收尾：一次上线过程的完整回顾

把整套配置串起来，我这边最终跑在生产线上的管道大概是这样的：

input { file { path => ["/var/log/nginx/access.log"] start_position => "beginning" sincedb_path => "/usr/local/logstash/sincedb/nginx_access.db" stat_interval => 3 } } filter { json { source => "message" } date { match => ["@timestamp", "ISO8601"] target => "@timestamp" } mutate { gsub => ["http_referer", "^$", "-"] gsub => ["http_x_forwarded_for", "^$", "-"] } fingerprint { source => ["client_ip", "@timestamp", "request_uri"] method => "SHA256" key => "nginx_log_dedup" } useragent { source => "http_user_agent" target => "ua" } geoip { source => "client_ip" target => "geo" } } output { if "_grokparsefailure" in [tags] { elasticsearch { hosts => ["http://127.0.0.1:9200"] index => "nginx-parse-failed-%{+YYYY.MM.dd}" document_id => "%{fingerprint}" } } else { elasticsearch { hosts => ["http://127.0.0.1:9200"] index => "nginx-access-%{+YYYY.MM.dd}" document_id => "%{fingerprint}" } } }

我用配置文件校验命令验证语法：

/usr/local/logstash/bin/logstash -t -f /etc/logstash/conf.d/nginx-to-es.conf

返回Configuration OK之后，再用systemd或者nohup方式启动进程，观察日志输出。第一次启动后，我会用Kibana DevTools查一下最新索引里的文档数量和字段类型：

GET nginx-access-2025.06.18/_count GET nginx-access-2025.06.18/_mapping

确认字段类型是预期的integer、float、date之后，再去Kibana创建索引模式，画第一个PV趋势图。

个人经验上，我最大的体会是：Logstash管道的复杂度不在插件多，而在你对线上数据形态的了解程度。JSON格式的日志、合理的索引模板、稳定的事件ID，这三件事做好，管道基本就不会出大乱子。如果你能把这一步跑顺，后面加Kafka、加多管线分流，都只是在input和output上做扩展，filter这块的资产永远不会浪费。