# Packed Backup CSV Export

`run_packed_backup_export.py` 是主入口：它在 `tidb-serverless` namespace 中协调
Pack、Export、Watermark 三个阶段，把 packed backup 导出为加密 CSV，并在成功后推进
`active_cluster_ids.s3_data_updated_at`。

入口会直接创建 Kubernetes 资源、读写 S3 并访问 MySQL，不是 dry-run 工具。同一流程的
重试应复用同一个 `RUN_ID`。

## 生成 CronJob manifest

[`generate_packed_backup_cronjob.py`](generate_packed_backup_cronjob.py) 可以完全离线地
生成定时执行所需的五个 Kubernetes 对象：MySQL Secret、Lark Secret、GPG 公钥
ConfigMap、workflow ConfigMap 和 CronJob。先复制并填写
[`example/cronjob-config.example.yaml`](example/cronjob-config.example.yaml)，然后显式指定
每个本地输入。精简示例只保留没有默认值的 CronJob 字段；全部字段、默认值和字段说明见
[`example/full-example.yaml`](example/full-example.yaml)；单独运行 workflow 时可从
[`example/workflow-spec.example.yaml`](example/workflow-spec.example.yaml) 开始：

```bash
python3 generate_packed_backup_cronjob.py \
  --config /secure/path/cronjob-config.yaml \
  --gpg-public-key /secure/path/pubkey.asc \
  --mysql-defaults-file /secure/path/mysql.cnf \
  --lark-webhook-file /secure/path/lark-webhook |
kubectl apply -f -
```

生成器默认把 manifest 写到 stdout，摘要写到 stderr，因此可直接交给管道消费者。需要保留
文件时显式传入 `--output /secure/path/cronjob.yaml`；文件以 `0600` 创建，已有文件只有在
同时传入 `--force` 时才会被截断重写。

Webhook 也可以通过 `--lark-webhook-env <ENV_NAME>` 从一个被明确点名的环境变量读取。
生成器没有 `~/.my.cnf`、默认 webhook 环境变量或远端 GPG object 等隐式来源，也不会
访问 MySQL、AWS 或 Kubernetes。它会提前检查配置 schema、文件类型和大小、GPG armor、
`my.cnf` 的 `[client]` 配置、HTTPS webhook 及资源名长度，并拒绝示例中尚未替换的
`<...>` 占位符；CronJob 直接挂载本地公钥生成的 ConfigMap，不使用 init container。
成功生成后，生成器会向 stderr 打印 OpenPGP fingerprint/holder 以及 MySQL
host、port、user、database，方便核对输入；不会打印 MySQL password 或 webhook。

输出 YAML 中的 Secret `data` 只是 Base64，不是加密。stdout 管道和 `--output` 文件都应
按凭据处理，避免进入日志、命令替换、版本库或不受信任的消费者。

Controller Job 名必须是 `<CRONJOB_NAME>-<EXECUTION_ID>`。Controller 精确移除配置的
CronJob 名和紧随其后的 `-`，因此 `on-demand-packed-export-reimport-0731` 得到的
execution ID 是 `reimport-0731`，而不是 `0731`。RUN_ID 为：

```text
RUN_ID = "cron-" + EXECUTION_ID + "-" + sha256(CONTROLLER_JOB_NAME)[0:8]
```

execution ID 必须是允许内部 `-` 的 Kubernetes DNS label。生成器会根据 CronJob 名、
shard 数和所有下游资源名计算长度预算；示例配置的预算是 14，并预留至少 10 个字符供
定时 Job 使用。同一个 Controller Job 名总是得到相同 RUN_ID，因此 retry 必须删除原
Job，再用完全相同的名字执行 `kubectl create job --from=cronjob/...`。Job 名不属于配置
的 CronJob 或 execution ID 不合法时，Controller 会直接失败，不存在旧规则或 fallback。

## 流程

