AWS Lambda部署FastAPI出现INIT_REPORT初始化超时问题求助

问题描述 #

在AWS Lambda上部署基于Python开发的FastAPI REST API时，遭遇初始化超时，访问API收到如下错误：

INIT_REPORT Init Duration: 10005.81 ms Phase: init Status: timeout
Init Duration: 9994.93 ms Phase: init Status: timeout(Request Time Out)

测试端点时同样返回请求超时。

代码实现 #

from fastapi import FastAPI, HTTPException, BackgroundTasks
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
import logging
import time
import os
import tempfile
import pydicom
import numpy as np
from PIL import Image
import firebase_admin
from firebase_admin import credentials, storage, db
from firebase_admin.exceptions import FirebaseError
from concurrent.futures import ThreadPoolExecutor
from ultralytics import YOLO
from threading import Lock
import asyncio
from aiofiles import open as aio_open
from aiofiles.os import remove as aio_remove
from mangum import Mangum  # Import Mangum for AWS Lambda

app = FastAPI()

# 启用CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 配置日志
logging.basicConfig(level=logging.INFO)

# 初始化线程池执行器
executor = ThreadPoolExecutor(max_workers=20)  # 增加工作线程数

# 存储处理状态的字典
processing_status = {}
processing_progress = {}

# Firebase初始化的线程安全锁
firebase_lock = Lock()

class PathData(BaseModel):
    path: str
    adminName1: str
    patientId: str

def initialize_firebase():
    with firebase_lock:
        if not firebase_admin._apps:
            cred = credentials.Certificate(r'Run AI backend\firebase-key.json')
            firebase_admin.initialize_app(cred, {
                'storageBucket': 'storagepath',
                'databaseURL': 'databaseurl'
            })

@app.get("/health")
async def health_check():
    return {"status": "ok"}

@app.post("/get_path")
async def get_path(data: PathData, background_tasks: BackgroundTasks):
    logging.info(f"接收数据: {data}")
    if not data.path or not data.adminName1 or not data.patientId:
        raise HTTPException(status_code=400, detail="未提供路径、adminName1或patientId")
    
    folder_path = data.path
    admin_name = data.adminName1
    patient_id = data.patientId
    logging.info(f"接收的文件夹路径: {folder_path}")
    
    # 初始化Firebase Admin SDK
    try:
        initialize_firebase()
    except FirebaseError as e:
        logging.error(f"Firebase初始化错误: {e}")
        raise HTTPException(status_code=500, detail="Firebase初始化失败")
    
    bucket = storage.bucket()

    blobs = bucket.list_blobs(prefix=folder_path)
    dcm_files = [blob.name for blob in blobs if blob.name.endswith('.dcm')]
    
    if not dcm_files:
        logging.error(f"指定路径下未找到DICOM文件: {folder_path}")
        raise HTTPException(status_code=404, detail="指定路径下未找到DICOM文件")
    
    dcm_file_path = dcm_files[0]
    logging.info(f"待处理的DICOM文件: {dcm_file_path}")

    # 初始化状态和进度
    task_id = f"{admin_name}_{patient_id}"
    processing_status[task_id] = {'status': '处理已启动'}
    processing_progress[task_id] = 0  # 初始进度0%

    background_tasks.add_task(process_dicom_file, dcm_file_path, folder_path, os.path.basename(dcm_file_path), admin_name, patient_id, task_id, bucket)
    
    return {"status": "处理已启动", "task_id": task_id}

@app.get("/status/{task_id}")
async def get_status(task_id: str):
    status = processing_status.get(task_id, {'status': '未知任务ID'})
    return status

@app.get("/progress/{task_id}")
async def get_progress(task_id: str):
    progress = processing_progress.get(task_id, 0)
    return {"progress": progress}

