为Nacos实现一个DNS服务发现

nacos

DNS协议

DNS协议其本身就是一个天然的分布式服务发现中心，通过域名映射IP，用户只需要记住域名即可访问对应的服务。

纵观Nacos、Eureka、Etcd以及Zookeeper，这些充当服务发现中心的组件，其都需要一个Client-SDK为上层应用提供根据服务名查找到一个机器IP的能力，而这就带来了一个问题，应用必须要集成这一些组件的客户端SDK，一旦组件的SDK出现问题需要升级时，就需要推动应用方的升级，这是不想见到的。而DNS协议，天然跨语言，因此Consul或者Kubernetes都选择使用DNS协议作为服务发现，直接解决了跨语言的问题。

Nacos的DNS

官网介绍

通过支持权重路由，动态DNS服务能让您轻松实现中间层负载均衡、更灵活的路由策略、流量控制以及简单数据中心内网的简单DNS解析服务。动态DNS服务还能让您更容易地实现以DNS协议为基础的服务发现，以消除耦合到厂商私有服务发现API上的风险。

目前，Nacos的DNS实现，是依赖了CoreDNS，其项目在nacos-coredns-plugin。

Java版本的DNS

由于目前Nacos-Client的SDK其功能比较完备的属于Java版本，因此考虑使用Java去实现一个Nacos DNS插件

依赖包

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
plugins {
    id 'java'
}

group 'com.alibaba.nacos.dns'
version '0.0.1'

sourceCompatibility = 1.8

jar {
    manifest {
        attributes 'Main-Class': 'com.conf.nacos.dns.Main'
    }
    from { configurations.compile.collect { it.isDirectory() ? it : zipTree(it) } }
    into('assets') {
        from 'assets'
    }
}

tasks.withType(JavaCompile) {
    options.encoding = 'UTF-8'
}

repositories {
    mavenCentral()
}

dependencies {
    compile group: 'com.alibaba.nacos', name: 'nacos-client', version: '1.3.0'
    compile group: 'dnsjava', name: 'dnsjava', version: '3.1.0'
    compile group: 'org.slf4j', name: 'slf4j-api', version: '1.7.30'
    testCompile group: 'junit', name: 'junit', version: '4.12'
}

实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
public class DnsServer {

    // 对 ByteBuffer 做一此缓存，进行复用
	private static final ThreadLocal<ByteBuffer> THREAD_LOCAL = ThreadLocal.withInitial(() -> ByteBuffer.allocate(1024));

    private static final Logger LOGGER = LoggerFactory.getLogger(DnsServer.class);
    
    // 与 Nacos-Server 的操作
    private static NacosDnsCore nacosDnsCore;
    
    // UDP
    private static DatagramChannel serverChannel;
    
    private static Selector selector = null;
    
    // 将请求分发到不同的线程去处理
	private static final Executor executor = Executors.newFixedThreadPool(
			Runtime.getRuntime().availableProcessors(), new ThreadFactory() {

				AtomicInteger workerId = new AtomicInteger();

				@Override
				public Thread newThread(Runnable r) {
					final Thread t = new Thread(r);
					t.setName("com.conf.nacos.dns.worker-" + workerId.getAndIncrement());
					t.setDaemon(true);
					return t;
				}
			});

	public static DnsServer create() throws NacosDnsException {
		return new DnsServer();
	}

	private DnsServer() throws NacosDnsException {
		try {
			init();
			nacosDnsCore = new NacosDnsCore();
		} catch (Throwable ex) {
			throw new NacosDnsException(Code.CREATE_DNS_SERVER_FAILED, ex);
		}
	}

