上一篇【第52篇】Elastic Stack全景解读——ES、Logstash、Beats与Kibana的协作
下一篇【第54篇】Elasticsearch Mapping设计最佳实践——从类型选择到性能优化

---

摘要

本文以Nginx Web服务器日志为案例，完整演示使用ELK Stack构建集约化日志管理平台的实战流程。从整体架构设计（Nginx→Filebeat→Logstash→Elasticsearch→Kibana）出发，逐项讲解Filebeat的日志采集配置、Logstash Pipeline的三大处理阶段（Input输入、Filter过滤、Output输出）以及Grok正则模式的构建方法。详细分析索引模板与ILM（索引生命周期管理）策略的配置，包括Hot-Warm-Cold-Delete四阶段的自动化管理。最后演示基于Kibana构建日志分析仪表板，涵盖日志搜索面板、时间分布直方图、错误统计T op N等核心分析组件的实现方式。

关键词：ELK Stack；Filebeat；Logstash Pipeline；Grok解析；ILM生命周期；日志管理平台；Nginx日志分析

---

一、整体架构设计

1.1 架构拓扑图

                    采集层              缓冲/处理层          存储层             展示层
                    
┌──────────────┐                      ┌───────────┐      ┌──────────────┐    ┌──────────┐
│  Nginx-01    │──┐                   │           │      │              │    │          │
│  /var/log/   │  │    ┌─────────┐    │           │      │ Elasticsearch│    │  Kibana  │
│  nginx/      │  │───→│Filebeat │───→│ Logstash  │─────→│              │───→│          │
└──────────────┘  │    └─────────┘    │           │      │  (集群)      │    │Dashboard │
                  │                   │ Pipeline  │      │              │    │          │
┌──────────────┐  │    ┌─────────┐    │           │      │  nginx-logs* │    │ Discover │
│  Nginx-02    │──┼───→│Filebeat │───→│ input/    │─────→│  error-*     │    │          │
│  /var/log/   │  │    └─────────┘    │ filter/   │      │  metricbeat* │    │ Alerting │
│  nginx/      │  │                   │ output    │      │              │    │          │
└──────────────┘  │                   └───────────┘      └──────────────┘    └──────────┘
                  │
┌──────────────┐  │    ┌─────────┐
│  App-03      │──┼───→│Filebeat │
│  /var/log/   │  │    └─────────┘
│  app/        │  │
└──────────────┘  │
                  ...

1.2 各组件职责

组件	版本建议	部署位置	职责描述
Nginx	1.20+	Web服务器	生产访问日志（access.log）和错误日志（error.log）
Filebeat	7.x/8.x	每台Nginx/App服务器	监控日志文件变化，采集新行并发送到Logstash
Logstash	7.x/8.x	独立服务器（可多台）	接收日志，Grok解析，字段转换，写入ES
Elasticsearch	7.x/8.x	集群（3+节点）	存储结构化日志，提供搜索和分析能力
Kibana	7.x/8.x	独立服务器	日志搜索、可视化仪表板、告警管理

二、日志格式定义

在开始采集之前，需要明确Nginx的日志格式。统一的日志格式是后续Grok解析的基础。

2.1 Nginx日志格式配置

# nginx.conf - 自定义日志格式
log_format json_combined escape=json '{'
    '"@timestamp": "$time_iso8601",'
    '"remote_addr": "$remote_addr",'
    '"remote_user": "$remote_user",'
    '"request": "$request",'
    '"status": $status,'
    '"body_bytes_sent": $body_bytes_sent,'
    '"request_time": $request_time,'
    '"http_referer": "$http_referer",'
    '"http_user_agent": "$http_user_agent",'
    '"http_x_forwarded_for": "$http_x_forwarded_for",'
    '"upstream_addr": "$upstream_addr",'
    '"upstream_response_time": $upstream_response_time,'
    '"host": "$host"'
'}';

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

2.2 日志格式对比

日志格式	解析难度	字段可读性	存储效率	推荐度
默认combined格式	需Grok正则	低	中	传统方案
JSON格式（推荐）	Logstash原生解析	高	中	强烈推荐
自定义分隔符	需编写解析规则	中	高	特殊场景

