┌─────────────────────────────────────────────────────────────────┐
│                      系統的兩套真實                              │
├─────────────────────────────────────────────────────────────────┤
│  第一套真實：程式碼（Code）                                      │
│  ├── 定義系統「應該」如何運作                                    │
│  └── 靜態、可控、可版本管理                                      │
│                                                                 │
│  第二套真實：日誌（Logs）                                        │
│  ├── 記錄系統「實際」如何運作                                    │
│  └── 動態、即時、反映真實行為                                    │
└─────────────────────────────────────────────────────────────────┘

問題：「為什麼今天下午 3 點交易失敗率突然上升？」

Step 1: Metrics（發現問題）
         └── Grafana 看到 error_rate 從 0.1% 跳到 5%

Step 2: Logs（定位原因）
         └── Kibana 搜尋該時段 ERROR logs
         └── 發現大量 "Connection timeout to payment-gateway"

Step 3: Tracing（追蹤路徑）
         └── Jaeger 查看失敗請求的完整呼叫鏈
         └── 發現 payment-gateway → bank-api 這段耗時異常

# log4j2.xml 範例
<Root level="DEBUG">
    <AppenderRef ref="Console"/>
    <AppenderRef ref="File"/>
</Root>

# 特點：
# - 輸出完整堆疊追蹤
# - 包含 SQL 查詢、HTTP 請求/回應 body
# - 可輸出到 Console 方便即時查看

# log4j2.xml 範例
<Root level="INFO">
    <AppenderRef ref="AsyncFile"/>
    <AppenderRef ref="Logstash"/>
</Root>

# 特點：
# - 使用 Async Appender 避免阻塞
# - 敏感資料必須遮蔽
# - 結構化 JSON 格式方便解析
# - 直接送到 Logstash/Kafka

2026-01-26 10:30:45 ERROR PaymentService - Payment failed for user john@example.com, amount 50000, card ending 1234

{
  "timestamp": "2026-01-26T10:30:45.123Z",
  "level": "ERROR",
  "logger": "com.bank.payment.PaymentService",
  "thread": "http-nio-8080-exec-15",
  "traceId": "abc123def456",
  "spanId": "789xyz",
  "service": "payment-service",
  "environment": "prod",
  "message": "Payment processing failed",
  "context": {
    "userId": "USR-78901",
    "transactionId": "TXN-20260126-001234",
    "amount": 50000,
    "currency": "TWD",
    "cardLast4": "****",
    "errorCode": "GATEWAY_TIMEOUT",
    "errorMessage": "Connection timeout after 30s"
  },
  "exception": {
    "class": "java.net.SocketTimeoutException",
    "message": "Read timed out",
    "stackTrace": "..."
  }
}

# /etc/logstash/conf.d/application-logs.conf

input {
  kafka {
    bootstrap_servers => "kafka-1:9092,kafka-2:9092,kafka-3:9092"
    topics => ["app-logs"]
    group_id => "logstash-app-logs"
    consumer_threads => 3
    codec => json
    decorate_events => true
  }
}