	private void init() throws Exception {
		AccessController.doPrivileged((PrivilegedAction<Void>) () -> {
			try {
				// 这里进行构建 DNS server，去监听 DNS 协议专用的端口 53
				selector = Selector.open();
				serverChannel = DatagramChannel.open();
                serverChannel.socket().bind(new InetSocketAddress("127.0.0.1", 53));
                // 使用非阻塞特性
                serverChannel.configureBlocking(false);
                // 关心链接的可读事件
				serverChannel.register(selector, SelectionKey.OP_READ);
			} catch (Throwable ex) {
				throw new NacosDnsException(Code.CREATE_DNS_SERVER_FAILED, ex);
			}
			return null;
		});
	}

public void start() {
		LOGGER.info("dns-server starting");
		final ByteBuffer buffer = ByteBuffer.allocate(1024);
		for ( ; ; ) {
			try {
                selector.select();
                // 获取当前可读的链接
				for (SelectionKey key : selector.selectedKeys()) {
					if (key.isReadable()) {
                        // 清空上一个链接的数据
                        buffer.clear();
                        // 进行数据的接受
                        SocketAddress client = serverChannel.receive(buffer);
                        // 指针反转，确保从头开始读数据
						buffer.flip();
						byte[] requestData = new byte[buffer.limit()];
                        buffer.get(requestData, 0, requestData.length);
                        // 利用不同的线程进行处理
						executor.execute(() -> handler(requestData, client));
					}
				}
			} catch (Throwable ex) {
				LOGGER.error("handler client request has error : {}", ExceptionUtil.getStackTrace(ex));
			}
		}
	}

    private void handler(final byte[] data, final SocketAddress client) {
        // 对池子里面获取一个 ByteBuffer 对象。
		final ByteBuffer buffer = THREAD_LOCAL.get();
		buffer.clear();
		try {
            // 利用 dnsjava 这个开源项目对 DNS 请求包进行解析（考虑自己实现DNS的解析）
			final Message message = new Message(data);
            final Record question = message.getQuestion();
            // 获取需要解析的域名，这里就是服务名信息
            final String domain = question.getName().toString();
            // 调用 NacosDnsCore，获取一个实例数据，并将其封装为 Record 对象
			final Record response = createRecordByQuery(question, domain);
            final Message out = message.clone();
            // 放入返回体
            out.addRecord(response, Section.ANSWER);
            // 写数据
            buffer.put(out.toWire());
            // 指针反转，确保能够从头开始写数据
            buffer.flip();
            // 数据发送
			serverChannel.send(buffer, client);
		} catch (Throwable ex) {
			LOGGER.error("response to client has error : {}", ExceptionUtil.getStackTrace(ex));
		} finally {
			THREAD_LOCAL.set(buffer);
		}
	}

	private static Record createRecordByQuery(final Record request, final String domain) {
        // NacosDnsCore 内部通过负载均衡策略返回一个 InstanceRecord 对象
        InstanceRecord record = nacosDnsCore.selectOne(domain);
        // 如果服务不存在，返回 NULLRecord
		if (record == null) {
			return NULLRecord.newRecord(request.getName(), request.getType(), request.getDClass());
        }
        // 根据实例的 IP、 Port 创建地址对象，并包装为 ARecord 对象
		InetSocketAddress address = new InetSocketAddress(record.getIp(), record.getPort());
		return new ARecord(request.getName(), request.getDClass(), request.getTTL(), address.getAddress());
	}
}

DnsServer就完成了DNS协议报文的处理以及如何回复，而NacosDnsCore是处理上层的DnsServer告知一个域名之后，怎么样去根据域名，找一个机器的IP地址返回。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
public class NacosDnsCore {

	private static final Logger logger = LoggerFactory.getLogger(NacosDnsCore.class);

	private static final Map<String, List<InstanceRecord>> serviceMap = new ConcurrentHashMap<>(
			32);

	private static String LOAD_BALANCER_NAME = Constants.RANDOM_LOAD_BALANCER;

	private static final Map<String, LoadBalancer> balancers = new HashMap<>();

	// 实际与 NacosServer 交互的客户端
	private static NamingService nacosClient;

