日志聚合分析：ELK Stack 在千万级日志处理中的实践与优化

# 日志规模统计
log_statistics:
  daily_volume: "10,000,000+ records"
  peak_throughput: "5,000 logs/second"
  average_throughput: "1,200 logs/second"
  
  service_distribution:
    user_service: "2,500,000 logs/day"
    order_service: "2,000,000 logs/day"
    payment_service: "1,800,000 logs/day"
    inventory_service: "1,500,000 logs/day"
    other_services: "2,200,000 logs/day"
  
  log_types:
    business_logs: "60%"
    error_logs: "15%"
    performance_logs: "20%"
    audit_logs: "5%"

# 传统日志系统的问题分析脚本
#!/bin/bash

echo "=== 传统日志系统问题分析 ==="

# 1. 磁盘空间占用分析
echo "1. 磁盘空间占用情况："
du -sh /var/log/* | sort -hr | head -10

# 2. 日志文件数量统计
echo "2. 日志文件数量："
find /var/log -name "*.log" | wc -l

# 3. 大文件识别
echo "3. 超大日志文件（>1GB）："
find /var/log -name "*.log" -size +1G -exec ls -lh {} \;

# 4. 日志查询性能测试
echo "4. 日志查询性能测试："
time grep "ERROR" /var/log/application.log | wc -l

# 5. 磁盘I/O压力分析
echo "5. 磁盘I/O使用率："
iostat -x 1 3 | grep -E "(Device|sda)"

# Elasticsearch集群配置
apiVersion: elasticsearch.k8s.elastic.co/v1
kind: Elasticsearch
metadata:
  name: elasticsearch-cluster
spec:
  version: 8.8.0
  
  # Master节点配置
  nodeSets:
  - name: master
    count: 3
    config:
      node.roles: ["master"]
      xpack.security.enabled: true
      xpack.security.transport.ssl.enabled: true
    podTemplate:
      spec:
        containers:
        - name: elasticsearch
          resources:
            requests:
              memory: 4Gi
              cpu: 2
            limits:
              memory: 4Gi
              cpu: 2
          env:
          - name: ES_JAVA_OPTS
            value: "-Xms2g -Xmx2g"
    
  # 数据节点配置
  - name: data
    count: 6
    config:
      node.roles: ["data", "ingest"]
      indices.memory.index_buffer_size: "30%"
      indices.memory.min_index_buffer_size: "96mb"
    podTemplate:
      spec:
        containers:
        - name: elasticsearch
          resources:
            requests:
              memory: 16Gi
              cpu: 4
            limits:
              memory: 16Gi
              cpu: 4
          env:
          - name: ES_JAVA_OPTS
            value: "-Xms8g -Xmx8g"
        volumes:
        - name: elasticsearch-data
          persistentVolumeClaim:
            claimName: elasticsearch-data-pvc
    volumeClaimTemplates:
    - metadata:
        name: elasticsearch-data
      spec:
        accessModes:
        - ReadWriteOnce
        resources:
          requests:
            storage: 1Ti
        storageClassName: fast-ssd

  # 协调节点配置
  - name: coordinating
    count: 2
    config:
      node.roles: []
    podTemplate:
      spec:
        containers:
        - name: elasticsearch
          resources:
            requests:
              memory: 8Gi
              cpu: 2
            limits:
              memory: 8Gi
              cpu: 2
          env:
          - name: ES_JAVA_OPTS
            value: "-Xms4g -Xmx4g"

{
  "index_patterns": ["logs-*"],
  "template": {
    "settings": {
      "number_of_shards": 3,
      "number_of_replicas": 1,
      "index.refresh_interval": "30s",
      "index.translog.flush_threshold_size": "1gb",
      "index.merge.policy.max_merge_at_once": 5,
      "index.merge.policy.segments_per_tier": 5,
      "index.codec": "best_compression"
    },
    "mappings": {
      "properties": {
        "@timestamp": {
          "type": "date",
          "format": "strict_date_optional_time||epoch_millis"
        },
        "service_name": {
          "type": "keyword"
        },
        "log_level": {
          "type": "keyword"
        },
        "message": {
          "type": "text",
          "analyzer": "standard"
        },
        "request_id": {
          "type": "keyword"
        },
        "user_id": {
          "type": "keyword"
        },
        "response_time": {
          "type": "long"
        },
        "status_code": {
          "type": "integer"
        }
      }
    }
  }
}

{
  "policy": {
    "phases": {
      "hot": {
        "actions": {
          "rollover": {
            "max_size": "10gb",
            "max_age": "1d",
            "max_docs": 10000000
          },
          "set_priority": {
            "priority": 100
          }
        }
      },
      "warm": {
        "min_age": "2d",
        "actions": {
          "allocate": {
            "number_of_replicas": 0
          },
          "forcemerge": {
            "max_num_segments": 1
          },
          "set_priority": {
            "priority": 50
          }
        }
      },
      "cold": {
        "min_age": "7d",
        "actions": {
          "allocate": {
            "include": {
              "box_type": "cold"
            }
          },
          "set_priority": {
            "priority": 0
          }
        }
      },
      "delete": {
        "min_age": "30d"
      }
    }
  }
}

# Elasticsearch性能优化脚本
#!/bin/bash

# 1. 操作系统级别优化
echo "=== 系统级别优化 ==="

# 禁用swap
swapoff -a
echo 'vm.swappiness=1' >> /etc/sysctl.conf

# 增加文件描述符限制
echo 'elasticsearch soft nofile 65536' >> /etc/security/limits.conf
echo 'elasticsearch hard nofile 65536' >> /etc/security/limits.conf

# 增加虚拟内存映射限制
echo 'vm.max_map_count=262144' >> /etc/sysctl.conf
sysctl -p

# 2. JVM优化
echo "=== JVM优化 ==="
cat > /etc/elasticsearch/jvm.options.d/custom.options << EOF
# 垃圾回收器优化
-XX:+UseG1GC
-XX:G1HeapRegionSize=32m
-XX:+UnlockExperimentalVMOptions
-XX:+UseCGroupMemoryLimitForHeap

# GC日志配置
-Xlog:gc*,gc+age=trace,safepoint:gc.log:time,level,tags

# 内存优化
-XX:+AlwaysPreTouch
-Xss1m
-Djava.awt.headless=true
-Dfile.encoding=UTF-8
-Djna.nosys=true
-Dio.netty.noUnsafe=true
-Dio.netty.noKeySetOptimization=true
EOF

# 3. 集群配置优化
echo "=== 集群配置优化 ==="
curl -X PUT "localhost:9200/_cluster/settings" -H 'Content-Type: application/json' -d'
{
  "persistent": {
    "cluster.routing.allocation.disk.threshold.enabled": true,
    "cluster.routing.allocation.disk.watermark.low": "85%",
    "cluster.routing.allocation.disk.watermark.high": "90%",
    "cluster.routing.allocation.disk.watermark.flood_stage": "95%",
    "indices.recovery.max_bytes_per_sec": "100mb",
    "cluster.routing.allocation.node_concurrent_recoveries": 2,
    "cluster.routing.allocation.cluster_concurrent_rebalance": 2
  }
}
'

# Logstash管道配置
# pipelines.yml
- pipeline.id: application-logs
  path.config: "/usr/share/logstash/pipeline/application.conf"
  pipeline.workers: 4
  pipeline.batch.size: 1000
  pipeline.batch.delay: 50

- pipeline.id: error-logs
  path.config: "/usr/share/logstash/pipeline/error.conf"
  pipeline.workers: 2
  pipeline.batch.size: 500
  pipeline.batch.delay: 10

- pipeline.id: performance-logs
  path.config: "/usr/share/logstash/pipeline/performance.conf"
  pipeline.workers: 2
  pipeline.batch.size: 2000
  pipeline.batch.delay: 100

- pipeline.id: audit-logs
  path.config: "/usr/share/logstash/pipeline/audit.conf"
  pipeline.workers: 1
  pipeline.batch.size: 100
  pipeline.batch.delay: 5

# application.conf
input {
  redis {
    host => "redis-cluster"
    port => 6379
    key => "application-logs"
    data_type => "list"
    batch_count => 1000
    threads => 4
  }
}

filter {
  # JSON解析
  json {
    source => "message"
    target => "parsed"
  }
  
  # 时间戳处理
  date {
    match => [ "[parsed][timestamp]", "ISO8601" ]
    target => "@timestamp"
  }
  
  # 字段提取和规范化
  mutate {
    add_field => {
      "service_name" => "%{[parsed][service]}"
      "log_level" => "%{[parsed][level]}"
      "request_id" => "%{[parsed][requestId]}"
    }
    remove_field => [ "parsed", "host", "agent" ]
  }
  
  # 性能指标提取
  if [service_name] == "api-gateway" {
    grok {
      match => { 
        "message" => "Response time: %{NUMBER:response_time:int}ms, Status: %{NUMBER:status_code:int}"
      }
    }
  }
  
  # 错误日志特殊处理
  if [log_level] == "ERROR" {
    mutate {
      add_tag => [ "error", "alert" ]
    }
    
    # 提取异常堆栈
    if [message] =~ /Exception/ {
      multiline {
        pattern => "^\s"
        what => "previous"
      }
    }
  }
  
  # 用户ID脱敏
  if [user_id] {
    mutate {
      gsub => [ "user_id", "(\d{4})\d{4}(\d{4})", "\1****\2" ]
    }
  }
}

output {
  elasticsearch {
    hosts => ["elasticsearch-cluster:9200"]
    index => "logs-application-%{+YYYY.MM.dd}"
    template_name => "logs-application"
    template_pattern => "logs-application-*"
    manage_template => true
    
    # 性能优化配置
    workers => 4
    flush_size => 1000
    idle_flush_time => 10
  }
  
  # 错误日志额外输出到告警系统
  if "error" in [tags] {
    http {
      url => "http://alert-manager:9093/api/v1/alerts"
      http_method => "post"
      format => "json"
      mapping => {
        "alerts" => [{
          "labels" => {
            "alertname" => "ApplicationError"
            "service" => "%{service_name}"
            "severity" => "critical"
          }
          "annotations" => {
            "summary" => "Application error detected"
            "description" => "%{message}"
          }
        }]
      }
    }
  }
}

#!/bin/bash
# Logstash性能监控脚本

echo "=== Logstash性能监控 ==="

# 1. 管道统计信息
echo "1. 管道处理统计："
curl -s "http://logstash:9600/_node/stats/pipelines" | jq '.pipelines | to_entries[] | {
  pipeline: .key,
  events_in: .value.events.in,
  events_out: .value.events.out,
  events_filtered: .value.events.filtered,
  duration_in_millis: .value.events.duration_in_millis
}'

# 2. JVM内存使用情况
echo "2. JVM内存使用："
curl -s "http://logstash:9600/_node/stats/jvm" | jq '.jvm.mem'

# 3. 队列统计
echo "3. 队列统计："
curl -s "http://logstash:9600/_node/stats/pipeline" | jq '.pipeline.queue'

# 4. 处理延迟分析
echo "4. 处理延迟分析："
curl -s "http://logstash:9600/_node/stats/events" | jq '{
  in: .events.in,
  out: .events.out,
  filtered: .events.filtered,
  duration_in_millis: .events.duration_in_millis,
  avg_duration_per_event: (.events.duration_in_millis / .events.in)
}'

# Redis集群配置
cluster-enabled yes
cluster-config-file nodes.conf
cluster-node-timeout 5000
appendonly yes
appendfsync everysec

# 内存优化
maxmemory 8gb
maxmemory-policy allkeys-lru

# 网络优化
tcp-keepalive 60
timeout 300

# 持久化优化
save 900 1
save 300 10
save 60 10000

# filebeat.yml
filebeat.inputs:
# 应用日志收集
- type: log
  enabled: true
  paths:
    - /var/log/applications/*.log
  fields:
    log_type: application
    environment: production
  fields_under_root: true
  multiline.pattern: '^\d{4}-\d{2}-\d{2}'
  multiline.negate: true
  multiline.match: after
  scan_frequency: 10s
  harvester_buffer_size: 16384
  max_bytes: 10485760

# 错误日志收集
- type: log
  enabled: true
  paths:
    - /var/log/applications/error/*.log
  fields:
    log_type: error
    environment: production
  fields_under_root: true
  include_lines: ['ERROR', 'FATAL', 'Exception']
  
# 性能日志收集
- type: log
  enabled: true
  paths:
    - /var/log/performance/*.log
  fields:
    log_type: performance
    environment: production
  fields_under_root: true
  json.keys_under_root: true
  json.add_error_key: true

# 容器日志收集
- type: container
  enabled: true
  paths:
    - /var/lib/docker/containers/*/*.log
  processors:
  - add_kubernetes_metadata:
      host: ${NODE_NAME}
      matchers:
      - logs_path:
          logs_path: "/var/lib/docker/containers/"

