{
  "@timestamp": "2024-05-20T08:30:45.123Z",
  "event": { "dataset": "meeting.client", "action": "join_meeting" },
  "service": { "name": "vc-client", "version": "3.2.1", "environment": "prod" },
  "host": { "hostname": "PC-DEV-001", "os": { "name": "Windows 11", "version": "23H2" } },
  "user": { "id": "u_123456", "tenant_id": "t_789" },
  "meeting": { "id": "m_abcdef", "type": "scheduled", "join_latency_ms": 842 },
  "network": { "rtt_ms": 45, "packet_loss_pct": 0.2, "bandwidth_kbps": 1200 },
  "client": { "cpu_pct": 12.3, "mem_mb": 312, "gpu": "NVIDIA RTX 3060" },
  "log": { "level": "INFO", "logger": "MeetingController" }
}

infra/
├── ansible/
│   ├── inventories/
│   │   ├── prod.yml
│   │   └── staging.yml
│   ├── roles/
│   │   ├── elasticsearch/
│   │   ├── fluent-bit/
│   │   ├── logstash/
│   │   └── kibana/
│   ├── site.yml
│   └── vault.yml          # 加密存储证书、密钥
├── terraform/
│   ├── modules/
│   │   ├── es-cluster/
│   │   └── lb-log-ingest/
│   ├── main.tf
│   └── variables.tf
└── helm/
    └── vc-client-logging/  # 客户端侧 Fluent Bit Helm Chart

# roles/fluent-bit/templates/fluent-bit.conf.j2
[SERVICE]
    Flush        5
    Log_Level    info
    Daemon       Off
    Parsers_File parsers.conf
    HTTP_Server  On
    HTTP_Listen  0.0.0.0
    HTTP_Port    2020

