Logstash日志清洗与过滤的核心是filter阶段，按预处理→解析→后处理三步展开：先用mutate清理控制符和空白，再用dissect或grok结构化解析，最后完成类型转换、字段裁剪、条件过滤与脱敏。
Logstash 做日志清洗与过滤，核心是围绕 input → filter → output 这条流水线展开。它不存储也不展示，专干一件事：把杂乱的日志“翻译”成结构清晰、字段明确、可直接用于查询或可视化的数据。
日志清洗与过滤的关键阶段
Logstash 的 filter 阶段承担全部清洗和结构化任务。实际操作中，通常按顺序完成三类动作：预处理 → 解析 → 后处理。
- 预处理：清理原始 message 中干扰解析的字符
- 解析：用 dissect 或 grok 抽取关键字段
- 后处理：类型转换、字段裁剪、条件分发、脱敏等
清洗原始日志（预处理）
很多日志在采集过程中混入控制符、零宽空格、多余空白，导致后续 grok 匹配失败且无明确报错。
必须在 grok 之前做净化：
filter {
mutate {
gsub => [
"message", "\x00|\x01|\x02|\x03|\x04|\x05|\x06|\x07|\x08|\x0b|\x0c|\x0e|\x0f|\x10|\x11|\x12|\x13|\x14|\x15|\x16|\x17|\x18|\x19|\x1a|\x1b|\x1c|\x1d|\x1e|\x1f", "",
"message", "[[:space:]]{2,}", " ",
"message", "[\u200b-\u200f\u2028\u2029\u202a-\u202e\u2060-\u2064\u2066-\u206f\ufeff]", ""
]
}
}-
\x00-\x1f是常见控制符，Windows/Linux/Java 日志中都可能出现 -
\u200b类零宽空格常来自复制粘贴，会卡住%{QS:request}等模式 - 多个连续空格统一压缩为单个，避免字段错位
解析日志（结构化）
推荐优先使用 dissect（性能高、语法简），再考虑 grok（灵活性强、模式丰富）。
-
Dissect 示例（Nginx access log）
dissect { mapping => { "message" => "%{clientip} - %{user} [%{timestamp}] \"%{method} %{request} %{protocol}\" %{status} %{bytes} \"%{referrer}\" \"%{agent}\" \"%{x_forwarded_for}\"" } } -
Grok 示例（兼容复杂变体）
grok { match => { "message" => "%{IPORHOST:clientip} - %{DATA:user} \[%{HTTPDATE:timestamp}\] \"%{WORD:method} %{URIPATHPARAM:request} %{DATA:protocol}\" %{NUMBER:status:int} %{NUMBER:bytes:int} \"%{DATA:referrer}\" \"%{DATA:agent}\" \"%{DATA:x_forwarded_for}\"" } } -
注意事项：
-
URIPATHPARAM能自动分离路径与 query 参数（如/api/v1/user?id=123→request_path="/api/v1/user"，request_query="id=123"） - 字段后加
:int或:float可自动转类型，省去 mutate convert 步骤 - 对 error.log 等非标准格式，建议用
if [type] == "nginx-error"加条件判断，避免误匹配
-
后处理（字段精炼与业务适配）
解析出字段后，还需进一步加工才能满足分析需求：
-
提取接口主路径（便于聚合统计）
if [request] { grok { match => ["request", "^%{URIPATH:request_path}(\?%{GREEDYDATA:request_query})?$"] } } -
转换数值类型（Elasticsearch 聚合必需）
mutate { convert => ["status", "integer"] } -
过滤无效日志（减少写入量）
if [status] !~ /^2|3\d{2}$/ { drop {} } -
脱敏敏感字段（如手机号、token）
mutate { gsub => ["message", "(token=)[^&\s]+", "\1***"] }
实际配置结构示意
一个典型的 Nginx 日志处理 pipeline：
input {
file { path => "/var/log/nginx/access.log" type => "nginx-access" }
}
filter {
# 预处理
mutate { gsub => [ ... ] }
# 解析
dissect { ... }
# 或 grok { ... }
# 后处理
if [request] { grok { ... } }
mutate { convert => ["status", "integer", "bytes", "integer"] }
if ![status] { drop {} }
}
output {
elasticsearch { hosts => ["http://es:9200"] index => "nginx-access-%{+YYYY.MM.dd}" }
}不复杂但容易忽略。
