优化云录制存储分层归档的冷热数据分级技巧

# Terraform 示例：阿里云 OSS 生命周期规则
resource "alicloud_oss_bucket" "record_bucket" {
  bucket = "corp-record-prod"
  lifecycle_rule {
    name     = "hot_to_warm_30d"
    enabled  = true
    prefix   = "live_teaching/"
    expiration {
      days = 30
    }
    transition {
      days          = 30
      storage_class = "IA"          # 低频访问型
    }
    transition {
      days          = 180
      storage_class = "Archive"     # 归档型
    }
    transition {
      days          = 1825          # 5 年
      storage_class = "ColdArchive" # 深度归档型
    }
  }
}

flowchart LR
    Client[业务服务] -->|HTTP Range 请求| Gateway[解冻网关]
    Gateway -->|HEAD Object| OSS[(对象存储)]
    OSS -->|x-oss-storage-class| Gateway
    Gateway -->|Archive? 异步提交 Restore| OSS
    Gateway -->|IA/Standard? 直接代理| OSS
    Gateway -->|302 Redirect / 预签名 URL| Client

sequenceDiagram
    participant Client as 解冻网关
    participant OSS as 对象存储
    participant KMS as 云KMS(托管CMK)
    participant LocalHSM as 本地HSM/CMK
    Client->>OSS: RestoreObject (Archive→Standard)
    Note over OSS: 异步解冻中...
    Client->>OSS: HeadObject (获取 x-oss-meta-dek-ciphertext)
    Client->>KMS: Decrypt(DEK_Ciphertext) -- 云KMS代理调用本地HSM
    KMS->>LocalHSM: Decrypt(CMK_ID, DEK_Ciphertext)
    LocalHSM-->>KMS: Plaintext_DEK
    KMS-->>Client: Plaintext_DEK
    Client->>Client: 本地内存缓存 DEK (TTL=24h, LRU)
    Client->>OSS: GetObject (Range请求)
    Client->>Client: AES-256-GCM 解密 (硬件加速 AES-NI)

type DEKCache struct {
    mu    sync.RWMutex
    cache *lru.Cache // key: objectETag, value: *DEKEntry
    kms   KMSClient
}

func (c *DEKCache) GetOrFetch(ctx context.Context, objMeta *ObjectMeta) ([]byte, error) {
    // 1. 读缓存
    c.mu.RLock()
    if entry, ok := c.cache.Get(objMeta.ETag); ok && !entry.Expired() {
        c.mu.RUnlock()
        atomic.AddInt64(&metrics.DEKCacheHit, 1)
        return entry.Plaintext, nil
    }
    c.mu.RUnlock()

    // 2. 单飞防护：同一对象并发解冻只解密一次
    c.mu.Lock()
    defer c.mu.Unlock()
    if entry, ok := c.cache.Get(objMeta.ETag); ok && !entry.Expired() {
        return entry.Plaintext, nil
    }

    // 3. 调用 KMS 解密 (含重试/熔断)
    plainDEK, err := c.kms.Decrypt(ctx, objMeta.DEKCiphertext)
    if err != nil {
        return nil, fmt.Errorf("kms decrypt failed: %w", err)
    }

    // 4. 入缓存
    c.cache.Add(objMeta.ETag, &DEKEntry{
        Plaintext: plainDEK,
        ExpireAt:  time.Now().Add(24 * time.Hour),
    })
    atomic.AddInt64(&metrics.DEKCacheMiss, 1)
    return plainDEK, nil
}

-- ClickHouse 物化视图：每日会话级成本聚合
CREATE MATERIALIZED VIEW mv_session_daily_cost
ENGINE = SummingMergeTree()
PARTITION BY toYYYYMM(billing_date)
ORDER BY (biz_tag, session_id, billing_date)
AS SELECT
    toDate(billing_timestamp) AS billing_date,
    JSONExtractString(resource_tags, 'biz_tag') AS biz_tag,
    JSONExtractString(resource_tags, 'session_id') AS session_id,
    sum(CASE WHEN charge_type = 'Storage' THEN amount ELSE 0 END) AS storage_cost,
    sum(CASE WHEN charge_type = 'Retrieval' THEN amount ELSE 0 END) AS retrieval_cost,
    sum(CASE WHEN charge_type = 'Request' THEN amount ELSE 0 END) AS request_cost,
    sum(CASE WHEN charge_type = 'Replication' THEN amount ELSE 0 END) AS replication_cost,
    sum(amount) AS total_cost
FROM billing_detail_table
WHERE product_code = 'OSS'
GROUP BY billing_date, biz_tag, session_id;

+-------------------------------------------------------+
|              Business Applications (SDK / SQL)        |
+-------------------------------------------------------+
|              Unified Namespace Gateway (Trino/StarRocks) |
|  - 解析 SQL -> Iceberg Manifest -> S3 Presigned URL   |
+-------------------------------------------------------+
|              Apache Iceberg Catalog (Hive/REST)       |
|  - 表: recordings (partition by biz_tag, dt)          |
|  - 隐藏分区: storage_tier (hot/ia/archive/cold)       |
|  - 字段: path, size, sha256, tier, tier_changed_at    |
+-------------------------------------------------------+
|  Hot (OSS Standard) | Warm (OSS IA) | Cold (OSS Archive) | DR Cloud (S3 Glacier) |
+---------------------+---------------+--------------------+-----------------------+