# 系统日志收集
- type: syslog
  protocol.udp:
    host: "0.0.0.0:514"
  fields:
    log_type: system
    environment: production

# 输出配置
output.redis:
  hosts: ["redis-cluster:6379"]
  password: "${REDIS_PASSWORD}"
  key: "%{[log_type]}-logs"
  db: 0
  timeout: 5s
  max_retries: 3
  backoff.init: 1s
  backoff.max: 60s
  bulk_max_size: 2048
  worker: 2

# 处理器配置
processors:
- add_host_metadata:
    when.not.contains.tags: forwarded
- add_docker_metadata: ~
- add_kubernetes_metadata: ~

# 性能优化配置
queue.mem:
  events: 4096
  flush.min_events: 512
  flush.timeout: 1s

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

# 自定义处理器配置
processors:
# 字段重命名和清理
- rename:
    fields:
      - from: "log.file.path"
        to: "source_file"
      - from: "container.name"
        to: "container_name"
    ignore_missing: true

# 时间戳标准化
- timestamp:
    field: "@timestamp"
    layouts:
      - '2006-01-02T15:04:05.000Z'
      - '2006-01-02 15:04:05'
    test:
      - '2023-06-15T10:30:45.123Z'

# 数据脱敏
- script:
    lang: javascript
    id: data_masking
    source: >
      function process(event) {
        var message = event.Get("message");
        if (message) {
          // 手机号脱敏
          message = message.replace(/(\d{3})\d{4}(\d{4})/g, "$1****$2");
          // 身份证号脱敏
          message = message.replace(/(\d{6})\d{8}(\d{4})/g, "$1********$2");
          // 邮箱脱敏
          message = message.replace(/(\w{2})\w+(@\w+\.\w+)/g, "$1***$2");
          event.Put("message", message);
        }
      }

