HTTP触发的Azure Function运行5分钟后停止向Service Bus发送消息

问题背景 #

有一个运行时版本为~4、基于Python 3.9的HTTP触发Azure Function应用，由POST请求触发，用于处理Azure Event Hubs流数据并将输出发送至Azure Service Bus，单次调用可能运行数小时。

应用运行前5分钟一切正常，但初始化异步Service Bus客户端约5分钟后，消息不再传输至Service Bus，且期间无任何错误、异常或警告。该问题仅在云端出现，本地运行无异常，且Azure Function资源无网络和防火墙限制。

查看Azure Function文件系统日志，发现关键错误：

Encountered a System.Threading.Tasks.TaskCanceledException exception after 300001.013ms with message: The request was canceled due to the configured HttpClient.Timeout of 300 seconds elapsing

日志序列片段：

http://0.0.0.0:8181'./appsvctmp/volatile/logs/runtime/95819576165bdac59778b1cfad868566f03e09116048c0c4ad5d3516e6e96024.log
2024-09-24T09:14:42.276924535Z: [INFO]  Updated CA certificates
2024-09-24T09:14:44.077950242Z: [INFO]  warn: Microsoft.AspNetCore.Hosting.Diagnostics[15]
2024-09-24T09:14:44.077984542Z: [INFO]        Overriding HTTP_PORTS '8080' and HTTPS_PORTS ''. Binding to values defined by URLS instead 'http://0.0.0.0:8181'.
2024-09-24T10:44:06.466821919Z: [INFO]  fail: Middleware[0]
2024-09-24T10:44:06.492710253Z: [INFO]        Failed to forward request to http://169.254.132.4. Encountered a System.Threading.Tasks.TaskCanceledException exception after 300001.013ms with message: The request was canceled due to the configured HttpClient.Timeout of 300 seconds elapsing.. Check application logs to verify the application is properly handling HTTP traffic./appsvctmp/volatile/logs/runtime/4a6c3e8d7bd7e27e90a3050994cc782ccd484a5ab6b1a7d91dc1dd71ef66ceb3.log
2024-09-24T10:48:52.067841706Z: [INFO]  Updated CA certificates
2024-09-24T10:48:55.093252187Z: [INFO]  warn: Microsoft.AspNetCore.Hosting.Diagnostics[15]
2024-09-24T10:48:55.093292687Z: [INFO]        Overriding HTTP_PORTS '8080' and HTTPS_PORTS ''. Binding to values defined by URLS instead 'http://0.0.0.0:8181'.Ending Log Tail of existing logs ---Starting Live Log Stream ---

相关代码与配置 #

Service Bus发送代码片段：

import async_timeout
from azure.servicebus.aio import ServiceBusClient


client = ServiceBusClient.from_connection_string(
    conn_str=CONN_STRING,
    logging_enable=True,
    retry_total=3,
    retry_backoff_max=10,
    retry_mode="fixed",
)

total_send_time: float = 0
message_count: int = 0
message: Optional[ServiceBusMessage] = None

async with client:
    sender = client.get_topic_sender(topic_name=TOPIC_NAME)
    async with sender:
        message = None

        while output_queue.qsize() > 0:
            try:
                async with async_timeout.timeout(10):
                    message = await output_queue.get()
            except asyncio.TimeoutError:
                continue

            tick = time.monotonic()
            await sender.send_messages(message, timeout=10)
            tock = time.monotonic()
            output_queue.task_done()
            message = None

            total_send_time += tock - tick
            message_count += 1
            if message_count % 20 == 0:
                average_send_time = total_send_time / 20
                logging.info(
                    f"Successfully sent {message_count} messages to Service Bus (average send time: {average_send_time:.4f} sec)"
                )
                total_send_time = 0

host.json配置：

{
  "version": "2.0",
  "logging": {
    "applicationInsights": {
      "samplingSettings": {
        "isEnabled": true,
        "excludedTypes": "Request"
      }
    }
  },
  "extensionBundle": {
    "id": "Microsoft.Azure.Functions.ExtensionBundle",
    "version": "[3.3.0, 4.0.0)"
  },
  "functionTimeout": "-1"
}

function.json配置：

