# PCH Billing 系统分析文档

## 📋 程序概述

`pch_billing.py` 是一个基于 **Apache Airflow** 的多云账单数据同步系统，主要用于自动化收集和处理来自 AWS、GCP、Azure 三大云厂商的账单数据，并统一存储到 MySQL 数据库中，为 PingCAP 公司提供统一的云成本管理平台。

## 🏗️ 系统架构

### 基础信息
- **框架**: Apache Airflow
- **调度频率**: 每12小时执行一次 (`0 */12 * * *`)
- **负责人**: zhuxiaopeng (xiaopeng.zhu@pingcap.com)
- **数据源**: Google BigQuery
- **目标存储**: MySQL 数据库

### 架构图
```
┌─────────────────┐    ┌──────────────────┐    ┌─────────────────┐
│   BigQuery      │    │   Airflow DAG    │    │     MySQL       │
│                 │    │                  │    │                 │
│ ┌─────────────┐ │    │ ┌──────────────┐ │    │ ┌─────────────┐ │
│ │ GCP Billing │ │───▶│ │ sync_gcp()   │ │───▶│ │cloud_product│ │
│ └─────────────┘ │    │ └──────────────┘ │    │ │   _cost     │ │
│                 │    │                  │    │ └─────────────┘ │
│ ┌─────────────┐ │    │ ┌──────────────┐ │    │                 │
│ │ AWS Billing │ │───▶│ │ sync_aws()   │ │───▶│                 │
│ └─────────────┘ │    │ └──────────────┘ │    │                 │
│                 │    │                  │    │                 │
│ ┌─────────────┐ │    │ ┌──────────────┐ │    │                 │
│ │Azure Billing│ │───▶│ │sync_azure()  │ │───▶│                 │
│ └─────────────┘ │    │ └──────────────┘ │    │                 │
└─────────────────┘    └──────────────────┘    └─────────────────┘
```

## 🔧 核心功能模块

### 1. 主控制函数 - `pch_billing()`

**作用**: 系统的主入口点，协调各云厂商数据同步流程

**工作流程**:
```python
def pch_billing():
    after_date = start_date(False)  # 获取45天前的日期
    bq_hook = bigquery.BigQueryHook(use_legacy_sql=False)
    
    # 1. 清理旧数据
    with get_connection() as cnx:
        with cnx.cursor() as cursor:
            cursor.execute(f"DELETE FROM cloud_product_cost WHERE date >= '{after_date}'")
    
    # 2. 依次同步三大云厂商数据
    cli = bq_hook.get_client()
    sync_aws(after_date, cli)
    sync_gcp(after_date, cli) 
    sync_azure(after_date, cli)
```

**关键特性**:
- ✅ 增量同步：只处理45天内的数据
- ✅ 数据清理：每次运行前删除指定日期后的数据，确保一致性
- ✅ 顺序执行：按 AWS → GCP → Azure 的顺序处理

### 2. GCP 账单同步 - `sync_gcp()`

**数据源**: `gcp_billing_v2.gcp_billing_export_v1_*`

**核心逻辑**:
```python
def sync_gcp(after_date, cli):
    # 构建复杂的 BigQuery SQL
    query_sql = contract_gcp_query(after_date)
    results = cli.query(query_sql).result()
    
    # 构建资源映射表
    resource_dict = {}  # cloudInsID -> resourceID
    
    # 处理查询结果
    for row in results:
        if row[3] in resource_dict:  # 只处理已知资源
            gcp_billing.append({
                "resource_id": resource_dict[row[3]],
                "product": row[2],
                "cost": row[1], 
                "date": datetime(row[0].year, row[0].month, row[0].day)
            })
```

**SQL 查询特点**:
- 📊 聚合计算：按项目ID、产品、日期分组
- 💰 费用计算：`(cost + credits_amount) * currency_conversion_rate`
- 🎯 过滤条件：只取费用大于0的记录
- 📅 时间范围：根据日期动态选择表（2023-04-19前后使用不同表）

