ELK日志平台 使用方式与典型场景

使用方式与典型场景

示例一：Filebeat 采集 Java 应用日志

filebeat.inputs:
  - type: filestream
    id: order-service
    paths:
      - /var/log/app/order-service/*.log
    fields:
      service.name: order
      env: prod
    fields_under_root: true
    # 把 Java 异常堆栈合并成一条事件，否则堆栈会被拆成几十行
    multiline:
      type: pattern
      pattern: '^\d{4}-\d{2}-\d{2}'
      negate: true
      match: after
      max_lines: 200
    # 只上报需要的，减少体积
    processors:
      - drop_fields:
          fields: ["agent.ephemeral_id", "log.offset"]

output.logstash:
  hosts: ["logstash.internal:5044"]
  # 开启压缩，跨机房时省带宽
  compression_level: 3

适合：日志写在文件里、格式以行为单位、需要保留原始上下文。不适合：应用直接往 socket 打日志且量大到不想落盘，那种场景可以直接用 Logstash 的 tcp input 或者 OpenTelemetry Collector。

示例二：Logstash 解析与富化

input {
  beats { port => 5044 }
}

filter {
  # 只对 JSON 行做解析，非 JSON 行原样保留
  json {
    source => "message"
    skip_on_invalid_json => true
    target => "doc"
  }
  # 从 message 里抽关键字段
  grok {
    match => { "message" => "%{TIMESTAMP_ISO8601:ts} %{LOGLEVEL:level} \[%{DATA:thread}\] %{DATA:logger} - %{GREEDYDATA:msg}" }
    tag_on_failure => ["_grok_parse_failed"]
  }
  # 用日志内时间覆盖采集时间，保证时间轴正确
  date {
    match => ["ts", "ISO8601"]
    target => "@timestamp"
    timezone => "Asia/Shanghai"
  }
  # 统一字段名、降体积
  mutate {
    rename => { "msg" => "log.message" }
    lowercase => ["level"]
    remove_field => ["ts", "thread", "host.os.version"]
    # 脱敏：把手机号中间四位打码
    gsub => ["log.message", "1[3-9]\d{9}", "[phone]"]
  }
}

output {
  elasticsearch {
    hosts => ["http://elasticsearch:9200"]
    user => "logstash_writer"
    password => "${ES_PASSWORD}"
    index => "app-logs-%{+yyyy.MM.dd}"
    ilm_enabled => true
    ilm_rollover_alias => "app-logs"
  }
}

注意 filter 的顺序：先 json 再 grok 再 date 再 mutate，因为 mutate 一旦把原始 message 删掉，后面的 grok 就没有输入了。

示例三：Kibana 检索与排障命令

# Kibana Discover / KQL 查询框里直接粘贴
level: "ERROR" and service.name: "order" and not message: "timeout"

# 想跨字段全文搜，用 Lucene 语法
message: ("NullPointerException" or "Connection refused")

# 看索引列表、文档数、体积（?v 带表头）
curl -u elastic:${ELASTIC_PASSWORD} 'localhost:9200/_cat/indices?v&s=index'

# 看某个索引处在 ILM 的哪个阶段、下一步什么时候执行
curl -u elastic:${ELASTIC_PASSWORD} 'localhost:9200/app-logs-000001/_ilm/explain?pretty'

# 磁盘到水位线后索引被置只读，清理完磁盘再解除 block
curl -u elastic:${ELASTIC_PASSWORD} -XPUT 'localhost:9200/_all/_settings' \
  -H 'Content-Type: application/json' \
  -d '{"index.blocks.read_only_allow_delete": null}'

# 看分片分布是否均匀、有没有未分配分片
curl -u elastic:${ELASTIC_PASSWORD} 'localhost:9200/_cat/allocation?v'

排障路径建议固定成三步：先在 Kibana 用 service.name 加时间范围缩小到具体服务，再用 trace id 或请求 id 拉出完整调用链，最后回到 _cat/indices 与 _ilm/explain 确认是不是索引层面的问题（写入被限流、分片失衡、生命周期卡住）。