后端云原生容器编排【免费下载链接】pythonOfficial Python client library for kubernetes项目地址https://gitcode.com/gh_mirrors/python1/python点击查看免费下载本文以kubernetes.aio.dynamic.discovery_test模块为主体深入剖析 Kubernetes 官方 Python 客户端异步版中动态客户端Discoverer的测试设计与底层实现。读者将掌握 Discoverer 的缓存复用、资源反序列化、主资源与子资源解析三大核心机制理解如何针对异步动态客户端编写有效的单元与集成测试并了解对应的源码级运行原理。模块定位一份测试模块如何撑起 API 文档在仓库的文档源码 doc/source/kubernetes.aio.dynamic.discovery_test.rst 中该页面对应的正文仅有寥寥几行 Sphinx 指令.. automodule:: kubernetes.aio.dynamic.discovery_test :members: :show-inheritance: :undoc-members:这并非内容贫瘠而是 Sphinxautomodule指令的典型用法文档内容由被引用的 Python 模块自动提取生成。真正的“文档主体”是 kubernetes/aio/dynamic/discovery_test.py 这一测试模块——它通过三个测试用例完整覆盖了异步动态客户端中资源发现器Discoverer的关键行为缓存文件在客户端重建时是否被复用不触发重复写缓存从缓存文件反序列化得到的Resource是否与内存中新发现的对象完全等价get_resources_for_api_version对主资源resource与子资源subresource的拆分解析是否正确。因此本文以该测试模块为主线逐条还原其测试意图并在源码层面对应到 kubernetes/aio/dynamic/discovery.py、kubernetes/aio/dynamic/resource.py 与 kubernetes/aio/dynamic/client.py 的具体实现形成“测试 → 源码 → 原理”的完整闭环。被测对象Discoverer 在异步动态客户端中的角色动态客户端DynamicClient见 kubernetes/aio/dynamic/client.py允许在不依赖静态生成的 API 客户端类的前提下通过apiVersion与kind动态发现并操作 Kubernetes API 资源。其核心组件是client.resources属性所指向的Discoverer资源发现器。Discoverer的类定义位于 discovery.py其职责可概括为发现向 API Server 请求全部 API 分组/apis与分组内各版本构建“分组 → 版本 → 资源”的层级容器检索按prefix、group、api_version、kind等条件查找匹配的Resource对象缓存将发现结果序列化到本地临时文件下次启动时直接加载避免重复的网络发现请求。Discoverer是抽象基类定义了api_groups、search、discover三个抽象成员discovery.py并由两种加载策略实现实现类加载策略对应discover()LazyDiscoverer惰性加载仅发现分组/版本骨架资源在首次检索时才请求discovery.pyEagerDiscoverer急切加载实例化时一次性发现全部资源discovery.pyDynamicClient.__init__默认使用LazyDiscovererclient.py这也是测试模块所覆盖的默认路径。测试基座异步测试框架与 e2e 配置来源discovery_test.py的测试类继承自unittest.IsolatedAsyncioTestCasediscovery_test.py这是 Python 3.8 标准库提供的异步测试基类——每个测试方法以async def编写由框架自动调度事件循环无需手工管理 loop。class TestDiscoverer(unittest.IsolatedAsyncioTestCase): classmethod def setUpClass(cls): cls.config base.get_e2e_configuration()setUpClass中通过base.get_e2e_configuration()kubernetes/aio/e2e_test/base.py获取连接配置其行为分两个分支若~/.kube/config由kube_config.KUBE_CONFIG_DEFAULT_LOCATION决定默认取KUBECONFIG环境变量或家目录下.kube/config见 kubernetes/aio/config/kube_config.py存在则通过load_kube_config异步加载集群凭据否则依次探测https://127.0.0.1:8443与http://127.0.0.1:8080两个本地端点全部不可达时直接unittest.SkipTest跳过整个测试类。这意味着前两个缓存测试属于真实 e2e 测试它们要求环境中存在一个可连接的 Kubernetes 集群或通过 kubeconfig 指向的集群否则测试会被优雅跳过。第三个测试则通过unittest.mock完全脱离真实集群属于纯单元测试。用例一test_init_cache_from_file——缓存文件复用验证该用例discovery_test.py验证的核心命题是当再次创建DynamicClient时若缓存文件已存在且版本一致不应重复执行发现写入流程。测试逻辑分两步async with api_client.ApiClient(configurationself.config) as apic: client await DynamicClient(apic) await client.resources.get(api_versionv1, kindNode) mtime1 os.path.getmtime(client.resources._Discoverer__cache_file) async with api_client.ApiClient(configurationself.config) as apic: client await DynamicClient(apic) await client.resources.get(api_versionv1, kindNode) mtime2 os.path.getmtime(client.resources._Discoverer__cache_file) self.assertTrue(mtime1 mtime2)第一次创建客户端 → 触发LazyDiscoverer初始化 → 首次查询v1/Node触发分组内资源加载 → 发现结果写入缓存文件记录其mtime第二次重新创建客户端 → 初始化时命中缓存 → 直接反序列化缓存 → 查询同样成功断言两次mtime相等证明第二次没有再次调用_write_cache覆盖缓存文件。要理解该断言为何成立需要回到Discoverer.__init_cache的实现discovery.pyasync def __init_cache(self, refreshFalse): if refresh or not os.path.exists(self.__cache_file): self._cache {library_version: __version__} refresh True else: try: with open(self.__cache_file, r) as f: self._cache json.load(f, clspartial(CacheDecoder, self.client)) if self._cache.get(library_version) ! __version__: # Version mismatch, need to refresh cache await self.invalidate_cache() except Exception as e: logger.error(load cache error: %s, e) await self.invalidate_cache() await self._load_server_info() await self.discover() if refresh: self._write_cache()关键设计有三点缓存文件存在且可正常反序列化时refresh保持Falsediscover()结束后不写缓存文件缓存中记录library_version即kubernetes.aio.__version__一旦库版本变化即视为缓存失效并刷新discovery.py加载/反序列化异常时也会静默回退到全量刷新保证客户端永远可用。缓存文件路径本身也值得注意Discoverer.__init__discovery.py以configuration.host为输入计算md5生成形如osrcp-md5.json的文件并放入系统临时目录tempfile.gettempdir()不同集群地址对应不同缓存文件天然实现了多集群缓存隔离。测试中直接访问私有属性_Discoverer__cache_file正是为了读取该路径的mtime。用例二test_cache_decoder_resource_and_subresource——缓存反序列化等价性该用例discovery_test.py验证的是CacheDecoder的正确性从缓存文件解码出的Resource对象应与内存中重新发现的对象在结构上完全一致。测试同样分两个客户端生命周期# 第一个客户端先 invalidate_cache 强制刷新得到内存中的 deploy1 async with api_client.ApiClient(configurationself.config) as apic: client await DynamicClient(apic) await client.resources.invalidate_cache() deploy1 await client.resources.get(kindDeployment, api_versionapps/v1) # 第二个客户端此时缓存文件已生成走 CacheDecoder 反序列化路径得到 deploy2 async with api_client.ApiClient(configurationself.config) as apic: client2 await DynamicClient(apic) deploy2 await client2.resources.get(kindDeployment, api_versionapps/v1) deploy2.client deploy1.client self.assertDictEqual(deploy1.to_dict(), deploy2.to_dict())其中invalidate_cache()的实现discovery.py是await self.__init_cache(refreshTrue)即强制走“重建缓存并写盘”路径确保第一个客户端拿到的是新鲜发现的结果。deploy2.client deploy1.client这一行是测试的“平衡操作”Resource.client持有各自的ApiClient实例to_dict()序列化结果不应包含 client 引用因此统一 client 后直接比较to_dict()即可。这里的关键在于CacheEncoder / CacheDecoder 的对称性序列化CacheEncoder.default统一调用对象的to_dict()discovery.pyResource.to_dict()resource.py输出包含_type标记字段与全部属性反序列化CacheDecoder.object_hook依据_type字段分发到Resource、ResourceList、ResourceGroup三类对象discovery.py。def object_hook(self, obj): if _type not in obj: return obj _type obj.pop(_type) if _type Resource: return Resource(clientself.client, **obj) elif _type ResourceList: return ResourceList(self.client, **obj) elif _type ResourceGroup: return ResourceGroup(obj[preferred], resourcesself.object_hook(obj[resources])) return obj注意CacheDecoder.__init__接收client参数并通过functools.partial注入discovery.py这正是反序列化时能为新Resource重新绑定当前ApiClient的关键——这也是为什么测试断言deploy1.to_dict() deploy2.to_dict()能够成立即使底层 client 不同两个对象在“发现层面”的结构化描述完全一致。用例三test_get_resources_for_api_version——主资源与子资源解析第三个用例discovery_test.py是纯单元测试通过patch替换真实网络调用直接验证Discoverer.get_resources_for_api_version的解析逻辑patch(kubernetes.aio.dynamic.discovery.Discoverer.get_resources_for_api_version, new_callableAsyncMock) async def test_get_resources_for_api_version(self, mock_get_resources): mock_get_resources.return_value { resources: [{name: pods, kind: Pod}], subresources: { virtualmachineinstances: { sev/fetchcertchain: {name: virtualmachineinstances/sev/fetchcertchain} } } } mock_client MagicMock() mock_client.configuration.host https://mock-host discoverer Discoverer(clientmock_client) response await discoverer.get_resources_for_api_version(api, v1, pods, True) self.assertEqual(response[resources][0][name], pods) self.assertEqual(response[resources][0][kind], Pod) self.assertIn(virtualmachineinstances, response[subresources]) self.assertIn(sev/fetchcertchain, response[subresources][virtualmachineinstances])这个用例实际上验证了API Server 返回结构中的两类条目如何被归类。真实的解析逻辑位于 discovery.py其核心是按资源名中是否包含/来区分主资源与子资源resources_raw list(filter(lambda r: / not in r[name], resources_response)) subresources_raw list(filter(lambda r: / in r[name], resources_response)) for subresource in subresources_raw: # Handle resources with 2 parts in their name resource, name subresource[name].split(/, 1) if not subresources.get(resource): subresources[resource] {} subresources[resource][name] subresource例如 Kubernetes 中pods/status、virtualmachineinstances/sev/fetchcertchain这类“资源名含/”的条目都属于子资源而pods、deployments等属于主资源。解析后主资源被包装为Resource对象discovery.py同时为每个 kind 额外生成对应的ResourceList子资源则按“父资源 → 子资源名”组织为嵌套字典供Resource.subresources使用见 resource.py 中对子资源的Subresource包装。另外值得注意get_resources_for_api_version在请求失败时会做容错处理——当 API Server 返回 503ServiceUnavailableError或内容类型异常ContentTypeError例如 503 时返回text/plain而非 JSON时按空资源列表处理discovery.py保证单个分组发现失败不会拖垮整个客户端初始化。底层机制纵览从发现到检索的完整链路分组发现parse_api_groups无论惰性还是急切策略Discoverer都会先执行parse_api_groupsdiscovery.py向/apis发起GET获得集群全部 API 分组及各组版本列表以default_groups为骨架api下的核心v1分组 apis前缀discovery.py将发现到的分组逐个填入结果为三层嵌套结构{prefix: {group: {version: ResourceGroup}}}并写入self._cache[resources]。资源检索search与getLazyDiscoverer.searchdiscovery.py是默认客户端走的路先在内存缓存树中按[prefix, group, api_version, kind, 额外参数]逐层查找若未命中触发invalidate_cache()全量刷新后重试一次这保证了新安装的 CRD 等资源能被“二次发现”找到若本次查找触发了新的分组资源加载会置位__update_cache并在结束时写缓存。Discoverer.getdiscovery.py则是search的“精确取一”版本多结果时优先匹配api_version再排除ResourceList类型最终唯一命中才返回零结果抛ResourceNotFoundError多结果抛ResourceNotUniqueError两者均定义于 kubernetes/aio/dynamic/exceptions.py。示例中的真实调用异步动态客户端的完整用法可以参考 examples_asyncio/dynamic-client/cluster_scoped_custom_resource.py其中展示了本文所述机制的组合使用async with api_client.ApiClient(configurationconfig) as apic: client await DynamicClient(apic) crd_api await client.resources.get( api_versionapiextensions.k8s.io/v1, kindCustomResourceDefinition )创建 CRD 后立即查询新 kind 会得到ResourceNotFoundError示例在except分支中await asyncio.sleep(2)等待发现层刷新cluster_scoped_custom_resource.py这正是LazyDiscoverer.search“未命中则失效缓存并重试”设计所支撑的容错语义。如何运行该测试discovery_test.py同时兼容直接执行与测试框架运行# 直接运行模块尾部自带 unittest.main() python -m kubernetes.aio.dynamic.discovery_test # 或通过 unittest 发现机制 python -m unittest kubernetes.aio.dynamic.discovery_test运行前提说明前两个缓存测试依赖可达的 Kubernetes 集群若本机存在~/.kube/config或设置了KUBECONFIG将加载其中配置否则回退到本地127.0.0.1:8443/8080探测全部失败时测试被unittest.SkipTest跳过第三个测试全程使用AsyncMock/MagicMock无需集群即可运行测试文件位于异步客户端目录下运行前需保证kubernetes.aio相关依赖如aiohttp已安装。小结kubernetes.aio.dynamic.discovery_test虽名为“测试”却是理解异步动态客户端资源发现机制的最佳入口。三个用例分别锚定三条核心契约缓存复用缓存文件在跨客户端生命周期内被复用且不产生多余写盘对应__init_cache的refresh分支设计反序列化等价CacheEncoder/CacheDecoder对称处理使缓存解码结果与全新发现结果保持一致对应_type标记与object_hook分发资源/子资源拆分API Server 返回中的资源名按是否含/归类为主资源与子资源对应get_resources_for_api_version的过滤逻辑。对照 discovery.py 与 resource.py 的源码可以进一步确认惰性发现、按需加载、版本驱动的缓存失效、以及失败容错共同构成了这套动态发现体系在生产环境可用的基础。对于希望在异步场景下动态操作 CRD 或集群资源的开发者理解这三条契约就能准确预测client.resources.get(...)的行为边界。赞分享后端云原生容器编排【免费下载链接】pythonOfficial Python client library for kubernetes项目地址https://gitcode.com/gh_mirrors/python1/python点击查看免费下载相关推荐Kubernetes Python 客户端异步 OIDC 认证Token 刷新机制与测试实现解析Kubernetes Python 客户端异步 OIDC 认证Token 刷新机制与测试实现解析 本篇文章以 kubernetes.aio.config.op后端云原生容器编排Kubernetes Python 异步客户端 ExecProvider 执行插件认证实现原理与测试剖析Kubernetes Python 异步客户端 ExecProvider 执行插件认证实现原理与测试剖析 本篇文章以官方 Kubernetes Python后端云原生容器编排Kubernetes Python 客户端异步动态客户端端到端测试深度解析kubernetes.aio.dynamic.client_testKubernetes Python 客户端异步动态客户端端到端测试深度解析kubernetes.aio.dynamic.client_test 本文以官方后端云原生容器编排上一篇OneUptime 工作流组件完全指南组件目录、配置参数与数据操作实战下一篇StringSifter功能全解析从命令行参数到批量处理的完整用户指南创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考