如果日志已经是JSON格式，Logstash可以跳过Grok解析直接使用json插件处理，大大降低解析复杂度。

三、Filebeat采集配置

3.1 Filebeat配置文件

# filebeat.yml - Nginx日志采集配置

# ====== 输入配置 ======
filebeat.inputs:
  - type: log
    enabled: true
    paths:
      - /var/log/nginx/access.log
    tags: ["nginx", "access", "production"]
    fields:
      log_type: nginx_access
      environment: production
      service: web_frontend
    fields_under_root: false
    
  - type: log
    enabled: true
    paths:
      - /var/log/nginx/error.log
    tags: ["nginx", "error", "production"]
    fields:
      log_type: nginx_error
      environment: production
      service: web_frontend
    fields_under_root: false

  - type: log
    enabled: true
    paths:
      - /var/log/app/*.log
    multiline.pattern: '^\d{4}-\d{2}-\d{2}'
    multiline.negate: true
    multiline.match: after
    tags: ["app", "production"]
    fields:
      log_type: application
      environment: production

# ====== 处理器配置 ======
processors:
  - add_host_metadata:
      when.not.contains.tags: forwarded
  - add_cloud_metadata: ~
  - drop_fields:
      fields: ["agent.ephemeral_id", "agent.id", "input.type"]
      ignore_missing: true

# ====== 输出配置 ======
output.logstash:
  hosts: ["logstash-01:5044", "logstash-02:5044"]
  loadbalance: true
  worker: 2
  compression_level: 3

# ====== 日志 ======
logging.level: info
logging.to_files: true
logging.files:
  path: /var/log/filebeat
  name: filebeat.log
  keepfiles: 7
  permissions: 0644

3.2 关键配置说明

配置项	值	说明
paths	/var/log/nginx/access.log	监控的日志文件路径，支持通配符
tags	["nginx", "access"]	给日志添加标签，Logstash可根据标签条件处理
fields	自定义字段	附加元数据，如环境、服务名
multiline.pattern	^\d{4}-\d{2}-\d{2}	多行日志合并规则（如Java异常堆栈）
output.logstash.loadbalance	true	多Logstash实例间的负载均衡
compression_level	3	传输压缩级别（1-9），平衡CPU和带宽

3.3 Filebeat处理器（Processors）详解

processors:
  # 添加服务器主机元数据（主机名、IP、操作系统信息等）
  - add_host_metadata: ~
  
  # 添加云平台元数据（AWS、GCP、Azure等）
  - add_cloud_metadata: ~
  
  # 解析User-Agent字符串，提取浏览器、操作系统信息
  - user_agent:
      fields: ["user_agent.original"]
      target: "user_agent"
  
  # 删除不需要的字段，减少数据传输量
  - drop_fields:
      fields: ["agent.ephemeral_id", "ecs.version"]
      ignore_missing: true
  
  # 添加自定义字段
  - add_fields:
      target: "environment"
      fields:
        dc: "beijing"
        rack: "A3"

四、Logstash Pipeline配置

4.1 完整Pipeline配置

# logstash.conf - Nginx日志处理Pipeline

# ====== Input 阶段 ======
input {
  beats {
    port => 5044
    host => "0.0.0.0"
    client_inactivity_timeout => 300
    ssl => true
    ssl_certificate => "/etc/logstash/certs/logstash.crt"
    ssl_key => "/etc/logstash/certs/logstash.key"
  }
}

# ====== Filter 阶段 ======
filter {
  # 根据tags区分日志类型
  if "nginx_access" in [tags] {
    
    # 如果是JSON格式，直接解析（无需Grok）
    if [message] =~ /^\{/ {
      json {
        source => "message"
        target => "nginx"
      }
    }
    
    # 如果不是JSON，使用Grok解析combined格式
    else {
      grok {
        match => { 
          "message" => '%{IPORHOST:nginx.remote_addr} - %{DATA:nginx.remote_user} \[%{HTTPDATE:nginx.timestamp}\] "%{WORD:nginx.method} %{DATA:nginx.url} HTTP/%{NUMBER:nginx.http_version}" %{NUMBER:nginx.status} (?:%{NUMBER:nginx.body_bytes_sent}|-) (?:"(?:%{DATA:nginx.http_referer}|-)"|%{QS:nginx.http_referer}) %{QS:nginx.http_user_agent}'
        }
      }
    }
    
    # 字段类型转换
    mutate {
      convert => {
        "nginx.status" => "integer"
        "nginx.body_bytes_sent" => "integer"
        "nginx.request_time" => "float"
      }
      add_field => {
        "log_category" => "access"
      }
      rename => {
        "nginx.remote_addr" => "client_ip"
      }
    }
    
    # GeoIP地理位置富化
    geoip {
      source => "client_ip"
      target => "geoip"
      database => "/etc/logstash/GeoLite2-City.mmdb"
      add_field => {
        "[geoip][coordinates]" => "%{[geoip][longitude]}"
        "[geoip][coordinates]" => "%{[geoip][latitude]}"
      }
    }
    
    # UserAgent解析
    useragent {
      source => "nginx.http_user_agent"
      target => "ua"
    }
  }
  
  # 错误日志处理
  if "nginx_error" in [tags] {
    grok {
      match => {
        "message" => '%{DATA:nginx.error_timestamp} \[%{WORD:nginx.error_level}\] %{NUMBER:nginx.pid}#%{NUMBER:nginx.tid}: (\*%{NUMBER:nginx.conn_id} )?%{GREEDYDATA:nginx.error_message}'
      }
    }
    
    mutate {
      add_field => {
        "log_category" => "error"
        "severity" => "%{nginx.error_level}"
      }
    }
  }
  
  # 应用日志处理
  if "app" in [tags] {
    grok {
      match => {
        "message" => '%{TIMESTAMP_ISO8601:app.timestamp} %{LOGLEVEL:app.level} %{DATA:app.logger} - %{GREEDYDATA:app.message}'
      }
    }
  }
  
  # ====== 通用处理 ======
  # 日期解析（用日志中的时间戳覆盖@timestamp）
  date {
    match => ["nginx.timestamp", "dd/MMM/yyyy:HH:mm:ss Z"]
    target => "@timestamp"
    tag_on_failure => ["_dateparsefailure"]
  }
  
  # 移除不需要的字段
  mutate {
    remove_field => ["message", "@version", "input", "ecs"]
  }
}

# ====== Output 阶段 ======
output {
  # 条件判断：根据日志类型写入不同索引
  if "nginx_access" in [tags] {
    elasticsearch {
      hosts => ["http://es-node-01:9200", "http://es-node-02:9200", "http://es-node-03:9200"]
      index => "nginx-access-%{+YYYY.MM.dd}"
      template => "/etc/logstash/templates/nginx-access-template.json"
      template_name => "nginx-access"
      template_overwrite => true
      manage_template => true
      user => "logstash_writer"
      password => "${ES_PASSWORD}"
    }
  }
  
  if "nginx_error" in [tags] {
    elasticsearch {
      hosts => ["http://es-node-01:9200", "http://es-node-02:9200", "http://es-node-03:9200"]
      index => "nginx-error-%{+YYYY.MM.dd}"
      user => "logstash_writer"
      password => "${ES_PASSWORD}"
    }
  }
  
  if "app" in [tags] {
    elasticsearch {
      hosts => ["http://es-node-01:9200", "http://es-node-02:9200", "http://es-node-03:9200"]
      index => "app-logs-%{+YYYY.MM.dd}"
      user => "logstash_writer"
      password => "${ES_PASSWORD}"
    }
  }
  
  # 错误日志也输出到文件备份
  if "nginx_error" in [tags] {
    file {
      path => "/var/log/logstash/error_backup-%{+YYYY-MM-dd}.log"
      codec => line { format => "%{message}" }
    }
  }
}

4.2 Pipeline三段式结构解析

Input阶段               Filter阶段                Output阶段
┌─────────────┐    ┌──────────────────┐    ┌──────────────────┐
│ beats {}    │    │ 1. grok/json解析  │    │ elasticsearch {} │
│             │    │ 2. mutate字段转换  │    │                  │
│ 从Beats接收  │───→│ 3. geoip富化      │───→│ 按日志类型写入    │
│ 日志事件     │    │ 4. useragent解析  │    │ 不同的ES索引      │
│             │    │ 5. date日期处理   │    │                  │
└─────────────┘    └──────────────────┘    │ file {} (备份)   │
                                           └──────────────────┘

五、Grok正则模式详解

5.1 Grok是什么

Grok是Logstash中最强大的文本解析工具，它将非结构化日志文本转换为结构化的键值对。Grok基于正则表达式，但通过预定义的模式库（Pattern Library）大大简化了正则表达式的编写。

5.2 内置模式库

Logstash内置了120+个预定义模式，涵盖最常见的日志格式需求：

模式类别	常用模式	匹配示例
IP/网络	IP, IPORHOST, HOSTNAME	192.168.1.1
时间日期	TIMESTAMP_ISO8601, HTTPDATE	2024-01-15T10:30:00Z
数字	NUMBER, INT, BASE10NUM	12345, -98.76
数据量	DATA, GREEDYDATA	任意字符串
URI/URL	URIPATH, URIPARAM, URI	/api/users?page=1
邮件	EMAILADDRESS, EMAILLOCALPART	user@example.com
系统日志	SYSLOGBASE, SYSLOGTIMESTAMP	syslog格式
HTTP	COMBINEDAPACHELOG, COMMONAPACHELOG	Nginx/Apache日志

5.3 Grok模式构建实例

场景一：解析Nginx默认combined格式

# Nginx combined格式日志行
# 192.168.1.100 - - [15/Jan/2024:10:30:45 +0800] "GET /api/users HTTP/1.1" 200 1234 "-" "Mozilla/5.0..."

# 对应Grok模式
filter {
  grok {
    match => {
      "message" => '%{IPORHOST:client_ip} %{USER:ident} %{USER:auth} \[%{HTTPDATE:timestamp}\] "%{WORD:method} %{URIPATHPARAM:request} HTTP/%{NUMBER:http_version}" %{NUMBER:response_code} %{NUMBER:body_bytes} "%{DATA:referrer}" "%{DATA:user_agent}"'
    }
  }
}

场景二：解析自定义应用日志

# 应用日志格式
# 2024-01-15 10:30:45.123 [http-nio-8080-exec-1] ERROR com.example.UserService - User not found: id=12345

filter {
  grok {
    match => {
      "message" => '%{TIMESTAMP_ISO8601:log_timestamp} \[%{DATA:thread}\] %{LOGLEVEL:log_level} %{JAVACLASS:logger} - %{GREEDYDATA:log_message}'
    }
  }
}

5.4 自定义Grok模式

当内置模式不满足需求时，可以创建自定义模式：

# custom_patterns.txt（自定义模式文件）
STATUS_CODE ([1-5][0-9]{2})
API_VERSION v[0-9]
REQUEST_ID [a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12}

# 在Logstash中引用自定义模式
filter {
  grok {
    patterns_dir => ["/etc/logstash/patterns"]
    match => {
      "message" => '%{REQUEST_ID:request_id} %{API_VERSION:api_version} %{STATUS_CODE:status}'
    }
  }
}

5.5 Grok调试技巧

方法	说明	适用场景
Grok Debugger	Kibana Dev Tools内置的Grok调试器	开发阶段，在线调试
_grokparsefailure	Grok匹配失败时的自动标签	生产监控，捕获解析失败
tag_on_failure	自定义失败标签	区分不同的匹配失败原因
break_on_match	匹配成功后不再继续尝试	提高性能

Kibana Grok Debugger使用步骤：

1. 打开Kibana → Dev Tools → Grok Debugger
2. 在Sample Data中输入示例日志行
3. 在Grok Pattern中输入Grok模式
4. 实时查看解析结果

# 生产环境中监控Grok解析成功率
filter {
  grok {
    match => { "message" => "%{COMBINEDAPACHELOG}" }
    tag_on_failure => ["_grokparsefailure"]
    overwrite => ["message"]
  }
  
  # 解析失败的日志仍然写入ES，便于排查
  if "_grokparsefailure" in [tags] {
    mutate {
      add_tag => ["grok_failed"]
      add_field => { "raw_message" => "%{message}" }
    }
  }
}

六、索引模板与ILM生命周期管理

6.1 索引模板配置

索引模板定义新创建索引的Mapping和Settings，确保每天自动创建的索引具有一致的结构。

PUT _index_template/nginx-access-template
{
  "index_patterns": ["nginx-access-*"],
  "priority": 200,
  "template": {
    "settings": {
      "number_of_shards": 3,
      "number_of_replicas": 1,
      "refresh_interval": "30s",
      "index.lifecycle.name": "nginx-logs-policy",
      "index.lifecycle.rollover_alias": "nginx-access"
    },
    "mappings": {
      "dynamic": "strict",
      "properties": {
        "@timestamp": {
          "type": "date"
        },
        "client_ip": {
          "type": "ip"
        },
        "nginx": {
          "properties": {
            "method": {
              "type": "keyword"
            },
            "url": {
              "type": "keyword"
            },
            "status": {
              "type": "short"
            },
            "body_bytes_sent": {
              "type": "integer"
            },
            "request_time": {
              "type": "scaled_float",
              "scaling_factor": 1000
            },
            "http_user_agent": {
              "type": "text",
              "fields": {
                "keyword": {
                  "type": "keyword",
                  "ignore_above": 256
                }
              }
            }
          }
        },
        "geoip": {
          "properties": {
            "country_name": {
              "type": "keyword"
            },
            "city_name": {
              "type": "keyword"
            },
            "location": {
              "type": "geo_point"
            }
          }
        },
        "ua": {
          "properties": {
            "name": {
              "type": "keyword"
            },
            "os": {
              "type": "keyword"
            },
            "device": {
              "type": "keyword"
            }
          }
        }
      }
    }
  }
}

6.2 ILM（索引生命周期管理）

ILM自动管理索引从创建到删除的完整生命周期，通过Hot-Warm-Cold-Delete四个阶段实现存储成本与查询性能的平衡。

PUT _ilm/policy/nginx-logs-policy
{
  "policy": {
    "phases": {
      "hot": {
        "min_age": "0ms",
        "actions": {
          "rollover": {
            "max_size": "50GB",
            "max_age": "1d",
            "max_docs": 100000000
          },
          "set_priority": {
            "priority": 100
          }
        }
      },
      "warm": {
        "min_age": "3d",
        "actions": {
          "forcemerge": {
            "max_num_segments": 1
          },
          "shrink": {
            "number_of_shards": 1
          },
          "allocate": {
            "number_of_replicas": 0,
            "require": {
              "data": "warm"
            }
          },
          "set_priority": {
            "priority": 50
          }
        }
      },
      "cold": {
        "min_age": "30d",
        "actions": {
          "allocate": {
            "number_of_replicas": 0,
            "require": {
              "data": "cold"
            }
          },
          "set_priority": {
            "priority": 0
          }
        }
      },
      "delete": {
        "min_age": "90d",
        "actions": {
          "delete": {
            "delete_searchable_snapshot": true
          }
        }
      }
    }
  }
}

6.3 ILM四阶段详解

阶段	进入时间	存储位置	副本数	主要操作	用途
Hot	立即	SSD	1-2	Rollover（滚动创建新索引）	接收新数据，高频查询
Warm	3天后	SSD/HDD	0-1	ForceMerge（合并段）、Shrink（缩小分片）	近期数据，中频查询
Cold	30天后	HDD/对象存储	0	冻结索引	历史数据，低频查询
Delete	90天后	—	—	删除	过期数据清理

6.4 ILM操作详解

{
  "ILM支持的Actions": {
    "Rollover": "当索引达到条件时滚动创建新索引",
    "Shrink": "减少主分片数量以降低资源消耗",
    "ForceMerge": "强制合并Lucene段，提升查询效率",
    "Allocate": "将索引分配到特定节点（warm/cold节点）",
    "ReadOnly": "将索引设为只读",
    "Freeze": "冻结索引以减少内存占用",
    "Delete": "删除过期索引",
    "Migrate": "迁移索引到其他层",
    "SearchableSnapshot": "挂载为可搜索快照"
  }
}

七、Kibana日志分析仪表板搭建

完成数据采集和处理后，最后一步是通过Kibana构建日志分析仪表板，让运维和开发团队直观了解系统运行状态。

7.1 仪表板组件设计

日志分析仪表板布局：

┌──────────────────────────────────────────────────────────────┐
│  时间选择器: [2024-01-15 ~ 2024-01-21]  [自动刷新: 30s]       │
├──────────────────────┬──────────────┬──────────────────────────┤
│  请求量统计           │  状态码分布    │  平均响应时间            │
│  (Metric - 总数)      │  (Pie图)     │  (Metric - ms)          │
│  1,234,567            │  2xx: 95%    │  156ms                   │
├──────────────────────┴──────────────┴──────────────────────────┤
│  请求量时间分布 (Line Chart - 1分钟间隔)                        │
│  ▁▂▃▅▆▇█▇▆▅▃▂▁▁▂▃▅▆▇█▇▆▅▃                                       │
├──────────────────────────────────────────────────────────────┤
│  Top 10 请求URL          │  Top 10 来源IP        │ 错误日志列表    │
│  (Bar Chart)             │  (Bar Chart)          │  (Data Table)  │
│  /api/users     ████▌   │  10.0.1.23    ███▌   │  时间 | 级别    │
│  /api/products  ███▌    │  10.0.1.45    ██▌    │  404错误详情     │
│  /api/orders    ██▌     │  10.0.2.11    █▌     │  500错误详情     │
└──────────────────────────────────────────────────────────────┘

7.2 核心可视化组件配置

请求量时间分布（Line Chart）

可视化类型: 折线图
索引模式: nginx-access-*
Y轴聚合: Count
X轴: @timestamp
时间间隔: 1分钟
分组:
  - 按 nginx.status 分组（区分正常/异常流量）

状态码分布（Pie Chart）

可视化类型: 饼图
索引模式: nginx-access-*
分片方式: Terms
字段: nginx.status
排序: Count, 降序
大小: 10
自定义标签颜色:
  2xx → 绿色 (#34a853)
  3xx → 蓝色 (#4285f4)
  4xx → 橙色 (#fbbc04)
  5xx → 红色 (#ea4335)

错误日志 Top N

可视化类型: 数据表
索引模式: nginx-error-*
Metrics:
  - Count (错误总数)
  按 nginx.error_level 分组
Split Rows:
  - Terms: nginx.error_message.keyword
  - Size: 20
  - 排序: Count, 降序

7.3 Dashboard控件配置

Controls:
  
  - 类型: Options List
    字段: nginx.method
    标签: "请求方法"
    选项: GET, POST, PUT, DELETE
    默认: 全部选择
  
  - 类型: Options List  
    字段: nginx.status
    标签: "状态码"
    选项: 200, 301, 302, 400, 403, 404, 500, 502, 503
    默认: 全部选择
  
  - 类型: Options List
    字段: nginx.host
    标签: "域名"
    默认: 全部选择

八、总结与最佳实践

核心要点回顾

1. 格式先行：在采集前统一日志格式（推荐JSON），可大幅降低解析复杂度
2. Filebeat采集、Logstash处理：职责清晰，各司其职，Beats做轻量采集，Logstash做复杂转换
3. Grok是解析核心：掌握Grok模式构建方法是处理非结构化日志的关键能力
4. ILM管理生命周期：自动化的Hot-Warm-Cold-Delete策略实现存储成本与性能的平衡
5. 仪表板是最终交付物：通过Kibana Dashboard让日志分析成果直接服务于业务团队

最佳实践清单

• 日志格式规范化：所有应用统一输出JSON格式日志，避免为每种格式编写不同的Grok模式
• Logstash Pipeline测试：使用logstash -t命令验证配置语法，使用stdin/stdout模式进行本地调试
• 索引滚动策略：生产环境配置ILM + Rollover，避免单个索引过大影响性能
• 解析失败监控：通过_grokparsefailure标签监控解析成功率，失败日志单独归档
• Pipeline模块化：将不同业务日志的处理逻辑拆分为独立的Pipeline文件，便于维护
• 资源规划：Logstash的pipeline.workers设为CPU核心数的1~2倍，pipeline.batch.size根据日志吞吐量调整

---

上一篇【第52篇】Elastic Stack全景解读——ES、Logstash、Beats与Kibana的协作
下一篇【第54篇】Elasticsearch Mapping设计最佳实践——从类型选择到性能优化

---