如何利用K8s lifecycle preStop让Python应用Pod完成任务后再终止？

要解决Pod在未完成请求时被终止的问题，需要从应用改造和K8s配置调整两方面入手，确保Pod能优雅处理现有请求后再退出，同时结合CPU/内存指标控制缩容时机：

一、Python应用层面：捕获信号并优雅收尾 #

修改Python代码，捕获K8s发送的SIGTERM信号（Pod终止时的默认信号），停止接收新的SQS消息，等待当前所有请求/任务处理完成后再退出。

示例代码（同步场景） #

import signal
import time
import boto3
from threading import Event

# 全局标志：是否收到终止信号
shutdown_event = Event()

def handle_sigterm(signum, frame):
    print("收到SIGTERM信号，停止接收新任务，处理完现有任务后退出")
    shutdown_event.set()

# 注册信号处理函数
signal.signal(signal.SIGTERM, handle_sigterm)

def process_sqs_message(message):
    # 替换为你的实际业务处理逻辑
    print(f"处理消息: {message.body}")
    time.sleep(10)  # 模拟耗时任务
    message.delete()

def main():
    sqs = boto3.resource('sqs')
    queue = sqs.get_queue_by_name(QueueName='your-queue-name')
    
    while not shutdown_event.is_set():
        # 拉取SQS消息（可调整为长轮询优化性能）
        messages = queue.receive_messages(MaxNumberOfMessages=10, WaitTimeSeconds=5)
        for msg in messages:
            if shutdown_event.is_set():
                # 已触发终止，不再处理新拉取的消息
                break
            process_sqs_message(msg)
    
    print("所有任务处理完成，应用退出")

if __name__ == "__main__":
    main()

异步/框架场景（如FastAPI、Celery） #

• FastAPI：使用关闭钩子停止接收新请求，等待现有请求处理完成：

  from fastapi import FastAPI
  import asyncio
  
  app = FastAPI()
  shutdown_flag = False
  
  @app.on_event("shutdown")
  async def shutdown_event():
      global shutdown_flag
      shutdown_flag = True
      print("启动优雅关闭流程，等待现有请求处理完成")
      # 替换为实际等待异步任务完成的逻辑
      await asyncio.sleep(30)
  
  @app.get("/process-sqs")
  async def process_sqs():
      if shutdown_flag:
          return {"status": "应用正在关闭，不再接收新请求"}
      # 替换为你的SQS消息处理逻辑
      return {"status": "处理中"}
  

• Celery：通过--soft-time-limit和--time-limit配置，结合信号处理实现Worker优雅关闭，确保任务完成后再退出。

二、K8s配置调整 #

1. 设置Pod优雅终止时间 #

在Deployment的Pod模板中配置terminationGracePeriodSeconds，给应用足够的时间处理完现有请求，值需大于你的任务最长处理时间：

apiVersion: apps/v1
kind: Deployment
metadata:
  name: your-app-deployment
spec:
  replicas: 1
  template:
    spec:
      terminationGracePeriodSeconds: 300  # 根据任务最长耗时调整
      containers:
      - name: your-app-container
        image: your-app-image:latest
        # 其他容器配置

2. 调整HPA配置：结合SQS消息数与CPU/内存指标 #

原有HPA仅基于SQS消息数扩容缩容，现在添加CPU和内存使用率作为补充指标，确保消息数为0但仍有任务在处理时，不会立即缩容Pod：

apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: your-app-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: your-app-deployment
  minReplicas: 0
  maxReplicas: 10
  metrics:
  # 原有SQS消息数指标（需提前通过外部指标暴露）
  - type: External
    external:
      metric:
        name: sqs_queue_messages_ready
        selector:
          matchLabels:
            queue: your-queue-name
      target:
        type: AverageValue
        averageValue: 50  # 消息数超过50时扩容
  # CPU使用率指标：低于70%时允许缩容
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 70
  # 内存使用率指标：低于75%时允许缩容
  - type: Resource
    resource:
      name: memory
      target:
        type: Utilization
        averageUtilization: 75
  # 配置缩容延迟，避免负载波动导致频繁缩容
  behavior:
    scaleDown:
      stabilizationWindowSeconds: 300
      policies:
      - type: Percent
        value: 50
        periodSeconds: 60

3. 可选：使用PreStop钩子（补充机制） #

如果应用对信号处理不敏感，可添加PreStop钩子主动通知应用启动收尾流程：

containers:
- name: your-app-container
  image: your-app-image:latest
  lifecycle:
    preStop:
      exec:
        command: ["/bin/sh", "-c", "curl -X POST http://localhost:8000/shutdown"]

关键注意事项 #

• 确保应用处理SIGTERM时不会立即退出，必须等待所有现有任务完成。
• terminationGracePeriodSeconds的值要大于任务最长处理时间，避免K8s强制杀死未完成任务的Pod。
• HPA的缩容延迟需设置合理，防止短时间内负载波动导致误缩容。

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