Locust TCP Socket压测请求统计不显示问题排查

核心问题原因 #

你的代码存在以下几个关键问题，导致Locust仪表盘无法正常展示请求统计：

• 方法拦截逻辑错误：FleetSyncLocustClient中__getattr__直接从FleetSyncClient.__dict__获取未绑定的实例方法，调用时缺少self参数，导致send_protobuf_message等核心方法未被正确包装，请求事件无法触发。
• 进程强制退出：connect_terminal和abort方法使用sys.exit(1)直接终止Locust用户进程，中断统计流程，导致Locust无法收集请求数据。
• Socket连接意外关闭：connect_terminal中用with socket.create_connection创建连接，退出with块时底层Socket会被自动关闭，导致self.ssock无效，后续发送消息会抛出异常。
• 响应长度未正确计算：当前response_length固定为0，无法提供真实的请求数据维度。

分步修复方案 #

1. 重构客户端包装类，替换动态拦截为显式方法包装 #

放弃不稳定的__getattr__动态拦截，直接重写需要统计的核心方法，确保每个请求都能触发Locust的请求事件。修改locust_fleetsync_wrapper.py：

import time
import ssl
import socket
from client.fleetsync_client import FleetSyncClient
from locust.exception import StopUser


class FleetSyncLocustClient(FleetSyncClient):
    def __init__(self, host, request_event):
        super().__init__(host)
        self._request_event = request_event

    def abort(self, text):
        # 替换sys.exit，用Locust原生异常终止用户实例，避免中断进程
        raise StopUser(f"###Error: {text}")

    def connect_terminal(self):
        try:
            context = ssl.create_default_context()
            context.check_hostname = False
            context.verify_mode = ssl.CERT_NONE   
            context.load_cert_chain(certfile=self.CERTIFICATE)

            # 移除with语句，避免socket被自动关闭
            sock = socket.create_connection((self.hostname, 443))
            try:
                self.ssock = context.wrap_socket(sock, server_hostname=self.hostname)
                print(f"Connected to FarmCentre server: {self.hostname}")
                return
            except ssl.CertificateError as e:
                sock.close()
                self.abort(f"Certificate validation error: {e}. Please check the server certificate")
            except ssl.SSLError as e:
                sock.close()
                self.abort(f"Connection with server {self.hostname} failed: {e}")
            except Exception as e:
                sock.close()
                self.abort(f"An error occurred: {e}")
        except Exception as err:
            self.abort(f"Connection with server {self.hostname} failed unexpected {err=}, {type(err)=}")

    def send_protobuf_message(self, message, request_name="send_protobuf_message"):
        request_meta = {
            "request_type": "socket",
            "name": request_name,
            "start_time": time.time(),
            "response_length": 0,
            "response": None,
            "context": {},
            "exception": None,
        }
        start_perf_counter = time.perf_counter()
        try:
            # 调用父类原始方法
            response = super().send_protobuf_message(message)
            # 计算真实响应长度
            request_meta["response_length"] = len(response.SerializeToString())
            request_meta["response"] = response
        except Exception as e:
            request_meta["exception"] = e
        finally:
            request_meta["response_time"] = (time.perf_counter() - start_perf_counter) * 1000
            # 触发Locust请求事件，写入统计数据
            self._request_event.fire(**request_meta)
        return request_meta["response"]

2. 区分不同类型请求（可选） #

如果需要在仪表盘上区分GPS和SOV消息的统计数据，可以在调用时传入自定义请求名称：

# 在locustfile.py的send_gps方法中修改调用
def send_gps(self, gps_coordinates: GpsCollection):
    pb_msg = ProtobufMsg(self.machine.terminal_id, self.machine.device_id)
    for gps in gps_coordinates.records:
        pb_msg.add_gps_coordinate(gps)
        self.send_message(pb_msg, request_name="send_gps")

# 在send_sovs方法中修改调用
def send_sovs(self, sovs):
    pb_msg = ProtobufMsg(self.machine.terminal_id, self.machine.device_id)
    for sov in sovs.records:
        pb_msg.add_sov_data(sov)
        self.send_message(pb_msg, request_name="send_sov")

# 更新send_message方法
def send_message(self, pb_msg: ProtobufMsg, request_name="send_protobuf_message"):
    if pb_msg.datapoints >= ProtobufMsg.datapoints_limit:
            response = self.client.send_protobuf_message(pb_msg.msg, request_name=request_name)
            print(f"Terminal: {self.machine.terminal_id} received: {response}")
            pb_msg.datapoints = 0
            pb_msg.update_request_msg()

3. 移除原客户端中的强制退出逻辑 #

修改fleetsync_client.py中的abort方法，避免直接终止进程：

def abort(self, text):
    print(f"###Error: {text}")
    self.disconnect_terminal()
    # 移除sys.exit(1)，由包装类的abort方法抛出StopUser异常

验证修改效果 #

1. 启动Locust服务，访问localhost:8089仪表盘。
2. 启动压测后，可在Statistics标签页看到send_protobuf_message（或自定义的send_gps/send_sov）的请求统计，包括响应时间、成功率、吞吐量等数据。
3. 若请求失败，失败信息会被捕获并展示在Failures标签页中。

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