flowchart LR
    A[前端发起导出] --> B[API 网关]
    B --> C[业务服务<br/>校验权限/参数]
    C --> D[写入任务记录表<br/>status=PENDING]
    C --> E[推送任务到 Broker]
    E --> F[Worker 消费<br/>并发生成报表]
    F --> G[写入对象存储<br/>生成临时下载链接]
    G --> H[更新任务记录<br/>status=SUCCESS/FAILED]
    H --> I[WebSocket/轮询通知前端]
    I --> J[前端下载文件]

# 伪代码：分片任务分发
def dispatch_export_chunks(main_task_id, meeting_id, filters, chunk_size=50000):
    total = count_meeting_records(meeting_id, filters)
    chunks = (total + chunk_size - 1) // chunk_size
    for i in range(chunks):
        offset = i * chunk_size
        chunk_task = ExportChunkTask(
            main_task_id=main_task_id,
            chunk_index=i,
            offset=offset,
            limit=chunk_size,
            filters=filters
        )
        app.send_task('export.chunk.generate', args=[chunk_task.to_dict()], 
                      queue='export.chunk', priority=5)

@shared_task(bind=True, max_retries=3, autoretry_for=(TransientError,))
def generate_chunk(self, task_data):
    task_id = task_data['main_task_id']
    if redis_client.get(f'export:cancel:{task_id}'):
        update_task_status(task_id, 'CANCELLED')
        cleanup_temp_files(task_data)
        return {'status': 'cancelled'}
    # ... 正常生成逻辑

# docker-compose / K8s Deployment 示例
services:
  worker-cpu:      # 计算密集型：分片生成、数据聚合、Excel 渲染
    command: celery -A proj worker -Q export.chunk,export.merge -c 7 --prefetch-multiplier=1 -l INFO
    deploy:
      resources:
        limits:
          cpus: '8'
          memory: 8G
        requests:
          cpus: '4'
          memory: 4G

  worker-io:       # IO 密集型：OSS 上传、Webhook 回调、数据库状态回写
    command: celery -A proj worker -Q export.upload,export.callback -c 500 -P gevent -l INFO
    deploy:
      resources:
        limits:
          cpus: '2'
          memory: 2G

# proj/tasks/base.py
from celery import Task
from celery.signals import task_prerun, task_postrun, task_failure
import structlog
import time
from contextvars import ContextVar

# 上下文变量传递 TraceID / TenantID / UserID
trace_id_var: ContextVar[str] = ContextVar('trace_id', default='')
tenant_id_var: ContextVar[int] = ContextVar('tenant_id', default=0)

logger = structlog.get_logger(__name__)

class BaseExportTask(Task):
    """所有导出任务基类：自动化上下文注入、耗时埋点、异常标准化"""
    abstract = True
    autoretry_for = (TransientExportError,)
    retry_backoff = True
    retry_backoff_max = 600  # 最大退避 10 分钟
    retry_jitter = True
    max_retries = 3

    # 任务级别的资源限流信号量（Redis 实现分布式信号量）
    _rate_limit_semaphore = None 

    def __call__(self, *args, **kwargs):
        # 1. 提取上下文
        ctx = kwargs.pop('_export_ctx', {})
        trace_id_var.set(ctx.get('trace_id', ''))
        tenant_id_var.set(ctx.get('tenant_id', 0))
        
        # 2. 租户级并发限流检查 (令牌桶)
        if not self._acquire_tenant_slot(tenant_id_var.get()):
            raise TenantQuotaExceeded(f"Tenant {tenant_id_var.get()} export concurrency limit reached")
        
        start_time = time.monotonic()
        try:
            return super().__call__(*args, **kwargs)
        finally:
            # 3. 释放槽位 & 记录慢任务
            self._release_tenant_slot(tenant_id_var.get())
            duration = time.monotonic() - start_time
            if duration > 60:  # 超过 60s 记录慢任务日志
                logger.warning("slow_export_task", task=self.name, duration=duration, **ctx)

    def _acquire_tenant_slot(self, tenant_id: int) -> bool:
        # Redis Lua 脚本原子执行：检查配额 + 扣减
        # KEYS[1]=quota_key, ARGV[1]=limit, ARGV[2]=ttl
        lua_script = """
        local current = redis.call('GET', KEYS[1])
        if current and tonumber(current) >= tonumber(ARGV[1]) then return 0 end
        local val = redis.call('INCR', KEYS[1])
        if val == 1 then redis.call('EXPIRE', KEYS[1], ARGV[2]) end
        return 1
        """
        return redis_client.eval(lua_script, 1, f"export:quota:{tenant_id}", 
                                 settings.EXPORT_TENANT_CONCURRENCY_LIMIT, 3600)

    def _release_tenant_slot(self, tenant_id: int):
        redis_client.decr(f"export:quota:{tenant_id}")