{
  "scriptFile": "__init__.py",
  "bindings": [
    { 
      "authLevel": "function",
      "type": "httpTrigger",
      "direction": "in",
      "name": "req",
      "methods": [
        "get",
        "post"
      ]
    },
    {
      "type": "http",
      "direction": "out",
      "name": "$return"
    }
  ]
}

问题根源 #

这个问题的核心是Azure Function HTTP触发的前端代理超时机制影响了进程内的共享网络资源：

1. 尽管设置了functionTimeout: "-1"允许函数长时间运行，但HTTP触发存在5分钟（300秒）的硬超时限制——这是前端代理层的超时，而非函数执行超时。
2. 当HTTP请求超过5分钟未返回响应时，前端代理会终止与函数进程的连接，并重置进程内的共享HttpClient实例（Azure SDK内部依赖该实例进行网络请求）。
3. 异步Service Bus客户端依赖这个共享HttpClient，一旦被重置，后续发送请求会因底层连接切断而静默失败，且代码未捕获到该层面的异常。
4. 日志中提到的HttpClient.Timeout of 300 seconds elapsing就是前端代理超时的直接体现，http://169.254.132.4是Azure Functions内部路由地址，说明代理层无法再将请求转发到函数进程。

解决方案 #

1. 架构重构（生产环境推荐） #

HTTP触发设计用于短请求处理，不适合数小时的长运行任务。标准解决方案是拆分流程为异步架构：

• HTTP触发函数：接收初始POST请求后，立即返回202 Accepted响应，同时将任务信息发送到Azure Queue Storage或Service Bus。
• 队列触发函数：从队列中获取任务，处理Event Hubs流数据并发送结果到Service Bus，该函数可通过functionTimeout: "-1"设置长时间运行。

这种架构完全避开HTTP触发的超时限制，符合Azure Functions的最佳实践。

2. 临时代码修复（非生产环境应急） #

如果暂时无法重构架构，可通过定期重置Service Bus客户端规避底层HttpClient被回收的问题：

import async_timeout
from azure.servicebus.aio import ServiceBusClient
import time
import asyncio

async def create_servicebus_client():
    return ServiceBusClient.from_connection_string(
        conn_str=CONN_STRING,
        logging_enable=True,
        retry_total=3,
        retry_backoff_max=10,
        retry_mode="fixed",
    )

total_send_time: float = 0
message_count: int = 0
message: Optional[ServiceBusMessage] = None
client = await create_servicebus_client()
last_reset_time = time.monotonic()

while output_queue.qsize() > 0:
    # 每4分钟重置一次客户端，提前于5分钟超时窗口
    if time.monotonic() - last_reset_time > 240:
        await client.close()
        client = await create_servicebus_client()
        last_reset_time = time.monotonic()
        logging.info("Reset Service Bus client to avoid timeout")

    try:
        async with async_timeout.timeout(10):
            message = await output_queue.get()
    except asyncio.TimeoutError:
        continue

    try:
        async with client:
            sender = client.get_topic_sender(topic_name=TOPIC_NAME)
            async with sender:
                tick = time.monotonic()
                await sender.send_messages(message, timeout=10)
                tock = time.monotonic()
    except Exception as e:
        logging.error(f"Failed to send message: {str(e)}")
        # 发送失败时立即重置客户端
        await client.close()
        client = await create_servicebus_client()
        last_reset_time = time.monotonic()
        continue

    output_queue.task_done()
    message = None

    total_send_time += tock - tick
    message_count += 1
    if message_count % 20 == 0:
        average_send_time = total_send_time / 20
        logging.info(
            f"Successfully sent {message_count} messages to Service Bus (average send time: {average_send_time:.4f} sec)"
        )
        total_send_time = 0

3. 辅助配置优化 #

更新扩展包版本至最新，避免潜在SDK bug：

{
  "version": "2.0",
  "logging": {
    "applicationInsights": {
      "samplingSettings": {
        "isEnabled": true,
        "excludedTypes": "Request"
      }
    }
  },
  "extensionBundle": {
    "id": "Microsoft.Azure.Functions.ExtensionBundle",
    "version": "[4.0.0, 5.0.0)"
  },
  "functionTimeout": "-1"
}

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