Apache作为广泛使用的Web服务器，每天会产生大量的访问日志和错误日志。这些文本里包含客户端IP、请求路径、状态码、响应时间、来源页面和浏览器信息，但当节点数量增多后，日志分散在各台机器上，单纯依靠tail、grep或awk已经很难快速定位问题。ELK Stack提供了一条完整的日志处理链路：Filebeat负责在每台Apache节点上轻量采集，Logstash负责解析和过滤，Elasticsearch负责存储和索引，Kibana负责检索和可视化。本文以Apache 2.4和ELK 8.x为例，介绍如何把Apache的access.log和error.log接入这套体系。
一、统一Apache日志格式
要让Logstash稳定解析日志，第一步是统一Apache的日志输出格式。Apache默认提供两种常用格式：common和combined。其中combined格式在common的基础上增加了Referer和User-Agent两个字段，适合直接用于访问分析。但如果还想记录请求耗时、请求唯一ID等信息，就需要自定义LogFormat。
Apache的日志格式通过百分号指令定义，例如%h表示客户端IP，%t表示时间，%r表示请求行，%>s表示最终状态码，%b表示响应字节数。可以在主配置文件的<VirtualHost>块或全局配置中增加一段LogFormat，把响应耗时和唯一请求标识也输出到日志中。这样每条访问日志都会带上更多可用于性能分析的信息。
LogFormat "%h %l %u %t \"%r\" %>s %b \"%{Referer}i\" \"%{User-Agent}i\" %D %{UNIQUE_ID}e" extended
CustomLog "logs/access.log" extended
错误日志的多行输出也是常见难题。PHP异常堆栈或Apache错误详情经常跨越多行，如果Filebeat按行采集，后续Logstash只能收到不完整的message。因此除了统一访问日志格式，还需要在Filebeat中启用多行合并，把以[开头的条目作为一条新日志的起始标志。
二、用Filebeat采集Apache日志
Filebeat是一个使用Go编写的轻量级日志采集器，部署在Apache所在的主机上，通过读取日志文件的新增内容并把事件发送到Logstash或Elasticsearch。它占用的内存和CPU很小，适合在每台Web节点上运行。对于滚动日志，Filebeat会自动跟踪文件偏移，并且在文件被轮转后继续读取新文件。
下面的配置示例定义了两个日志输入：访问日志和错误日志。访问日志按行读取，错误日志启用了多行合并，使用multiline.pattern匹配以[开头的行，multiline.negate设为true表示不匹配该模式的行会合并到上一条事件中，multiline.match设为after保证后续行追加到前一行之后。同时通过fields给日志打上服务名和环境标记，便于在Logstash中区分来源。
filebeat.inputs:
- type: log
enabled: true
paths:
- /var/log/apache2/access.log
fields:
service: apache
log_type: access
fields_under_root: true
- type: log
enabled: true
paths:
- /var/log/apache2/error.log
fields:
service: apache
log_type: error
fields_under_root: true
multiline.pattern: '^\['
multiline.negate: true
multiline.match: after
output.logstash:
hosts: ["192.168.10.20:5044"]
Filebeat也提供了Apache模块，可以自动设置采集路径和解析规则，但如果后面使用Logstash做自定义过滤，通常会关闭模块自带的ingest pipeline，采用上边这种原始文本采集方式。原始文本更具灵活性，不会因为模块版本变化影响字段结构。
三、用Logstash解析和过滤日志
Logstash接收Filebeat发来的事件后，需要把原始文本拆分成结构化字段。核心工具是grok表达式，它使用预定义的正则模式来匹配不同部分。对于Apache combined格式，Logstash内置了COMBINEDAPACHELOG模式，可以直接解析出clientip、timestamp、verb、request、response、bytes、referrer和agent等字段。如果前面在LogFormat中追加了%D和%{UNIQUE_ID}e，则需要在grok模式末尾继续拼接对应的模式。
下面的pipeline配置展示了完整的input、filter和output。filter中先通过if [fields][log_type] == "access"判断事件来源，然后对访问日志执行grok解析。解析成功后，date插件会把timestamp字符串转换为@timestamp时间字段，mutate插件把response转换为整数类型，并重命名字段以符合后续索引命名规范。
input {
beats {
port => 5044
}
}
filter {
if [fields][log_type] == "access" {
grok {
match => { "message" => "%{COMBINEDAPACHELOG} %{NUMBER:response_time_micros:float} %{UNIQUE_ID:unique_id}" }
}
date {
match => [ "timestamp", "dd/MMM/yyyy:HH:mm:ss Z" ]
target => "@timestamp"
timezone => "Asia/Shanghai"
}
mutate {
convert => { "response" => "integer" }
rename => { "bytes" => "response_bytes" }
}
}
if [fields][log_type] == "error" {
grok {
match => { "message" => "\[%{HTTPDATE:error_timestamp}\] \[%{DATA:module}:%{LOGLEVEL:loglevel}\] %{GREEDYDATA:error_message}" }
}
date {
match => [ "error_timestamp", "EEE MMM dd HH:mm:ss.SSSSSS yyyy" ]
target => "@timestamp"
timezone => "Asia/Shanghai"
}
}
}
output {
elasticsearch {
hosts => ["http://192.168.10.30:9200"]
index => "apache-logs-%{+YYYY.MM.dd}"
}
}
grok解析失败是接入阶段最常见的问题。可以在Logstash本机使用--config.test_and_exit参数检查配置语法，或借助Kibana中的Grok Debugger工具，用一条真实日志反复调整模式。如果某些字段在特定情况下缺失，例如请求头Referer为空，COMBINEDAPACHELOG仍然可以匹配，因为缺失值通常由Apache输出为-。字段类型转换失败时，Logstash默认会添加_grokparsefailure标签，可以利用这个标签在output中把失败事件写入单独的dead letter index，避免丢失原始日志。
四、在Elasticsearch中创建索引并配置映射
Logstash写入Elasticsearch时使用按天滚动的索引名apache-logs-YYYY.MM.dd，这种方式便于清理历史数据和限定查询范围。为了让所有日期索引拥有统一的字段类型，需要预先创建索引模板。模板中可以指定分片数、副本数，以及字段映射，例如把response和response_bytes定义为整数，把response_time_micros定义为浮点数，避免动态映射把数字误判为字符串。
PUT _index_template/apache_logs
{
"index_patterns": ["apache-logs-*"],
"template": {
"settings": {
"number_of_shards": 3,
"number_of_replicas": 1
},
"mappings": {
"properties": {
"clientip": { "type": "ip" },
"response": { "type": "integer" },
"response_bytes": { "type": "integer" },
"response_time_micros": { "type": "float" },
"timestamp": { "type": "date", "format": "dd/MMM/yyyy:HH:mm:ss Z" },
"request": { "type": "text", "fields": { "keyword": { "type": "keyword", "ignore_above": 256 } } }
}
}
}
}
在Kibana中创建Data View时，索引模式填写apache-logs-*，时间字段选择@timestamp。之后就可以在Discover页面输入KQL查询，例如response:500查看所有5xx错误，或者response_time_micros > 1000000找出响应超过1秒的慢请求。Kibana的Lens和TSVB可以基于这些字段快速生成状态码分布饼图、响应时间趋势折线图、客户端IP访问量Top榜等图表。
针对错误日志，可以在仪表盘上添加一个数据表，筛选loglevel:error，并同时展示error_message字段。这样运维人员不需要登录服务器就能查看最近发生的错误详情。如果需要监控特定错误关键字，还可以创建基于Elasticsearch alerting的规则，当错误频率超过阈值时发送通知。
五、多节点扩展与性能优化
当Apache节点数量增加时，Filebeat本身几乎不会成为瓶颈，但Logstash需要合理调整并发。可以在一台Logstash服务器上增加pipeline.workers参数，让多个工作线程并行处理事件，也可以部署多台Logstash组成处理层，由Filebeat输出到多个地址实现负载均衡。如果担心Logstash宕机导致日志积压或丢失，可以在Filebeat和Logstash之间加入Kafka或Redis作为缓冲队列。
Elasticsearch层需要根据日志量设计分片。对于中小规模环境，每天一个主分片和一个副本通常够用；如果每天写入数十GB以上，可以扩大分片数，但要注意单分片过大或过小都会影响性能。日志数据默认属于写多读少场景，建议使用索引生命周期管理（ILM），把超过30天的索引自动迁移到冷节点或直接删除，控制集群磁盘占用。
时区是日志分析中容易被忽略的环节。Apache默认使用服务器本地时间写日志，而Elasticsearch的@timestamp字段要求UTC时间。Logstash的date过滤器可以通过timezone参数把原始时间正确转换为UTC，避免了Kibana中图表横轴偏移若干小时的问题。另外，如果多个节点的系统时区不一致，也会导致解析出的时间混乱，所以建议在节点初始化时统一设置为同一时区，并在Logstash中显式指定timezone。
最后，部署完成后建议持续关注_grokparsefailure标签出现次数。当Apache日志格式发生调整，例如新增了一个头信息或改变了分隔符，旧的grok模式可能无法匹配新格式。这时候需要根据失败样本更新Logstash配置，并重新加载pipeline。借助ELK的集中化能力，这类问题可以在几分钟内发现并修复，而不用逐台登录服务器查看日志。
ELK StackApache日志Filebeat采集修改时间：2026-10-05 11:27:15