[INPUT]
    Name              tail
    Path              /var/log/vc-client/*.json
    Parser            json
    Tag               vc.client.*
    Refresh_Interval  10
    Mem_Buf_Limit     50MB
    Skip_Long_Lines   On

[FILTER]
    Name                record_modifier
    Match               vc.client.*
    Record              k8s_cluster ${CLUSTER_NAME}
    Record              environment ${ENVIRONMENT}

[OUTPUT]
    Name            http
    Match           vc.client.*
    Host            {{ logstash_lb_vip }}
    Port            8080
    URI             /vc-logs
    Format          json
    Header          Authorization Bearer {{ vault_logstash_token }}
    Retry_Limit     5
    Compress        gzip

# .gitlab-ci.yml
stages:
  - validate
  - plan
  - deploy-staging
  - smoke-test
  - deploy-prod

validate:
  stage: validate
  image: ansible/ansible:latest
  script:
    - ansible-lint ansible/
    - yamllint ansible/
    - terraform fmt -check terraform/

plan:
  stage: plan
  script:
    - cd terraform && terraform init && terraform plan -out=tfplan
  artifacts:
    reports:
      terraform: terraform/tfplan

deploy-staging:
  stage: deploy-staging
  needs: [plan]
  environment: staging
  script:
    - cd terraform && terraform apply -auto-approve tfplan
    - ansible-playbook -i ansible/inventories/staging.yml ansible/site.yml

smoke-test:
  stage: smoke-test
  needs: [deploy-staging]
  script:
    - curl -sf https://kibana-staging.example.com/api/status | jq -e '.status.overall.level=="available"'
    - curl -sf "https://es-staging.example.com/_cat/indices/vc-client-*?h=index,docs.count" | grep -q "vc-client-"

deploy-prod:
  stage: deploy-prod
  needs: [smoke-test]
  when: manual
  environment: production
  script:
    - cd terraform && terraform apply -auto-approve
    - ansible-playbook -i ansible/inventories/prod.yml ansible/site.yml

PUT _ilm/policy/vc-client-logs
{
  "policy": {
    "phases": {
      "hot": {
        "min_age": "0ms",
        "actions": {
          "rollover": { "max_size": "50gb", "max_age": "1d" },
          "set_priority": { "priority": 100 }
        }
      },
      "warm": {
        "min_age": "3d",
        "actions": {
          "readonly": {},
          "forcemerge": { "max_num_segments": 1 },
          "shrink": { "number_of_shards": 1 },
          "set_priority": { "priority": 50 }
        }
      },
      "cold": {
        "min_age": "14d",
        "actions": {
          "searchable_snapshot": {
            "snapshot_repository": "vc-logs-s3-repo",
            "force_merge_index": true
          },
          "set_priority": { "priority": 10 }
        }
      },
      "delete": {
        "min_age": "90d",
        "actions": { "delete": {} }
      }
    }
  }
}

groups:
- name: vc-client-logs
  rules:
  - alert: ClientLogIngestLatencyHigh
    expr: histogram_quantile(0.95, rate(fluentbit_output_proc_duration_seconds_bucket[5m])) > 2
    for: 5m
    labels: { severity: warning }
    annotations:
      summary: "客户端日志入库 P95 延迟 > 2s"
      runbook: "https://wiki.example.com/runbook/log-ingest-latency"

  - alert: ClientErrorRateSpike
    expr: |
      sum(rate(vc_client_logs_total{level=~"ERROR|FATAL"}[5m])) by (service_version)
      /
      sum(rate(vc_client_logs_total[5m])) by (service_version) > 0.02
    for: 3m
    labels: { severity: critical }
    annotations:
      summary: "版本 {{ $labels.service_version }} 错误率突破 2%"

// ILogTransport.h
class ILogTransport {
public:
    virtual ~ILogTransport() = default;
    // 非阻塞写入，返回写入字节数或错误码
    virtual Result<size_t> Write(const StructuredLogEvent& event) noexcept = 0;
    // 显式刷盘/上报，供 App 进入后台/退出时调用
    virtual void Flush(FlushPolicy policy = FlushPolicy::kBestEffort) = 0;
    // 运行时动态调整上报策略（如切换 Wi-Fi/4G）
    virtual void UpdateNetworkPolicy(const NetworkPolicy& policy) = 0;
};

graph LR
    A[客户端代码仓] -->|CI 阶段: schema_lint| B(Schema Registry)
    C[Logstash/Fluentd Pipeline] -->|启动时: fetch_schema| B
    D[Elasticsearch Index Template] -->|自动生成 mapping| B
    E[Kibana Dashboard] -->|字段下拉源| B

# .gitlab-ci.yml 片段
schema_validate:
  stage: validate
  image: node:20-alpine
  script:
    - npm i -g @apicurio/data-models @asyncapi/validator
    # 1. 语法校验
    - find schemas/ -name "*.json" -exec ajv validate -s {} -d {} ;
    # 2. 兼容性校验 (BACKWARD/TRANSITIVE)
    - curl -X POST "$SCHEMA_REGISTRY_URL/subjects/vc.client.log-value/versions" 
      -H "Content-Type: application/vnd.schemaregistry.v1+json" 
      -d @schemas/vc_client_log.json 
      | jq -e '.compatibility == "BACKWARD"'
  rules:
    - changes: [schemas/**/*]

- name: DLQ 自动修复 - 字段类型冲突自动重映射
  hosts: logstash
  tasks:
    - name: 扫描 DLQ 索引获取冲突字段
      uri:
        url: "http://es:9200/vc-client-dlq-*/_search?q=error.type:mapper_parsing_exception&size=100"
      register: dlq_errors

    - name: 生成动态映射修正脚本
      set_fact:
        fix_script: |
          ctx._source.{{ item.field }} = Long.parseLong(ctx._source.{{ item.field }}.toString())
      loop: "{{ dlq_errors.json.hits.hits | json_query('[]._source.error.caused_by.reason') | regex_findall('field \[(.+?)\]') | unique }}"

    - name: 执行 _update_by_query 修复并重入主索引
      uri:
        method: POST
        url: "http://es:9200/vc-client-dlq-*/_update_by_query?conflicts=proceed"
        body: "{{ fix_script }}"
        body_format: json

PUT _index_template/vc-client-multi-tenant
{
  "index_patterns": ["vc-client-*"],
  "template": {
    "settings": {
      "number_of_shards": 3,
      "number_of_replicas": 1,
      "routing.allocation.include._tier": "data_hot",
      "index.routing_path": "{{tenant_id}}/{{env}}/{{service_version}}"
    },
    "mappings": {
      "dynamic": "strict",
      "properties": {
        "tenant_id": { "type": "keyword" },
        "environment": { "type": "keyword" },
        "service_version": { "type": "keyword" }
      }
    }
  }
}

# chaos-experiment/log-ingest-failure.yaml
apiVersion: chaos-mesh.org/v1alpha1
kind: PodChaos
metadata:
  name: logstash-network-partition
