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. ［Stack Management］→［Index Patterns］に移動します
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 - コンプライアンスログ