尧图网页设计 
OpenMetadata 压测实战用 Locust 为 Data Quality API 构建自动化负载测试与阈值断言体系【免费下载链接】OpenMetadataThe Open Context Layer for Data and AI , OpenMetadata is the open platform for building trusted data context and business semantics for humans, AI assistants, and agents.项目地址: https://gitcode.com/GitHub_Trending/op/OpenMetadata本篇介绍 OpenMetadata ingestion 模块中基于 Locust 的 API 负载测试框架如何在 ingestion/tests/load 目录下为新的服务端资源编写压测任务、如何通过manifest.yaml声明响应时间阈值以及整套「Locust 无头压测 pytest 断言」的自动化调用链读完后可直接照此为任意/api/v1/...端点补充可重复、可回归的性能测试。一、这套压测体系解决什么问题OpenMetadata 的服务端 API默认监听http://localhost:8585需要持续的性能基线保障否则一次慢查询回归可能无声地拖垮 UI 上的数据质量、目录浏览等页面。ingestion/tests/load正是为此设计的负载测试套件其核心组成test_load.pypytest 入口负责拼装 Locust 无头headless命令行参数、执行压测、再把 CSV 统计结果与阈值清单逐项比对test_resources/all_resources.pyLocust 用户与任务自动发现入口test_resources/tasks/每个待压测资源一个*.py任务文件当前已有 test_case_tasks.py 与 test_case_result_tasks.pytest_resources/manifest.yaml各请求的指标阈值清单utils.pyrun_load_test封装在 pytest 进程内直接调用 Locust 的main.main()。压测执行流程可以概括为三步Locust headless 发起压力 → 导出_stats.csv统计 → pytest 将每个指标与manifest.yaml中的阈值做assertLessEqual比较任一超标即用例失败从而把「性能是否退化」变成 CI 里可自动判定的断言。依赖上Locust 版本在 setup.py 中锁定为locust~2.32.0test_load.py 通过skipIf(sys.version_info (3, 9), ...)声明 Locust 仅在 Python 3.9 运行3.8 环境会自动跳过该用例。二、自动发现机制新增任务文件无需注册理解扩展方式之前先看 all_resources.py 如何把tasks/目录变成压测任务集。get_all_tasks_set()会遍历test_resources/tasks/*.py跳过base_前缀文件用importlib动态加载模块再经inspect.getmembers收集该模块中定义的所有类作为父 TaskSet 的tasksclass AllResources(TaskSet): Execute tasks for all resources classmethod def set_tasks(cls): tasks get_all_tasks_set() cls.tasks set(tasks) class All(HttpUser): host http://localhost:8585 wait_time constant(1) # closed workload AllResources.set_tasks() tasks [AllResources]几个值得注意的实现细节host硬编码为http://localhost:8585即要求本地或容器内OpenMetadata 服务端已启动且使用默认端口wait_time constant(1)表示每个虚拟用户执行完任务后固定等待 1 秒注释里称其为closed workload闭式工作负载——请求总量由并发用户数决定适合同步比对不同资源间的相对耗时由于是自动发现新增资源压测不需要修改all_resources.py只要把任务类放进tasks/即可。三、为新资源编写 Locust 任务3.1 文件与命名约定按照 README 的约定在test_resources/tasks下新增一个*.py文件文件名本身不限但社区习惯使用 Java 侧 Resource 类名按下划线分隔的写法例如TestCaseResource对应test_case_tasks.py。文件中最少需要导入两个 Locust 符号from locust import task, TaskSettask是装饰器把一个方法标记为压测任务其可选的整数参数表示任务权重——权重越大该任务被虚拟用户选中执行的概率越高任务类必须继承TaskSet它会被 all_resources.py 自动发现并挂载到用户类上。3.2 一个完整的任务定义示例以下示例完整展示了 README 中的任务写法带权重的时间范围查询、按参数分组命名请求class TestCaseResultTasks(TaskSet): Test case result resource load test def _list_test_case_results(self, start_ts: int, end_ts: int, days_range: str): List test case results for a given time range Args: start_ts (int): start timestamp end_ts (int): end timestamp range (str): for test_case in self.test_cases: fqn test_case.get(fullyQualifiedName) if fqn: self.client.get( f{TEST_CASE_RESULT_RESOURCE_PATH}/{fqn}, params{ # type: ignore startTs: start_ts, endTs: end_ts, }, authself.bearer, namef{TEST_CASE_RESULT_RESOURCE_PATH}/[fqn]/{days_range} ) task(3) def list_test_case_results_30_days(self): List test case results for the last 30 days. Weighted 3 now datetime.now() last_30_days int((now - timedelta(days30)).timestamp() * 1000) self._list_test_case_results(last_30_days, int(now.timestamp() * 1000), 30_days)实际仓库中的 test_case_result_tasks.py 正是这个模式的落地30_days权重 3、60_days权重 2、180_days权重 1让高频的短窗口查询在压测中占比更高更贴近真实 UI 的使用分布。3.3 鉴权on_start 钩子与 login_user压测请求如果带鉴权需要先把 token 放进self.bearer。标准做法是利用 Locust 的on_start钩子调用仓库内置的登录助手 login_userfrom _openmetadata_testutils.helpers.login_user import login_user class TestCaseResultTasks(TaskSet): Test case result resource load test [...] def on_start(self): Get a list of test cases to fetch results for self.bearer login_user(self.client) resp self.client.get(f{TEST_CASE_RESOURCE_PATH}, params{limit: 100}, authself.bearer) json resp.json() self.test_cases json.get(data, [])从源码看login_user会向/api/v1/users/login发送adminopen-metadata.org/ base64 密码YWRtaW4的登录请求取出accessToken并包装为BearerAuth对象实现见 login_user.py。这意味着压测前置条件是一个使用默认管理员账号的 OpenMetadata 实例且on_start里顺手预取了 100 条 test case 作为后续结果查询的数据来源——预取不计入task任务因此不会污染任务统计口径。3.4 必须定义 stop 方法README 特别强调IMPORTANT每个TaskSet类必须定义一个stop任务把控制权交还父级用户类class TestCaseResultTasks(TaskSet): Test case result resource load test [...] task def stop(self): self.interrupt()self.interrupt()会终止当前 TaskSet使虚拟用户回到AllResources层去随机选择下一个资源任务缺失它会导致用户永远困在第一个 TaskSet 里其他资源分不到流量。3.5 用 name 参数聚合带参请求的统计如果 URL 本身带变量如/api/v1/dataQuality/testCases/testCaseResults/{fqn}每条真实请求的 URL 都不同统计报告里会碎片化成上千行。解法是在self.client.get上传入name参数把同一模式的请求归并到同一统计项self.client.get( f{TEST_CASE_RESULT_RESOURCE_PATH}/{fqn}, params{ # type: ignore startTs: start_ts, endTs: end_ts, }, authself.bearer, namef{TEST_CASE_RESULT_RESOURCE_PATH}/[fqn]/{days_range} )name即请求在统计报告中显示的聚合名。README 给出的一份真实统计摘要 CSV节选即按此分组Type,Name,Request Count,Failure Count,Median Response Time,Average Response Time,Min Response Time,Max Response Time,Average Content Size,Requests/s,Failures/s,50%,66%,75%,80%,90%,95%,98%,99%,99.9%,99.99%,100% GET,/api/v1/dataQuality/testCases/testCaseResults/[fqn]/60_days,3510,0,13,16.2354597524217,5.146791999997902,100.67633299999557,84567.57407407407,49.30531562959204,0.0,13,17,20,21,28,35,45,56,92,100,100这个Name值同时也是 manifest.yaml 中声明阈值的 key两者必须严格一致断言逻辑才能命中。四、manifest.yaml把性能要求变成可断言的清单压测的最后一步是在 test_resources/manifest.yaml 中登记被测资源及其指标阈值。README 中的示例/api/v1/dataQuality/testCases/testCaseResults/[fqn]/30_days: type: GET 99%: 100 /api/v1/dataQuality/testCases/testCaseResults/[fqn]/60_days: type: GET 99%: 100语义是「该 GET 请求 99% 的请求耗时低于 100ms」。当前仓库中实际生效的阈值如下时间单位均为毫秒/api/v1/dataQuality/testCases/testCaseResults/[fqn]/30_days: type: GET 99%: 1000 /api/v1/dataQuality/testCases/testCaseResults/[fqn]/60_days: type: GET 99%: 1000 /api/v1/dataQuality/testCases/testCaseResults/[fqn]/180_days: type: GET 99%: 1000 /api/v1/dataQuality/testCases?limit10: type: GET 99%: 6000清单文件头注释也说明了取值规则key 可以是summaries/all__stats.csv中出现的任一指标列时间以毫秒表达1000ms 1s。4.1 断言是如何执行的test_load.py 读取压测产出的{summary_file}_stats.csvcsv.DictReader对每一行按Name查 manifest命中后先用type字段核对 HTTP 方法一致再遍历该资源的每个「指标 → 阈值」项用self.subTest逐项assertLessEqual(int(stat), threshold)。任一指标超标会输出形如99% for /api/v1/dataQuality/testCases?limit10 is greater than threshold的断言失败信息。由于使用了subTest单个指标超标不会中断其余指标的校验一次运行能暴露全部退化点。4.2 可用的指标清单README 列出了 manifest 中可声明的全部指标对应 Locust 导出 CSV 的列名Request Count请求总数Failure Count失败总数Median Response Time中位响应时间Average Response Time平均响应时间Min / Max Response Time最小 / 最大响应时间Average Content Size平均响应体大小Requests/s、Failures/s每秒请求 / 失败数50%、66%、75%、80%、90%、95%、98%、99%、99.9%、99.99%、100%各分位响应时间五、运行压测与关键参数pytest 用例 TestAllResources.test_all_resources 会把压测参数拼装成如下 Locust 命令行经 run_load_test 在进程内执行并断言退出码为 0locust --headless \ -H http://localhost:8585 \ --user 50 \ --spawn-rate 1 \ -f ingestion/tests/load/test_resources/all_resources.py \ --run-time 1m \ --only-summary \ --csv ingestion/tests/load/summaries/all_各参数含义与可调项参数当前值说明--headless固定无界面模式适合 CI-Hhttp://localhost:8585被测服务端地址与 all_resources.py 中的host保持一致--user50并发虚拟用户数可用环境变量LOCUST_USER覆盖--spawn-rate1每秒启动的用户数--run-time1m压测时长可用环境变量LOCUST_RUNTIME覆盖--csvsummaries/all_统计 CSV 输出前缀pytest 随后读取{前缀}_stats.csv做断言因此手动复现或调参时只需先起一个 OpenMetadata 服务端端口 8585并导入足够的数据例如 test case 与结果数据可用 ingestion 的示例管道灌入再按需修改LOCUST_USER/LOCUST_RUNTIME运行 pytest 用例即可得到与 CI 一致的压力画像与阈值判定。六、小结ingestion/tests/load提供了一条「压测即可执行测试」的完整闭环在test_resources/tasks/下新增任务文件用task(权重)定义带参数的请求on_start中完成登录与数据预取name参数聚合统计并务必实现stop任务在 manifest.yaml 中为该请求的聚合名声明指标阈值毫秒级由 test_load.py 自动完成 Locust headless 执行、CSV 导出与逐项断言性能退化会在测试阶段被直接捕获。由于 all_resources.py 的自动发现机制这套流程对新资源是零侵入的加文件、加清单条目即可把任意 API 纳入回归压测基线。【免费下载链接】OpenMetadataThe Open Context Layer for Data and AI , OpenMetadata is the open platform for building trusted data context and business semantics for humans, AI assistants, and agents.项目地址: https://gitcode.com/GitHub_Trending/op/OpenMetadata创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 OpenMetadata 压测实战：用 Locust 为 Data Quality API 构建自动化负载测试与阈值断言体系
OpenMetadata 压测实战用 Locust 为 Data Quality API 构建自动化负载测试与阈值断言体系【免费下载链接】OpenMetadataThe Open Context Layer for Data and AI , OpenMetadata is the open platform for building trusted data context and business semantics for humans, AI assistants, and agents.项目地址: https://gitcode.com/GitHub_Trending/op/OpenMetadata本篇介绍 OpenMetadata ingestion 模块中基于 Locust 的 API 负载测试框架如何在 ingestion/tests/load 目录下为新的服务端资源编写压测任务、如何通过manifest.yaml声明响应时间阈值以及整套「Locust 无头压测 pytest 断言」的自动化调用链读完后可直接照此为任意/api/v1/...端点补充可重复、可回归的性能测试。一、这套压测体系解决什么问题OpenMetadata 的服务端 API默认监听http://localhost:8585需要持续的性能基线保障否则一次慢查询回归可能无声地拖垮 UI 上的数据质量、目录浏览等页面。ingestion/tests/load正是为此设计的负载测试套件其核心组成test_load.pypytest 入口负责拼装 Locust 无头headless命令行参数、执行压测、再把 CSV 统计结果与阈值清单逐项比对test_resources/all_resources.pyLocust 用户与任务自动发现入口test_resources/tasks/每个待压测资源一个*.py任务文件当前已有 test_case_tasks.py 与 test_case_result_tasks.pytest_resources/manifest.yaml各请求的指标阈值清单utils.pyrun_load_test封装在 pytest 进程内直接调用 Locust 的main.main()。压测执行流程可以概括为三步Locust headless 发起压力 → 导出_stats.csv统计 → pytest 将每个指标与manifest.yaml中的阈值做assertLessEqual比较任一超标即用例失败从而把「性能是否退化」变成 CI 里可自动判定的断言。依赖上Locust 版本在 setup.py 中锁定为locust~2.32.0test_load.py 通过skipIf(sys.version_info (3, 9), ...)声明 Locust 仅在 Python 3.9 运行3.8 环境会自动跳过该用例。二、自动发现机制新增任务文件无需注册理解扩展方式之前先看 all_resources.py 如何把tasks/目录变成压测任务集。get_all_tasks_set()会遍历test_resources/tasks/*.py跳过base_前缀文件用importlib动态加载模块再经inspect.getmembers收集该模块中定义的所有类作为父 TaskSet 的tasksclass AllResources(TaskSet): Execute tasks for all resources classmethod def set_tasks(cls): tasks get_all_tasks_set() cls.tasks set(tasks) class All(HttpUser): host http://localhost:8585 wait_time constant(1) # closed workload AllResources.set_tasks() tasks [AllResources]几个值得注意的实现细节host硬编码为http://localhost:8585即要求本地或容器内OpenMetadata 服务端已启动且使用默认端口wait_time constant(1)表示每个虚拟用户执行完任务后固定等待 1 秒注释里称其为closed workload闭式工作负载——请求总量由并发用户数决定适合同步比对不同资源间的相对耗时由于是自动发现新增资源压测不需要修改all_resources.py只要把任务类放进tasks/即可。三、为新资源编写 Locust 任务3.1 文件与命名约定按照 README 的约定在test_resources/tasks下新增一个*.py文件文件名本身不限但社区习惯使用 Java 侧 Resource 类名按下划线分隔的写法例如TestCaseResource对应test_case_tasks.py。文件中最少需要导入两个 Locust 符号from locust import task, TaskSettask是装饰器把一个方法标记为压测任务其可选的整数参数表示任务权重——权重越大该任务被虚拟用户选中执行的概率越高任务类必须继承TaskSet它会被 all_resources.py 自动发现并挂载到用户类上。3.2 一个完整的任务定义示例以下示例完整展示了 README 中的任务写法带权重的时间范围查询、按参数分组命名请求class TestCaseResultTasks(TaskSet): Test case result resource load test def _list_test_case_results(self, start_ts: int, end_ts: int, days_range: str): List test case results for a given time range Args: start_ts (int): start timestamp end_ts (int): end timestamp range (str): for test_case in self.test_cases: fqn test_case.get(fullyQualifiedName) if fqn: self.client.get( f{TEST_CASE_RESULT_RESOURCE_PATH}/{fqn}, params{ # type: ignore startTs: start_ts, endTs: end_ts, }, authself.bearer, namef{TEST_CASE_RESULT_RESOURCE_PATH}/[fqn]/{days_range} ) task(3) def list_test_case_results_30_days(self): List test case results for the last 30 days. Weighted 3 now datetime.now() last_30_days int((now - timedelta(days30)).timestamp() * 1000) self._list_test_case_results(last_30_days, int(now.timestamp() * 1000), 30_days)实际仓库中的 test_case_result_tasks.py 正是这个模式的落地30_days权重 3、60_days权重 2、180_days权重 1让高频的短窗口查询在压测中占比更高更贴近真实 UI 的使用分布。3.3 鉴权on_start 钩子与 login_user压测请求如果带鉴权需要先把 token 放进self.bearer。标准做法是利用 Locust 的on_start钩子调用仓库内置的登录助手 login_userfrom _openmetadata_testutils.helpers.login_user import login_user class TestCaseResultTasks(TaskSet): Test case result resource load test [...] def on_start(self): Get a list of test cases to fetch results for self.bearer login_user(self.client) resp self.client.get(f{TEST_CASE_RESOURCE_PATH}, params{limit: 100}, authself.bearer) json resp.json() self.test_cases json.get(data, [])从源码看login_user会向/api/v1/users/login发送adminopen-metadata.org/ base64 密码YWRtaW4的登录请求取出accessToken并包装为BearerAuth对象实现见 login_user.py。这意味着压测前置条件是一个使用默认管理员账号的 OpenMetadata 实例且on_start里顺手预取了 100 条 test case 作为后续结果查询的数据来源——预取不计入task任务因此不会污染任务统计口径。3.4 必须定义 stop 方法README 特别强调IMPORTANT每个TaskSet类必须定义一个stop任务把控制权交还父级用户类class TestCaseResultTasks(TaskSet): Test case result resource load test [...] task def stop(self): self.interrupt()self.interrupt()会终止当前 TaskSet使虚拟用户回到AllResources层去随机选择下一个资源任务缺失它会导致用户永远困在第一个 TaskSet 里其他资源分不到流量。3.5 用 name 参数聚合带参请求的统计如果 URL 本身带变量如/api/v1/dataQuality/testCases/testCaseResults/{fqn}每条真实请求的 URL 都不同统计报告里会碎片化成上千行。解法是在self.client.get上传入name参数把同一模式的请求归并到同一统计项self.client.get( f{TEST_CASE_RESULT_RESOURCE_PATH}/{fqn}, params{ # type: ignore startTs: start_ts, endTs: end_ts, }, authself.bearer, namef{TEST_CASE_RESULT_RESOURCE_PATH}/[fqn]/{days_range} )name即请求在统计报告中显示的聚合名。README 给出的一份真实统计摘要 CSV节选即按此分组Type,Name,Request Count,Failure Count,Median Response Time,Average Response Time,Min Response Time,Max Response Time,Average Content Size,Requests/s,Failures/s,50%,66%,75%,80%,90%,95%,98%,99%,99.9%,99.99%,100% GET,/api/v1/dataQuality/testCases/testCaseResults/[fqn]/60_days,3510,0,13,16.2354597524217,5.146791999997902,100.67633299999557,84567.57407407407,49.30531562959204,0.0,13,17,20,21,28,35,45,56,92,100,100这个Name值同时也是 manifest.yaml 中声明阈值的 key两者必须严格一致断言逻辑才能命中。四、manifest.yaml把性能要求变成可断言的清单压测的最后一步是在 test_resources/manifest.yaml 中登记被测资源及其指标阈值。README 中的示例/api/v1/dataQuality/testCases/testCaseResults/[fqn]/30_days: type: GET 99%: 100 /api/v1/dataQuality/testCases/testCaseResults/[fqn]/60_days: type: GET 99%: 100语义是「该 GET 请求 99% 的请求耗时低于 100ms」。当前仓库中实际生效的阈值如下时间单位均为毫秒/api/v1/dataQuality/testCases/testCaseResults/[fqn]/30_days: type: GET 99%: 1000 /api/v1/dataQuality/testCases/testCaseResults/[fqn]/60_days: type: GET 99%: 1000 /api/v1/dataQuality/testCases/testCaseResults/[fqn]/180_days: type: GET 99%: 1000 /api/v1/dataQuality/testCases?limit10: type: GET 99%: 6000清单文件头注释也说明了取值规则key 可以是summaries/all__stats.csv中出现的任一指标列时间以毫秒表达1000ms 1s。4.1 断言是如何执行的test_load.py 读取压测产出的{summary_file}_stats.csvcsv.DictReader对每一行按Name查 manifest命中后先用type字段核对 HTTP 方法一致再遍历该资源的每个「指标 → 阈值」项用self.subTest逐项assertLessEqual(int(stat), threshold)。任一指标超标会输出形如99% for /api/v1/dataQuality/testCases?limit10 is greater than threshold的断言失败信息。由于使用了subTest单个指标超标不会中断其余指标的校验一次运行能暴露全部退化点。4.2 可用的指标清单README 列出了 manifest 中可声明的全部指标对应 Locust 导出 CSV 的列名Request Count请求总数Failure Count失败总数Median Response Time中位响应时间Average Response Time平均响应时间Min / Max Response Time最小 / 最大响应时间Average Content Size平均响应体大小Requests/s、Failures/s每秒请求 / 失败数50%、66%、75%、80%、90%、95%、98%、99%、99.9%、99.99%、100%各分位响应时间五、运行压测与关键参数pytest 用例 TestAllResources.test_all_resources 会把压测参数拼装成如下 Locust 命令行经 run_load_test 在进程内执行并断言退出码为 0locust --headless \ -H http://localhost:8585 \ --user 50 \ --spawn-rate 1 \ -f ingestion/tests/load/test_resources/all_resources.py \ --run-time 1m \ --only-summary \ --csv ingestion/tests/load/summaries/all_各参数含义与可调项参数当前值说明--headless固定无界面模式适合 CI-Hhttp://localhost:8585被测服务端地址与 all_resources.py 中的host保持一致--user50并发虚拟用户数可用环境变量LOCUST_USER覆盖--spawn-rate1每秒启动的用户数--run-time1m压测时长可用环境变量LOCUST_RUNTIME覆盖--csvsummaries/all_统计 CSV 输出前缀pytest 随后读取{前缀}_stats.csv做断言因此手动复现或调参时只需先起一个 OpenMetadata 服务端端口 8585并导入足够的数据例如 test case 与结果数据可用 ingestion 的示例管道灌入再按需修改LOCUST_USER/LOCUST_RUNTIME运行 pytest 用例即可得到与 CI 一致的压力画像与阈值判定。六、小结ingestion/tests/load提供了一条「压测即可执行测试」的完整闭环在test_resources/tasks/下新增任务文件用task(权重)定义带参数的请求on_start中完成登录与数据预取name参数聚合统计并务必实现stop任务在 manifest.yaml 中为该请求的聚合名声明指标阈值毫秒级由 test_load.py 自动完成 Locust headless 执行、CSV 导出与逐项断言性能退化会在测试阶段被直接捕获。由于 all_resources.py 的自动发现机制这套流程对新资源是零侵入的加文件、加清单条目即可把任意 API 纳入回归压测基线。【免费下载链接】OpenMetadataThe Open Context Layer for Data and AI , OpenMetadata is the open platform for building trusted data context and business semantics for humans, AI assistants, and agents.项目地址: https://gitcode.com/GitHub_Trending/op/OpenMetadata创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考