日志系统全栈：Loki、EFK、Vector 与日志架构设计

Loki 与传统 Elasticsearch 的核心区别：

Elasticsearch：
  ├── 倒排索引所有字段
  ├── 搜索极快
  ├── 存储体积大（原始日志的 2-3x）
  └── 成本高

Loki：
  ├── 只索引标签（label）
  ├── 日志内容按时间顺序存储在对象存储中
  ├── 搜索时需抓取块并扫描
  ├── 存储体积小（接近原始日志）
  └── 成本低 10x+

权衡：
  - 已知标签过滤 → Loki 高效
  - 全文模糊搜索 → Loki 慢（需用 LogQL 管道过滤）

# loki-values.yaml (Helm)
loki:
  auth_enabled: false
  commonConfig:
    replication_factor: 1

  storage:
    type: s3
    s3:
      endpoint: s3.amazonaws.com
      region: us-east-1
      bucketnames: loki-chunks
      access_key_id: ${AWS_ACCESS_KEY_ID}
      secret_access_key: ${AWS_SECRET_ACCESS_KEY}

distributor:
  replicas: 1

ingester:
  replicas: 1
  persistence:
    enabled: true
    size: 10Gi

querier:
  replicas: 1
  persistence:
    enabled: true
    size: 10Gi

queryFrontend:
  replicas: 1

compactor:
  replicas: 1
  persistence:
    enabled: true
    size: 10Gi

# fluentd.conf
<source>
  @type tail
  path /var/log/containers/*.log
  pos_file /var/log/fluentd.pos
  tag kubernetes.*
  <parse>
    @type json
    time_key time
    time_format %Y-%m-%dT%H:%M:%S.%NZ
  </parse>
</source>

<filter kubernetes.**>
  @type kubernetes_metadata
</filter>

<filter kubernetes.**>
  @type record_transformer
  <record>
    cluster ${ENV['CLUSTER_NAME']}
    environment production
  </record>
</filter>

<match kubernetes.**>
  @type elasticsearch
  host elasticsearch
  port 9200
  logstash_format true
  logstash_prefix k8s-logs
  flush_interval 10s
</match>

// ILM (Index Lifecycle Management)
PUT _ilm/policy/logs_policy
{
  "policy": {
    "phases": {
      "hot": {
        "min_age": "0ms",
        "actions": {
          "rollover": {
            "max_size": "50GB",
            "max_age": "1d"
          }
        }
      },
      "warm": {
        "min_age": "3d",
        "actions": {
          "shrink": { "number_of_shards": 1 },
          "forcemerge": { "max_num_segments": 1 }
        }
      },
      "cold": {
        "min_age": "7d",
        "actions": {
          "freeze": {}
        }
      },
      "delete": {
        "min_age": "30d",
        "actions": {
          "delete": {}
        }
      }
    }
  }
}

# vector.yaml
api:
  enabled: true
  address: 0.0.0.0:8686

sources:
  kubernetes_logs:
    type: kubernetes_logs
    pod_annotation_fields:
      container_image: "container_image"
      container_name: "container_name"
      pod_labels: "pod_labels"

  host_metrics:
    type: host_metrics
    collectors: [cpu, memory, disk]

transforms:
  parse_json:
    type: remap
    inputs: [kubernetes_logs]
    source: |
      . = parse_json!(.message)
      .timestamp = to_timestamp!(.time)
      .level = downcase!(.level ?? "info")

  filter_errors:
    type: filter
    inputs: [parse_json]
    condition: '.level == "error" || .level == "fatal"'

sinks:
  loki:
    type: loki
    inputs: [parse_json]
    endpoint: http://loki:3100
    labels:
      app: "{{ kubernetes.pod_labels.app }}"
      namespace: "{{ kubernetes.pod_namespace }}"
      level: "{{ level }}"
    encoding:
      codec: json
    healthcheck:
      enabled: true

  elasticsearch:
    type: elasticsearch
    inputs: [filter_errors]
    endpoints: [http://elasticsearch:9200]
    index: "error-logs-%Y.%m.%d"
    bulk:
      index: "error-logs"

  console:
    type: console
    inputs: [filter_errors]
    encoding:
      codec: json

{
  "timestamp": "2026-08-13T10:32:01.234Z",
  "level": "ERROR",
  "service": "payment-svc",
  "version": "1.2.3",
  "trace_id": "abc123def456",
  "span_id": "span789",
  "message": "Payment processing failed",
  "error": {
    "type": "TimeoutError",
    "message": "Connection to bank API timed out after 30s",
    "stack": "..."
  },
  "http": {
    "method": "POST",
    "route": "/api/payment",
    "status_code": 504,
    "duration_ms": 30000
  },
  "user": {
    "id": "user_789",
    "tier": "premium"
  },
  "request": {
    "order_id": "order_456",
    "amount": 299.99,
    "currency": "CNY",
    "payment_method": "wechat_pay"
  },
  "context": {
    "retry_count": 3,
    "bank_api": "https://bank.example.com/v2/charge",
    "region": "ap-east-1"
  }
}

// Go：自动脱敏敏感字段
func sanitizeFields(data map[string]interface{}) map[string]interface{} {
    sensitiveKeys := []string{"password", "token", "secret", "credit_card", "ssn"}
    for _, key := range sensitiveKeys {
        if _, ok := data[key]; ok {
            data[key] = "[REDACTED]"
        }
    }
    // 递归处理嵌套
    for k, v := range data {
        if nested, ok := v.(map[string]interface{}); ok {
            data[k] = sanitizeFields(nested)
        }
    }
    return data
}

// 或使用 Vector VRL
// . = redact!(., filters: ["credit_card", "ssn"])