上周有个运维同事半夜找我，说线上某个接口持续返回500，业务方问了一堆问题：是哪个来源IP在刷？请求参数什么样？从什么时候开始的？他只能登录服务器一行行翻access.log，翻了半小时也没看明白。我让他把日志接进ELK，两分钟就在Kibana里筛出了异常请求的特征。这种效率差，靠的是一条稳定的日志链路：nginx负责产生日志，Logstash负责读取和解析，Elasticsearch负责存储与检索，Kibana负责把数据变成能看懂的图表。

这个系列前面几篇已经把Elasticsearch和Kibana的基础讲得差不多了，这篇专门补上中间那台“搬运工”：Logstash读取nginx日志并写入ES。我不打算再把安装部署从头列一遍，重点放在配置、正则解析和真实踩过的坑上。适合已经装好ES和Kibana、正准备把nginx访问日志接进来的读者，也适合已经在跑采集、但打开Kibana发现每条日志只有一长串message字符串的人。

1. 日志链路选型：Logstash在这个位置到底值不值

1.1 先确认nginx日志落在哪里

nginx的访问日志默认写在/var/log/nginx/access.log，错误日志在error.log。生产环境最常见的接法有三类：

• 每台nginx节点安装Filebeat，由Filebeat读取文件后发给Logstash或直连ES；
• 日志被统一收集到某台中心机上挂载目录，Logstash通过file插件直接读取；
• Logstash直接以分布式方式读取多台机器的共享存储路径。

本文场景对应第二种，也就是标题里说的“Logstash读取nginx日志”。但你要知道，Filter和Output阶段对Filebeat发过来的数据同样适用，差别只在于input插件不同。换句话说，你把input从file换成beats，后面的grok、geoip、日期解析、字段清洗一步都不用改，这是一条很重要的迁移路径。

1.2 和Filebeat直出ES的取舍

很多刚接触ELK的人会问：既然Filebeat这么轻量，为什么不让它直接把日志写到ES，非要中间夹一个Logstash？

我的建议很直接：

• 单机、日志量一天不到几个GB：Filebeat直接输出ES，或者Logstash直接读文件，都行，怎么省事怎么来；
• 日志量到了一天几十GB、nginx节点分散在四五台以上：Filebeat负责采集传输，Logstash负责集中解析，这是最稳的架构；
• 不建议在ES端用ingest pipeline做复杂grok：虽然ES的ingest节点也能干这事，但调试麻烦，而且让ES白白吃CPU。Logstash解析完再写入，ES只做存储和检索，负担更清晰。

Logstash直接读文件最大的优势是少装一层Agent，你不需要在每台nginx上维护Filebeat的配置和版本；最大的劣势是每台要部署一个JVM进程，内存占用不低。如果nginx节点特别多，部署和配置分发的成本会上升，那时候就应该把采集职责交给Filebeat。一句话总结：规模小选直接读，规模大了再过渡到采集端分离。

2. 先把input段拉通：file插件里的细节

2.1 关键参数与默认行为

直接看一段我在生产环境常用的input配置：

input { file { path => ["/var/log/nginx/access.log"] type => "nginx-access" start_position => "beginning" sincedb_path => "/var/lib/logstash/sincedb/nginx-access" sincedb_write_interval => 15 stat_interval => 1 codec => plain { charset => "UTF-8" } } }

几个参数逐个解释：

• path：支持通配符，比如"/var/log/nginx/*.log"。但你最好只匹配access.log，不要用通配符把error.log一起吞进来，因为这两种日志格式完全不同，后面grok解析会打架。
• start_position：只有当你这台机器第一次读这个文件、且sincedb里没有记录时才有效。很多新手以为它永远从头读，不是的，文件一旦被读取过，重启后会按sincedb记录的偏移量继续。
• sincedb_path：sincedb负责记录每个文件当前的读取位置，默认存在Logstash用户目录下。我建议显式指定位移文件，比如放到/var/lib/logstash/sincedb/，这样想清理历史位置时直接删这个文件就好，不会误删别的状态。
• codec => plain：默认就是plain，不写也行，但显式写出来你可以顺带指定charset。如果nginx日志里出现中文而且是GBK编码，得设成"GBK"，否则进ES就是乱码。
• stat_interval：每隔多久刷新一次文件状态，默认1秒已经很快。如果压得太小，文件在频繁写入时会额外消耗CPU。

