elk-stack

ELK Stack

使用 Elasticsearch、Logstash 和 Kibana 集中管理并分析日志。

适用场景

在以下情况下使用此技能：

• 集中管理来自多个来源的日志
• 构建日志搜索与分析平台
• 创建基于日志的仪表板和警报
• 为日志实现全文搜索
• 处理和转换日志数据

前提条件

• Docker 或服务器基础设施
• 足够的日志存储磁盘空间
• 日志来源可通过网络访问

Docker 部署

# docker-compose.yml
version: '3.8'

services:
  elasticsearch:
    image: docker.elastic.co/elasticsearch/elasticsearch:8.11.0
    environment:
      - discovery.type=single-node
      - xpack.security.enabled=false
      - "ES_JAVA_OPTS=-Xms1g -Xmx1g"
    ports:
      - "9200:9200"
    volumes:
      - elasticsearch-data:/usr/share/elasticsearch/data

  logstash:
    image: docker.elastic.co/logstash/logstash:8.11.0
    volumes:
      - ./logstash/pipeline:/usr/share/logstash/pipeline
      - ./logstash/config:/usr/share/logstash/config
    ports:
      - "5044:5044"
      - "5000:5000"
    depends_on:
      - elasticsearch

  kibana:
    image: docker.elastic.co/kibana/kibana:8.11.0
    ports:
      - "5601:5601"
    environment:
      - ELASTICSEARCH_HOSTS=http://elasticsearch:9200
    depends_on:
      - elasticsearch

  filebeat:
    image: docker.elastic.co/beats/filebeat:8.11.0
    user: root
    volumes:
      - ./filebeat/filebeat.yml:/usr/share/filebeat/filebeat.yml:ro
      - /var/lib/docker/containers:/var/lib/docker/containers:ro
      - /var/run/docker.sock:/var/run/docker.sock:ro
    depends_on:
      - logstash

volumes:
  elasticsearch-data:

Elasticsearch 配置

索引模板

PUT _index_template/logs-template
{
  "index_patterns": ["logs-*"],
  "template": {
    "settings": {
      "number_of_shards": 1,
      "number_of_replicas": 1,
      "index.lifecycle.name": "logs-policy"
    },
    "mappings": {
      "properties": {
        "@timestamp": { "type": "date" },
        "message": { "type": "text" },
        "level": { "type": "keyword" },
        "service": { "type": "keyword" },
        "host": { "type": "keyword" },
        "trace_id": { "type": "keyword" }
      }
    }
  }
}

索引生命周期管理

PUT _ilm/policy/logs-policy
{
  "policy": {
    "phases": {
      "hot": {
        "min_age": "0ms",
        "actions": {
          "rollover": {
            "max_size": "50GB",
            "max_age": "1d"
          }
        }
      },
      "warm": {
        "min_age": "7d",
        "actions": {
          "shrink": { "number_of_shards": 1 },
          "forcemerge": { "max_num_segments": 1 }
        }
      },
      "cold": {
        "min_age": "30d",
        "actions": {
          "freeze": {}
        }
      },
      "delete": {
        "min_age": "90d",
        "actions": {
          "delete": {}
        }
      }
    }
  }
}

Logstash 管道

基本管道

# logstash/pipeline/main.conf
input {
  beats {
    port => 5044
  }
  
  tcp {
    port => 5000
    codec => json_lines
  }
}

