OpenYurt 之 Yurthub 数据过滤框架解析

简介：OpenYurt 是业界首个非侵入的边缘计算云原生开源项目，通过边缘自治，云边协同，边缘单元化，边缘流量闭环等能力为用户提供云边一体化的使用体验。在 Openyurt 里边缘网络可以使用数据过滤框架在不同节点池里实现边缘流量闭环能力。

作者：应健健，新华智云计算中心

OpenYurt 是业界首个非侵入的边缘计算云原生开源项目，通过边缘自治，云边协同，边缘单元化，边缘流量闭环等能力为用户提供云边一体化的使用体验。在 Openyurt 里边缘网络可以使用数据过滤框架在不同节点池里实现边缘流量闭环能力。

Yurthub 数据过滤框架解析

Yurthub 本质上是一层 kube-apiserver 的代理，在代理的基础上加了一层 cache，一来保证边缘节点离线的情况下可以使用本地 cache 保证业务稳定性,有效的解决了边缘自治的问题。二来可以降低大量的 list & watch 操作对云上 api 产生一定的负载。

Yurthub 的数据过滤通过节点上的 pod 以及 kubelet 的请求通过 Load Balancer 发送给 kube-apiserver，代理接收到响应消息进行数据过滤处理，之后再将过滤后的数据返回给请求方。如果节点是边缘节点会根据请求类型对响应请求体中的资源进行本地缓存，如果是云端节点考虑到网络状态良好不进行本地缓存。

Yurthub 的过滤框架实现原理图：

Yurthub 目前包含四种过滤规则，通过 addons 请求的 user-agent，resource，verb 判断经过那个过滤器进行相应的数据过滤。

四种过滤规则功能及实现

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_serviceTopology=coredns/endpointslices#list,kube-proxy/services#list;watch --filter_endpointsFilter=nginx-ingress-controller/endpoints#list;watch

2、无法热更新，每次修改配置都需要重启 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
}