# 信号量自动埋点：接入 OpenTelemetry / SkyWalking
@task_prerun.connect
def on_task_prerun(sender=None, task_id=None, task=None, args=None, kwargs=None, **_):
    # 启动 Span、注入 Baggage
    pass

@task_failure.connect
def on_task_failure(sender=None, exception=None, traceback=None, **_):
    # 统一错误码映射、告警路由
    pass

# tasks/export_merge.py
import pyarrow.parquet as pq
import pyarrow as pa
import xlsxwriter
import io

@app.task(bind=True, base=BaseExportTask, queue='export.merge')
def merge_parquet_to_excel(self, main_task_id: str, chunk_paths: list, output_filename: str):
    """
    流式合并：不将全量数据加载内存，利用 PyArrow Scanner 迭代器 + xlsxwriter 顺序写入
    """
    # 1. 打开 OSS 输出流 (假设使用 MinIO/S3 SDK 支持的 upload_fileobj)
    output_buffer = io.BytesIO()  # 或直接包装 OSS 上传流
    workbook = xlsxwriter.Workbook(output_buffer, {'in_memory': True, 'constant_memory': True})
    worksheet = workbook.add_worksheet('MeetingData')
    
    # 2. 写表头 (从第一个分片 Schema 获取)
    first_schema = pq.read_schema(chunk_paths[0])
    headers = first_schema.names
    for col_idx, header in enumerate(headers):
        worksheet.write(0, col_idx, header)
    
    # 3. 流式合并写入
    row_idx = 1
    for chunk_path in chunk_paths:
        # 使用 PyArrow Scanner 分批读取 (batch_size=65536)
        scanner = pq.ParquetFile(chunk_path).iter_batches(batch_size=65536)
        for batch in scanner:
            # batch 是 pyarrow.Table，转 numpy/pandas 极快，零拷贝
            df_chunk = batch.to_pandas(types_mapper=pd.ArrowDtype) 
            # 批量写入 xlsxwriter (write_row 循环较慢，建议用 worksheet.write_rows 或 pandas to_excel 但需注意内存)
            # 此处伪代码演示核心：流式写入，内存不随数据量增长
            for _, row in df_chunk.iterrows():
                worksheet.write_row(row_idx, 0, row.tolist())
                row_idx += 1
    
    workbook.close()
    # 4. 上传最终文件
    output_buffer.seek(0)
    final_url = oss_client.upload_fileobj(output_buffer, f"exports/{main_task_id}/{output_filename}")
    return {'final_url': final_url, 'format': 'xlsx', 'rows': row_idx - 1}

graph TD
    A[API 网关] --> B{租户分级<br/>Platinum/Gold/Standard/Free}
    B -->|Platinum| C[专属 Worker Pool<br/>预留 8 核<br/>高优先级队列]
    B -->|Gold| D[共享高优池<br/>权重 50%<br/>限并发 5 任务/租户]
    B -->|Standard/Free| E[共享低优池<br/>权重 20%<br/>限并发 1 任务/租户<br/>单日总导出行数配额]
    
    C --> F[Redis 租户配额桶<br/>Token Bucket 算法]
    D --> F
    E --> F
    F --> G[Celery Worker Autoscaler<br/>基于队列积压 + 租户权重扩缩容]

# proj/router.py
class TenantAwareRouter:
    def route_for_task(self, task, args, kwargs, options, **_):
        ctx = kwargs.get('_export_ctx', {})
        tier = ctx.get('tenant_tier', 'free')
        
        if tier == 'platinum':
            return {'queue': 'export.platinum', 'priority': 9}
        elif tier == 'gold':
            return {'queue': 'export.gold', 'priority': 5}
        else:
            return {'queue': 'export.standard', 'priority': 1}

# proj/autoscaler.py (自定义 Autoscaler 继承 celery.worker.autoscale.Autoscaler)
class TenantFairAutoscaler(Autoscaler):
    def _get_worker_queue_load(self):
        """重写负载计算：不仅看队列长度，看『加权积压任务数』"""
        inspect = self.app.control.inspect()
        active_queues = inspect.active_queues()
        # 获取各队列积压
        reserved = inspect.reserved() or {}
        scheduled = inspect.scheduled() or {}
        
        weighted_load = 0
        for q_name, tasks in reserved.items():
            weight = QUEUE_WEIGHTS.get(q_name, 1) # platinum=3, gold=2, standard=1
            weighted_load += len(tasks) * weight
        # ... 结合 scheduled 计算总负载
        return weighted_load
    
    def _maybe_scale(self):
        # 扩容判断：加权负载 / 当前进程数 > 阈值
        # 缩容判断：空闲 Worker > N 且 低优队列无积压
        pass

