OpenYurt 之 Yurthub 数据过滤软件系统解析

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 host

if 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 node

currentNode, 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 node

currentNode, 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.EndpointSubset

for 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 = newEpSubsets

if len(endpoints.Subsets) == 0 {

// this endpoints has no nodepool valid addresses for ingress controller, return nil to ignore it

return nil

}

}

return endpoints

}

MasterServiceFilter

针对 services 下的DNS顺利顺利完成 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 ServiceList

serviceList, ok := list.(*v1.ServiceList)

if !ok {

return b, nil

}

// mutate master service

for i := range serviceList.Items {

if serviceList.Items[i].Namespace == MasterServiceNamespace && serviceList.Items[i].Name == MasterServiceName {

serviceList.Items[i].Spec.ClusterIP = fh.host

for j := range serviceList.Items[i].Spec.Ports {

if serviceList.Items[i].Spec.Ports[j].Name == MasterServicePortName {

serviceList.Items[i].Spec.Ports[j].Port = fh.port

break

}

}

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 serviceList

return 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.Service

for i := range serviceList.Items {

nsName := fmt.Sprintf("%s/%s", serviceList.Items[i].Namespace, serviceList.Items[i].Name)

// remove lb service

if 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 service

if _, 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 = svcNew

return fh.serializer.Encode(serviceList)

}

return b, nil

}

滤网方法论现状

现在的滤网方法论相对僵大块，将人力资源滤网大块编码器至字符中都，只能是已持有人的人力资源才能顺利顺利完成都可的滤网,为了本土化解这个弊端，能够对滤网方法论顺利顺利完成都可的改造。

高效率

设计方案一：

常用参数或者配置文件的形式自定义滤网可用，但是这种方式有以下症结：

1、可用简单能够将所以能够自定义的可用写入到叫停参数或者复制到配置文件 例如下PDF：

---filter_serviceTopology=coredns/endpointslices#list,kube-proxy/services#list;watch ---filter_endpointsFilter=nginx-ingress-controller/endpoints#list;watch

2、未能热更为新，每次修改可用都能够较重新启动 Yurthub 终止。

设计方案二：

1、常用 configmap 的形式自定义滤网可用降低可用log可用PDF(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 URL是 Yurthub 的改由URL，那么 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 毫无疑问是破碎终尾端上的带有信息CPU能力也的反向改由。而是对 Kubernetes 终尾端应用领域一般来说管理纳了一层新的积体电路，备有破碎测算所能够的核心维护能力也。

2、YurtHub 毫无疑问限于于破碎测算情节，或许可以作为终尾端外侧的一个常备组件，限于于常用 Kubernetes 的取参数情节。相信这也不会驱动 YurtHub 向更为高性能，更为高反应性发展。

本文为阿中都的云原创内容，未经允许不得转载。

。

武汉妇科医院哪好
株洲男科医院哪个比较好
南昌白癜风医院哪家专业
银川白癜风医院电话
宁波妇科医院哪个比较好
先诺欣
咳嗽有痰吃什么药
频繁放屁
大同在线
太极急支糖浆治咳嗽效果怎么样