Locust中如何标记超时轮询步骤失败并统计总耗时

问题背景 #

我有一个执行步骤流的Locust测试，代码如下：

@task
def execute_tasks():
   test = TestFlow(self.client)
   test.step1()
   test.step2()
   ...

其中一个步骤涉及状态轮询，代码如下：

def test4(self, timeout_seconds=420, interval_seconds=5):
    start = time.time()
    is_done = False
    while not is_done and time.time() - start < timeout_seconds:
        time.sleep(interval_seconds)
        response = self.client.get("url", name="wait_for_test")
        test_status = response.json()
        is_done = test_status["status"] == "COMPLETED"

    if not is_done:
        raise Exception("Polling timed out!")

问题1：当轮询超时时，如何在Locust中标记该步骤为失败，而非仅抛出异常？ #

Locust提供两种手动标记失败的方式，无需抛出异常中断任务：

1. 调用client.fail()方法：直接通过客户端对象标记失败，该操作会被Locust统计到失败请求中，且不影响后续任务执行。
2. 触发events.request.fire()事件：自定义失败统计条目，适合需要单独追踪轮询超时事件的场景。

修改后的代码示例：

def test4(self, timeout_seconds=420, interval_seconds=5):
    start_time = time.time()
    is_done = False
    
    while not is_done and time.time() - start_time < timeout_seconds:
        time.sleep(interval_seconds)
        response = self.client.get("url", name="wait_for_test")
        test_status = response.json()
        is_done = test_status["status"] == "COMPLETED"

    if not is_done:
        # 方式1：直接标记请求失败，纳入Locust失败统计
        self.client.fail(f"Polling timed out after {timeout_seconds} seconds")
        # 方式2：自定义失败统计条目（可选，需提前导入events）
        # from locust import events
        # events.request.fire(
        #     request_type="POLL",
        #     name="polling_timeout",
        #     response_time=int((time.time() - start_time)*1000),
        #     response_length=0,
        #     exception=Exception(f"Polling timed out after {timeout_seconds} seconds")
        # )

问题2：如何统计从轮询开始到收到“COMPLETED”状态的总耗时，而非仅跟踪单个GET请求？ #

可以通过Locust的events.request.fire()事件，手动生成自定义统计条目，记录整个轮询流程的总耗时：

1. 记录轮询启动的时间戳
2. 轮询结束后计算总耗时（转换为毫秒，匹配Locust默认统计单位）
3. 触发自定义请求事件，将总耗时纳入统计面板。

修改后的完整代码示例：

from locust import events
import time

def test4(self, timeout_seconds=420, interval_seconds=5):
    start_time = time.time()
    is_done = False
    polling_success = False
    
    while not is_done and time.time() - start_time < timeout_seconds:
        time.sleep(interval_seconds)
        response = self.client.get("url", name="wait_for_test")
        test_status = response.json()
        is_done = test_status["status"] == "COMPLETED"
        if is_done:
            polling_success = True
            break

    # 计算总耗时（转成毫秒）
    total_response_time = int((time.time() - start_time) * 1000)
    
    if polling_success:
        # 触发成功的轮询总耗时统计
        events.request.fire(
            request_type="POLL",
            name="polling_total_success",
            response_time=total_response_time,
            response_length=0,  # 无响应长度时设为0
            exception=None
        )
    else:
        # 触发失败的轮询总耗时统计
        events.request.fire(
            request_type="POLL",
            name="polling_total_timeout",
            response_time=total_response_time,
            response_length=0,
            exception=Exception(f"Polling timed out after {timeout_seconds} seconds")
        )
        # 同时标记客户端请求失败（可选）
        self.client.fail(f"Polling timed out after {timeout_seconds} seconds")

说明 #

• 自定义的POLL类型请求会在Locust统计面板中单独展示，能清晰看到轮询的整体耗时分布，而非每次轮询的单个GET请求耗时。
• 若无需区分成功/失败的总耗时，可将统计名称统一设为polling_total_time，通过异常字段区分结果。

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