# 地理位置信息添加
- add_locale:
    format: offset

# 条件处理
- if:
    contains:
      log_type: "error"
  then:
    - add_tags:
        tags: [error, alert]
    - add_fields:
        fields:
          priority: high
          alert_channel: slack

# 丢弃无用字段
- drop_fields:
    fields: ["agent", "ecs", "host.architecture", "host.os.family"]

{
  "dashboard": {
    "title": "Application Performance Overview",
    "panels": [
      {
        "title": "Log Volume Trend",
        "type": "line",
        "query": {
          "bool": {
            "filter": [
              {"range": {"@timestamp": {"gte": "now-24h"}}},
              {"term": {"log_type": "application"}}
            ]
          }
        },
        "aggregations": {
          "logs_over_time": {
            "date_histogram": {
              "field": "@timestamp",
              "interval": "1h"
            }
          }
        }
      },
      {
        "title": "Error Rate by Service",
        "type": "pie",
        "query": {
          "bool": {
            "filter": [
              {"range": {"@timestamp": {"gte": "now-24h"}}},
              {"term": {"log_level": "ERROR"}}
            ]
          }
        },
        "aggregations": {
          "services": {
            "terms": {
              "field": "service_name",
              "size": 10
            }
          }
        }
      },
      {
        "title": "Response Time Distribution",
        "type": "histogram",
        "query": {
          "bool": {
            "filter": [
              {"range": {"@timestamp": {"gte": "now-1h"}}},
              {"exists": {"field": "response_time"}}
            ]
          }
        },
        "aggregations": {
          "response_time_histogram": {
            "histogram": {
              "field": "response_time",
              "interval": 100
            }
          }
        }
      }
    ]
  }
}