async def process_dicom_file(dcm_file_path, original_folder_path, original_filename, admin_name, patient_id, task_id, bucket):
    # 初始化YOLO模型
    model = YOLO(r'Run AI backend\yolov8_segmentation_fractureAtlas.pt')

    blob = bucket.blob(dcm_file_path)
    with tempfile.NamedTemporaryFile(delete=False) as temp_file:
        blob.download_to_filename(temp_file.name)
        temp_dcm_path = temp_file.name
    
    processing_progress[task_id] = 10

    ds = pydicom.dcmread(temp_dcm_path)
    pixel_array = ds.pixel_array
    image = Image.fromarray(pixel_array).convert("L")
    with tempfile.NamedTemporaryFile(delete=False, suffix='.png') as temp_image_file:
        image.save(temp_image_file.name)
        image_path = temp_image_file.name

    processing_progress[task_id] = 30

    processing_status[task_id]['status'] = '正在对X光片运行AI检测'
    await asyncio.sleep(2)

    results = model(image_path, conf=0.25)
    prediction = results[0]
    fracture_detected = len(prediction.boxes) > 0
    message = "阳性" if fracture_detected else "阴性"

    plot_array = prediction.plot()
    plot_img = Image.fromarray(np.uint8(plot_array))

    processed_pixel_array = np.array(plot_img)
    if len(processed_pixel_array.shape) == 3 and processed_pixel_array.shape[2] == 3:
        processed_pixel_array = np.mean(processed_pixel_array, axis=2).astype(ds.pixel_array.dtype)
    ds.PixelData = processed_pixel_array.tobytes()
    with tempfile.NamedTemporaryFile(delete=False, suffix='.dcm') as temp_processed_dcm_file:
        ds.save_as(temp_processed_dcm_file.name)
        processed_dcm_path = temp_processed_dcm_file.name

    processing_progress[task_id] = 70

    processed_folder_path = os.path.join(original_folder_path, 'AI_Results').replace("\\", "/")
    processed_blob_path = os.path.join(processed_folder_path, original_filename).replace("\\", "/")
    logging.info(f"处理后的DICOM文件将上传至: {processed_blob_path}")
    
    processing_status[task_id]['status'] = '正在上传生成结果'
    await asyncio.sleep(2)

    processed_blob = bucket.blob(processed_blob_path)
    processed_blob.upload_from_filename(processed_dcm_path)

    await aio_remove(image_path)
    await aio_remove(temp_dcm_path)
    await aio_remove(processed_dcm_path)

    processing_progress[task_id] = 100  # 更新进度为100%

    processing_status[task_id]['status'] = '完成'
    processing_status[task_id]['message'] = message
    processing_status[task_id]['fracture_detected'] = fracture_detected
    processing_status[task_id]['processed_image_path'] = processed_blob_path

    patient_ref = db.reference(f'superadmin/admins/{admin_name}/patients/{patient_id}')
    patient_ref.update({'AI_result': 'true'})

    return {"message": message, "fracture_detected": fracture_detected, "processed_image_path": processed_blob_path}

# 添加Mangum处理程序适配AWS Lambda
handler = Mangum(app)

AWS Cloud Logs日志 #

INFO: Started server process [8]
2024-08-11T20:13:37.314+05:30
INFO: Waiting for application startup.
2024-08-11T20:13:37.314+05:30
INFO: Application startup complete.
2024-08-11T20:13:37.315+05:30
INFO: Uvicorn running on http://0.0.0.0:8080 (Press CTRL+C to quit)
2024-08-11T20:13:41.644+05:30
INIT_REPORT Init Duration: 9994.93 ms Phase: init Status: timeout
2024-08-11T20:13:45.492+05:30
INFO: Started server process [8]
2024-08-11T20:13:45.492+05:30
INFO: Waiting for application startup.
2024-08-11T20:13:45.492+05:30
INFO: Application startup complete.
2024-08-11T20:13:45.493+05:30
INFO: Uvicorn running on http://0.0.0.0:8080 (Press CTRL+C to quit)

解决方案 #

1. 延迟资源初始化，规避冷启动阻塞 #

当前代码在模块加载阶段就创建了ThreadPoolExecutor，部分依赖初始化会占用初始化时间。将资源初始化延迟到首次请求时执行：

• 线程池改为懒加载模式：

  executor = None
  
  def get_executor():
      global executor
      if executor is None:
          executor = ThreadPoolExecutor(max_workers=20)
      return executor
  

• YOLO模型初始化保留在process_dicom_file中即可，无需提前到模块级别。

2. 优化依赖包体积，加快加载速度 #

• 使用Lambda层分离大依赖：将ultralytics、pydicom、firebase-admin等体积较大的依赖打包到Lambda层，减少部署包大小，提升初始化效率。
• 清理冗余依赖：只保留运行必需的包，避免安装不必要的依赖项。
• 瘦身Python依赖：使用pip install --target . --no-deps仅安装核心依赖，排除冗余子包。

3. 调整Lambda配置参数 #

• 延长初始化超时时间：Lambda默认初始化超时为10秒，刚好与当前超时时间接近，可在控制台将初始化超时调整为15-30秒。
• 升级运行时内存：更高内存的Lambda实例对应更高的CPU和网络带宽，可加快依赖加载与初始化速度，比如从128MB升级至512MB或1GB。

4. 优化Firebase初始化流程 #

• 避免读取本地证书文件：将Firebase证书内容存储为环境变量，代码中直接读取环境变量构建证书：

  import json
  
  def initialize_firebase():
      with firebase_lock:
          if not firebase_admin._apps:
              cred_json = json.loads(os.environ.get("FIREBASE_CREDENTIALS"))
              cred = credentials.Certificate(cred_json)
              firebase_admin.initialize_app(cred, {
                  'storageBucket': os.environ.get("FIREBASE_STORAGE_BUCKET"),
                  'databaseURL': os.environ.get("FIREBASE_DATABASE_URL")
              })
  

  此方式既避免了本地文件IO开销，也提升了安全性。

5. 减少模块级别操作 #

模块加载阶段的所有操作都会计入初始化时间，需进行优化：

• 移除模块级别的logging.basicConfig，改为在函数内配置或直接使用Lambda自带日志系统。
• 禁止在模块级别执行IO操作、计算密集型操作。

6. 预热Lambda函数 #

通过CloudWatch Events定期触发/health健康检查端点，保持函数处于暖状态，避免冷启动时的初始化超时。

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