CREATE TABLE IF NOT EXISTS lakehouse.recordings (
    meeting_id        STRING,
    start_time        TIMESTAMP(3),
    biz_tag           STRING,
    source_path       STRING,          -- s3://bucket/prefix/xxx.mp4
    transcoded_paths  MAP<STRING, STRING>, -- {'720p': 's3://...', 'audio': 's3://...'}
    file_size_bytes   BIGINT,
    sha256            STRING,
    storage_tier      STRING,          -- 'STANDARD' | 'IA' | 'ARCHIVE' | 'COLD_ARCHIVE'
    tier_changed_at   TIMESTAMP(3),
    retention_expire  DATE,
    encryption_cmk_id STRING,
    -- 业务扩展字段
    speakers          ARRAY<STRING>,
    keywords          ARRAY<STRING>
)
PARTITIONED BY (
    days(start_time),           -- 物理分区：按天管理文件生命周期
    biz_tag,                    -- 业务隔离
    storage_tier                -- 隐藏分区：自动路由到对应存储后端
)
TBLPROPERTIES (
    'write.target-file-size-bytes'='134217728',  -- 128MB 合并小文件
    'write.metadata.compression-codec'='zstd',
    'history.expire.max-snapshot-age-ms'='2592000000' -- 保留 30 天快照支持时间旅行
);

# Python SDK 示例：配合 OSS 生命周期回调事件，更新 Iceberg 分区值
def on_oss_tier_change_event(event):
    # event: { "object_key": "live/2024/05/20/abc.mp4", "new_tier": "IA", "timestamp": "..." }
    table = catalog.load_table("lakehouse.recordings")
    
    # 1. 定位受影响的数据文件
    # 利用 Iceberg 的 partition pruning 仅扫描相关分区
    scan = table.scan(
        row_filter=("source_path == 's3://bucket/" + event['object_key'] + "'")
    ).select("file_path", "file_format", "record_count", "file_size_in_bytes")
    
    # 2. 重写清单，仅修改 storage_tier 字段值
    # 利用 RewriteDataFilesAction 实现零拷贝元数据变更
    with table.update_spec() as update:
        update.rewrite_data_files(
            filter=scan.filter,
            sort_order=table.spec().sort_order,
            # 核心：通过 case when 修改隐藏分区值，触发新快照生成
            transforms={ "storage_tier": f"'{event['new_tier']}'" } 
        )
    
    # 3. 发布新快照，Trino/StarRocks 秒级感知
    logger.info(f"Iceberg snapshot updated for {event['object_key']} -> {event['new_tier']}")

# .gitlab-ci.yml 片段：归档恢复演练 Job
archive_dr_drill_l2:
  stage: dr_drill
  image: python:3.11-slim
  rules:
    - if: $CI_PIPELINE_SOURCE == "schedule" && $DRILL_LEVEL == "L2"
  variables:
    TARGET_BIZ_TAG: "security_audit"
    DRILL_DATE_RANGE: "2024-02-01/2024-04-30"
    RESTORE_TIER: "Standard" # 归档->标准
  script:
    - pip install -r requirements-dr.txt
    - |
      python dr_drill.py 
        --biz-tag $TARGET_BIZ_TAG 
        --date-range $DRILL_DATE_RANGE 
        --target-tier $RESTORE_TIER 
        --concurrency 50 
        --verify-sha256 
        --output-report drill_report_${CI_PIPELINE_ID}.json
  artifacts:
    reports:
      junit: drill_report_*.json
    expire_in: 90d
  timeout: 6h

# 1. 一键诊断分层健康度
$ recctl doctor --bucket corp-record-prod --deep-check
[OK] 冷数据占比 48.2% (目标 >45%)
[WARN] IA 存储中 12% 对象存储时长 < 30 天，面临最小存储时长罚金风险
[ERROR] 发现 3,452 个对象未打 biz_tag，无法匹配生命周期规则
[ACTION] 建议执行: recctl tag-fix --auto --dry-run

# 2. 模拟成本优化收益
$ recctl simulate --strategy aggressive --start-date 2024-01-01
模拟结果 (基于近 90 天访问日志):
  当前月均成本: ¥ 1,245,000
  优化后月均成本: ¥ 682,000 (降本 45.2%)
  预估额外取回费: ¥ 18,000/月
  净节省: ¥ 545,000/月
  受影响高频对象: 12 个 (建议加入白名单)

# 3. 发起合规取证包 (生成仅含预签名 URL 的清单，不下载数据)
$ recctl evidence-pack --case-id CASE-2024-05-001 
    --meeting-ids "M1001,M1002,M1005" 
    --output-format csv 
    --url-expiry 168h
已生成 evidence_CASE-2024-05-001.csv，包含 15 个对象预签名下载链接
已自动发起深度归档解冻，预计 2024-05-21 06:00 就绪