Dubbo-kubernetes 基于 Informer 服务发现优化之路
阿
丛国庆约 6 分钟
丛国庆约 6 分钟
List/Watch 机制介绍
Aliware
List / Watch 机制是 Kubernetes 中实现集群控制模块最核心的设计之一，它采用统一的异步消息处理机制，保证了消息的实时性、可靠性、顺序性和性能等，为声明式风格的API奠定了良好的基础。
$ curl -i http://{kube-api-server-ip}:8080/api/v1/watch/endpoints?watch=yesHTTP/1.1 200 OKContent-Type: application/jsonTransfer-Encoding: chunkedDate: Thu, 14 Seo 2022 20:22:59 GMTTransfer-Encoding: chunked{"type":"ADDED", "object":{"kind":"Endpoints","apiVersion":"v1",...}}{"type":"ADDED", "object":{"kind":"Endpoints","apiVersion":"v1",...}}{"type":"MODIFIED", "object":{"kind":"Endpoints","apiVersion":"v1",...}}
Dubbo 基于 Watch 的服务发现
Aliware
private void watchEndpoints(ServiceInstancesChangedListener listener, String serviceName) {Watch watch = kubernetesClient.endpoints().inNamespace(namespace).withName(serviceName).watch(new Watcher<Endpoints>() {// 资源更改的事件回调@Overridepublic void eventReceived(Action action, Endpoints resource) {notifyServiceChanged(serviceName, listener);...}});...}private void notifyServiceChanged(String serviceName, ServiceInstancesChangedListener listener) {ServiceInstancesChangedEvent event = new ServiceInstancesChangedEvent(serviceName, getInstances(serviceName));...listener.onEvent(event);...}监听到资源变化后，调用 notifyServiceChanged 方法从 kube-apiserver 全量拉取资源 list 数据，保持 Dubbo 本地侧服务列表。
@Overridepublic List<ServiceInstance> getInstances(String serviceName){// 直接调用kube-apiserverEndpoints endpoints = kubernetesClient.endpoints().inNamespace(namespace).withName(serviceName).get();return toServiceInstance(endpoints, serviceName);}
Informer 机制介绍
Aliware
Informer 模块是 Kubernetes 中的基础组件，以 List/Watch 为基础，负责各组件与 kube-apiserver 的资源与事件同步。 Kubernetes 中的组件，如果要访问 Kubernetes 中的 Object，绝大部分情况下会使用 Informer 中的 Lister()方法，而非直接调用 kube-apiserver。
-
Informer 在初始化时，Reflector 会先调用 List 获得所有的 Pod，同时调用 Watch 长连接监听 kube-apiserver。 -
Reflector 拿到全部 Pod 后，将 Add Pod 这个事件发送到 DeltaFIFO。 -
DeltaFIFO 随后 pop 这个事件到 Informer 处理。 -
Informer 向 Indexer 发布 Add Pod 事件。 -
Indexer 接到通知后，直接操作 Store 中的数据（key->value 格式）。 -
Informer 触发 EventHandler 回调。 -
将 key 推到 Workqueue 队列中。 -
从 WorkQueue 中 pop 一个 key。 -
然后根据 key 去 Indexer 取到 val。根据当前的 EventHandler 进行 Add Pod 操作(用户自定义的回调函数)。 -
随后当 Watch 到 kube-apiserver 资源有改变的时候，再重复 2-9 步骤。
Informer 关键设计
-
本地缓存 ：Informer 只会调用 K8s List 和 Watch 两种类型的 API。Informer 在初始化的时，先调用 List 获得某种 resource 的全部 Object，缓存在内存中; 然后，调用 Watch API 去 watch 这种 resource，去维护这份缓存; 最后，Informer 就不再调用 kube-apiserver。Informer 抽象了 cache 这个组件，并且实现了 store 接口，后续获取资源直接通过本地的缓存来进行获取。 -
无界队列 ：为了协调数据生产与消费的不一致状态，在客户端中通过实现了一个无界队列 DeltaFIFO 来进行数据的缓冲，当 reflector 获取到数据之后，只需要将数据推到到 DeltaFIFO 中，则就可以继续 watch 后续事件，从而减少阻塞时间，如上图 2-3 步骤所示。 -
事件去重 ：在 DeltaFIFO 中，如果针对某个资源的事件重复被触发，则就只会保留相同事件最后一个事件作为后续处理，有 resourceVersion 唯一键保证，不会重复消费。 -
复用连接 ：每一种资源都实现了 Informer 机制，允许监控不同的资源事件。为了避免同一个资源建立多个 Informer，每个 Informer 使用一个 Reflector 与 apiserver 建立链接，导致 kube-apiserver 负载过高的情况，K8s 中抽象了 sharedInformer 的概念，即共享的 Informer, 可以使同一类资源 Informer 共享一个 Reflector。内部定义了一个 map 字段，用于存放所有 Infromer 的字段。针对同一资源只建立一个连接，减小 kube-apiserver 的负载。
Dubbo 引入 Informer 机制后的服务发现
Aliware
Dubbo 3.1.1 后引入 Informer 机制，Informer 组件会利用其特性在 consumer 侧内存中维护 Kubernetes 环境中的所有地址列表。
资源监听由 Watch API 更换为 Informer API
/*** 监听Endpoints*/private void watchEndpoints(ServiceInstancesChangedListener listener, String serviceName) {SharedIndexInformer<Endpoints> endInformer = kubernetesClient.endpoints().inNamespace(namespace).withName(serviceName).inform(new ResourceEventHandler<Endpoints>() {@Overridepublic void onUpdate(Endpoints oldEndpoints, Endpoints newEndpoints) {notifyServiceChanged(serviceName, listener, toServiceInstance(newEndpoints, serviceName));}// 省略掉onAdd和onDelete...});...}/*** 通知订阅者Service改变*/private void notifyServiceChanged(String serviceName, ServiceInstancesChangedListener listener, List<ServiceInstance> serviceInstanceList) {ServiceInstancesChangedEvent event = new ServiceInstancesChangedEvent(serviceName, serviceInstanceList);// 发布事件listener.onEvent(event);}
getInstances() 优化
引入 Informer 后，无需直接调用 List 接口，而是直接从 Informer 的 store 中获取，减少对 kube-apiserver 的直接调用。
public List<ServiceInstance> getInstances(String serviceName) {Endpoints endpoints = null;SharedIndexInformer<Endpoints> endInformer = ENDPOINTS_INFORMER.get(serviceName);if (endInformer != null) {// 直接从informer的store中获取Endpoints信息List<Endpoints> endpointsList = endInformer.getStore().list();if (endpointsList.size() > 0) {endpoints = endpointsList.get(0);}}// 如果endpoints经过上面处理仍为空，属于异常情况，那就从kube-apiserver拉取if (endpoints == null) {endpoints = kubernetesClient.endpoints().inNamespace(namespace).withName(serviceName).get();}return toServiceInstance(endpoints, serviceName);}
结论
Aliware