2.2 sincedb的工作机制与日志切割

sincedb记的不是文件名，而是文件的inode、设备号、字节偏移量三元组。这样做的好处是文件改名后还能继续读旧文件。但nginx日志天天被logrotate切割，这里有个典型的坑：如果切割方式是把access.log重命名成access.log.1，再新建一个access.log，新文件是新的inode，Logstash会把它当成一个全新文件从头读，可能把历史数据又读一遍，造成重复。

解决办法：

• 在logrotate配置里用copytruncate：先复制文件内容到目标文件，再清空原文件。Logstash盯着同一个inode，只会读到新写入的数据。
• 或者把rotate参数改成旧文件加时间戳、新文件立刻创建，再让Logstash的path只匹配当前活跃文件。

我见过不少团队把两种方式混着用，结果Kibana里数据出现时重时漏。建议统一用copytruncate，这是最不容易出错的做法。

2.3 多行日志进不来怎么办

nginx的access.log基本是单行一条，但error.log不同，特别是配置了debug级别、出现堆栈信息时，一个错误会跨多行。如果你硬要把error.log也接进来，需要用multiline codec，把属于同一条事件的多行拼成一行：

input { file { path => ["/var/log/nginx/error.log"] type => "nginx-error" codec => multiline { pattern => "^\d{4}/\d{2}/\d{2} \d{2}:\d{2}:\d{2}" negate => true what => "previous" } } }

这里的意思是以日期开头的行作为新事件开头，非日期开头的行追加到前一条事件继续。实际nginx错误的开头格式不一定是2025/04/18 12:00:00，你要先看一眼自己的error.log开头是什么格式再定pattern，别照抄。

3. 解析nginx日志的硬骨头：grok正则实战

3.1 先对照nginx日志格式

默认情况下nginx的日志格式一般是这样定义的：

log_format main '$remote_addr - $remote_user [$time_local] "$request" ' '$status $body_bytes_sent "$http_referer" ' '"$http_user_agent" "$http_x_forwarded_for"';

对应的真实一行大概长这样：

192.168.10.11 - - [18/Apr/2025:14:23:05 +0800] "GET /api/user/list?page=1 HTTP/1.1" 200 5326 "https://example.com/list" "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0 Safari/537.36" "-"

我要解析的目标很清楚：客户端IP、请求时间、请求方法、完整URL、状态码、响应体大小、来源页面、User-Agent、真实客户端IP。不要在message里保留一长串字符串，而是要拆成一个一个字段，这样Kibana才能按字段筛选和聚合。

3.2 用Logstash内置正则搭出第一版

Logstash的grok过滤器自带几百个正则模式，其中几个是解析nginx的常客：

第一版grok可以这样写：

filter { if [type] == "nginx-access" { grok { match => { "message" => "%{IPORHOST:clientip} - (%{USER:remote_user}|-) \[%{HTTPDATE:timestamp}\] \"%{WORD:method} %{URIPATHPARAM:request} HTTP/%{NUMBER:httpversion}\" %{INT:status} %{INT:body_bytes_sent} \"(?:%{URI:http_referer}|-)\" \"%{DATA:user_agent}\"" } break_on_match => false named_captures_only => true } } }

注意几个细节：

• remote_user默认是-，但如果你开启了HTTP Basic认证，它会变成用户名。所以这里写成了(%{USER:remote_user}|-)，两种都兼容。
• HTTP referer在没有来源时会输出"-"，直接用URI模式匹配会失败，所以要加一个(?:%{URI:http_referer}|-)的分支兜底。
• User-Agent里的内容可能包含空格、引号、括号等任意字符，用DATA比QUOTEDSTRING更稳。但要注意grok的贪婪匹配问题：UA如果用GREEDYDATA，它会一口气吃到行尾，可能把后面真正的"x-forwarded-for"吃掉，所以这里我用了非贪婪的DATA。

3.3 真实日志里的那些边界情况

