手机网站制作方案,网站建设佰首选金手指三十,wordpress即时新闻,中国高定十大品牌简介#xff1a;OpenYurt 是业界首个非侵入的边缘计算云原生开源项目#xff0c;通过边缘自治#xff0c;云边协同#xff0c;边缘单元化#xff0c;边缘流量闭环等能力为用户提供云边一体化的使用体验。在 Openyurt 里边缘网络可以使用数据过滤框架在不同节点池里实现边缘…简介OpenYurt 是业界首个非侵入的边缘计算云原生开源项目通过边缘自治云边协同边缘单元化边缘流量闭环等能力为用户提供云边一体化的使用体验。在 Openyurt 里边缘网络可以使用数据过滤框架在不同节点池里实现边缘流量闭环能力。
作者应健健新华智云计算中心
OpenYurt 是业界首个非侵入的边缘计算云原生开源项目通过边缘自治云边协同边缘单元化边缘流量闭环等能力为用户提供云边一体化的使用体验。在 Openyurt 里边缘网络可以使用数据过滤框架在不同节点池里实现边缘流量闭环能力。
Yurthub 数据过滤框架解析
Yurthub 本质上是一层 kube-apiserver 的代理在代理的基础上加了一层 cache一来保证边缘节点离线的情况下可以使用本地 cache 保证业务稳定性,有效的解决了边缘自治的问题。二来可以降低大量的 list watch 操作对云上 api 产生一定的负载。
Yurthub 的数据过滤通过节点上的 pod 以及 kubelet 的请求通过 Load Balancer 发送给 kube-apiserver代理接收到响应消息进行数据过滤处理之后再将过滤后的数据返回给请求方。如果节点是边缘节点会根据请求类型对响应请求体中的资源进行本地缓存如果是云端节点考虑到网络状态良好不进行本地缓存。
Yurthub 的过滤框架实现原理图 Yurthub 目前包含四种过滤规则通过 addons 请求的 user-agentresourceverb 判断经过那个过滤器进行相应的数据过滤。
四种过滤规则功能及实现
ServiceTopologyFilter
主要针对 EndpointSlice 资源进行数据过滤, 但 Endpoint Slice 特性需要在 Kubernetes v1.18 或以上版本才能支持如果在 1.18 版本以下建议使用 endpointsFilter 过滤器。当经过该过滤器首先通过 kubernetes.io/service-name 找到 endpointSlice 资源所对应的 services 资源之后判断 servces 资源是否存在 openyurt.io/topologyKeys 这个 Annotations如果存在那么通过这个 Annotations 的值判断数据过滤规则最后更新 response data 返回给 addons。
Annotations 的值分为两大类
1、kubernetes.io/hostname:只过滤出相同节点的 endpoint ip
2、openyurt.io/nodepool 或者 kubernetes.io/zone: 通过这个 Annotations 获取对应节点池最后遍历 endpointSlice 资源通过 endpointSlice 里的 topology 字段中的 kubernetes.io/hostname 字段在 endpointSlice 对象里找到对应的 Endpoints之后重组 endpointSlice 里的 Endpoints 后返回给 addons。
代码实现
func (fh *serviceTopologyFilterHandler) reassembleEndpointSlice(endpointSlice *discovery.EndpointSlice) *discovery.EndpointSlice {var serviceTopologyType string// get the service Topology typeif svcName, ok : endpointSlice.Labels[discovery.LabelServiceName]; ok {svc, err : fh.serviceLister.Services(endpointSlice.Namespace).Get(svcName)if err ! nil {klog.Infof(skip reassemble endpointSlice, failed to get service %s/%s, err: %v, endpointSlice.Namespace, svcName, err)return endpointSlice}if serviceTopologyType, ok svc.Annotations[AnnotationServiceTopologyKey]; !ok {klog.Infof(skip reassemble endpointSlice, service %s/%s has no annotation %s, endpointSlice.Namespace, svcName, AnnotationServiceTopologyKey)return endpointSlice}}var newEps []discovery.Endpoint// if type of service Topology is kubernetes.io/hostname// filter the endpoint just on the local hostif serviceTopologyType AnnotationServiceTopologyValueNode {for i : range endpointSlice.Endpoints {if endpointSlice.Endpoints[i].Topology[v1.LabelHostname] fh.nodeName {newEps append(newEps, endpointSlice.Endpoints[i])}}endpointSlice.Endpoints newEps} else if serviceTopologyType AnnotationServiceTopologyValueNodePool || serviceTopologyType AnnotationServiceTopologyValueZone {// if type of service Topology is openyurt.io/nodepool// filter the endpoint just on the node which is in the same nodepool with current nodecurrentNode, err : fh.nodeGetter(fh.nodeName)if err ! nil {klog.Infof(skip reassemble endpointSlice, failed to get current node %s, err: %v, fh.nodeName, err)return endpointSlice}if nodePoolName, ok : currentNode.Labels[nodepoolv1alpha1.LabelCurrentNodePool]; ok {nodePool, err : fh.nodePoolLister.Get(nodePoolName)if err ! nil {klog.Infof(skip reassemble endpointSlice, failed to get nodepool %s, err: %v, nodePoolName, err)return endpointSlice}for i : range endpointSlice.Endpoints {if inSameNodePool(endpointSlice.Endpoints[i].Topology[v1.LabelHostname], nodePool.Status.Nodes) {newEps append(newEps, endpointSlice.Endpoints[i])}}endpointSlice.Endpoints newEps}}return endpointSlice
}EndpointsFilter
针对 endpoints 资源进行相应的数据过滤首先判断 endpoint 是否存在对应的 service通过 node 的 label: apps.openyurt.io/nodepool 获取节点池之后获取节点池下的所有节点遍历 endpoints.Subsets 下的资源找出同一个节点池的 Ready pod address 以及 NotReady pod address 重组成新的 endpoints 之后返回给 addons。
func (fh *endpointsFilterHandler) reassembleEndpoint(endpoints *v1.Endpoints) *v1.Endpoints {svcName : endpoints.Name_, err : fh.serviceLister.Services(endpoints.Namespace).Get(svcName)if err ! nil {klog.Infof(skip reassemble endpoints, failed to get service %s/%s, err: %v, endpoints.Namespace, svcName, err)return endpoints}// filter the endpoints on the node which is in the same nodepool with current nodecurrentNode, err : fh.nodeGetter(fh.nodeName)if err ! nil {klog.Infof(skip reassemble endpoints, failed to get current node %s, err: %v, fh.nodeName, err)return endpoints}if nodePoolName, ok : currentNode.Labels[nodepoolv1alpha1.LabelCurrentNodePool]; ok {nodePool, err : fh.nodePoolLister.Get(nodePoolName)if err ! nil {klog.Infof(skip reassemble endpoints, failed to get nodepool %s, err: %v, nodePoolName, err)return endpoints}var newEpSubsets []v1.EndpointSubsetfor i : range endpoints.Subsets {endpoints.Subsets[i].Addresses filterValidEndpointsAddr(endpoints.Subsets[i].Addresses, nodePool)endpoints.Subsets[i].NotReadyAddresses filterValidEndpointsAddr(endpoints.Subsets[i].NotReadyAddresses, nodePool)if endpoints.Subsets[i].Addresses ! nil || endpoints.Subsets[i].NotReadyAddresses ! nil {newEpSubsets append(newEpSubsets, endpoints.Subsets[i])}}endpoints.Subsets newEpSubsetsif len(endpoints.Subsets) 0 {// this endpoints has no nodepool valid addresses for ingress controller, return nil to ignore itreturn nil}}return endpoints
}MasterServiceFilter
针对 services 下的域名进行 ip 以及端口替换这个过滤器的场景主要在于边缘端的 pod 无缝使用 InClusterConfig 访问集群资源。
func (fh *masterServiceFilterHandler) ObjectResponseFilter(b []byte) ([]byte, error) {list, err : fh.serializer.Decode(b)if err ! nil || list nil {klog.Errorf(skip filter, failed to decode response in ObjectResponseFilter of masterServiceFilterHandler, %v, err)return b, nil}// return data un-mutated if not ServiceListserviceList, ok : list.(*v1.ServiceList)if !ok {return b, nil}// mutate master servicefor i : range serviceList.Items {if serviceList.Items[i].Namespace MasterServiceNamespace serviceList.Items[i].Name MasterServiceName {serviceList.Items[i].Spec.ClusterIP fh.hostfor j : range serviceList.Items[i].Spec.Ports {if serviceList.Items[i].Spec.Ports[j].Name MasterServicePortName {serviceList.Items[i].Spec.Ports[j].Port fh.portbreak}}klog.V(2).Infof(mutate master service into ClusterIP:Port%s:%d for request %s, fh.host, fh.port, util.ReqString(fh.req))break}}// return the mutated serviceListreturn fh.serializer.Encode(serviceList)
}DiscardCloudService
该过滤器针对两种 service 其中的一种类型是 LoadBalancer因为边缘端无法访问 LoadBalancer 类型的资源所以该过滤器会将这种类型的资源直接过滤掉。另外一种是针对 kube-system 名称空间下的 x-tunnel-server-internal-svc这个 services 主要存在 cloud 节点用于访问 yurt-tunnel-server对于 edge 节点会直接过滤掉该 service。
func (fh *discardCloudServiceFilterHandler) ObjectResponseFilter(b []byte) ([]byte, error) {list, err : fh.serializer.Decode(b)if err ! nil || list nil {klog.Errorf(skip filter, failed to decode response in ObjectResponseFilter of discardCloudServiceFilterHandler %v, err)return b, nil}serviceList, ok : list.(*v1.ServiceList)if ok {var svcNew []v1.Servicefor i : range serviceList.Items {nsName : fmt.Sprintf(%s/%s, serviceList.Items[i].Namespace, serviceList.Items[i].Name)// remove lb serviceif serviceList.Items[i].Spec.Type v1.ServiceTypeLoadBalancer {if serviceList.Items[i].Annotations[filter.SkipDiscardServiceAnnotation] ! true {klog.V(2).Infof(load balancer service(%s) is discarded in ObjectResponseFilter of discardCloudServiceFilterHandler, nsName)continue}}// remove cloud clusterIP serviceif _, ok : cloudClusterIPService[nsName]; ok {klog.V(2).Infof(clusterIP service(%s) is discarded in ObjectResponseFilter of discardCloudServiceFilterHandler, nsName)continue}svcNew append(svcNew, serviceList.Items[i])}serviceList.Items svcNewreturn fh.serializer.Encode(serviceList)}return b, nil
}过滤框架现状
目前的过滤框架比较僵硬将资源过滤硬编码至代码中只能是已注册的资源才能进行相应的过滤,为了解决这个问题需要对过滤框架进行相应的改造。
解决方案
方案一
使用参数或者环境变量的形式自定义过滤配置但是这种方式有以下弊端
1、配置复杂需要将所以需要自定义的配置写入到启动参数或者读取环境变量 例如下格式
--filter_serviceTopologycoredns/endpointslices#list,kube-proxy/services#list;watch --filter_endpointsFilternginx-ingress-controller/endpoints#list;watch2、无法热更新每次修改配置都需要重启 Yurthub 生效。
方案二
1、使用 configmap 的形式自定义过滤配置降低配置复杂度配置格式(user-agent/resource#list,watch) 多个资源通过逗号隔开。如下所示
filter_endpoints: coredns/endpoints#list;watch,test/endpoints#list;watch
filter_servicetopology: coredns/endpointslices#list;watch
filter_discardcloudservice:
filter_masterservice: 2、利用 Informer 机制保证配置实时生效
综合以上两点在 OpenYurt 中我们选择了解决方案二。
开发过程中遇到的问题
在边缘端 Informer watch 的 api 地址是 Yurthub 的代理地址那么 Yurthub 在启动代理端口之前都是无法保证 configmap 的数据是正常的。如果在启动完成之后 addons 的请求先于 configmap 数据更新 这个时候会导致数据在没有过滤的情况下就返回给了 addons这样会导致很多预期以外的问题。
为了解决这个问题 我们需要在 apporve 中加入 WaitForCacheSync 保证数据同步完成之后才能返回相应的过滤数据但是在 apporve 中加入 WaitForCacheSync 也直接导致 configmap 进行 watch 的时候也会被阻塞所以需要在 WaitForCacheSync 之前加入一个白名单机制当 Yurthub 使用 list watch 访问 configmap 的时候我们直接不进行数据过滤相应的代码逻辑如下
func (a *approver) Approve(comp, resource, verb string) bool {if a.isWhitelistReq(comp, resource, verb) {return false}if ok : cache.WaitForCacheSync(a.stopCh, a.configMapSynced); !ok {panic(wait for configMap cache sync timeout)}a.Lock()defer a.Unlock()for _, requests : range a.nameToRequests {for _, request : range requests {if request.Equal(comp, resource, verb) {return true}}}return false
}总结
1、通过上述的扩展能力可以看出YurtHub 不仅仅是边缘节点上的带有数据缓存能力的反向代理。而是对 Kubernetes 节点应用生命周期管理加了一层新的封装提供边缘计算所需要的核心管控能力。
2、YurtHub 不仅仅适用于边缘计算场景其实可以作为节点侧的一个常备组件适用于使用 Kubernetes 的任意场景。相信这也会驱动 YurtHub 向更高性能更高稳定性发展。
原文链接
本文为阿里云原创内容未经允许不得转载。
