Nacos注册中心客户端容灾

news2024/11/24 18:50:41

目前Nacos客户端有一个FailoverReactor来进行容灾文件的管理，可以通过在指定磁盘文件里写入容灾数据来进行客户端使用数据的覆盖。FailoverReactor目前会拦截Nacos客户端查询接口调用，以getAllInstances接口为例，目前FailoverReactor的工作流程如下图：

这里主要涉及到两个组件：

FailoverReactor：容灾数据管理类，负责容灾开关和数据的加载和刷新；
ServiceInfoHoler：默认的服务数据管理类，持有一份内存服务数据缓存，负责处理Nacos服务端的推送的最新数据；

FailoverReactor和ServiceInfoHoler的交互机制如下：

这里和客户端容灾的相关逻辑主要是3个：

FailoverReactor会定期读取磁盘容灾开关文件；
当容灾开关打开时，FailoverReactor会从磁盘里加载容灾数据文件，同时用户调用查询请求时就会优先使用容灾文件里的数据；
FailoverReactor会定期从ServiceInfoHolder拿到最新的内存数据，保存到容灾磁盘数据文件里；

目前的方式有四个问题：

无法人工覆盖容灾数据，当前是定期将ServiceInfoHolder里的数据进行持久化作为容灾数据。即使手动修改了容灾数据磁盘文件内容，也会被覆盖为ServiceInfoHolder里的数据；
基于磁盘的容灾缓存有一些限制，比如服务的订阅者有多个机器实例时，如果需要打开容灾开关，需要运维批量修改机器的文件；
subscribe接口不会使用FailoverReactor里的数据；
容灾数据的可观测性也需要优化。

实现方案

上面提到的问题1、3和4都比较好解决，下面会一一阐述。针对问题2，一般在生产环境中，我们会考虑使用中心化的数据存储来进行容灾数据的存储和管理。我们可以将FailoverReactor依赖的数据来源抽象为一个SPI接口FailoverDataSource，这个接口默认实现还是本地磁盘，但是用户可以实现这个SPI接口来使用自定义的容灾数据源。

接口和数据结构定义

FailoverDataSource的定义为：

public interface FailoverDataSource {

     FailoverSwitch getSwitch();

     Map<String, FailoverData> getFailoverData();
}

各个方法的作用为：

FailoverSwitch getSwitch()：获取当前的容灾开关；
Map<String, FailoverData> getFailoverData()：获取当前的容灾数据，返回一个map，key是服务名，value为对应服务的容灾数据；

FailoverSwitch的定义建议如下：

public class FailoverSwitch {

    private boolean enabled;
}

enabled：容灾是否打开；

FailoverData的定义建议为：

public abstract class FailoverData {

    private DataType dataType;
    
    private Object data;
    
    protected FailoverData(Object data, DataType dataType) {
        this.data = data;
        this.dataType = dataType;
    }

    enum DataType {
        naming,
        config
    }
}

dataType：容灾数据类型，这里因为需要综合考虑配置模块的容灾，所以使用抽象类定义了naming和config两种容灾数据类型；
data：容灾数据，子类设置具体类型；

以naming模块为例，NamingFailoverData扩展FailoverData：

public class NamingFailoverData extends FailoverData {

    private NamingFailoverData(ServiceInfo serviceInfo) {
        super(serviceInfo, DataType.naming);
    }

    public static NamingFailoverData newNamingFailoverData(ServiceInfo serviceInfo) {
        return new NamingFailoverData(serviceInfo);
    }
}

NamingFailoverData里的容灾数据类型为ServiceInfo。

交互流程

FailoverReactor内部流程

FailoverReactor和FailoverDataSource、ServiceInfoHoler的交互机制优化为：

各个组件职责说明如下：

FailoverReactor：存储容灾缓存数据，管理容灾开关定时任务和容灾数据更新定时任务；
FailoverSwitchRefresher：定时（每5秒）从容灾数据源查询容灾开关，根据容灾开关的值进行相应操作：
- 若容灾打开：从容灾数据源查询容灾数据FailoverDataSource，保存到FailoverReactor的内存容灾数据map里；
- 若容灾关闭：清空FailoverReactor的内存容灾数据map；
FailoverDataSource：存储容灾数据，前文已经提及；
ServiceInfoHoler：默认情况下服务数据的管理，前文已经提及；

客户端查询请求流程

对于客户端的查询请求，其流程优化为：

这里的流程和之前的变化为：当从failoverReactor里拿不到容灾数据的时候，还是会去serviceInfoHolder里读取数据。这么做的目的是因为我们可能只配置部分服务进行容灾，其他的服务还是走serviceInfoHolder。

订阅接口事件通知流程

对于订阅接口，之前是不会受到容灾开关的影响，现在则也会在容灾开启时停止数据更新通知：

public ServiceInfo processServiceInfo(ServiceInfo serviceInfo) {
        ...
        if (changed) {
            NAMING_LOGGER.info("current ips:({}) service: {} -> {}", serviceInfo.ipCount(), serviceInfo.getKey(),
                    JacksonUtils.toJson(serviceInfo.getHosts()));
            // 判断容灾开关是否打开，打开时不发布事件：
            if (!failoverDataSource.getFailoverSwitch().isEnabled()) {
                NotifyCenter.publishEvent(new InstancesChangeEvent(notifierEventScope, serviceInfo.getName(), serviceInfo.getGroupName(),
                        serviceInfo.getClusters(), serviceInfo.getHosts()));
            }
            DiskCache.write(serviceInfo, cacheDir);
        }
        return serviceInfo;
    }

同时，在容灾关闭时，我们需要根据容灾期间数据是否发生变化来决定要不要触发订阅事件通知，我们可以把这个逻辑驾到上面提到的FailoverSwitchRefresher里：

在这里当FailoverSwitchRefresher轮询发现容灾关闭时，在清空FailoverReactor的内存数据之前，会触发FailoverReactor和ServiceInfoHolder的数据比较，如果发现数据不一致，则会触发ServiceInfoHolder发布对应的服务变更事件。

可观测性

我们可以定义个MultiGauge来存储FailoverReactor里目前生效的容灾数据内容，统计粒度为每个服务当前有多少个实例：

MultiGauge failoverInstanceCounts = MultiGauge.builder("nacos_naming_client_failover_instances").register(Metrics.globalRegistry);

Set<String> serviceNames = failoverDataSource.getSwitch().getServiceNames();
Map<String, FailoverData> failoverDataMap = failoverDataSource.getFailoverData();

failoverInstanceCounts.register(serviceNames.stream().map(serviceName -> MultiGauge.Row.of(Tags.of("service_name", serviceName), ((ServiceInfo)failoverDataMap.get(serviceName)).ipCount())).collect(Collectors.toList()), true);