Locust作为库运行时test_start钩子无法获取parsed_options

我正尝试在分布式环境中使用Locust自定义参数功能，参考了官方的自定义命令行参数示例，但将Locust作为库运行时遇到了问题：

• 解析后的选项未被填充
• 从UI传入自定义参数时，在test_start事件钩子里拿不到值，environment.parsed_options为None

我的代码如下：

@events.init_command_line_parser.add_listener
def _(parser):
    parser.add_argument("--my-argument", type=str, env_var="LOCUST_MY_ARGUMENT", default="", help="It's working", include_in_web_ui=True)

@events.test_start.add_listener
def _(environment, **kw):
    print(f"Custom argument supplied: {environment.parsed_options.my_argument}")

class Master:
    env = Environment(
        user_classes=[Class1, Class2, Class3],
        events=events,
        available_user_classes={
            "Class1": Class1,
            "Class2": Class2,
            "Class3": Class3,
        },
    )

    def _run_master(self):
        self.env.step_load = True
        self.env.create_master_runner()
        self.env.create_web_ui("0.0.0.0", 8089, userclass_picker_is_active=True)
        gevent.spawn(stats_history, self.env.runner)
        self.env.runner.greenlet.join()
        self.env.web_ui.stop()

    def run(self):
        """
        Wrapper for running master locust node

        Parameters: None

        Returns: None
        """
        get_required_env("LOADTESTSERVICE_NAMESPACE")
        self._run_master()

当手动初始化Environment把Locust当库用的时候，系统不会自动解析命令行或UI传入的参数，得自己处理参数解析和赋值：

1. 手动触发命令行参数解析
   在启动master runner之前，调用parse_command_line_args()方法，让环境完成参数解析：

   def _run_master(self):
       # 先解析命令行参数，生成parsed_options
       self.env.parse_command_line_args()
       self.env.step_load = True
       self.env.create_master_runner()
       self.env.create_web_ui("0.0.0.0", 8089, userclass_picker_is_active=True)
       gevent.spawn(stats_history, self.env.runner)
       self.env.runner.greenlet.join()
       self.env.web_ui.stop()
   

2. 获取UI传入的参数
   UI提交的参数不会自动同步到parsed_options，可以从web_ui的parsed_args里取，结合命令行参数做降级：

   @events.test_start.add_listener
   def _(environment, **kw):
       # 优先用UI传入的参数，没有则用命令行/环境变量的值
       my_arg = environment.web_ui.parsed_args.get("my_argument", environment.parsed_options.my_argument)
       print(f"Custom argument supplied: {my_arg}")
   

3. 分布式环境下同步参数到Worker
   如果是分布式架构，Master需要把自定义参数主动发给Worker节点：

   # Master节点发送参数
   @events.test_start.add_listener
   def _(environment, **kw):
       my_arg = environment.parsed_options.my_argument
       print(f"Custom argument supplied on master: {my_arg}")
       if environment.runner and environment.runner.master:
           # 给所有Worker发消息传递参数
           environment.runner.send_message("custom_argument", my_arg)
   
   # Worker节点接收并保存参数
   @events.message.add_listener
   def _(environment, msg, **kwargs):
       if msg.type == "custom_argument":
           # 手动给parsed_options赋值
           environment.parsed_options.my_argument = msg.data
           print(f"Custom argument received on worker: {msg.data}")
   

注意：手动创建的Environment实例，parsed_options不会自动生成，必须调用parse_command_line_args()或者手动创建Namespace对象赋值给它。

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