K8s Pod缩容导致Celery任务被终止，如何实现任务重试？

我们使用Celery(v5.4.0)处理CPU密集型长任务，以Redis作为消息代理（broker）和结果后端（backend），部署在基于CPU利用率自动扩缩容的Kubernetes集群中。

遇到的核心问题：

• 当Pod因低利用率触发缩容时，部分正在Pod内执行的任务被直接终止，且未自动在活跃Pod上重新入队。

已尝试的优化措施：

• 将Kubernetes的terminationGracePeriodSeconds设置为较大值，但仍有部分执行时间超长的异常任务被终止。
• 在任务级别配置了acks_late=True和reject_on_worker_lost=True，期望任务完成后再确认、Pod异常退出时任务能重新入队，但该配置未达到预期效果。

测试验证场景：
我们将Celery Worker运行在2个Docker容器中，在其中一个容器执行任务时手动杀死它，结果因SIGKILL终止的任务并未在健康容器上重新入队。

任务示例代码：

@celery_worker.task(
    name="primitive-task",
    bind=True,
    ignore_result=False,
    reject_on_worker_lost=True,
    acks_late=True,
)
def _add_task(self, x: int, y: int, task_id: int) -> int:
    sleep_time = random.randint(5, 10)

    print(
        f"[PID: {os.getpid()}]add task sleeping for {sleep_time} seconds for"
        f" {x=} | {y=} | {task_id=}"
    )
    time.sleep(sleep_time)

    print(f"result for {x=} | {y=} is {x+y} | {task_id=}")
    return x + y

1. 让Celery Worker优雅处理Kubernetes终止信号 #

Kubernetes缩容时会先发送SIGTERM信号，等待terminationGracePeriodSeconds时长后发送SIGKILL。Celery默认不会优雅响应SIGTERM，需要启动Worker时配置超时参数：

• --soft-time-limit：设置任务允许的最大执行时长，超时后Worker会发送SIGUSR1让任务主动退出并重新入队
• --time-limit：硬超时限制，确保超时任务被强制终止
  示例启动命令：

celery -A your_app worker --loglevel=info --soft-time-limit=300 --time-limit=360 --acks-late --reject-on-worker-lost

2. 调整Redis代理的可见性超时 #

Redis作为Broker时，任务被Worker取走后会进入“不可见”状态，默认超时为300秒。如果任务执行时长超过这个值，Redis会自动将任务放回队列；若Worker被SIGKILL杀死，也需要该超时确保任务能重新入队。
在Celery配置中添加：

CELERY_BROKER_TRANSPORT_OPTIONS = {
    'visibility_timeout': 7200  # 2小时，需大于任务最长预期执行时间
}

3. 配置Kubernetes Pod的PreStop钩子 #

给Worker Pod添加PreStop钩子，在收到终止信号时主动通知Celery停止接受新任务，并等待正在运行的任务完成（或超时）：

apiVersion: apps/v1
kind: Deployment
metadata:
  name: celery-worker
spec:
  template:
    spec:
      containers:
      - name: worker
        image: your-worker-image
        lifecycle:
          preStop:
            exec:
              command: ["celery", "-A", "your_app", "control", "shutdown"]
      terminationGracePeriodSeconds: 3600  # 设置为覆盖最长任务执行时间的数值

该钩子会让Celery Worker优雅关闭，不再接收新任务，同时等待现有任务完成。若任务在超时时间内未完成，Kubernetes仍会发送SIGKILL，但结合Redis可见性超时，任务会被重新入队。

4. 确保reject_on_worker_lost全局生效 #

reject_on_worker_lost=True需要在Worker级别和任务级别同时配置才能确保生效，建议启动Worker时也加上--reject-on-worker-lost参数，避免仅任务级别配置导致的失效。

5. 任务级别添加重试兜底逻辑 #

在任务中显式添加重试机制，即使任务被意外终止，也能通过重试逻辑重新执行：

@celery_worker.task(
    name="primitive-task",
    bind=True,
    ignore_result=False,
    reject_on_worker_lost=True,
    acks_late=True,
    max_retries=3,
    retry_backoff=True,
)
def _add_task(self, x: int, y: int, task_id: int) -> int:
    try:
        sleep_time = random.randint(5, 10)
        print(
            f"[PID: {os.getpid()}]add task sleeping for {sleep_time} seconds for"
            f" {x=} | {y=} | {task_id=}"
        )
        time.sleep(sleep_time)
        print(f"result for {x=} | {y=} is {x+y} | {task_id=}")
        return x + y
    except Exception as e:
        self.retry(exc=e)

内容的提问来源于stack exchange，提问作者radioactive11