```mermaid
%%{init: {"theme":"base","themeVariables":{"background":"#FFFFFF","primaryTextColor":"#243447","lineColor":"#708090","fontFamily":"Arial, sans-serif"}}}%%
flowchart TD
    A[主入口<br/>固定 RUN_ID] --> B[发现 backup<br/>得到 backup UTC time]
    B --> C[(MySQL<br/>选择待导出且未 blocklist 的 cluster)]
    C -->|cluster IDs| D[(S3<br/>cluster-ids 文本对象)]
    D -->|URI、SHA-256、数量| E[Kubernetes ConfigMap<br/>固化 workflow 输入]
    E -->|backup name + dfs:object-key| F[Pack Jobs<br/>cse-ctl pack-backup]
    F -->|Job annotation| G[Pack task JSON<br/>task_name + pack_backup_prefix]
    G -->|stdin| H[Export ConfigMap + Jobs<br/>Dumpling CSV + GPG]
    H -->|版本目录 + complete marker| I[(S3<br/>共享 export.outPrefix)]
    G -->|嵌入 pack-tasks.json| J[Watermark ConfigMap + Job]
    I -->|逐个 HEAD complete| J
    J -->|本轮 complete<br/>条件 UPDATE + COMMIT| K[(MySQL<br/>s3_data_updated_at)]
    J -->|complete 缺失或来自更早导出<br/>INSERT IGNORE + COMMIT| L[(MySQL<br/>cluster_blocklist)]

    classDef controller fill:#EAF2F8,stroke:#52789B,color:#23384D,stroke-width:1.5px;
    classDef k8s fill:#EAF5EF,stroke:#4F8067,color:#244334,stroke-width:1.5px;
    classDef s3 fill:#FFF4E5,stroke:#B7793D,color:#51371F,stroke-width:1.5px;
    classDef db fill:#F1ECF8,stroke:#76609A,color:#352B48,stroke-width:1.5px;
    classDef data fill:#F4F6F8,stroke:#7B8794,color:#2F3A45,stroke-width:1.5px;

    class A,B controller;
    class E,F,H,J k8s;
    class D,I s3;
    class C,K,L db;
    class G data;
```

若没有 cluster 满足选择条件，流程成功结束，不创建 cluster-id 对象、workflow
ConfigMap 或下游 Job。已有同名资源时，入口会校验并复用，只补建缺失的 Job。

### 1. 固化输入

1. 入口在 `tikv-api` Pod 内调用 backup API，取返回的第一条 backup；名称须为
   `YYYYMMDD/HHMMSS.meta`，其中时间作为本次 cutoff。
2. 入口从 `StatefulSet/tikv-api` 读取 `DFS_S3_BUCKET`、`DFS_S3_REGION`，并用本地
   MySQL option file 查询待导出 cluster。
3. cluster ID 以 UTF-8 文本写入 S3，每行一个；workflow ConfigMap 只保存 backup
   名、对象 URI、region、SHA-256、字节数和 cluster 数。

### 2. Pack

- 创建 `pack.shards` 个 Job，并等待全部完成。
- 每个 Job 执行 `/cse-ctl pack-backup`；cluster 列表通过
  `--keyspace-select.cluster-id-in dfs:<cluster-id-object-key>` 传入。
- 每个 Job 把 packed-backup prefix 写在 annotation
  `cse.tidbcloud.com/pack-backup-prefix`。该 annotation 是 Pack 阶段交接的必要输出；
  任一已完成 Job 缺少它或值无效时，Pack task 汇总失败。入口把已完成 Job 汇总为：

```json
[{"task_name":"<pack-job>","pack_backup_prefix":"s3://<bucket>/<prefix>/"}]
```

这份 JSON 是后续 Export 与 Watermark 共用的输入。

### 3. Export

- 每个 Pack task 对应一个 Export Job；task prefix 通过 Job 参数
  `--s3-prefix <pack_backup_prefix>meta` 传入。
- Job 列举 `meta/*.meta`，为每个 `<cluster_id>_<keyspace_name>` identity 启动
  Dumpling。导出格式固定为 CSV，过滤 `mysql.*`、`sys.*`，并使用本地传入的 GPG
  公钥加密。
- Dumpling 成功后写入该 identity 的 `complete` marker；marker 是 Watermark 阶段的
  完成凭证。

### 4. Watermark

- 原 Pack task JSON 被写入 Watermark 输入 ConfigMap；MySQL option file 从生成的
  Kubernetes Secret 挂载。
- Job 重新列举各 Pack prefix 下的 `meta/*.meta`，按 identity 还原 cluster ID。
- 仅当一个 cluster 的全部 identity 都有匹配的 `complete` marker 时，才推进该
  cluster 的 MySQL watermark。更晚的 S3 marker 或 MySQL watermark 不会被回退。
- `complete` 缺失或属于更早的导出时，worker 以幂等写入把 cluster 加入
  `cluster_blocklist`；缺少 marker Metadata 或 `backup-time` 也按更早的导出处理。
- Worker 把单行 JSON 结果写入 `/dev/termination-log`；Controller 从 Pod 的 terminated
  container status 读取结果，`stdout` 和 `stderr` 只保存诊断日志。