filter {
  # 解析 JSON（如果還不是）
  if [message] =~ /^\{/ {
    json {
      source => "message"
      target => "parsed"
    }
    mutate {
      rename => { "[parsed][timestamp]" => "@timestamp" }
      rename => { "[parsed][level]" => "level" }
      rename => { "[parsed][service]" => "service" }
      rename => { "[parsed][traceId]" => "traceId" }
    }
  }
  
  # 敏感資料遮蔽
  mutate {
    gsub => [
      "[parsed][context][email]", "(?<=.{3}).(?=.*@)", "*",
      "[parsed][context][phone]", "(?<=.{4}).(?=.{4})", "*"
    ]
  }
  
  # 添加處理時間戳
  ruby {
    code => "event.set('processed_at', Time.now.utc.iso8601(3))"
  }
  
  # 移除不需要的欄位
  mutate {
    remove_field => ["message", "@version", "host"]
  }
}

output {
  elasticsearch {
    hosts => ["es-1:9200", "es-2:9200", "es-3:9200"]
    index => "app-logs-%{[service]}-%{+YYYY.MM.dd}"
    user => "${ES_USER}"
    password => "${ES_PASSWORD}"
    ssl => true
    cacert => "/etc/logstash/certs/ca.crt"
  }
  
  # Dead Letter Queue for failed events
  if "_jsonparsefailure" in [tags] or "_grokparsefailure" in [tags] {
    elasticsearch {
      hosts => ["es-1:9200"]
      index => "dlq-app-logs-%{+YYYY.MM.dd}"
    }
  }
}

索引命名規則：{類型}-{服務}-{日期}

範例：
├── app-logs-payment-service-2026.01.26
├── app-logs-user-service-2026.01.26
├── access-logs-nginx-2026.01.26
├── audit-logs-all-2026.01
└── security-logs-waf-2026.01.26

每日 Log 量：100 GB
建議 Shard 數：100 GB ÷ 25 GB = 4 Primary Shards
Replica：1（生產環境至少 1）
總 Shard 數：4 × 2 = 8 Shards

PUT _index_template/app-logs-template
{
  "index_patterns": ["app-logs-*"],
  "priority": 100,
  "template": {
    "settings": {
      "number_of_shards": 3,
      "number_of_replicas": 1,
      "refresh_interval": "5s",
      "index.lifecycle.name": "logs-policy",
      "index.lifecycle.rollover_alias": "app-logs"
    },
    "mappings": {
      "dynamic": "strict",
      "properties": {
        "@timestamp": { "type": "date" },
        "level": { "type": "keyword" },
        "service": { "type": "keyword" },
        "traceId": { "type": "keyword" },
        "message": { "type": "text" },
        "context": {
          "type": "object",
          "dynamic": true
        },
        "exception": {
          "properties": {
            "class": { "type": "keyword" },
            "message": { "type": "text" },
            "stackTrace": { "type": "text", "index": false }
          }
        }
      }
    }
  }
}

# /etc/logstash/pipelines.yml

# Application Logs Pipeline
- pipeline.id: app-logs
  path.config: "/etc/logstash/conf.d/app-logs.conf"
  pipeline.workers: 4
  pipeline.batch.size: 1000
  pipeline.batch.delay: 50
  queue.type: persisted
  queue.max_bytes: 4gb

# Access Logs Pipeline
- pipeline.id: access-logs
  path.config: "/etc/logstash/conf.d/access-logs.conf"
  pipeline.workers: 2
  pipeline.batch.size: 2000

# Security Logs Pipeline（高優先級）
- pipeline.id: security-logs
  path.config: "/etc/logstash/conf.d/security-logs.conf"
  pipeline.workers: 2
  pipeline.batch.size: 500
  queue.type: persisted

filter {
  grok {
    match => {
      "message" => '%{IPORHOST:client_ip} - %{DATA:user} \[%{HTTPDATE:timestamp}\] "%{WORD:method} %{URIPATHPARAM:request} HTTP/%{NUMBER:http_version}" %{NUMBER:status:int} %{NUMBER:bytes:int} "%{DATA:referrer}" "%{DATA:user_agent}" %{NUMBER:request_time:float}'
    }
    tag_on_failure => ["_grokparsefailure_nginx"]
  }
  
  # 解析時間戳
  date {
    match => ["timestamp", "dd/MMM/yyyy:HH:mm:ss Z"]
    target => "@timestamp"
    remove_field => ["timestamp"]
  }
  
  # GeoIP 解析
  geoip {
    source => "client_ip"
    target => "geo"
    fields => ["country_name", "city_name", "location"]
  }
  
  # User-Agent 解析
  useragent {
    source => "user_agent"
    target => "ua"
  }
}

# /etc/logstash/patterns/custom

# 銀行交易格式
BANK_TXN_ID TXN-%{YEAR}%{MONTHNUM}%{MONTHDAY}-%{NUMBER:6}
BANK_ACCOUNT [0-9]{3}-[0-9]{2}-[0-9]{7}-[0-9]
MASKED_CARD \*{12}[0-9]{4}

# 使用範例
filter {
  grok {
    patterns_dir => ["/etc/logstash/patterns"]
    match => {
      "message" => "%{BANK_TXN_ID:txn_id} %{BANK_ACCOUNT:account} %{MASKED_CARD:card}"
    }
  }
}

filter {
  # 解析 JSON
  json {
    source => "message"
    target => "data"
    skip_on_invalid_json => true
    tag_on_failure => ["_jsonparsefailure"]
  }
  
  # 處理巢狀 JSON
  if [data][nested_json] {
    json {
      source => "[data][nested_json]"
      target => "[data][parsed_nested]"
    }
  }
  
  # 扁平化（flatten）
  ruby {
    code => '
      def flatten_hash(hash, parent_key = "", sep = "_")
        hash.each_with_object({}) do |(k, v), h|
          new_key = parent_key.empty? ? k : "#{parent_key}#{sep}#{k}"
          if v.is_a?(Hash)
            h.merge!(flatten_hash(v, new_key, sep))
          else
            h[new_key] = v
          end
        end
      end
      
      if event.get("[data]").is_a?(Hash)
        flattened = flatten_hash(event.get("[data]"))
        event.set("[flat]", flattened)
      end
    '
  }
}

filter {
  mutate {
    # 重新命名欄位
    rename => {
      "[data][userName]" => "[user][name]"
      "[data][userEmail]" => "[user][email]"
    }
    
    # 型別轉換
    convert => {
      "[data][amount]" => "float"
      "[data][count]" => "integer"
      "[data][success]" => "boolean"
    }
    
    # 字串處理
    lowercase => ["[user][email]"]
    strip => ["[data][message]"]
    gsub => [
      # 遮蔽信用卡號
      "[data][card_number]", "^(\d{4})\d{8}(\d{4})$", "\1********\2",
      # 遮蔽 Email
      "[user][email]", "(?<=.{3}).(?=.*@)", "*"
    ]
    
    # 移除欄位
    remove_field => ["message", "host", "@version"]
    
    # 添加欄位
    add_field => {
      "[meta][processed_by]" => "logstash-prod-01"
      "[meta][pipeline_version]" => "2.1.0"
    }
  }
}

# /etc/logstash/logstash.yml

# Worker 數量（建議 = CPU 核心數）
pipeline.workers: 8

# Batch 設定
pipeline.batch.size: 1000
pipeline.batch.delay: 50

# 記憶體設定
# /etc/logstash/jvm.options
-Xms4g
-Xmx4g

# 持久化 Queue（防止資料丟失）
queue.type: persisted
queue.max_bytes: 8gb
queue.checkpoint.writes: 1024

# 查看 Pipeline 統計
curl -s localhost:9600/_node/stats/pipelines?pretty

# 關鍵指標
{
  "pipelines": {
    "app-logs": {
      "events": {
        "in": 1000000,           # 輸入事件數
        "filtered": 1000000,     # 過濾後事件數
        "out": 999500,           # 輸出事件數
        "duration_in_millis": 60000,
        "queue_push_duration_in_millis": 5000
      },
      "queue": {
        "events_count": 500,     # Queue 中等待的事件
        "max_queue_size_in_bytes": 8589934592
      }
    }
  }
}

# oracle-slow-query.conf
input {
  jdbc {
    jdbc_driver_library => "/opt/oracle/ojdbc8.jar"
    jdbc_driver_class => "Java::oracle.jdbc.OracleDriver"
    jdbc_connection_string => "jdbc:oracle:thin:@//db-host:1521/ORCL"
    jdbc_user => "${ORACLE_USER}"
    jdbc_password => "${ORACLE_PASSWORD}"
    schedule => "*/5 * * * *"
    statement => "
      SELECT 
        sql_id,
        sql_text,
        elapsed_time/1000000 as elapsed_seconds,
        executions,
        buffer_gets,
        disk_reads,
        TO_CHAR(last_active_time, 'YYYY-MM-DD HH24:MI:SS') as last_active_time
      FROM v$sql 
      WHERE elapsed_time/1000000 > 5
      AND last_active_time > SYSDATE - 1/24
    "
    tracking_column => "last_active_time"
    use_column_value => true
  }
}

filter {
  mutate {
    add_field => { "log_type" => "oracle_slow_query" }
    add_field => { "database" => "ORCL" }
  }
}

# kafka-consumer-lag.conf
input {
  exec {
    command => "/opt/kafka/bin/kafka-consumer-groups.sh --bootstrap-server kafka:9092 --describe --all-groups 2>/dev/null | grep -v CONSUMER-ID"
    interval => 60
    codec => plain
  }
}

filter {
  split { field => "message" }
  
  grok {
    match => {
      "message" => "%{NOTSPACE:consumer_group}\s+%{NOTSPACE:topic}\s+%{NUMBER:partition:int}\s+%{NUMBER:current_offset:int}\s+%{NUMBER:log_end_offset:int}\s+%{NUMBER:lag:int}\s+%{NOTSPACE:consumer_id}\s+%{NOTSPACE:host}\s+%{NOTSPACE:client_id}"
    }
  }
  
  mutate {
    add_field => { "log_type" => "kafka_consumer_lag" }
  }
  
  # Lag 超過閾值標記
  if [lag] > 10000 {
    mutate {
      add_tag => ["high_lag"]
      add_field => { "alert_level" => "warning" }
    }
  }
}

格式：{環境}-{類型}-{服務}-{時間粒度}

範例：
├── prod-app-logs-payment-service-2026.01.26    # 每日
├── prod-app-logs-user-service-2026.01.26       # 每日
├── prod-audit-logs-all-2026.01                 # 每月（稽核需長期保留）
├── prod-metrics-system-2026.01.26              # 每日
└── prod-security-logs-waf-2026.01.26           # 每日

PUT _ilm/policy/logs-lifecycle-policy
{
  "policy": {
    "phases": {
      "hot": {
        "min_age": "0ms",
        "actions": {
          "rollover": {
            "max_age": "1d",
            "max_size": "30gb",
            "max_docs": 10000000
          },
          "set_priority": {
            "priority": 100
          }
        }
      },
      "warm": {
        "min_age": "1d",
        "actions": {
          "shrink": {
            "number_of_shards": 1
          },
          "forcemerge": {
            "max_num_segments": 1
          },
          "set_priority": {
            "priority": 50
          },
          "allocate": {
            "require": {
              "data": "warm"
            }
          }
        }
      },
      "cold": {
        "min_age": "7d",
        "actions": {
          "set_priority": {
            "priority": 0
          },
          "allocate": {
            "require": {
              "data": "cold"
            }
          },
          "freeze": {}
        }
      },
      "delete": {
        "min_age": "90d",
        "actions": {
          "delete": {}
        }
      }
    }
  }
}

PUT _index_template/optimized-logs-template
{
  "index_patterns": ["prod-app-logs-*"],
  "template": {
    "settings": {
      "number_of_shards": 3,
      "number_of_replicas": 1,
      "index.mapping.total_fields.limit": 500,
      "index.mapping.depth.limit": 5,
      "index.mapping.nested_fields.limit": 20
    },
    "mappings": {
      "dynamic": "strict",
      "_source": {
        "enabled": true,
        "excludes": ["stackTrace"]
      },
      "properties": {
        "@timestamp": {
          "type": "date",
          "format": "strict_date_optional_time||epoch_millis"
        },
        "level": {
          "type": "keyword",
          "ignore_above": 16
        },
        "service": {
          "type": "keyword",
          "ignore_above": 64
        },
        "traceId": {
          "type": "keyword",
          "ignore_above": 64
        },
        "message": {
          "type": "text",
          "analyzer": "standard",
          "fields": {
            "keyword": {
              "type": "keyword",
              "ignore_above": 256
            }
          }
        },
        "context": {
          "type": "object",
          "dynamic": true
        },
        "exception": {
          "properties": {
            "class": { "type": "keyword" },
            "message": { "type": "text" },
            "stackTrace": {
              "type": "text",
              "index": false,
              "doc_values": false
            }
          }
        }
      }
    }
  }
}

# Hot Node 配置
# elasticsearch.yml
node.name: hot-node-1
node.roles: [data_hot, data_content]
node.attr.data: hot
path.data: /ssd/elasticsearch/data

# Warm Node 配置
node.name: warm-node-1
node.roles: [data_warm]
node.attr.data: warm
path.data: /ssd/elasticsearch/data

# Cold Node 配置
node.name: cold-node-1
node.roles: [data_cold, data_frozen]
node.attr.data: cold
path.data: /hdd/elasticsearch/data

# Master Node 配置
node.name: master-1
node.roles: [master]

// ✅ 好的查詢：使用 filter context（可快取）
GET app-logs-*/_search
{
  "query": {
    "bool": {
      "filter": [
        { "term": { "level": "ERROR" } },
        { "term": { "service": "payment-service" } },
        { "range": { "@timestamp": { "gte": "now-1h" } } }
      ],
      "must": [
        { "match": { "message": "timeout" } }
      ]
    }
  },
  "size": 100,
  "_source": ["@timestamp", "level", "service", "message", "traceId"],
  "sort": [{ "@timestamp": "desc" }]
}

// ❌ 不好的查詢：wildcard 開頭、未限制時間範圍
GET app-logs-*/_search
{
  "query": {
    "wildcard": { "message": "*timeout*" }
  }
}

// 錯誤率統計（按服務、每小時）
GET app-logs-*/_search
{
  "size": 0,
  "query": {
    "bool": {
      "filter": [
        { "range": { "@timestamp": { "gte": "now-24h" } } }
      ]
    }
  },
  "aggs": {
    "by_service": {
      "terms": {
        "field": "service",
        "size": 20
      },
      "aggs": {
        "by_hour": {
          "date_histogram": {
            "field": "@timestamp",
            "calendar_interval": "hour"
          },
          "aggs": {
            "error_count": {
              "filter": { "term": { "level": "ERROR" } }
            },
            "total_count": {
              "value_count": { "field": "_id" }
            },
            "error_rate": {
              "bucket_script": {
                "buckets_path": {
                  "errors": "error_count._count",
                  "total": "total_count"
                },
                "script": "params.total > 0 ? params.errors / params.total * 100 : 0"
              }
            }
          }
        }
      }
    }
  }
}

日寫入量（原始）：R GB/day
壓縮比：約 10:1（JSON Log）
索引 Overhead：約 1.5x

每日儲存需求 = R × 0.1 × 1.5 = R × 0.15 GB

範例：
- 每日原始 Log：1 TB
- 壓縮後：100 GB
- 含索引：150 GB
- 保留 90 天：150 × 90 = 13.5 TB
- 含 Replica：13.5 × 2 = 27 TB

✅ 好的 Dashboard：
├── 清楚的標題與說明
├── 從高階到細節的資訊流
├── 一致的顏色編碼（紅=錯誤、黃=警告、綠=正常）
├── 合理的時間範圍預設
├── 可點擊深入（Drill-down）
└── 不超過 15 個視覺化元件

❌ 不好的 Dashboard：
├── 塞滿整個螢幕的圖表
├── 沒有上下文的數字
├── 不一致的時間範圍
├── 過度複雜的聚合
└── 沒有標示單位

# 基本搜尋
level: ERROR and service: payment-service

# 時間範圍
@timestamp >= "2026-01-26T00:00:00" and @timestamp < "2026-01-27T00:00:00"

# 萬用字元
message: *timeout* and service: payment*

# 排除
level: ERROR and not service: test-service

# 組合條件
(level: ERROR or level: WARN) and service: (payment-service or user-service)

# 存在性檢查
exception.class: * and level: ERROR

# 數值範圍
context.response_time > 1000 and context.response_time <= 5000

# 正規表達式
message: /.*Connection refused.*/

# 模糊搜尋（容錯 1 個字元）
message: timout~1

# 鄰近搜尋（timeout 和 connection 相距 5 個字以內）
message: "timeout connection"~5

# 欄位存在
_exists_: exception.stackTrace

# 範圍查詢
context.amount: [1000 TO 50000]

// Kibana Alerting Rule 範例：錯誤率超標告警
{
  "name": "Payment Service High Error Rate",
  "consumer": "alerts",
  "rule_type_id": ".es-query",
  "schedule": { "interval": "1m" },
  "params": {
    "index": ["app-logs-payment-service-*"],
    "timeField": "@timestamp",
    "searchType": "esQuery",
    "esQuery": {
      "query": {
        "bool": {
          "filter": [
            { "term": { "service": "payment-service" } },
            { "term": { "level": "ERROR" } },
            { "range": { "@timestamp": { "gte": "now-5m" } } }
          ]
        }
      }
    },
    "threshold": [50],
    "thresholdComparator": ">"
  },
  "actions": [
    {
      "group": "threshold met",
      "id": "slack-connector-id",
      "params": {
        "message": "🚨 Payment Service 錯誤率過高！\n過去 5 分鐘錯誤數：{{context.value}}\n請立即檢查：{{context.link}}"
      }
    },
    {
      "group": "threshold met",
      "id": "pagerduty-connector-id",
      "params": {
        "severity": "critical",
        "summary": "Payment Service High Error Rate Alert"
      }
    }
  ]
}

┌────────────────────────────────────────────────────────────────┐
│  Service Health Overview                    [Last 24 hours ▼] │
├────────────────────────────────────────────────────────────────┤
│  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐       │
│  │ 99.95%   │  │  1.2M    │  │  0.05%   │  │  125ms   │       │
│  │ Uptime   │  │ Requests │  │ Error %  │  │ P95 Lat  │       │
│  └──────────┘  └──────────┘  └──────────┘  └──────────┘       │
├────────────────────────────────────────────────────────────────┤
│  [Request Volume & Error Rate - Line Chart]                    │
│  ▁▂▃▄▅▆▇█▇▆▅▄▃▂▁▂▃▄▅▆▇█▇▆▅▄▃▂▁                               │
├────────────────────────────────────────────────────────────────┤
│  ┌─────────────────────────┐  ┌─────────────────────────────┐ │
│  │ Error by Service        │  │ Error by Type               │ │
│  │ [Bar Chart]             │  │ [Donut Chart]               │ │
│  │ payment: ████████ 450   │  │  Timeout: 45%               │ │
│  │ user:    ████ 200       │  │  DB Error: 30%              │ │
│  │ order:   ██ 100         │  │  Validation: 25%            │ │
│  └─────────────────────────┘  └─────────────────────────────┘ │
├────────────────────────────────────────────────────────────────┤
│  [Top 10 Errors - Data Table]                                  │
│  Exception                  | Service  | Count | Last Seen    │
│  ───────────────────────────┼──────────┼───────┼─────────────  │
│  SocketTimeoutException     │ payment  │  230  │ 2 min ago    │
│  DataIntegrityViolation     │ order    │   85  │ 5 min ago    │
└────────────────────────────────────────────────────────────────┘

┌────────────────────────────────────────────────────────────────┐
│  Transaction Monitoring                     [Last 1 hour ▼]   │
├────────────────────────────────────────────────────────────────┤
│  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐       │
│  │  15,234  │  │ NT$ 2.3B │  │  99.8%   │  │    12    │       │
│  │ Total Tx │  │  Amount  │  │ Success  │  │ Failures │       │
│  └──────────┘  └──────────┘  └──────────┘  └──────────┘       │
├────────────────────────────────────────────────────────────────┤
│  [Transaction Volume by Channel - Stacked Area]                │
│  █ Mobile  █ Web  █ ATM  █ Branch                             │
│  ▁▂▃▄▅▆▇█▇▆▅▄▃▂▁▂▃▄▅▆▇█▇▆▅▄▃▂▁                               │
├────────────────────────────────────────────────────────────────┤
│  ┌─────────────────────────┐  ┌─────────────────────────────┐ │
│  │ Tx by Type              │  │ Response Time Distribution  │ │
│  │ Transfer:  ████████ 45% │  │ [Histogram]                 │ │
│  │ Payment:   ██████ 30%   │  │ < 100ms:  ████████████ 80%  │ │
│  │ Inquiry:   ████ 20%     │  │ 100-500ms: ████ 15%         │ │
│  │ Other:     █ 5%         │  │ > 500ms:   █ 5%             │ │
│  └─────────────────────────┘  └─────────────────────────────┘ │
├────────────────────────────────────────────────────────────────┤
│  [Failed Transactions - Table with drill-down]                 │
│  Time        | TxnId          | Type     | Error      | Amount │
│  ────────────┼────────────────┼──────────┼────────────┼─────── │
│  10:35:22    │ TXN-2026012... │ Transfer │ TIMEOUT    │ 50,000 │
│  10:33:15    │ TXN-2026012... │ Payment  │ DECLINED   │ 12,000 │
└────────────────────────────────────────────────────────────────┘

我需要一個 Elasticsearch 查詢，條件如下：
- Index: app-logs-*
- 時間範圍: 過去 24 小時
- 條件: level 是 ERROR，service 是 payment-service
- 排序: 按時間降序
- 只回傳: @timestamp, level, message, traceId
- 最多 100 筆

請給我完整的 ES Query JSON。

GET app-logs-*/_search
{
  "query": {
    "bool": {
      "filter": [
        { "term": { "level": "ERROR" } },
        { "term": { "service": "payment-service" } },
        { "range": { "@timestamp": { "gte": "now-24h" } } }
      ]
    }
  },
  "size": 100,
  "_source": ["@timestamp", "level", "message", "traceId"],
  "sort": [{ "@timestamp": "desc" }]
}

我需要分析過去 7 天的錯誤趨勢：
1. 按服務分組
2. 每個服務按小時統計錯誤數
3. 計算每小時的錯誤率（錯誤數/總數）
4. 只看錯誤率超過 1% 的時段

請給我 ES Query。

以下是過去 1 小時 payment-service 的 ERROR logs 摘要：

1. SocketTimeoutException - 230 次
   - 主要發生在 10:30-10:45
   - 都是連接 payment-gateway.bank.com:443

2. DataIntegrityViolationException - 85 次
   - 分散在整個小時
   - duplicate key value violates unique constraint "pk_transaction"

3. IllegalArgumentException - 45 次
   - Invalid currency code: null

請分析：
1. 這些錯誤的可能根本原因
2. 優先處理順序
3. 建議的排查步驟

## 錯誤分析報告

### 1. SocketTimeoutException（最高優先）
**可能原因**：
- payment-gateway 在 10:30-10:45 期間發生問題
- 可能是網路問題、下游服務過載或部署

**排查步驟**：
1. 檢查該時段 payment-gateway 的 metrics/logs
2. 確認網路連線狀態（ping, traceroute）
3. 檢查是否有該時段的變更部署

### 2. DataIntegrityViolationException（中優先）
**可能原因**：
- 重複交易 ID 產生邏輯有問題
- 可能有重試機制未正確處理

**排查步驟**：
1. 檢查交易 ID 產生邏輯（是否有 race condition）
2. 確認重試機制是否有 idempotency key

### 3. IllegalArgumentException（低優先）
**可能原因**：
- 上游傳入的資料缺少 currency code
- 輸入驗證不完整

**排查步驟**：
1. 追蹤這些請求的來源（traceId）
2. 確認 API contract 是否清楚定義 currency 必填

{"@timestamp":"2026-01-26T10:30:45.123Z","level":"ERROR","logger":"com.bank.payment.service.PaymentGatewayClient","thread":"http-nio-8080-exec-15","traceId":"abc123","spanId":"def456","service":"payment-service","environment":"prod","message":"Failed to process payment","context":{"userId":"USR-78901","transactionId":"TXN-20260126-001234","amount":50000,"currency":"TWD","merchantId":"MER-001","channel":"MOBILE","retryCount":3},"exception":{"class":"java.net.SocketTimeoutException","message":"Read timed out","stackTrace":"java.net.SocketTimeoutException: Read timed out\n\tat java.base/sun.nio.ch.NioSocketImpl.timedRead...(省略 50 行)"}}

## 錯誤摘要
- 時間: 2026-01-26 10:30:45
- 服務: payment-service
- 錯誤類型: SocketTimeoutException
- TraceId: abc123

## 關鍵上下文
- 交易ID: TXN-20260126-001234
- 金額: 50,000 TWD
- 通路: MOBILE
- 已重試: 3 次

## 錯誤訊息
Read timed out（連接 payment gateway 逾時）

## 我需要知道
1. 這個錯誤的可能原因
2. 如何進一步診斷
3. 暫時緩解措施

請分析以下 Log 模式：

## 時間範圍
2026-01-26 10:00 - 11:00

## Log 統計
| 錯誤類型 | 數量 | 首次出現 | 最後出現 |
|---------|------|---------|---------|
| SocketTimeoutException | 230 | 10:30:12 | 10:45:33 |
| NullPointerException | 45 | 10:15:00 | 10:55:00 |
| SQLException | 12 | 10:20:00 | 10:22:00 |

## 關聯事件
- 10:28: payment-gateway 開始回應變慢（從 P95 100ms → 2000ms）
- 10:30: 部署 payment-service v2.3.1
- 10:45: 手動重啟 payment-gateway

## 問題
1. 這些錯誤是否相關？
2. 根本原因可能是什麼？
3. 如何避免再次發生？

告警內容：
- 時間: 2026-01-26 03:00
- 告警: Payment Success Rate < 90%
- 當前值: 85%
- 影響服務: payment-service, order-service

請整理成 Incident Summary，包含：
1. 影響範圍
2. 時間線
3. 關鍵指標變化

以下是 payment-service 在 02:55-03:10 的 ERROR logs 統計：

| 時間 | SocketTimeout | DBConnection | Validation |
|------|---------------|--------------|------------|
| 02:55 | 5 | 0 | 2 |
| 03:00 | 150 | 0 | 3 |
| 03:05 | 200 | 45 | 2 |
| 03:10 | 180 | 80 | 1 |

同時段的基礎設施事件：
- 02:58: DB failover 開始
- 03:02: DB failover 完成
- 03:08: DB connection pool 重建完成

請分析錯誤模式與事件的關聯。

請根據以下資訊撰寫 Post-mortem 報告：

## 事件摘要
- 時間: 2026-01-26 03:00-03:15
- 影響: 支付成功率下降到 85%
- 根本原因: 資料庫 failover 導致連線中斷

## 時間線
（詳細時間線...）

## 需要涵蓋的章節
1. Executive Summary
2. Impact Analysis
3. Root Cause
4. Resolution
5. Action Items
6. Lessons Learned

@Component
public class RateLimitedLogger {
    
    private final Logger log = LoggerFactory.getLogger(this.getClass());
    private final LoadingCache<String, RateLimiter> limiters = CacheBuilder.newBuilder()
        .expireAfterAccess(10, TimeUnit.MINUTES)
        .build(new CacheLoader<>() {
            @Override
            public RateLimiter load(String key) {
                return RateLimiter.create(1.0); // 每秒最多 1 條
            }
        });
    
    public void error(String key, String message, Object... args) {
        try {
            RateLimiter limiter = limiters.get(key);
            if (limiter.tryAcquire()) {
                log.error(message, args);
            } else {
                // 可選：記錄被限流的次數
                log.debug("Log rate limited for key: {}", key);
            }
        } catch (ExecutionException e) {
            log.error(message, args);
        }
    }
}

// 使用方式
@Service
public class PaymentService {
    
    @Autowired
    private RateLimitedLogger rateLimitedLogger;
    
    public void processPayment() {
        try {
            // 業務邏輯
        } catch (SocketTimeoutException e) {
            // 同類錯誤每秒最多記錄 1 次
            rateLimitedLogger.error(
                "payment-gateway-timeout",
                "Payment gateway timeout: {}",
                e.getMessage()
            );
        }
    }
}

1. 立即確認影響範圍
   └── ES Cluster 健康狀態、磁碟空間、Index 狀態

2. 暫時緩解
   ├── 動態調高 Log Level（INFO → WARN）
   ├── Logstash 啟動 Sampling（每 10 筆取 1 筆）
   └── 關閉非必要的 Log 來源

3. 根本解決
   ├── 找出爆量的 Log 來源（哪個服務、哪種類型）
   ├── 修正程式（關閉 Debug、修復迴圈）
   └── 部署修正版本

4. 事後處理
   ├── 清理多餘的 Index
   ├── 檢討 Log 規範
   └── 建立監控告警（Log Volume Alert）

# 查看各 Index 大小
GET _cat/indices?v&s=store.size:desc&h=index,docs.count,store.size

# 查看 Shard 分布
GET _cat/shards?v&s=store:desc

# 查看 Mapping 欄位數
GET app-logs-*/_mapping?pretty | jq 'to_entries | .[0].value.mappings.properties | keys | length'

# 查看 Index 設定
GET app-logs-*/_settings

# 1. 刪除過舊的 Index（謹慎操作！）
DELETE app-logs-*-2025.10.*

# 2. 強制合併 Segment（Warm/Cold Index）
POST app-logs-2025.11.*/_forcemerge?max_num_segments=1

# 3. 縮減 Shard 數（需要先設定 read-only）
PUT app-logs-2025.11.01/_settings
{
  "index.blocks.write": true
}

POST app-logs-2025.11.01/_shrink/app-logs-2025.11.01-shrunk
{
  "settings": {
    "index.number_of_shards": 1,
    "index.number_of_replicas": 1,
    "index.codec": "best_compression"
  }
}

# 4. 套用 ILM Policy 到現有 Index
PUT app-logs-*/_settings
{
  "index.lifecycle.name": "logs-policy"
}

filter {
  # 信用卡號遮蔽
  mutate {
    gsub => [
      # 保留前 4 後 4
      "[context][card_number]", "^(\d{4})\d{8}(\d{4})$", "\1********\2",
      # 完整遮蔽
      "[context][cvv]", ".*", "***"
    ]
  }
  
  # Email 遮蔽
  mutate {
    gsub => [
      "[user][email]", "(?<=.{3}).(?=.*@)", "*"
    ]
  }
  
  # 身分證字號遮蔽（台灣）
  mutate {
    gsub => [
      "[user][id_number]", "^([A-Z])(\d{2})\d{5}(\d{2})$", "\1\2*****\3"
    ]
  }
  
  # 移除絕對不應記錄的欄位
  mutate {
    remove_field => [
      "[context][password]",
      "[context][cvv]",
      "[context][pin]",
      "[request][body][password]"
    ]
  }
}

@Component
public class SensitiveDataMasker {
    
    private static final Pattern CARD_PATTERN = 
        Pattern.compile("\\b(\\d{4})\\d{8}(\\d{4})\\b");
    private static final Pattern EMAIL_PATTERN = 
        Pattern.compile("(?<=.{3}).(?=.*@)");
    private static final Pattern TW_ID_PATTERN = 
        Pattern.compile("\\b([A-Z])(\\d{2})\\d{5}(\\d{2})\\b");
    
    public String mask(String input) {
        if (input == null) return null;
        
        String result = input;
        result = CARD_PATTERN.matcher(result).replaceAll("$1********$2");
        result = EMAIL_PATTERN.matcher(result).replaceAll("*");
        result = TW_ID_PATTERN.matcher(result).replaceAll("$1$2*****$3");
        
        return result;
    }
}

// Log4j2 Layout 整合
@Plugin(name = "MaskingPatternLayout", category = Node.CATEGORY, 
        elementType = Layout.ELEMENT_TYPE, printObject = true)
public class MaskingPatternLayout extends AbstractStringLayout {
    
    private final SensitiveDataMasker masker = new SensitiveDataMasker();
    
    @Override
    public String toSerializable(LogEvent event) {
        String message = event.getMessage().getFormattedMessage();
        return masker.mask(message);
    }
}

// 稽核 Log 範例結構
{
  "@timestamp": "2026-01-26T10:30:45.123Z",
  "audit": {
    "event_type": "DATA_ACCESS",
    "action": "READ",
    "result": "SUCCESS",
    "actor": {
      "user_id": "EMP-001234",
      "user_name": "王小明",
      "department": "財務部",
      "ip_address": "10.1.2.100",
      "session_id": "sess-abc123"
    },
    "resource": {
      "type": "CUSTOMER_ACCOUNT",
      "id": "ACC-78901234",
      "fields_accessed": ["balance", "transaction_history"]
    },
    "context": {
      "application": "core-banking",
      "module": "account-inquiry",
      "transaction_id": "TXN-20260126-001234",
      "reason": "客戶臨櫃查詢",
      "ticket_number": "SR-2026012600123"
    }
  },
  "integrity": {
    "hash": "sha256:abc123...",
    "previous_hash": "sha256:xyz789..."
  }
}

@Service
public class AuditLogService {
    
    private final MessageDigest digest = MessageDigest.getInstance("SHA-256");
    private String previousHash = "";
    
    public void writeAuditLog(AuditEvent event) {
        // 計算 Hash Chain
        String content = objectMapper.writeValueAsString(event);
        String currentHash = calculateHash(previousHash + content);
        
        event.getIntegrity().setHash(currentHash);
        event.getIntegrity().setPreviousHash(previousHash);
        
        // 寫入 Log
        auditLogger.info(objectMapper.writeValueAsString(event));
        
        // 更新 Chain
        previousHash = currentHash;
        
        // 同時寫入不可竄改的儲存（如 WORM Storage）
        wormStorage.append(event);
    }
    
    private String calculateHash(String input) {
        byte[] hash = digest.digest(input.getBytes(StandardCharsets.UTF_8));
        return "sha256:" + bytesToHex(hash);
    }
}

# 標準欄位命名（團隊共用）

# 時間相關
timestamp: ISO 8601 格式，UTC 時區
processed_at: Logstash 處理時間

# 識別相關
trace_id: 分散式追蹤 ID（小寫底線）
span_id: Span ID
request_id: 請求 ID
transaction_id: 交易 ID
session_id: Session ID

# 來源相關
service: 服務名稱（小寫橫線，如 payment-service）
environment: 環境（prod, staging, dev）
version: 服務版本
host: 主機名稱
instance: 實例 ID

# 使用者相關
user.id: 使用者 ID
user.type: 使用者類型（customer, employee, system）
user.ip: 來源 IP

# 業務相關
context.*: 業務上下文（自由定義）

# 錯誤相關
error.type: 錯誤類型
error.code: 錯誤代碼
error.message: 錯誤訊息
error.stack_trace: 堆疊追蹤（僅 ERROR level）

// logback-spring.xml
<configuration>
    <springProperty name="SERVICE_NAME" source="spring.application.name"/>
    <springProperty name="ENV" source="spring.profiles.active"/>
    
    <appender name="JSON" class="ch.qos.logback.core.ConsoleAppender">
        <encoder class="net.logstash.logback.encoder.LogstashEncoder">
            <includeMdcKeyName>traceId</includeMdcKeyName>
            <includeMdcKeyName>spanId</includeMdcKeyName>
            <includeMdcKeyName>userId</includeMdcKeyName>
            <customFields>
                {"service":"${SERVICE_NAME}","environment":"${ENV}"}
            </customFields>
            <fieldNames>
                <timestamp>@timestamp</timestamp>
                <version>[ignore]</version>
                <levelValue>[ignore]</levelValue>
            </fieldNames>
        </encoder>
    </appender>
    
    <root level="INFO">
        <appender-ref ref="JSON"/>
    </root>
</configuration>

// 建立角色
PUT _security/role/developer
{
  "cluster": ["monitor"],
  "indices": [
    {
      "names": ["dev-*", "staging-*"],
      "privileges": ["read", "view_index_metadata"],
      "field_security": {
        "except": ["user.password", "context.card_number"]
      }
    }
  ],
  "applications": [
    {
      "application": "kibana-.kibana",
      "privileges": ["feature_discover.read", "feature_visualize.all"],
      "resources": ["space:dev-space"]
    }
  ]
}

// 建立使用者
PUT _security/user/dev_user_001
{
  "password": "...",
  "roles": ["developer"],
  "full_name": "Developer 001",
  "email": "dev001@company.com",
  "metadata": {
    "department": "Engineering",
    "team": "Payment"
  }
}

# GitLab CI 範例
deploy_production:
  stage: deploy
  script:
    - kubectl apply -f k8s/
    # 發送部署事件到 Elasticsearch
    - |
      curl -X POST "https://es-cluster:9200/deploy-events/_doc" \
        -H "Content-Type: application/json" \
        -d '{
          "@timestamp": "'$(date -u +%Y-%m-%dT%H:%M:%SZ)'",
          "event_type": "deployment",
          "service": "'$CI_PROJECT_NAME'",
          "version": "'$CI_COMMIT_TAG'",
          "environment": "production",
          "deployer": "'$GITLAB_USER_LOGIN'",
          "commit_sha": "'$CI_COMMIT_SHA'",
          "pipeline_url": "'$CI_PIPELINE_URL'"
        }'