第一版能跑通绝大多数场景，但生产环境总会给你惊喜，我简单列几个几乎一定会遇到的：

第一，日志里有IPv6地址。IPv6写成2001:db8::1，用%{IP}会匹配失败，%{IPORHOST}可以。所以上面第一版用的是IPORHOST，不要为了“看起来专业”改成IP。

第二，请求行里的URL带空格或中文。例如GET /api/v2/order%20list HTTP/1.1，%{NOTSPACE}会在空格处截断，用URIPATHPARAM会比较好，因为它能处理URL编码。如果URL非常规，可以再加一个分支(?<request>.+?) HTTP/兜底。

第三，多级代理下$http_x_forwarded_for会是"1.1.1.1, 2.2.2.2"这样的列表。如果直接把整串作为client IP，后续geoip会失败。我习惯在后续filter里用mutate的split功能按逗号切一下，只保留第一个IP：

mutate { split => { "x_forwarded_for" => "," } replace => { "clientip" => "%{[x_forwarded_for][0]}" } }

第四，日志格式本身有变化。有人会在log_format里加$request_time，有人加响应头字段。一旦格式变了，grok就会静默失败，事件会被打上_grokparsefailure标签。所以我强烈建议在format变更时不要只改nginx配置文件，还要同步改Logstash这边的正则，最好在测试环境先验证一版再切。

3.4 调试正则的正确姿势

调试grok不要在生产配置上反复重启Logstash，先把数据拉出来试。我用的套路是三步：

1. 先在Logstash配置里加一个stdout输出，让event以rubydebug格式打印，用它看原始message和解析结果；
2. 挑几条有代表性的真实日志，比如一个客户端IP异常的、一个referer为-的、一个UA超长的，拼成一个样例文件，在测试pipeline里跑；
3. 如果还是匹配不上，再打开Kibana的Dev Tools里的Grok Debugger功能，直接贴样例和正则，实时看分组结果。