### 3. AWS 账单同步 - `sync_aws()`

**并发处理架构**:
```python
def sync_aws(sdate, cli):
    # 获取所有 AWS 账户
    aws_accounts = []  # 从 cloud_resource 表查询
    
    # 使用线程池并发处理（最多10个线程）
    with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
        future_results = []
        for (id, cloudInsID) in aws_accounts:
            future_results.append(
                executor.submit(get_aws_billing_by_id, id, cloudInsID, sdate, cli)
            )
```

**数据源格式**: `aws_nonprod_billing_native.{account_id}_aws_billing_data`

**查询策略**:
```sql
-- 普通账单项目
SELECT day, cost, product FROM table WHERE 
  line_item_line_item_type <> 'SavingsPlanNegation' 
  AND line_item_product_code <> 'ComputeSavingsPlans'

UNION ALL

-- Savings Plans 费用
SELECT day, cost, 'SavingsPlans' as product FROM table WHERE
  line_item_line_item_type = 'SavingsPlanNegation' 
  OR line_item_product_code = 'ComputeSavingsPlans'
```

### 4. Azure 账单同步 - `sync_azure()`

**多区域支持**:
- 🇨🇳 中国区 (`azure_billing_cost_cn*`)
- 🇺🇸 美国区 (`azure_billing_cost_us*`)

**处理逻辑**:
```python
def sync_azure(sdate, cli):
    sync_azure_by_region(sdate, cli, 'cn')   # 中国区
    sync_azure_by_region(sdate, cli, 'us')   # 美国区

def sync_azure_by_region(sdate, cli, region):
    provider_code = 7 if region == 'cn' else 4  # 不同区域对应不同 providerID
    
    # 并发处理多个订阅
    with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
        # ... 处理逻辑
```

## 📊 数据流程分析

### 数据处理流程
```mermaid
graph TD
    A[开始执行] --> B[获取45天前日期]
    B --> C[删除MySQL中旧数据]
    C --> D[连接BigQuery]
    D --> E[并发处理AWS账户]
    D --> F[处理GCP项目]
    D --> G[处理Azure订阅]
    E --> H[构建AWS费用数据]
    F --> I[构建GCP费用数据]
    G --> J[构建Azure费用数据]
    H --> K[批量插入MySQL]
    I --> K
    J --> K
    K --> L[完成同步]
```

### 数据库表结构

**MySQL 目标表**: `cloud_product_cost`
| 字段名 | 类型 | 说明 |
|--------|------|------|
| createdAt | DATETIME | 记录创建时间 |
| resourceID | INT | 内部资源ID（外键） |
| productName | VARCHAR | 云产品/服务名称 |
| cost | DECIMAL | 费用金额 |
| date | DATE | 费用发生日期 |

**资源映射表**: `cloud_resource`
| 字段名 | 类型 | 说明 |
|--------|------|------|
| id | INT | 内部资源ID |
| cloudInsID | VARCHAR | 云厂商实例ID |
| providerID | INT | 云厂商标识（1=AWS, 2=GCP, 4=Azure US, 7=Azure CN） |
| classType | VARCHAR | 资源类型 |
| isDel | TINYINT | 是否删除 |

## 🚀 技术特点

### 1. 性能优化
- **并发处理**: AWS 和 Azure 使用线程池并发处理多个账户/订阅
- **增量同步**: 只处理45天内数据，避免全量处理
- **批量插入**: 使用 `executemany()` 批量插入数据库

### 2. 错误处理
```python
try:
    cursor.executemany("INSERT INTO ...", data_to_insert)
    cnx.commit()
except Exception as e:
    cnx.rollback()  # 事务回滚
    raise e
```

### 3. 资源管理
```python
# 使用 context manager 确保连接正确关闭
with get_connection() as cnx:
    with cnx.cursor() as cursor:
        # 数据库操作
```