# proj/telemetry.py
from opentelemetry import trace
from opentelemetry.trace import SpanKind
from celery.signals import before_task_publish, task_prerun, task_postrun, task_failure

tracer = trace.get_tracer("meeting.export")

@before_task_publish.connect
def inject_trace_context(headers=None, **kwargs):
    """Producer 端：注入 Trace Context 到消息 Header"""
    if headers is None: headers = {}
    carrier = {}
    trace.get_current_span().get_span_context() # 获取当前 SpanContext
    trace.propagate.inject(carrier)
    headers.update(carrier) # 传给 Broker

@task_prerun.connect
def start_consumer_span(sender=None, task_id=None, task=None, args=None, kwargs=None, **_):
    """Consumer 端：从 Header 提取上下文，启动 Consumer Span"""
    ctx = kwargs.get('_export_ctx', {})
    # 从 headers 恢复 context (Celery 会自动将 headers 放入 request.headers)
    parent_context = trace.propagate.extract(sender.request.headers) 
    span_name = f"celery.task.{task.name}"
    span = tracer.start_span(span_name, context=parent_context, kind=SpanKind.CONSUMER)
    span.set_attribute("celery.task_id", task_id)
    span.set_attribute("tenant.id", ctx.get('tenant_id'))
    # 存入任务实例供后续使用
    sender.request._otel_span = span

@task_postrun.connect
def end_consumer_span(sender=None, task_id=None, task=None, args=None, kwargs=None, retval=None, state=None, **_):
    span = getattr(sender.request, '_otel_span', None)
    if span:
        span.set_attribute("celery.task_state", state)
        if isinstance(retval, dict):
            span.set_attribute("export.rows", retval.get('rows', 0))
            span.set_attribute("export.format", retval.get('format', ''))
        span.end()

@task_failure.connect
def record_exception(sender=None, exception=None, traceback=None, **_):
    span = getattr(sender.request, '_otel_span', None)
    if span:
        span.record_exception(exception)
        span.set_status(trace.Status(trace.StatusCode.ERROR, str(exception)))

// 统一日志格式 (JSON Lines)
{
  "timestamp": "2023-10-25T10:00:00.123Z",
  "level": "INFO",
  "logger": "tasks.export.chunk",
  "trace_id": "a1b2c3d4e5f6",
  "span_id": "1234567890abcdef",
  "tenant_id": 10086,
  "task_id": "celery-task-uuid-...",
  "chunk_index": 3,
  "event": "chunk_generated",
  "metrics": {
    "rows": 50000,
    "duration_ms": 1250,
    "memory_peak_mb": 180,
    "output_size_bytes": 4500000
  }
}

## 故障复盘报告：[日期] 会议导出服务大面积超时/失败

### 1. 故障现象
- 时间范围：2023-10-25 14:00 - 14:45
- 影响范围：Standard 租户导出成功率跌至 60%，P99 入队延迟 120s
- 客户投诉单：12 单

### 2. 时间线
- 14:00 监控告警 `export_queue_lag > 5000`
- 14:05 值班同学登录 Flower，发现 `worker-cpu` 仅 3/16 存活，其余 OOMKilled
- 14:10 执行手动扩容 + 杀掉卡死任务
- 14:25 队列消化完毕，成功率恢复

### 3. 根因分析 (5 Why)
1. **直接原因**：Worker 进程 OOM Kill 导致并发度骤降。
2. **为什么 OOM**：单个 Chunk 任务处理 50 万行宽表 (200 列)，Pandas `read_sql` 未分批，内存峰值 4GB+。
3. **为什么未分批**：代码逻辑 `pd.read_sql(query, conn)` 无 `chunksize` 参数；Code Review 未覆盖大宽表场景。
4. **为什么测试未发现**：压测数据仅 10 列，未模拟宽表场景；无内存 Profile 基线对比。
5. **系统性缺陷**：缺乏「大宽表」专项测试用例；Worker 缺乏进程级内存硬限制 (`memory_limit` cgroup) 保护机制。

### 4. 改进措施 (Action Items)
| 编号 | 措施 | 负责人 | 截止日期 | 状态 |
|------|------|--------|----------|------|
| ACT-01 | 修改 Chunk Task：强制 `chunksize=10000` 流式读取 + `dtype` 指定压缩类型 | @Backend-Lead | 10/27 | Done |
| ACT-02 | K8s Deployment 增加 `resources.limits.memory: 6Gi` + `livenessProbe` 检查内存 | @SRE-Lead | 10/26 | Done |
| ACT-03 | CI 集成 `memray` 内存基线测试，PR 必检内存峰值 < 500MB | @QA-Lead | 11/01 | Doing |
| ACT-04 | 补充宽表/长文本/高基数字段压测场景 | @Perf-Team | 11/10 | Todo |