## Kubernetes 资源

所有资源都位于 `tidb-serverless`。Controller Pod 使用配置中的
`cronJob.serviceAccountName`；Pack、Export 和 Watermark Pod 使用代码固定引用的已有
`backup-restore` ServiceAccount。这两个身份职责不同，且都不会由本项目的 generator
创建。

| 阶段 | Pod ServiceAccount | 创建的资源 |
| --- | --- | --- |
| 固化输入 | `cronJob.serviceAccountName`（示例为 `backup-restore-manus`） | ConfigMap `pack-backup-offline-${RUN_ID}-cluster-ids` |
| Pack | `backup-restore` | `pack.shards` 个 Job：`pack-backup-offline-${RUN_ID}-shard-<i>-of-<N>`；配置 `pack.ebs.request` 时，每个 Pod 还会请求 ephemeral PVC |
| Export | `backup-restore` | ConfigMap `dumpling-packed-export-${RUN_ID}-init`；每个 Pack task 一个 Job：`dumpling-packed-export-${RUN_ID}-task-<i>-of-<N>` |
| Watermark | `backup-restore` | ConfigMap `packed-backup-watermark-${RUN_ID}-input`；Job `packed-backup-watermark-${RUN_ID}` |

Pack 生成器从 workflow 配置或 `tikv-api` 数据卷模板取得 StorageClass 名称，并将它写入
ephemeral PVC 模板；StorageClass 解析和动态供盘由 Kubernetes storage control plane 完成。

入口只读取、不创建：`StatefulSet/tikv-api` 和固定用于 backup discovery 的
`tikv-api-0` Pod。Controller 和下游 Pod 引用的 ServiceAccount 及其 RBAC 必须预先存在。

## S3 布局

`[${DFS_PREFIX}/]` 表示可选的 DFS prefix；cluster-id 对象不使用它。

```text
s3://${DFS_S3_BUCKET}/packed-backup-csv-export/cluster-id-lists/${RUN_ID}/cluster-ids
  # 每行一个 cluster ID
  # metadata: workflow-name, backup-name, sha256, byte-count, cluster-count

s3://${DFS_S3_BUCKET}/[${DFS_PREFIX}/]packed_backup/native/${backup_name}/${pack_job}/
└── meta/
    └── <cluster_id>_<keyspace_name>.meta

${export.outPrefix}/<meta_identity>/
├── <YYYY-MM-DDTHH:MM:SSZ>:dumpling-packed-export-${RUN_ID}/
│   └── <Dumpling 加密 CSV 产物>
└── complete
    # body: <YYYY-MM-DDTHH:MM:SSZ>:dumpling-packed-export-${RUN_ID}
    # metadata: job-instance-name, backup-time
```

## Watermark 数据库访问

入口和 Watermark worker 都先执行 `SET time_zone = '+00:00'`，使用参数化 SQL，并显式
提交或回滚事务。

入口使用本地 `mysql.cnf` 选择 cluster：

```sql
SELECT active_clusters.cluster_id
FROM active_cluster_ids AS active_clusters
WHERE active_clusters.updated_at > active_clusters.s3_data_updated_at
  AND active_clusters.updated_at <= :backup_cutoff
  AND NOT EXISTS (
    SELECT 1
    FROM cluster_blocklist AS blocklist
    WHERE blocklist.cluster_id = active_clusters.cluster_id
  )
ORDER BY active_clusters.cluster_id;
```

Watermark Job 使用 Secret 中的 `mysql.cnf`，并发校验 cluster；marker 全部通过后执行：

```sql
UPDATE active_cluster_ids
SET s3_data_updated_at = :backup_time
WHERE cluster_id = :cluster_id
  AND (s3_data_updated_at IS NULL OR s3_data_updated_at < :backup_time);
```

若条件更新未修改行，worker 会读取当前 `s3_data_updated_at`，确认它已等于或晚于本次
backup；不会降低现有 watermark。

marker 未验证且归类为缺失或更早导出时执行：

```sql
INSERT IGNORE INTO cluster_blocklist (cluster_id, used_by, description)
VALUES (
  :cluster_id,
  'br_export',
  'export_instance=<export-instance>, backup_time=<UTC-backup-time>'
);
```

`cluster_blocklist.cluster_id` 的主键同时支持入口反连接和 Watermark 的并发幂等写入。
`description` 使用 worker 已有的 Export instance 和 UTC backup time 提供简短定位上下文。