spec:
  action: network-partition
  mode: one
  selector:
    namespaces: [logging]
    labelSelectors:
      app: logstash
  direction: both
  target:
    selector:
      namespaces: [logging]
      labelSelectors:
        app: elasticsearch
  duration: "5m"
  scheduler:
    cron: "@once"
---
# 验证指标：演练期间 Fluent Bit 本地缓冲区增长曲线、DLQ 入队速率、告警触发时效

graph TB
    subgraph Hot Tier (NVMe SSD)
        H1[Index: vc-client-*-000001] -->|ILM Rollover| H2[Index: vc-client-*-000002]
    end
    subgraph Warm Tier (HDD/SSD)
        W1[Searchable Snapshot: vc-client-*-000001] -->|Force Merge 1 shard| W2[Read Only]
    end
    subgraph Cold Tier (S3/MinIO)
        C1[Fully Mounted Snapshot] -->|Glacier/Archive| C2[Cost: $0.004/GB/mo]
    end
    H2 -->|min_age: 3d| W1
    W2 -->|min_age: 14d| C1

# Fluent Bit Lua Filter: adaptive_sampling.lua
local cjson = require "cjson"
local sampler = require "sampler"  -- 自定义共享库

function adaptive_sampling(tag, timestamp, record)
    local level = record["log"]["level"]
    local service = record["service"]["name"]
    local key = service .. ":" .. level
    
    -- 从 Redis/共享内存获取当前采样率 (由控制平面下发)
    local rate = sampler.get_rate(key)  -- 返回 0.0 ~ 1.0
    
    if level == "DEBUG" or level == "TRACE" then
        if math.random() > rate then
            return -1, nil, nil  -- 丢弃
        end
        record["_sampled"] = true
        record["_sample_rate"] = rate
    end
    return 0, timestamp, record
end

# docker-compose.dev.yml
version: '3.8'
services:
  elasticsearch:
    image: docker.elastic.co/elasticsearch/elasticsearch:8.13.0
    environment:
      - discovery.type=single-node
      - xpack.security.enabled=false
      - "ES_JAVA_OPTS=-Xms1g -Xmx1g"
    ports: ["9200:9200"]
    volumes: [es_data:/usr/share/elasticsearch/data]

  logstash:
    image: docker.elastic.co/logstash/logstash:8.13.0
    volumes:
      - ./pipelines:/usr/share/logstash/pipeline:ro
      - ./schemas:/schemas:ro
    ports: ["8080:8080", "5044:5044"]
    depends_on: [elasticsearch]

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

volumes:
  es_data:

// log_contract_test.go
func TestClientLogSchemaCompatibility(t *testing.T) {
    // 1. 从 Registry 拉取最新 PROD 兼容 Schema
    schema, err := registry.GetLatestSchema("vc.client.log-value", "BACKWARD")
    require.NoError(t, err)

    // 2. 遍历本地测试用例 (JSON 文件)
    testCases := loadTestCases("testdata/logs/")
    for _, tc := range testCases {
        // 3. 校验 payload 符合 Schema
        err := jsonschema.Validate(schema, tc.Payload)
        assert.NoError(t, err, "Case %s: schema violation: %v", tc.Name, err)

        // 4. 校验关键字段语义 (非 Schema 能力)
        assert.Greater(t, tc.Payload.Get("meeting.join_latency_ms").Int(), int64(0))
        assert.Contains(t, []string{"wifi", "4g", "5g", "ethernet"}, tc.Payload.Get("network.type").String())
    }
}

flowchart LR
    A[Logstash Filter] --> B{字段规则引擎}
    B -->|匹配 user.phone| C[打标: PII:PHONE]
    B -->|匹配 meeting.id| D[打标: BIZ:MEETING_META]
    B -->|匹配 client.ip| E[打标: NET:IP]
    C --> F[写入 ES _meta.sensitivity=high]
    D --> F
    E --> F
    F --> G[ILM 策略关联敏感度]
    G --> H[高敏感: 30天强制删除/脱敏]
    G --> I[低敏感: 90天归档]

rules:
  - id: pii_phone
    field: "user.phone"
    pattern: "^1[3-9]\d{9}$"
    action: 
      - set_field: { "_meta.sensitivity": "high" }
      - set_field: { "_meta.pii_type": "phone" }
      - script: "ctx._source.user.phone = hash(ctx._source.user.phone)"  # 落盘即脱敏