### 4. 配置管理
```python
# 通过环境变量管理敏感信息
conf = {
    'user': os.getenv("username"),
    'password': os.getenv("password"),
    'host': os.getenv("host"),
    'database': os.getenv("database"),
}
```

## 📈 业务价值

### 1. 成本可视化
- 🔍 统一查看三大云厂商费用
- 📊 按产品、日期、资源维度分析
- 📈 识别成本趋势和异常

### 2. 运营效率
- ⏰ 自动化数据收集，减少人工操作
- 🔄 每12小时更新，保证数据时效性
- 🛡️ 错误重试机制，提高系统可靠性

### 3. 决策支持
- 💡 为财务团队提供准确成本数据
- 🎯 支持预算规划和成本优化
- 📋 生成各类成本报表

## ⚠️ 潜在问题与改进建议

### 1. 代码质量问题
```python
# 当前问题：
import datetime
from datetime import datetime, timedelta  # 重复导入

# 建议改进：
from datetime import datetime, timedelta
```

### 2. 硬编码问题
```python
# 当前：硬编码的表名和 SQL
table = "`gcp_billing_v2.gcp_billing_export_v1_01A169_D3D9BF_FB82A2`"

# 建议：配置化管理
GCP_BILLING_TABLE = os.getenv("GCP_BILLING_TABLE", "default_table")
```

### 3. 错误处理增强
```python
# 建议添加更详细的日志记录
import logging

def sync_gcp(after_date, cli):
    try:
        logging.info(f"开始同步GCP数据，起始日期：{after_date}")
        # ... 处理逻辑
        logging.info(f"GCP数据同步完成，共处理 {count} 条记录")
    except Exception as e:
        logging.error(f"GCP数据同步失败：{str(e)}")
        raise
```

### 4. 性能优化建议
- 🔄 考虑使用异步处理框架（如 asyncio）
- 📝 添加数据处理进度监控
- 🚀 优化 BigQuery 查询性能
- 💾 考虑使用连接池管理数据库连接

### 5. 监控告警
- 📊 添加数据质量检查
- 🔔 异常情况邮件/企微告警
- 📈 处理时长监控

## 🔍 使用示例

### 手动执行测试
```python
# 在 main 部分
if __name__ == '__main__':
    dag.test()  # 测试整个 DAG
```

### 查看同步状态
```sql
-- 查看最近同步的数据
SELECT 
    DATE(date) as cost_date,
    COUNT(*) as record_count,
    SUM(cost) as total_cost
FROM cloud_product_cost 
WHERE createdAt >= DATE_SUB(NOW(), INTERVAL 1 DAY)
GROUP BY DATE(date)
ORDER BY cost_date DESC;
```

### 按云厂商查看费用
```sql
-- 通过 resourceID 关联查看各云厂商费用
SELECT 
    CASE cr.providerID 
        WHEN 1 THEN 'AWS'
        WHEN 2 THEN 'GCP' 
        WHEN 4 THEN 'Azure-US'
        WHEN 7 THEN 'Azure-CN'
    END as provider,
    DATE(cpc.date) as cost_date,
    SUM(cpc.cost) as daily_cost
FROM cloud_product_cost cpc
JOIN cloud_resource cr ON cpc.resourceID = cr.id
WHERE cpc.date >= DATE_SUB(CURRENT_DATE, INTERVAL 7 DAY)
GROUP BY provider, DATE(cpc.date)
ORDER BY cost_date DESC, daily_cost DESC;
```

## 📚 相关文档

- [Apache Airflow 官方文档](https://airflow.apache.org/docs/)
- [Google Cloud BigQuery Python Client](https://cloud.google.com/bigquery/docs/reference/libraries)
- [MySQL Connector/Python](https://dev.mysql.com/doc/connector-python/en/)

---

**维护信息**:
- 创建时间: 2024年
- 维护团队: PingCAP DevOps Team
- 联系人: xiaopeng.zhu@pingcap.com