{
  "trigger": {
    "schedule": {
      "interval": "1m"
    }
  },
  "input": {
    "search": {
      "request": {
        "search_type": "query_then_fetch",
        "indices": ["logs-*"],
        "body": {
          "query": {
            "bool": {
              "filter": [
                {"range": {"@timestamp": {"gte": "now-5m"}}},
                {"term": {"log_level": "ERROR"}}
              ]
            }
          },
          "aggs": {
            "error_count": {
              "cardinality": {
                "field": "request_id"
              }
            },
            "services": {
              "terms": {
                "field": "service_name",
                "size": 5
              }
            }
          }
        }
      }
    }
  },
  "condition": {
    "compare": {
      "ctx.payload.aggregations.error_count.value": {
        "gt": 100
      }
    }
  },
  "actions": {
    "send_slack_alert": {
      "webhook": {
        "scheme": "https",
        "host": "hooks.slack.com",
        "port": 443,
        "method": "post",
        "path": "/services/YOUR/SLACK/WEBHOOK",
        "params": {},
        "headers": {
          "Content-Type": "application/json"
        },
        "body": """
        {
          "channel": "#alerts",
          "username": "ElasticSearch Watcher",
          "text": "High error rate detected: {{ctx.payload.aggregations.error_count.value}} errors in the last 5 minutes",
          "attachments": [
            {
              "color": "danger",
              "fields": [
                {
                  "title": "Time Range",
                  "value": "Last 5 minutes",
                  "short": true
                },
                {
                  "title": "Error Count",
                  "value": "{{ctx.payload.aggregations.error_count.value}}",
                  "short": true
                }
              ]
            }
          ]
        }
        """
      }
    },
    "send_email_alert": {
      "email": {
        "profile": "standard",
        "to": ["ops-team@company.com"],
        "subject": "Critical: High Error Rate Alert",
        "body": {
          "html": """
          <h2>High Error Rate Detected</h2>
          <p>Error count: <strong>{{ctx.payload.aggregations.error_count.value}}</strong></p>
          <p>Time range: Last 5 minutes</p>
          <p>Top affected services:</p>
          <ul>
          {{#ctx.payload.aggregations.services.buckets}}
            <li>{{key}}: {{doc_count}} errors</li>
          {{/ctx.payload.aggregations.services.buckets}}
          </ul>
          """
        }
      }
    }
  }
}