正则调试最烦人的地方是转义。在grok里匹配左中括号，写的是\[，在匹配message字符串时还要分辨哪些是正则元字符。我的建议是先从最核心的\[%{HTTPDATE:timestamp}\]开始，一点一点往外扩，别一次写完整个长正则。只有这样你才能知道到底是哪个字段让整条匹配断掉。

4. 写入ES前的那几道隐形工序

4.1 @timestamp的时区与接收时间坑

很多人把nginx日志接进ES之后发现一个诡异现象：Kibana时间轴上看到的事件时间，和自己访问接口的时间差了8小时。问题出在@timestamp。默认情况下，事件进入Logstash时，@timestamp存的是Logstash接收这条日志的时间，不是你访问接口的时间。

要让@timestamp变成日志里的访问时间，必须用date过滤器把它“覆盖”掉：

date { match => ["timestamp", "dd/MMM/yyyy:HH:mm:ss Z"] locale => "en" target => "@timestamp" }

这里有两个容易踩的细节：

• dd/MMM/yyyy:HH:mm:ss Z的格式必须和nginx完全一致，MMM是英文缩写月份。如果Logstash所在系统的locale不是英文，解析可能失败，所以要显式指定locale => "en"。
• 原来的timestamp字段保留一份也没关系，但为了省存储，大多数人会在date处理完成之后用mutate把它删掉。

另外，如果你想在Kibana里按时间聚合，注意索引的@timestamp会被映射成date类型。如果你之前用旧版本Logstash建过索引，字段类型已经定型，新的解析可能没办法直接写入，需要在模板里做一次date类型映射。

4.2 geoip和useragent：让数据一眼变“厚”

日志解析之后，虽然字段拆开了，但clientip还只是一个IP字符串。如果你想知道访问者大概来自哪个地区，或者想按操作系统、浏览器类型分析用户，需要用两个filter：

geoip { source => "clientip" target => "geoip" database => "/etc/logstash/GeoLite2-City.mmdb" } useragent { source => "user_agent" target => "ua" }

geoip的输出会包含国家、地区、城市、经纬度等字段，Kibana里直接就能画地图热点。但这块有个比较大的坑：新版Logstash不再自带GeoLite2-City.mmdb数据库文件，你得自己下载数据库并指定database路径，否则插件会提示找不到数据库，字段写不进去。如果日志来源大多是内网IP，geoip解析出来全是null，还白白增加处理时间，可以直接删掉geoip段。

useragent的解析结果是一个嵌套对象，包含ua.name、ua.os、ua.device这些子字段，比让Kibana直接对UA字符串做keyword聚合要准得多。注意，如果你打算在索引模板里给ua字段设置dynamic mapping，别让它映射成text，否则后续没法做精确聚合。

4.3 mutate把字段收拾干净

字段拆开之后，下一步是清洗和规范。我通常会在grok之后接一段mutate：

mutate { remove_field => ["timestamp", "message", "event.original"] convert => { "status" => "integer" "body_bytes_sent" => "integer" } rename => { "clientip" => "client_ip" } }

status和body_bytes_sent默认会被grok解析成字符串，如果后续要算平均耗时、聚合状态码分布，字符串类型会让ES头疼，所以直接转成整数。

remove_field里我常删的是原始message。如果你能接受日志丢失原始行，可以删，能省掉不少存储。但谨慎起见，我建议在初期保留message一段时间，等确认字段都在、排障不再需要原始字符串时再删。

如果URL里有查询参数，而你想按参数维度统计，可以再补一段kv过滤器。不过要先解决URL编码问题，有些参数是中文会被encode，kv解析出来是乱码，需要用urldecode先解一遍。这块默认不开，按需加。

5. output与ES侧配置：索引、模板、生命周期

5.1 elasticsearch插件的关键选项

filter处理完后，最后一步是输出到ES：

output { if [type] == "nginx-access" { elasticsearch { hosts => ["http://es-node-1:9200", "http://es-node-2:9200"] index => "nginx-access-%{+YYYY.MM.dd}" user => "elastic" password => "${ES_PASSWORD}" ilm_on_off => false manage_template => false } } }

几个选项逐个拆：

• index里的%{+YYYY.MM.dd}用的是事件时间，也就是date filter修正过的@timestamp。如果你没做date修正，这里会按Logstash接收时间分索引，日志时间晚一天就会进错索引。
• user和password：ES开启安全认证时必须配置。不要把明文密码直接写死在配置里，我一般用环境变量引用，或者用Logstash的keystore功能。
• manage_template => false：默认Logstash会自己管理logstash-*索引模板。如果你自己已经定义了模板和ILM策略，建议关掉，避免Logstash的模板覆盖你的配置。模板冲突是ELK里很常见的坑，两边各写各的，最后mapping一脸懵。

5.2 索引模板与ILM策略

nginx访问日志是典型的追加型数据，单索引如果一直写，一段时间后shard会非常大，检索变慢、恢复也慢。我建议从第一天就上ILM策略，按大小或时间做rollover：

PUT _ilm/policy/nginx-access-policy { "policy": { "phases": { "hot": { "min_age": "0ms", "actions": { "rollover": { "max_size": "50gb", "max_age": "7d" } } }, "warm": { "min_age": "30d", "actions": { "forcemerge": { "max_num_segments": 1 } } }, "delete": { "min_age": "180d", "actions": { "delete": {} } } } } }

然后把这个策略挂到索引模板上。有了rollover之后，索引名后面一定要带时间戳和序号，比如nginx-access-2025.04.18-000001，别再用纯粹的按天索引，否则ILM没法自动翻滚。

要注意的是：如果你已经写入了不少数据，再改模板只对新建索引生效，老索引还是旧mapping。所以模板、ILM这些要在一开始就定好，中途改的话通常要重建索引。

5.3 认证与重试背后的细节

ES集群如果开了HTTPS，Logstash这边还要配CA证书路径，否则会报SSL证书校验失败。自签名证书的场景下，常见做法是：

elasticsearch { hosts => ["https://es-node-1:9200"] ssl_certificate_authorities => ["/etc/logstash/certs/ca.crt"] ssl_verification_mode => "full" }

Logstash写ES默认是批量提交，它会把批次攒到一定程度再发，吞吐量已经不错。如果你压测发现写入速度上不去，可以调Logstash自身的pipeline.batch.size和pipeline.batch.delay，但别盲目调大，批量太大ES的bulk压力也大，容易导致reject。

ES节点暂时不可用时，Logstash会对写入失败的数据进行重试。默认情况重试节奏是退避式的，如果ES长时间不可用，积压的数据可能把Logstash的队列撑爆。所以生产环境我建议开启持久化队列。

6. 全链路排障思路与日常维护

6.1 分段验证，别一步直接上ES

我见过不少新手把input、filter、output一次写完，然后发现Kibana里没数据，排查半天不知道是读文件的问题还是正则的问题。正确做法是分段验证。

第一阶段，用stdin配合stdout，先在终端手动粘贴两行样例日志，确认grok和date解析正常。第二阶段，output改回stdout，让它读真实文件，确认文件路径、字符集、日志内容真的被读到。第三阶段，再接ES。三个阶段都通了，问题才能确认在哪个环节。

一套相对完整的生产配置大概是下面这个样子，你可以按这个骨架改：

input { file { path => ["/var/log/nginx/access.log"] type => "nginx-access" start_position => "beginning" sincedb_path => "/var/lib/logstash/sincedb/nginx-access" sincedb_write_interval => 15 stat_interval => 1 codec => plain { charset => "UTF-8" } } } filter { if [type] == "nginx-access" { grok { match => { "message" => "%{IPORHOST:clientip} - (%{USER:remote_user}|-) \[%{HTTPDATE:timestamp}\] \"%{WORD:method} %{URIPATHPARAM:request} HTTP/%{NUMBER:httpversion}\" %{INT:status} %{INT:body_bytes_sent} \"(?:%{URI:http_referer}|-)\" \"%{DATA:user_agent}\"" } break_on_match => false named_captures_only => true } date { match => ["timestamp", "dd/MMM/yyyy:HH:mm:ss Z"] locale => "en" target => "@timestamp" } geoip { source => "clientip" target => "geoip" } useragent { source => "user_agent" target => "ua" } mutate { remove_field => ["timestamp", "message"] convert => { "status" => "integer" "body_bytes_sent" => "integer" } } } } output { elasticsearch { hosts => ["http://es-node-1:9200"] index => "nginx-access-%{+YYYY.MM.dd}" user => "elastic" password => "${ES_PASSWORD}" manage_template => false } }

调试时要改的只有一点：把output临时换成stdout { codec => rubydebug }，然后观察打印的event。确认没问题再换回ES，这个习惯能省下一整晚的排障时间。

6.2 常见问题对照表

6.3 性能与稳定性：队列、workers和内存

Logstash的grok解析是CPU密集操作，正则写得太长、字段太多，单pipeline吞吐会明显下降。遇到日志量大、解析吃紧的情况，优化顺序是这样：

先检查pipeline.workers。Logstash 8默认workers等于CPU核数，但如果你在容器里跑，默认值可能不对，需要显式设置。

再检查持久化队列。这个尤其重要。默认Logstash用内存队列，进程一重启，积压的数据直接丢。nginx日志本身就是不可再生的，丢了就找不回来。在logstash.yml里开启：

queue.type: persisted queue.max_bytes: 8gb queue.checkpoint.writes: 1024

开启之后，Logstash重启、甚至整机宕机，缓冲区里的日志都会在服务恢复后继续写入ES。代价是占用磁盘，但对日志系统来说，这个代价完全值得。

如果同一台Logstash同时接收nginx日志、业务日志、数据库慢日志，我建议拆成多个pipeline，每种日志一个pipeline配置文件，独立worker和队列。这样nginx解析的正则有问题时，不会拖累其他日志流。

最后聊一个维护层面的习惯：每次上线或修改Logstash配置，先执行一次配置校验，bin/logstash -t -f /etc/logstash/conf.d/nginx.conf，它会告诉你配置文件有没有语法错误。但要注意，这种校验不会帮你验证grok到底能不能匹配真实日志，所以测试环境和生产环境各跑一段实际数据，才是真正靠谱的方式。我自己在实操中还有个坚持了很久的做法：在一台测试机上把nginx日志格式改一版，就同步在Logstash配置里改一版，然后对比两个版本的字段差异。等真的切到生产那天，你才会感激这个习惯。