filter {
  # Parse JSON logs
  if [message] =~ /^\{/ {
    json {
      source => "message"
    }
  }
  
  # Parse timestamp
  date {
    match => ["timestamp", "ISO8601", "yyyy-MM-dd HH:mm:ss"]
    target => "@timestamp"
  }
  
  # Add environment tag
  mutate {
    add_field => { "environment" => "production" }
  }
  
  # Grok pattern for nginx logs
  if [type] == "nginx" {
    grok {
      match => {
        "message" => '%{IPORHOST:client_ip} - %{USER:user} \[%{HTTPDATE:timestamp}\] "%{WORD:method} %{URIPATHPARAM:request} HTTP/%{NUMBER:http_version}" %{NUMBER:status} %{NUMBER:bytes}'
      }
    }
  }
}

output {
  elasticsearch {
    hosts => ["elasticsearch:9200"]
    index => "logs-%{+YYYY.MM.dd}"
  }
}

高级筛选

filter {
  # Parse application logs
  grok {
    match => {
      "message" => "%{TIMESTAMP_ISO8601:timestamp} %{LOGLEVEL:level} \[%{DATA:service}\] %{GREEDYDATA:log_message}"
    }
  }
  
  # Extract trace ID from message
  if [log_message] =~ /trace_id=/ {
    grok {
      match => { "log_message" => "trace_id=%{UUID:trace_id}" }
    }
  }
  
  # GeoIP lookup
  if [client_ip] {
    geoip {
      source => "client_ip"
      target => "geoip"
    }
  }
  
  # Drop debug logs in production
  if [level] == "DEBUG" and [environment] == "production" {
    drop {}
  }
  
  # Enrich with lookup
  translate {
    field => "status"
    destination => "status_description"
    dictionary => {
      "200" => "OK"
      "404" => "Not Found"
      "500" => "Internal Server Error"
    }
  }
}

Filebeat 配置

# filebeat/filebeat.yml
filebeat.inputs:
  - type: container
    paths:
      - '/var/lib/docker/containers/*/*.log'
    processors:
      - add_docker_metadata:
          host: "unix:///var/run/docker.sock"

  - type: log
    enabled: true
    paths:
      - /var/log/nginx/*.log
    tags: ["nginx"]
    fields:
      type: nginx

output.logstash:
  hosts: ["logstash:5044"]

logging.level: info
logging.to_files: true
logging.files:
  path: /var/log/filebeat
  name: filebeat
  keepfiles: 7

Elasticsearch 查询

基本查询

// Search all logs
GET logs-*/_search
{
  "query": {
    "match_all": {}
  }
}

// Search by keyword
GET logs-*/_search
{
  "query": {
    "match": {
      "message": "error"
    }
  }
}

// Filter by field
GET logs-*/_search
{
  "query": {
    "bool": {
      "must": [
        { "match": { "level": "ERROR" } },
        { "range": { "@timestamp": { "gte": "now-1h" } } }
      ],
      "filter": [
        { "term": { "service": "api-gateway" } }
      ]
    }
  }
}

聚合

// Count by log level
GET logs-*/_search
{
  "size": 0,
  "aggs": {
    "log_levels": {
      "terms": { "field": "level" }
    }
  }
}

// Error rate over time
GET logs-*/_search
{
  "size": 0,
  "aggs": {
    "errors_over_time": {
      "date_histogram": {
        "field": "@timestamp",
        "fixed_interval": "5m"
      },
      "aggs": {
        "error_count": {
          "filter": { "term": { "level": "ERROR" } }
        }
      }
    }
  }
}

Kibana 设置

索引模式

1. 前往“堆栈管理”→“索引模式”
2. 创建模式：logs-*
3. 设置时间字段：@timestamp

已保存的搜索

为常见查询创建已保存的搜索：

• level:ERROR - 所有错误
• service:api-gateway AND level:ERROR - API 网关错误
• response_time:>1000 - 慢请求

可视化

常见的可视化类型：

• 折线图：随时间变化的错误率
• 饼图：按日志级别划分的分布
• 数据表：错误消息排名
• 指标：错误总数

仪表板示例

创建包含以下内容的仪表板：

1. 日志总数（指标）
2. 错误率趋势（折线图）
3. 按服务划分的日志（饼图）
4. 最近的错误（数据表）
5. 日志流（Discover 面板）

告警

Watcher（X-Pack）

PUT _watcher/watch/error_alert
{
  "trigger": {
    "schedule": { "interval": "5m" }
  },
  "input": {
    "search": {
      "request": {
        "indices": ["logs-*"],
        "body": {
          "query": {
            "bool": {
              "must": [
                { "match": { "level": "ERROR" } },
                { "range": { "@timestamp": { "gte": "now-5m" } } }
              ]
            }
          }
        }
      }
    }
  },
  "condition": {
    "compare": { "ctx.payload.hits.total.value": { "gt": 100 } }
  },
  "actions": {
    "notify_slack": {
      "webhook": {
        "scheme": "https",
        "host": "hooks.slack.com",
        "port": 443,
        "method": "post",
        "path": "/services/xxx",
        "body": "{\"text\": \"High error rate detected: {{ctx.payload.hits.total.value}} errors in last 5 minutes\"}"
      }
    }
  }
}

常见问题

问题：磁盘使用率过高

问题：Elasticsearch 占用过多磁盘空间 解决方案：实施 ILM 策略，缩短保留期限

问题：搜索缓慢

问题：查询耗时过长 解决方案：优化索引设置、增加分片、使用筛选器

问题：日志解析失败

问题：日志未正确解析 解决方案：测试 grok 模式，检查日志格式是否发生变化

问题：内存压力

问题：Elasticsearch 出现 OOM 错误 解决方案：增加堆大小（最多为 RAM 的 50%），限制字段数据

最佳实践

• 实施索引生命周期管理
• 使用索引模板确保映射一致
• 在摄取时解析日志
• 限制存储字段以减少存储空间
• 对时间序列数据使用数据流
• 监控集群健康状况
• 实施适当的安全措施（X-Pack）
• 定期维护索引

相关技能

• loki-logging - 另一种日志堆栈
• prometheus-grafana - 指标监控
• audit-logging - 合规日志记录