• 对文中开头所讲的,有些个人的理解,如下:

    • 工具也好、自动化也好,开展之前的价值度量,这是在动手前需要明确的,如果只是讲提升效率,我感觉这不属于价值度量,因为工具、自动化本身自带效率提升的buff,更多的关注的是效率提升背后的东西,它的持续性,同时效率提升转化成了什么。

    • 价值度量之后,是围绕着某个问题在某些场景下的解决方案设计,最终的工具、自动化只是一种解决问题的对外表现形式,真正的难点是解决方案设计是否真的解决痛点,甚至是类似场景下的痛点,以便推广是更具普适性。如果一上来就着手工具开发,容易忽略工具、自动化功能实现背后的真正痛点。工具、自动化本身最大的挑战不是实现,而是它的持续性,一个东西做出来,持续的时间越长,收益越高。然而往往有些工具的使用成本、维护成本太高,导致它很“短寿”,自然也很难体现出它的价值。

    • 最后,摸索出适合本地化的工程效率组与业务组的合作模式,度量双方的价值产出,明确双方职责。在此之前基建很重要(基础的自动化、工具链平台建设),否则会发现没有一定基建的基础上很多东西,难以落地或者开展成本很高。

    • 最后技术栈尽量统一,工具、自动化开发设计、代码尽量规范。

  • 求助一个算法问题 at July 20, 2019

  • 集合点

    看测试脚本,先说明下此集合点只会在每个用户(协程)启动时,集合一次,执行过程是没有集合动作的。如下是locust源码,通过locust().run(runner=self)起对应协程执行各用户任务。

    # runners.py 
    def hatch():
    sleep_time = 1.0 / self.hatch_rate
    while True:
    if not bucket:
    logger.info("All locusts hatched: %s" % ", ".join(["%s: %d" % (name, count) for name, count in six.iteritems(occurence_count)]))
    events.hatch_complete.fire(user_count=self.num_clients)
    return

    locust = bucket.pop(random.randint(0, len(bucket)-1))
    occurence_count[locust.__name__] += 1
    def start_locust(_):
    try:
    locust().run(runner=self)
    except GreenletExit:
    pass
    new_locust = self.locusts.spawn(start_locust, locust)
    if len(self.locusts) % 10 == 0:
    logger.debug("%i locusts hatched" % len(self.locusts))
    gevent.sleep(sleep_time)

    再看run(),每个用户只会进行一次on_start()初始化准备工作,剩下的时间在while (True)中循环执行。同时也可以看到locust内部是没有集合点的,通过gevent.sleep(sleep_time)控制每秒加载用户,在此之前加载的用户已经在各自协程中执行任务了。

    # core.py
    def run(self, *args, **kwargs):
    self.args = args
    self.kwargs = kwargs

    try:
    if hasattr(self, "on_start"):
    self.on_start()
    except InterruptTaskSet as e:
    if e.reschedule:
    six.reraise(RescheduleTaskImmediately, RescheduleTaskImmediately(e.reschedule), sys.exc_info()[2])
    else:
    six.reraise(RescheduleTask, RescheduleTask(e.reschedule), sys.exc_info()[2])

    while (True):
    try:
    if self.locust.stop_timeout is not None and time() - self._time_start > self.locust.stop_timeout:
    return

    if not self._task_queue:
    self.schedule_task(self.get_next_task())

    try:
    self.execute_next_task()
    except RescheduleTaskImmediately:
    pass
    except RescheduleTask:
    self.wait()
    else:
    self.wait()

    楼主的集合点放到了on_start()中,只会集合一次,就是初始化的时候。另外再看locust 用户任务执行。

    请求数量

    locust每个协程(用户)运行任务时,具体任务执行有两种方式:@task修饰的权重随机任务(TaskSet),和@seq_task优先级顺序任务(TaskSequence),楼主采用的是@task方式

    # core.py

    class TaskSetMeta(type):
    """
    Meta class for the main Locust class. It's used to allow Locust classes to specify task execution
    ratio using an {task:int} dict, or a [(task0,int), ..., (taskN,int)] list.
    """


    def __new__(mcs, classname, bases, classDict):
    new_tasks = []
    for base in bases:
    if hasattr(base, "tasks") and base.tasks:
    new_tasks += base.tasks

    if "tasks" in classDict and classDict["tasks"] is not None:
    tasks = classDict["tasks"]
    if isinstance(tasks, dict):
    tasks = six.iteritems(tasks)

    for task in tasks:
    if isinstance(task, tuple):
    task, count = task
    for i in xrange(0, count):
    new_tasks.append(task)
    else:
    new_tasks.append(task)

    for item in six.itervalues(classDict):
    if hasattr(item, "locust_task_weight"):
    for i in xrange(0, item.locust_task_weight):
    new_tasks.append(item)

    classDict["tasks"] = new_tasks

    return type.__new__(mcs, classname, bases, classDict)

    源码如上可以看到,上下文源码不贴了。

    以楼主代码为例 每个用户的任务集为[cms_queryUserList,cms_queryUserList,cms_login],这是根据@task权重生成的,旨在达到权重越大被执行的概率越高的效果,每次执行随机从这个列表中取一个执行,如下源码,所以出现了楼主所说请求数量不一致。

    def get_next_task(self):
    return random.choice(self.tasks)

    如何解决

    可以尝试使用@seq_task修饰器,确定执行顺序,然后顺序执行,如下:

    coding=utf-8
    import requests
    from locust import HttpLocust,TaskSet,task
    import os
    from locust import events
    from gevent._semaphore import Semaphore
    all_locusts_spawned = Semaphore()
    all_locusts_spawned.acquire()
    def on_hatch_complete(**kwargs):
    all_locusts_spawned.release() #创建钩子方法

    events.hatch_complete += on_hatch_complete #挂载到locust钩子函数(所有的Locust实例产生完成时触发)

    class Cms(TaskSequence):
    # 访问cms后台首页
    @seq_task(1)
    def cms_login(self):
    # 定义请求头
    header = {
    "User-Agent": "Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/70.0.3538.67 Safari/537.36"
    }

    req = self.client.get("/cms/manage/loginJump.do?userAccount=admin&loginPwd=123456", headers=header, verify=False)
    if req.status_code == 200:
    print("success")
    else:
    print("fails")

    @seq_task(2)
    def cms_queryUserList(self):
    header = {
    "User-Agent":"Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/70.0.3538.67 Safari/537.36"
    }
    queryUserList_data = {
    "startCreateDate": "",
    "endCreateDate": "",
    "searchValue": "",
    "page": "1"
    }
    req = self.client.post("/cms/manage/queryUserList.do",data=queryUserList_data)
    queryUserList_url = 'http://192.168.23.132:8080/cms/manage/queryUserList.do'
    if req.status_code == 200: #u'查询用户成功!'
    print("success")
    else:
    print("fails")

    def on_start(self):
    """ on_start is called when a Locust start before any task is scheduled """
    # self.cms_login()
    # self.cms_queryUserList()
    all_locusts_spawned.wait() #限制在所有用户准备完成前处于等待状态

    class websitUser(HttpLocust):
    task_set = Cms
    min_wait = 3000 # 单位为毫秒
    max_wait = 6000 # 单位为毫秒

    if __name__ == "__main__":
    import os
    os.system("locust -f lesson8.py --host=http://192.168.23.134:8080")


    '''
    semaphore是一个内置的计数器:
    每当调用acquire()时,内置计数器-1
    每当调用release()时,内置计数器+1
    计数器不能小于0,当计数器为0时,acquire()将阻塞线程直到其他线程调用release()
    '''
  • 这里的代码驱动的意思应该是关键字驱动吧,一般关键字驱动和数据驱动是一起使用的。

    去年也是基于 Python Requests 开发请求发送、响应校验的通用自定义关键字以及上下游自定义关键字,比如Oralce、Mysql、Redis、Mongo数据库操作关键字,基本满足目前的接口自动化需要,同时使用Python Locust(采用Request库作为客户端)实现了单接口性能的自动化关键字开发,快速构建微服务接口的功能、性能自动化测试用例。

    不过基于RF关键字驱动的自动化测试,用例本身的维护性需要着重考虑,所以关键字设计时需要着重对可维护性需要把关,比如,代码设计复杂度、异常易于定位,输出指向明确、用例组织简洁。目前基本控制在了4、5行覆盖一条接口用例(包含了数据库验证)。

  • 高压力测试 分布式 都是需要堆硬件的。

  • locust 默认了1s等待时间,min_wait=0,max_wait=0试试看。

  • locust 默认了1s等待时间,min_wait=0,max_wait=0试试看。

  • 技术能力做的是将业务测试能力水平复制。

  • 目前做的大概就是这里说的基于算法模型的测试、大致分图像检测、图像检索(识别)两大类吧,结合实践说说自己的理解:

    算法测试的核心

    基于算法模型的两个核心:一是测试数据、二是评估指标 ,大致划分为三个测试类型功能测试、性能测试、效果测试(模型评估指标)。

    测试数据

    • 特征的覆盖:分析显性特征、隐形特征,进行特征标注。
    • 数据量的覆盖:在测试图像检测算法时,根据特征收敛曲线来确定数据量,结合经验来看3000张左右基本收敛,但同时考虑真实生产环境特征分布,我们一般采用1~2万张;图像检索算法,没有什么比较好的选取依据,一般采用1000组,每组20张(Top20)。

    为此,开发了一个算法数仓平台,主要用于算法测试数据的框图、特征标注、存储、查询、对接算法微服务自动化平台。

    算法测试数据尽量采用真实生产环境下的数据,尽可能的了解真实环境的特征分布,同时需要的注意的是,算法测试数据肯定是持续更新的,更新频率、方式需要结合真实数据源情况,比如之前系统接入都是办公室摄像头数据,现在接入的是路边摄像头数据,那么算法模型的效果肯定是不同的,因此我们需要感知真实环境下的数据变化。

    评估指标

    • 明确不同算法评估指标的差异性,不能一概而论 :比如图像检测,基于混淆矩阵(精确率、召回率、虚警率、F1等)、ROC(AUG)、PR(AP/MAP);目标检索评估指标不同于此,还需要采用其他指标评估,比如TAR,FAR,FRR,ERR、前N连续正确率等。

    算法测试覆盖的测试类型

    功能测试

    我们的算法模型通过微服务接口对外提供服务,因此功能性验证是基于接口功能的验证,与普通微服务接口测试策略一致,我们采用Python Request 基于关键字驱动、数据驱动实现了微服务接口的功能性测试,测试数据从上面介绍的算法数仓平台获取。

    性能测试

    大多数算法模型本身难以支持并发,比如人脸检索,数据库数据量10亿以上,虽然通过某些算法减少了计算量,但加载到内存的数据仍然几十、上百GB。对于算法模型大多数做的是稳定性测试和版本性能波动感知(版本间的性能差异感知)。

    使用Python Locust 基于关键字驱动、数据驱动,实现了算法微服务的性能测试自动化,测试数据从上面介绍的算法数仓平台获取。并且每次测试的执行策略(多少用户并发、每秒加载几个,运行多久),以及执行结果,如平均响应时间,*%响应时间,TPS等指标都会记录入数据库,每次运行均可以看到历史结果,并进行横向比较。

    效果测试

    这就是指的算法模型评估指标的计算了,使用Python封装各个指标计算方法作为自定义关键字,上传到自动化平台,根据不同算法模型指标评估需要组装对应指标计算方法即可,测试数据从上面介绍的算法数仓平台获取。每次执行记录各个指标,版本,入库存储,输出历史该接口的指标计算结果,进行横向比较。

    同时效果测试还包含竞品效果的比对测试。

    除了给出基本的指标计算结果,我们还尝试通过算法模型运行结果结合数据标注信息,分析出哪些特征对算法效果有直接影响,结合实际生产环境特征分布,来评估影响程度等。

    因此通过自研的算法数仓平台+自动化测试平台,基本实现了算法模型微服务的功能、性能、指标评估自动化测试,不仅关注每次执行结果,更关注版本间的差异性,感知持续优化程度。

    转载请说明出处,谢谢。

  • Setup teardown。

TestOps