#!/bin/bash
# ELK Stack性能基准测试脚本

echo "=== ELK Stack性能基准测试 ==="

# 1. Elasticsearch写入性能测试
echo "1. Elasticsearch写入性能测试"
curl -X POST "elasticsearch:9200/_bulk" -H 'Content-Type: application/json' --data-binary @bulk_test_data.json

# 2. 查询性能测试
echo "2. 查询性能测试"
for i in {1..100}; do
  start_time=$(date +%s%N)
  curl -s -X GET "elasticsearch:9200/logs-*/_search" -H 'Content-Type: application/json' -d'
  {
    "query": {
      "bool": {
        "filter": [
          {"range": {"@timestamp": {"gte": "now-1h"}}},
          {"term": {"service_name": "user-service"}}
        ]
      }
    },
    "size": 100
  }' > /dev/null
  end_time=$(date +%s%N)
  duration=$((($end_time - $start_time) / 1000000))
  echo "Query $i: ${duration}ms"
done

# 3. 聚合查询性能测试
echo "3. 聚合查询性能测试"
curl -X GET "elasticsearch:9200/logs-*/_search" -H 'Content-Type: application/json' -d'
{
  "size": 0,
  "aggs": {
    "services": {
      "terms": {
        "field": "service_name",
        "size": 50
      },
      "aggs": {
        "error_rate": {
          "filter": {
            "term": {"log_level": "ERROR"}
          }
        },
        "avg_response_time": {
          "avg": {
            "field": "response_time"
          }
        }
      }
    }
  }
}'