### 5. 经验沉淀
- 更新《导出任务开发规范 v2.1》：新增「大宽表处理模式」、「内存 Profile 必做项」。
- 将「Chunk 分片大小」纳入配置中心动态下发，无需发版调整。

# proj/celery_config.py
from kombu import Queue, Exchange

# Broker & Backend
broker_url = 'redis://:password@redis-cluster:6379/0'  # 生产建议 Redis Cluster
result_backend = 'redis://:password@redis-cluster:6379/1'
result_expires = 86400  # 结果保留 24h
result_compression = 'zstd'

# 序列化
task_serializer = 'json'
result_serializer = 'json'
accept_content = ['json']
timezone = 'Asia/Shanghai'
enable_utc = True

# 队列定义 (Topic Exchange 支持通配符路由)
task_queues = (
    Queue('export.platinum', Exchange('export', type='topic'), routing_key='export.platinum.#', priority=10),
    Queue('export.gold', Exchange('export', type='topic'), routing_key='export.gold.#', priority=5),
    Queue('export.standard', Exchange('export', type='topic'), routing_key='export.standard.#', priority=1),
    Queue('export.chunk', Exchange('export', type='topic'), routing_key='export.chunk.#', priority=3),
    Queue('export.merge', Exchange('export', type='topic'), routing_key='export.merge.#', priority=2),
    Queue('export.upload', Exchange('export', type='topic'), routing_key='export.upload.#', priority=1),
    Queue('export.dlx', Exchange('export.dlx', type='direct'), routing_key='export.dlx'), # 死信队列
)

# 路由
task_routes = {
    'tasks.export.full_report': {'queue': 'export.platinum', 'priority': 9},
    'tasks.export.chunk.generate': {'queue': 'export.chunk', 'priority': 3},
    'tasks.export.merge': {'queue': 'export.merge', 'priority': 2},
    'tasks.export.upload': {'queue': 'export.upload', 'priority': 1},
}

# Worker 优化
worker_prefetch_multiplier = 1  # 关键：长任务防抢占
worker_max_tasks_per_child = 100  # 防内存泄漏，每 100 任务重启进程
worker_disable_rate_limits = False
task_acks_late = True  # 任务执行完成后再 ACK，防止 Worker 崩溃丢任务
task_reject_on_worker_lost = True  # Worker 失联时任务重新入队
task_track_started = True  # 任务开始时更新状态为 STARTED

# 监控
worker_send_task_events = True
task_send_sent_event = True

# 死信配置 (需 RabbitMQ 或 Redis Stream 支持，Redis 列表模式需手动实现)
# task_queues 中定义 dlx，任务声明时指定 queue_arguments={'x-dead-letter-exchange': 'export.dlx'}

# k8s/worker-cpu-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: celery-worker-cpu
spec:
  replicas: 3
  selector:
    matchLabels:
      app: celery-worker-cpu
  template:
    spec:
      containers:
      - name: worker
        image: my-registry/export-worker:v1.2.3
        command: ["celery", "-A", "proj", "worker", "-Q", "export.chunk,export.merge", "-c", "7", "--prefetch-multiplier=1", "-l", "INFO", "--autoscale=10,3"] # max 10, min 3
        resources:
          requests:
            cpu: "3500m"
            memory: "6Gi"
          limits:
            cpu: "7000m"
            memory: "8Gi"
        env:
        - name: PYTHONUNBUFFERED
          value: "1"
        - name: OTEL_EXPORTER_OTLP_ENDPOINT
          value: "http://otel-collector:4317"
---
# k8s/hpa-worker-cpu.yaml
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: celery-worker-cpu-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: celery-worker-cpu
  minReplicas: 3
  maxReplicas: 30
  metrics:
  - type: External
    external:
      metric:
        name: celery_queue_messages_ready
        selector:
          matchLabels:
            queue: "export.chunk"
      target:
        type: AverageValue
        averageValue: "50" # 每个 Pod 积压 50 个分片任务触发扩容
  behavior:
    scaleDown:
      stabilizationWindowSeconds: 600 # 缩容冷却 10 分钟
      policies:
      - type: Percent
        value: 10
        periodSeconds: 60
    scaleUp:
      stabilizationWindowSeconds: 60
      policies:
      - type: Percent
        value: 100
        periodSeconds: 30
      - type: Pods
        value: 4
        periodSeconds: 30
      selectPolicy: Max