尧图网站建设 
KubeEdge 多 cloudcore 高可用下的 Router 消息路由Node 注解标记 请求二次转发实现【免费下载链接】kubeedgeKubernetes Native Edge Computing Framework (project under CNCF)项目地址: https://gitcode.com/GitHub_Trending/ku/kubeedge在 KubeEdge 中当 cloudcore 以多副本高可用方式部署时云端应用通过 Router 向边缘端发送的自定义消息可能被负载均衡转发到任意一个 cloudcore 实例而该实例未必是目标 edgecore 实际建立会话的那一个消息因此无法送达。本文基于设计提案 ha_router.md 与当前仓库源码完整解析这套“Node 注解记录归属 Router 请求二次转发”的高可用消息路由机制包括 Node 注解cloudcore的写入与修正时机、Router REST 监听器中判断归属并转发的完整调用链以及关键参数端口、超时、重试、消息体上限的默认值与来源。背景与问题提案原文docs/proposals/sig-scalability/ha_router.md给出的 Motivation 是当 cloudcore 采用高可用部署、多个 cloudcore 实例同时运行时如果用户应用想从云端向某个边缘节点发送自定义消息并不知道该从哪个 cloudcore 发出才能到达对应的 edgecore。用户即使通过 VIP 或某个特定 cloudcore 的 IP 发送消息也无法保证消息被正确投递。设计目标是在 cloudcore 高可用部署场景下用户从 VIP 或任意特定 cloudcore IP 发出消息都能确保消息最终被投递到目标 edgecore。这一问题的根源在于 cloudcore 高可用部署的连接模型。按照提案 Background 部分的描述并参考 cloudcore 高可用设计 的背景典型部署方式是 haproxy 等负载均衡组件后面挂多个 cloudcore所有实例都接受 edgecore 连接并工作而每个 edgecore 在同一时刻只会成功连接到一个 cloudcore且由于存在负载均衡具体连接到了哪个 cloudcore 是不确定的。三类自定义消息中需要处理的只有两个下行方向KubeEdge Router 承载三类用户自定义消息流见 router 模块cloud app - edge app云端应用到边缘应用cloud app - edge mqtt云端应用到边缘端 MQTT 消息总线edge mqtt - cloud app边缘 MQTT 到云端应用。提案指出第三类边缘到云端在多 cloudcore 场景下不存在问题因为每个 edgecore 只连接一个 cloudcore边缘侧消息天然通过该 edgecore 已建立的连接上报到对应的 cloudcore再由云端转发给云端应用。因此只需要解决cloud app - edge app和cloud app - edge mqtt这两个下行方向的“找对 cloudcore”问题这正是下面设计细节要覆盖的路径。设计总览三步完成高可用路由提案 Design Details 给出的方案由三个步骤构成节点创建时打标边缘节点加入集群时cloudcore 会在 apiserver 上游创建 Node 资源创建 Node 之前为该 Node 添加注解cloudcore: {cloudcore-ip}其中{cloudcore-ip}是 edgecore 当前连接到的这个 cloudcore 实例的 IP 地址。会话建立后校订cloudhub 为边缘节点建立会话后通过 apiserver 获取该节点资源读取cloudcore注解的值判断其是否与当前实际连接的 cloudcore IP 一致若不一致例如 edgecore 重连后落到了另一个 cloudcore则将注解值更新为当前 cloudcore 的 IP 并写回 apiserver。请求转发时查表云端应用向目标边缘节点的 app 或 mqtt 发起请求先经过高可用组件VIP/负载均衡转发到多个 cloudcore 中的某一个该 cloudcore 从 URL 中解析出 nodeName向 apiserver 查询该 Node 的注解读取cloudcore字段的值与本机 IP 比较——一致则本地处理不一致则把请求转发到注解所指的那个 cloudcore。下面结合当前仓库源码逐条印证这三步的实现。实现一创建 Node 时写入cloudcore注解对应设计步骤 1。Node 注解的 key 由公共常量定义位于 common/constants/default.goCloudConfigMapName cloudcore EdgeMappingCloudKey cloudcoreNode 的创建逻辑在 edgecontroller 的上游控制器 upstream.go 中cloudcore 在Nodes().Create()之前先用util.GetLocalIP取得本机当前 cloudcore 实例IP并写入注解node.Name name hostnameOverride : kubeedgeutil.GetHostname() localIP, err : kubeedgeutil.GetLocalIP(hostnameOverride) if err ! nil { return nil, fmt.Errorf(failed to get cloudcore localIP with err:%v, err) } if node.Annotations nil { node.Annotations make(map[string]string) } node.Annotations[common.EdgeMappingCloudKey] localIP node, err uc.kubeClient.CoreV1().Nodes().Create(utilcontext.WithEdgeNode(context.Background(), nodeID), node, metaV1.CreateOptions{})也就是说Node 资源上的cloudcore注解始终记录的是“最近一次为该边缘节点建立连接/创建节点的那个 cloudcore 实例 IP”它与提案中cloudcore: {cloudcore-ip}的语义一一对应。实现二会话建立后由 cloudhub 触发注解校订对应设计步骤 2。校订逻辑封装为独立函数UpdateAnnotation位于 upstream.gofunc UpdateAnnotation(ctx context.Context, nodeName string) error { node, err : client.GetKubeClient().CoreV1().Nodes().Get(ctx, nodeName, metaV1.GetOptions{}) if err ! nil { return fmt.Errorf(failed to get node:%s,err:%v, nodeName, err) } hostnameOverride : kubeedgeutil.GetHostname() localIP, err : kubeedgeutil.GetLocalIP(hostnameOverride) if err ! nil { return fmt.Errorf(failed to get cloudcore localIP with err:%v, err) } if value, ok : node.Annotations[comconstants.EdgeMappingCloudKey]; ok { if value localIP { return nil } } if node.Annotations nil { node.Annotations make(map[string]string) } node.Annotations[comconstants.EdgeMappingCloudKey] localIP _, err client.GetKubeClient().CoreV1().Nodes().Update(ctx, node, metaV1.UpdateOptions{}) ... }其行为与提案完全一致先 Get 节点、比较注解值与本机 IP相同则直接返回避免无意义的 apiserver 写放大不同才 Update 覆盖。触发时机在 cloudhub 的消息处理链上。从源码调用关系看message_handler.go 在处理 edgecore 上线/会话消息时调用controller.UpdateAnnotation(context.TODO(), nodeID)——即提案所述“cloudhub 为边缘节点创建会话之后”的校订时机。这样当 edgecore 断开重连并落到另一个 cloudcore 时节点注解会被新实例修正为新的 IP保证注解始终反映“当前哪个 cloudcore 持有该边缘节点”的事实。实现三Router 监听器中的归属判断与二次转发对应设计步骤 3也是整个机制中最核心的部分。Router 的 REST 监听器实现位于 cloud/pkg/router/listener/http.go其httpHandler的处理流程与提案描述的判定逻辑一一对应。从 URL 中解析 nodeName入口httpHandlerhttp.go首先校验 URL 分段数然后把路径的第一段uriSections[1]当作 edgeNodeName——这正是提案中“从 URL 中提取 nodeName”的落地。请求体通过http.MaxBytesReader限制在MaxMessageBytes 12 * (1 20)即 12 MiB 的上限防止超大消息。查询 Node 注解并比较 IP解析出节点名后调用GetEdgeToCloudCoreIP查询目标节点归属http.gofunc GetEdgeToCloudCoreIP(ctx context.Context, nodeName string) (string, error) { node, err : client.GetKubeClient().CoreV1().Nodes().Get(ctx, nodeName, metav1.GetOptions{}) if err ! nil { return , fmt.Errorf(failed to get node:%s,err:%v, nodeName, err) } cloudCoreIP, ok : node.Annotations[constants.EdgeMappingCloudKey] if !ok { return , fmt.Errorf(no corresponding cloudcore was found for edgeNode:%s, nodeName) } return cloudCoreIP, nil }可以看到通过 kube client 按 nodeName 从 apiserver 取 Node读取cloudcore注解若注解缺失节点未走 KubeEdge 接入流程直接报错。随后httpHandler用util.GetLocalIP取本机 IP进入核心分支http.go注解 IP ! 本机 IP构造转发请求协议跟随原始请求r.TLS ! nil时用https否则http目标地址为scheme://targetCloudCoreIP:port 原始 RequestURI并原样拷贝请求方法与请求头然后调用requestForward把整包请求转发给目标 cloudcore。注解 IP 本机 IP本地匹配路由规则matchedPath从 handler 注册表中取出对应处理器组装messageID、原始请求、超时时间与消息体等参数交给 handler 完成本地投递。转发细节与重试机制requestForwardhttp.go使用http.Client发起转发请求把上游 cloudcore 返回的响应头逐项写回给调用方非 200 状态时读取响应体作为错误信息向上抛出200 时直接io.Copy透传响应体并记录forwarded request to %s successfully日志。值得注意的是整个“查注解 → 判归属 → 转发或本地处理”的流程被包裹在retry-go的重试框架内http.go最多 3 次尝试、固定 1 秒间隔retry.Attempts(3)、retry.Delay(1*time.Second)、retry.DelayType(retry.FixedDelay)。这一设计可以推断用于吸收 apiserver 短暂抖动或目标 cloudcore 瞬时不可用等异常重试全部失败后以500 Internal Server Error响应客户端。相关配置参数默认值Router REST 监听器的初始化在 http.go 的InitHandler中完成从 router 配置 读取参数关键默认值如下参数默认值说明Port9443Router REST 监听端口高可用组件VIP/负载均衡通常将 9443 端口指向全部 cloudcore 实例转发请求也打向目标实例的同一端口RestTimeout60 秒交给消息处理器的本地投递超时rh.restTimeout消息体上限12 MiBMaxMessageBytes请求体与响应体均受此限制重试策略3 次 / 固定 1sretry-go包裹的归属判定与投递流程适用前提与已知限制基于当前仓库代码使用该机制时应注意以下前提与限制Node 注解是路由依据只有走 KubeEdge 接入流程、由 cloudcore 创建并打过cloudcore注解的节点才能被正确路由对未打注解的节点GetEdgeToCloudCoreIP会返回no corresponding cloudcore was found错误。高可用组件负责第一跳该设计假定 VIP/负载均衡如 haproxy已把云端请求均分或转发到各 cloudcore本机制解决的是“落到非归属实例后的二次转发”两者配合才构成完整链路。TLS 尚未启用Serve方法中存在// TODO: add tls for router注释当前ListenAndServe以明文 HTTP 启动转发时协议跟随原始请求因此目前整条链路以 HTTP 为主生产环境需结合现有 cloudcore 高可用方案自行评估网络层安全。节点名校验待完善isNodeName当前恒返回true源码中带有// TODO: check node name标记即 URL 第一段默认即按节点名处理调用方需保证 URL 格式规范。上行方向不在机制范围内如前所述edge mqtt - cloud app方向依赖 edgecore 与其唯一连接 cloudcore 之间的既有连接不经过本转发逻辑。小结KubeEdge 的 Router 消息高可用方案用三个动作闭合了多 cloudcore 场景下的消息投递问题创建 Node 时写入cloudcore注解upstream.go、cloudhub 会话建立后由UpdateAnnotation校订归属upstream.go message_handler.go、Router 监听器按注解 IP 做本地处理或二次转发http.go。整套机制以 apiserver 中的 Node 注解作为“边缘节点归属”的单一事实来源使任意 cloudcore 实例都能成为有效的消息入口对上游用户应用完全透明。【免费下载链接】kubeedgeKubernetes Native Edge Computing Framework (project under CNCF)项目地址: https://gitcode.com/GitHub_Trending/ku/kubeedge创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 KubeEdge 多 cloudcore 高可用下的 Router 消息路由：Node 注解标记 + 请求二次转发实现
KubeEdge 多 cloudcore 高可用下的 Router 消息路由Node 注解标记 请求二次转发实现【免费下载链接】kubeedgeKubernetes Native Edge Computing Framework (project under CNCF)项目地址: https://gitcode.com/GitHub_Trending/ku/kubeedge在 KubeEdge 中当 cloudcore 以多副本高可用方式部署时云端应用通过 Router 向边缘端发送的自定义消息可能被负载均衡转发到任意一个 cloudcore 实例而该实例未必是目标 edgecore 实际建立会话的那一个消息因此无法送达。本文基于设计提案 ha_router.md 与当前仓库源码完整解析这套“Node 注解记录归属 Router 请求二次转发”的高可用消息路由机制包括 Node 注解cloudcore的写入与修正时机、Router REST 监听器中判断归属并转发的完整调用链以及关键参数端口、超时、重试、消息体上限的默认值与来源。背景与问题提案原文docs/proposals/sig-scalability/ha_router.md给出的 Motivation 是当 cloudcore 采用高可用部署、多个 cloudcore 实例同时运行时如果用户应用想从云端向某个边缘节点发送自定义消息并不知道该从哪个 cloudcore 发出才能到达对应的 edgecore。用户即使通过 VIP 或某个特定 cloudcore 的 IP 发送消息也无法保证消息被正确投递。设计目标是在 cloudcore 高可用部署场景下用户从 VIP 或任意特定 cloudcore IP 发出消息都能确保消息最终被投递到目标 edgecore。这一问题的根源在于 cloudcore 高可用部署的连接模型。按照提案 Background 部分的描述并参考 cloudcore 高可用设计 的背景典型部署方式是 haproxy 等负载均衡组件后面挂多个 cloudcore所有实例都接受 edgecore 连接并工作而每个 edgecore 在同一时刻只会成功连接到一个 cloudcore且由于存在负载均衡具体连接到了哪个 cloudcore 是不确定的。三类自定义消息中需要处理的只有两个下行方向KubeEdge Router 承载三类用户自定义消息流见 router 模块cloud app - edge app云端应用到边缘应用cloud app - edge mqtt云端应用到边缘端 MQTT 消息总线edge mqtt - cloud app边缘 MQTT 到云端应用。提案指出第三类边缘到云端在多 cloudcore 场景下不存在问题因为每个 edgecore 只连接一个 cloudcore边缘侧消息天然通过该 edgecore 已建立的连接上报到对应的 cloudcore再由云端转发给云端应用。因此只需要解决cloud app - edge app和cloud app - edge mqtt这两个下行方向的“找对 cloudcore”问题这正是下面设计细节要覆盖的路径。设计总览三步完成高可用路由提案 Design Details 给出的方案由三个步骤构成节点创建时打标边缘节点加入集群时cloudcore 会在 apiserver 上游创建 Node 资源创建 Node 之前为该 Node 添加注解cloudcore: {cloudcore-ip}其中{cloudcore-ip}是 edgecore 当前连接到的这个 cloudcore 实例的 IP 地址。会话建立后校订cloudhub 为边缘节点建立会话后通过 apiserver 获取该节点资源读取cloudcore注解的值判断其是否与当前实际连接的 cloudcore IP 一致若不一致例如 edgecore 重连后落到了另一个 cloudcore则将注解值更新为当前 cloudcore 的 IP 并写回 apiserver。请求转发时查表云端应用向目标边缘节点的 app 或 mqtt 发起请求先经过高可用组件VIP/负载均衡转发到多个 cloudcore 中的某一个该 cloudcore 从 URL 中解析出 nodeName向 apiserver 查询该 Node 的注解读取cloudcore字段的值与本机 IP 比较——一致则本地处理不一致则把请求转发到注解所指的那个 cloudcore。下面结合当前仓库源码逐条印证这三步的实现。实现一创建 Node 时写入cloudcore注解对应设计步骤 1。Node 注解的 key 由公共常量定义位于 common/constants/default.goCloudConfigMapName cloudcore EdgeMappingCloudKey cloudcoreNode 的创建逻辑在 edgecontroller 的上游控制器 upstream.go 中cloudcore 在Nodes().Create()之前先用util.GetLocalIP取得本机当前 cloudcore 实例IP并写入注解node.Name name hostnameOverride : kubeedgeutil.GetHostname() localIP, err : kubeedgeutil.GetLocalIP(hostnameOverride) if err ! nil { return nil, fmt.Errorf(failed to get cloudcore localIP with err:%v, err) } if node.Annotations nil { node.Annotations make(map[string]string) } node.Annotations[common.EdgeMappingCloudKey] localIP node, err uc.kubeClient.CoreV1().Nodes().Create(utilcontext.WithEdgeNode(context.Background(), nodeID), node, metaV1.CreateOptions{})也就是说Node 资源上的cloudcore注解始终记录的是“最近一次为该边缘节点建立连接/创建节点的那个 cloudcore 实例 IP”它与提案中cloudcore: {cloudcore-ip}的语义一一对应。实现二会话建立后由 cloudhub 触发注解校订对应设计步骤 2。校订逻辑封装为独立函数UpdateAnnotation位于 upstream.gofunc UpdateAnnotation(ctx context.Context, nodeName string) error { node, err : client.GetKubeClient().CoreV1().Nodes().Get(ctx, nodeName, metaV1.GetOptions{}) if err ! nil { return fmt.Errorf(failed to get node:%s,err:%v, nodeName, err) } hostnameOverride : kubeedgeutil.GetHostname() localIP, err : kubeedgeutil.GetLocalIP(hostnameOverride) if err ! nil { return fmt.Errorf(failed to get cloudcore localIP with err:%v, err) } if value, ok : node.Annotations[comconstants.EdgeMappingCloudKey]; ok { if value localIP { return nil } } if node.Annotations nil { node.Annotations make(map[string]string) } node.Annotations[comconstants.EdgeMappingCloudKey] localIP _, err client.GetKubeClient().CoreV1().Nodes().Update(ctx, node, metaV1.UpdateOptions{}) ... }其行为与提案完全一致先 Get 节点、比较注解值与本机 IP相同则直接返回避免无意义的 apiserver 写放大不同才 Update 覆盖。触发时机在 cloudhub 的消息处理链上。从源码调用关系看message_handler.go 在处理 edgecore 上线/会话消息时调用controller.UpdateAnnotation(context.TODO(), nodeID)——即提案所述“cloudhub 为边缘节点创建会话之后”的校订时机。这样当 edgecore 断开重连并落到另一个 cloudcore 时节点注解会被新实例修正为新的 IP保证注解始终反映“当前哪个 cloudcore 持有该边缘节点”的事实。实现三Router 监听器中的归属判断与二次转发对应设计步骤 3也是整个机制中最核心的部分。Router 的 REST 监听器实现位于 cloud/pkg/router/listener/http.go其httpHandler的处理流程与提案描述的判定逻辑一一对应。从 URL 中解析 nodeName入口httpHandlerhttp.go首先校验 URL 分段数然后把路径的第一段uriSections[1]当作 edgeNodeName——这正是提案中“从 URL 中提取 nodeName”的落地。请求体通过http.MaxBytesReader限制在MaxMessageBytes 12 * (1 20)即 12 MiB 的上限防止超大消息。查询 Node 注解并比较 IP解析出节点名后调用GetEdgeToCloudCoreIP查询目标节点归属http.gofunc GetEdgeToCloudCoreIP(ctx context.Context, nodeName string) (string, error) { node, err : client.GetKubeClient().CoreV1().Nodes().Get(ctx, nodeName, metav1.GetOptions{}) if err ! nil { return , fmt.Errorf(failed to get node:%s,err:%v, nodeName, err) } cloudCoreIP, ok : node.Annotations[constants.EdgeMappingCloudKey] if !ok { return , fmt.Errorf(no corresponding cloudcore was found for edgeNode:%s, nodeName) } return cloudCoreIP, nil }可以看到通过 kube client 按 nodeName 从 apiserver 取 Node读取cloudcore注解若注解缺失节点未走 KubeEdge 接入流程直接报错。随后httpHandler用util.GetLocalIP取本机 IP进入核心分支http.go注解 IP ! 本机 IP构造转发请求协议跟随原始请求r.TLS ! nil时用https否则http目标地址为scheme://targetCloudCoreIP:port 原始 RequestURI并原样拷贝请求方法与请求头然后调用requestForward把整包请求转发给目标 cloudcore。注解 IP 本机 IP本地匹配路由规则matchedPath从 handler 注册表中取出对应处理器组装messageID、原始请求、超时时间与消息体等参数交给 handler 完成本地投递。转发细节与重试机制requestForwardhttp.go使用http.Client发起转发请求把上游 cloudcore 返回的响应头逐项写回给调用方非 200 状态时读取响应体作为错误信息向上抛出200 时直接io.Copy透传响应体并记录forwarded request to %s successfully日志。值得注意的是整个“查注解 → 判归属 → 转发或本地处理”的流程被包裹在retry-go的重试框架内http.go最多 3 次尝试、固定 1 秒间隔retry.Attempts(3)、retry.Delay(1*time.Second)、retry.DelayType(retry.FixedDelay)。这一设计可以推断用于吸收 apiserver 短暂抖动或目标 cloudcore 瞬时不可用等异常重试全部失败后以500 Internal Server Error响应客户端。相关配置参数默认值Router REST 监听器的初始化在 http.go 的InitHandler中完成从 router 配置 读取参数关键默认值如下参数默认值说明Port9443Router REST 监听端口高可用组件VIP/负载均衡通常将 9443 端口指向全部 cloudcore 实例转发请求也打向目标实例的同一端口RestTimeout60 秒交给消息处理器的本地投递超时rh.restTimeout消息体上限12 MiBMaxMessageBytes请求体与响应体均受此限制重试策略3 次 / 固定 1sretry-go包裹的归属判定与投递流程适用前提与已知限制基于当前仓库代码使用该机制时应注意以下前提与限制Node 注解是路由依据只有走 KubeEdge 接入流程、由 cloudcore 创建并打过cloudcore注解的节点才能被正确路由对未打注解的节点GetEdgeToCloudCoreIP会返回no corresponding cloudcore was found错误。高可用组件负责第一跳该设计假定 VIP/负载均衡如 haproxy已把云端请求均分或转发到各 cloudcore本机制解决的是“落到非归属实例后的二次转发”两者配合才构成完整链路。TLS 尚未启用Serve方法中存在// TODO: add tls for router注释当前ListenAndServe以明文 HTTP 启动转发时协议跟随原始请求因此目前整条链路以 HTTP 为主生产环境需结合现有 cloudcore 高可用方案自行评估网络层安全。节点名校验待完善isNodeName当前恒返回true源码中带有// TODO: check node name标记即 URL 第一段默认即按节点名处理调用方需保证 URL 格式规范。上行方向不在机制范围内如前所述edge mqtt - cloud app方向依赖 edgecore 与其唯一连接 cloudcore 之间的既有连接不经过本转发逻辑。小结KubeEdge 的 Router 消息高可用方案用三个动作闭合了多 cloudcore 场景下的消息投递问题创建 Node 时写入cloudcore注解upstream.go、cloudhub 会话建立后由UpdateAnnotation校订归属upstream.go message_handler.go、Router 监听器按注解 IP 做本地处理或二次转发http.go。整套机制以 apiserver 中的 Node 注解作为“边缘节点归属”的单一事实来源使任意 cloudcore 实例都能成为有效的消息入口对上游用户应用完全透明。【免费下载链接】kubeedgeKubernetes Native Edge Computing Framework (project under CNCF)项目地址: https://gitcode.com/GitHub_Trending/ku/kubeedge创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考