# 4. 索引性能监控
echo "4. 索引性能监控"
curl -X GET "elasticsearch:9200/_cat/indices/logs-*?v&s=store.size:desc"

# 5. 集群健康状态检查
echo "5. 集群健康状态"
curl -X GET "elasticsearch:9200/_cluster/health?pretty"

# 成本优化配置
cost_optimization:
  # 1. 冷热数据分离
  hot_tier:
    node_count: 3
    storage_type: "SSD"
    retention: "7 days"
    
  warm_tier:
    node_count: 2
    storage_type: "SSD"
    retention: "23 days"
    
  cold_tier:
    node_count: 2
    storage_type: "HDD"
    retention: "90 days"
  
  # 2. 索引压缩策略
  compression:
    codec: "best_compression"
    force_merge_segments: 1
    
  # 3. 自动化运维
  automation:
    index_lifecycle_management: true
    snapshot_policy: "daily"
    monitoring_alerts: true

# Prometheus监控配置
global:
  scrape_interval: 15s
  evaluation_interval: 15s

rule_files:
  - "elk_alerts.yml"

scrape_configs:
# Elasticsearch监控
- job_name: 'elasticsearch'
  static_configs:
    - targets: ['elasticsearch:9200']
  metrics_path: /_prometheus/metrics
  scrape_interval: 30s

# Logstash监控
- job_name: 'logstash'
  static_configs:
    - targets: ['logstash:9600']
  metrics_path: /_node/stats
  scrape_interval: 30s

# Kibana监控
- job_name: 'kibana'
  static_configs:
    - targets: ['kibana:5601']
  metrics_path: /api/status
  scrape_interval: 60s

# Redis监控
- job_name: 'redis'
  static_configs:
    - targets: ['redis:9121']
  scrape_interval: 30s

# 系统监控
- job_name: 'node-exporter'
  static_configs:
    - targets: ['node-exporter:9100']
  scrape_interval: 15s

# elk_alerts.yml
groups:
- name: elasticsearch
  rules:
  - alert: ElasticsearchClusterRed
    expr: elasticsearch_cluster_health_status{color="red"} == 1
    for: 5m
    labels:
      severity: critical
    annotations:
      summary: "Elasticsearch cluster status is RED"
      description: "Elasticsearch cluster {{ $labels.cluster }} status is RED"

  - alert: ElasticsearchHighMemoryUsage
    expr: elasticsearch_jvm_memory_used_bytes / elasticsearch_jvm_memory_max_bytes > 0.9
    for: 10m
    labels:
      severity: warning
    annotations:
      summary: "Elasticsearch high memory usage"
      description: "Elasticsearch node {{ $labels.node }} memory usage is above 90%"

  - alert: ElasticsearchLowDiskSpace
    expr: elasticsearch_filesystem_data_available_bytes / elasticsearch_filesystem_data_size_bytes < 0.1
    for: 5m
    labels:
      severity: critical
    annotations:
      summary: "Elasticsearch low disk space"
      description: "Elasticsearch node {{ $labels.node }} has less than 10% disk space available"

