如何编写FaaS大冷启动测试代码？仅对接Azure Blob Service

用户提问 #

我希望通过测量实例数量增加时的响应时间，找到优化资源配置的方法。我编写了一个生成5MB数据的Activity Function（代码如下），但测试结果未出现明显冷启动。请提供一段能产生更大冷启动时间的代码示例，若可能，仅对接Azure Blob Service即可。

原代码（用户提供）：

import azure.functions as func
import azure.durable_functions as df
import numpy as np
import pandas as pd
import time

# Client functions are mostly irrelevant, so some are omitted.
app = df.DFApp(http_auth_level=func.AuthLevel.ANONYMOUS)
@app.route(route="orchestrators/client_function")
@app.durable_client_input(client_name="client")
async def client_function(req: func.HttpRequest, client: df.DurableOrchestrationClient) -> func.HttpResponse:
    instance_id = await client.start_new("orchestrator", None, {})
    await client.wait_for_completion_or_create_check_status_response(req, instance_id)
    status = await client.get_status(instance_id)
    return f"runtime: {status.runtime_status}\n\noutput:\n {output_str}" 


@app.orchestration_trigger(context_name="context")
def orchestrator(context: df.DurableOrchestrationContext) -> dict:
    start = time.perf_counter()
    activity_func = context.call_activity(f"activity{i}", size)
    return time.perf_counter() - start


@app.activity_trigger(input_name="size")
def activity1(size: int) -> int:
    data = np.random.rand(size) 
    df  =  pd.DataFrame(data) 
    df_ = df.to_dict()
    return len(df_)

要放大冷启动时间，核心是增加首次执行时的初始化开销和外部服务交互成本。以下代码将Blob客户端初始化放在Activity函数内部（避免宿主缓存），并执行大文件上传操作，能显著拉长冷启动时的响应时间：

import azure.functions as func
import azure.durable_functions as df
from azure.storage.blob import BlobServiceClient
import time
import os

app = df.DFApp(http_auth_level=func.AuthLevel.ANONYMOUS)

# 客户端函数保留原有逻辑
@app.route(route="orchestrators/client_function")
@app.durable_client_input(client_name="client")
async def client_function(req: func.HttpRequest, client: df.DurableOrchestrationClient) -> func.HttpResponse:
    # 传入Blob容器名和测试文件大小（示例为50MB）
    instance_id = await client.start_new("orchestrator", None, {"container_name": "test-container", "file_size_mb": 50})
    response = await client.wait_for_completion_or_create_check_status_response(req, instance_id)
    status = await client.get_status(instance_id)
    output_str = status.output.get("execution_time", "N/A")
    return func.HttpResponse(f"runtime: {status.runtime_status}\n\nexecution time: {output_str} seconds")

@app.orchestration_trigger(context_name="context")
def orchestrator(context: df.DurableOrchestrationContext) -> dict:
    input_data = context.get_input()
    start = time.perf_counter()
    # 调用Blob操作的Activity
    await context.call_activity("blob-heavy-activity", input_data)
    execution_time = time.perf_counter() - start
    return {"execution_time": round(execution_time, 2)}

@app.activity_trigger(input_name="input_data")
def blob_heavy_activity(input_data: dict) -> str:
    # 关键：将Blob客户端初始化放在函数内部，冷启动时需重新创建
    connect_str = os.environ["AZURE_STORAGE_CONNECTION_STRING"]
    blob_service_client = BlobServiceClient.from_connection_string(connect_str)
    
    container_name = input_data["container_name"]
    file_size_mb = input_data["file_size_mb"]
    
    # 确保容器存在（首次执行时会创建容器，增加额外开销）
    container_client = blob_service_client.get_container_client(container_name)
    container_client.create_container(exists_ok=True)
    
    # 生成大字节流并分块上传，放大IO和网络开销
    blob_name = f"test-file-{int(time.time())}.bin"
    blob_client = container_client.get_blob_client(blob_name)
    chunk_size = 1024 * 1024  # 1MB chunks
    total_chunks = file_size_mb
    
    for _ in range(total_chunks):
        blob_client.upload_blob(
            b"\x00" * chunk_size,
            blob_type="BlockBlob",
            append_block=True
        )
    
    return f"Uploaded {file_size_mb}MB file to Blob storage"

代码说明 #

1. Blob客户端内部初始化：将BlobServiceClient的创建放在Activity函数内部，避免宿主提前缓存客户端实例，冷启动时必须重新完成认证、网络握手等步骤，增加启动开销。
2. 容器创建校验：首次执行时会创建Blob容器（如果不存在），额外增加一次服务端交互。
3. 大文件分块上传：生成并上传几十MB的文件，分块操作会触发多次网络请求，显著拉长首次执行的时间，让冷启动和热启动的时间差更易观测。
4. 环境变量配置：需在Azure Functions配置中添加AZURE_STORAGE_CONNECTION_STRING，指向你的Blob存储账户。

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