	public NacosDnsCore() throws Throwable {
		// 加载 Nacos-Client 配置
		try (InputStream stream = NacosDnsCore.class.getClassLoader()
				.getResourceAsStream("nacos-dns.properties")) {
			Properties properties = new Properties();
			properties.load(stream);

			Properties config = new Properties();
			for (String[] keys : Constants.NACOS_PEOPERTIES_KEY) {
				MapUtils.putIfValNoNull(config, keys[0], properties.getProperty(keys[1]));
			}

			nacosClient = NacosFactory.createNamingService(config);
			// 初始化负载均衡实现
			initLoadBalancer();
		}
	}

	private void initLoadBalancer() {
		ServiceLoader<LoadBalancer> loader = ServiceLoader.load(LoadBalancer.class);
		loader.forEach(loadBalancer -> balancers.put(loadBalancer.name(), loadBalancer));
	}

	// 根据 DnsServer 传来的域名信息，查询一个 InstanceRecord 数据
	public Optional<InstanceRecord> selectOne(final String domain) {
		List<InstanceRecord> list = findAllInstanceByServiceName(domain);
		if (list.isEmpty()) {
			return Optional.empty();
		}
		// 如果服务存在，则根据负载均衡选择一个进行返回
		return Optional.of(balancers.get(LOAD_BALANCER_NAME).selectOne(list));
	}

	// 根据域名查询所有的机器实例
	private List<InstanceRecord> findAllInstanceByServiceName(final String serviceName) {
		if (!serviceMap.containsKey(serviceName)) {
			// 如果当前缓存不存在，则像服务端进行拉取
			obtainServiceFromRemoteServer(serviceName);
		}
		// 从缓存中获取
		return serviceMap.getOrDefault(serviceName, Collections.emptyList());
	}

	private static void obtainServiceFromRemoteServer(final String serviceName) {
		serviceMap.computeIfAbsent(serviceName, name -> {
			try {
				// 对域名进行一些预处理，
				String domain = serviceName.substring(0, serviceName.length() - 1).replace("\\@\\@", "@@");
				// 获取到服务名称
				final String _serviceName = NamingUtils.getServiceName(domain);
				// 获取服务分组信息
				final String _groupName = NamingUtils.getGroupName(domain);
				// 从 NacosServer 获取到了该服务的所有实例数据
				List<Instance> instances = nacosClient.getAllInstances(_serviceName, _groupName);
				// 注册一个监听器监听服务实例的变化
				registerInstanceChangeObserver(serviceName);
				// 转为更小的对象
				return parseToInstanceRecord(instances);
			}
			catch (Throwable ex) {
				logger.error(
						"An error occurred querying the service instance remotely : {}",
						ExceptionUtil.getStackTrace(ex));
				return Collections.emptyList();
			}
		});
	}

	private static void registerInstanceChangeObserver(final String serviceName)
			throws NacosException {
		nacosClient.subscribe(serviceName, event -> {
			NamingEvent namingEvent = (NamingEvent) event;
			final String name = namingEvent.getServiceName();
			final List<Instance> newInstances = namingEvent.getInstances();
			// 进行服务实例数据的更新操作
			serviceMap.computeIfPresent(name, (s, instanceRecords) -> {
				// 清空原本数据内的数据
				instanceRecords.clear();
				return parseToInstanceRecord(newInstances);
			});
		});
	}

	private static List<InstanceRecord> parseToInstanceRecord(List<Instance> instances) {
		int maxSize = 10_000;
		Stream<Instance> stream;
		if (instances.size() < maxSize) {
			stream = instances.stream();
		}
		else {
			stream = instances.parallelStream();
		}

		return stream.map(instance -> InstanceRecord.builder().ip(instance.getIp())
				.port(instance.getPort()).healthy(instance.isHealthy())
				.enabled(instance.isEnabled()).weight(instance.getWeight())
				.metadata(instance.getMetadata()).build())
				.collect(CopyOnWriteArrayList::new, CopyOnWriteArrayList::add,
						CopyOnWriteArrayList::addAll);
	}

}

验证

Nacos 控制台

Nacos 控制台

DNS 查询服务

DNS 查询服务