- name: logstash
  rules:
  - alert: LogstashHighEventLatency
    expr: logstash_pipeline_events_duration_in_millis / logstash_pipeline_events_in > 1000
    for: 15m
    labels:
      severity: warning
    annotations:
      summary: "Logstash high event processing latency"
      description: "Logstash pipeline {{ $labels.pipeline }} has high processing latency"

  - alert: LogstashPipelineStalled
    expr: increase(logstash_pipeline_events_out[5m]) == 0
    for: 10m
    labels:
      severity: critical
    annotations:
      summary: "Logstash pipeline stalled"
      description: "Logstash pipeline {{ $labels.pipeline }} has not processed any events in 10 minutes"

#!/bin/bash
# ELK Stack故障诊断和处理脚本

echo "=== ELK Stack故障诊断工具 ==="

# 1. 集群健康检查
check_cluster_health() {
    echo "1. 检查Elasticsearch集群健康状态"
    health=$(curl -s "http://elasticsearch:9200/_cluster/health" | jq -r '.status')
    
    case $health in
        "green")
            echo "✅ 集群状态正常"
            ;;
        "yellow")
            echo "⚠️  集群状态警告 - 检查副本分片"
            curl -s "http://elasticsearch:9200/_cat/shards?v" | grep UNASSIGNED
            ;;
        "red")
            echo "❌ 集群状态严重 - 立即处理"
            curl -s "http://elasticsearch:9200/_cluster/allocation/explain?pretty"
            ;;
    esac
}

# 2. 索引状态检查
check_index_status() {
    echo "2. 检查索引状态"
    curl -s "http://elasticsearch:9200/_cat/indices?v&health=red"
    
    # 检查未分配的分片
    unassigned=$(curl -s "http://elasticsearch:9200/_cat/shards" | grep UNASSIGNED | wc -l)
    if [ $unassigned -gt 0 ]; then
        echo "发现 $unassigned 个未分配分片"
        # 尝试重新分配
        curl -X POST "http://elasticsearch:9200/_cluster/reroute?retry_failed=true"
    fi
}

# 3. 性能问题诊断
diagnose_performance() {
    echo "3. 性能问题诊断"
    
    # 检查慢查询
    curl -s "http://elasticsearch:9200/_nodes/stats/indices/search" | jq '.nodes[].indices.search'
    
    # 检查JVM堆内存使用
    curl -s "http://elasticsearch:9200/_nodes/stats/jvm" | jq '.nodes[].jvm.mem.heap_used_percent'
    
    # 检查磁盘使用情况
    curl -s "http://elasticsearch:9200/_nodes/stats/fs" | jq '.nodes[].fs.total'
}

# 4. Logstash问题诊断
diagnose_logstash() {
    echo "4. Logstash问题诊断"
    
    # 检查管道状态
    curl -s "http://logstash:9600/_node/stats/pipelines" | jq '.pipelines'
    
    # 检查死信队列
    if [ -d "/var/lib/logstash/dead_letter_queue" ]; then
        dlq_size=$(du -sh /var/lib/logstash/dead_letter_queue | cut -f1)
        echo "死信队列大小: $dlq_size"
    fi
}

# 5. 自动修复尝试
auto_fix() {
    echo "5. 尝试自动修复"
    
    # 清理过期索引
    curl -X DELETE "http://elasticsearch:9200/logs-*" -H 'Content-Type: application/json' -d'
    {
      "query": {
        "range": {
          "@timestamp": {
            "lt": "now-30d"
          }
        }
      }
    }'
    
    # 强制合并小分片
    curl -X POST "http://elasticsearch:9200/logs-*/_forcemerge?max_num_segments=1"
    
    # 重启有问题的Logstash管道
    curl -X PUT "http://logstash:9600/_node/pipelines/application-logs/reload"
}

# 执行诊断流程
check_cluster_health
check_index_status
diagnose_performance
diagnose_logstash

# 如果发现问题，询问是否执行自动修复
read -p "是否执行自动修复? (y/n): " -n 1 -r
echo
if [[ $REPLY =~ ^[Yy]$ ]]; then
    auto_fix
fi