// 確保 TraceId 傳遞到 Log
@Configuration
public class TracingConfig {
    
    @Bean
    public Filter traceIdFilter() {
        return (request, response, chain) -> {
            String traceId = tracer.currentSpan().context().traceId();
            MDC.put("traceId", traceId);
            MDC.put("spanId", tracer.currentSpan().context().spanId());
            try {
                chain.doFilter(request, response);
            } finally {
                MDC.clear();
            }
        };
    }
}

# Logstash: 安全事件轉發到 SIEM
output {
  if [level] == "ERROR" and [context][security_event] {
    # 發送到 SIEM
    syslog {
      host => "siem.company.com"
      port => 514
      protocol => "tcp"
      severity => "warning"
      facility => "local0"
      message => "%{message}"
    }
  }
  
  # 同時保留在 ES
  elasticsearch {
    hosts => ["es-cluster:9200"]
    index => "security-logs-%{+YYYY.MM.dd}"
  }
}

// 1. 搜尋特定時間範圍的 ERROR
GET app-logs-*/_search
{
  "query": {
    "bool": {
      "filter": [
        { "term": { "level": "ERROR" } },
        { "range": { "@timestamp": { "gte": "now-1h" } } }
      ]
    }
  }
}

// 2. 統計各服務錯誤數
GET app-logs-*/_search
{
  "size": 0,
  "query": {
    "bool": {
      "filter": [
        { "term": { "level": "ERROR" } },
        { "range": { "@timestamp": { "gte": "now-24h" } } }
      ]
    }
  },
  "aggs": {
    "by_service": {
      "terms": { "field": "service", "size": 20 }
    }
  }
}

// 3. 追蹤特定 TraceId
GET app-logs-*/_search
{
  "query": {
    "term": { "traceId": "abc123def456" }
  },
  "sort": [{ "@timestamp": "asc" }]
}

// 4. 搜尋特定異常類型
GET app-logs-*/_search
{
  "query": {
    "bool": {
      "filter": [
        { "term": { "exception.class": "java.net.SocketTimeoutException" } }
      ]
    }
  }
}

# 基本條件
level: ERROR and service: payment-service

# 時間範圍
@timestamp >= "2026-01-26" and @timestamp < "2026-01-27"

# 模糊搜尋
message: *timeout* and service: payment*

# 組合條件
(level: ERROR or level: WARN) and context.amount > 10000

# 排除
level: ERROR and not exception.class: "ValidationException"

# 存在性
exception.stackTrace: * and level: ERROR