tellus_bootstrap_dns/lib.rs
1//! DNS seed discovery for tellus cluster bootstrap: [DnsSeeds] implements
2//! [SeedDiscovery], resolving seed addresses from SRV records,
3//! which carry address and port (e.g. a Kubernetes headless service), or from A/AAAA records
4//! paired with a configured port. Pass it to [bootstrap](tellus::cluster::bootstrap), which
5//! resolves it every `resolve_interval` and joins through the resolved addresses on every
6//! resolution, retrying resolve failures, e.g. records still appearing while an orchestrator
7//! starts the other nodes. Only forming a new cluster waits for the view to settle.
8//! SRV targets resolve concurrently; a failing target is skipped while any other resolves, and a
9//! set resolving no address at all remains an error so bootstrap does not settle on it. Nothing
10//! here distinguishes a stale target from a transient failure, because the resolver does not
11//! either, so both are skipped alike. A partial view does settle, so a target failing throughout
12//! the settle window shrinks the universe a
13//! [FormationProvider](tellus::cluster::formation::FormationProvider) counts its majority against.
14//!
15//! Whatever is queried must resolve to one address per node, since bootstrap counts addresses and
16//! a peer is only admitted at the address it advertises; on Kubernetes that makes the headless
17//! service single stack, see this crate's README.
18//!
19//! The default resolver keeps no response cache. Bootstrap polls DNS to see nodes come and go, and
20//! a cached answer would hide a node registered after it for the record's TTL, e.g. a sibling
21//! restarting a moment later than this one. A resolver given via [DnsSeeds::with_resolver] is
22//! used as configured.
23
24#![warn(missing_docs, clippy::missing_errors_doc)]
25
26use futures_util::future::join_all;
27use hickory_resolver::{TokioResolver, net::NetError, proto::rr::RData};
28use std::{net::SocketAddr, num::NonZeroU16};
29use tellus::cluster::SeedDiscovery;
30use thiserror::Error;
31
32/// Seed discovery via DNS, the [SeedDiscovery] for clusters whose nodes are named by DNS
33/// records instead of configured addresses.
34#[derive(Debug)]
35pub struct DnsSeeds {
36 resolver: TokioResolver,
37 query: Query,
38}
39
40impl DnsSeeds {
41 /// Discovery via the given [Query], which a config file can provide. Resolves with the
42 /// system resolver configuration; see [with_resolver](DnsSeeds::with_resolver).
43 ///
44 /// # Errors
45 /// Fails if the system resolver configuration cannot be read.
46 pub fn new(query: Query) -> Result<Self, DnsSeedsError> {
47 Ok(Self {
48 resolver: system_resolver()?,
49 query,
50 })
51 }
52
53 /// Discovery via the SRV records at the given name, e.g.
54 /// `_tellus._udp.svc.cluster.example`: every record contributes its port paired with each
55 /// address its target resolves to, so nodes can advertise differing ports. Resolves with
56 /// the system resolver configuration; see [with_resolver](DnsSeeds::with_resolver).
57 ///
58 /// # Errors
59 /// Fails if the system resolver configuration cannot be read.
60 pub fn srv(name: impl Into<String>) -> Result<Self, DnsSeedsError> {
61 Self::new(Query::Srv(name.into()))
62 }
63
64 /// Discovery via the A/AAAA records at the given name, every address paired with the given
65 /// port: for clusters whose nodes all advertise one well known port. Resolves with the
66 /// system resolver configuration; see [with_resolver](DnsSeeds::with_resolver).
67 ///
68 /// # Errors
69 /// Fails if the system resolver configuration cannot be read.
70 pub fn ip(name: impl Into<String>, port: NonZeroU16) -> Result<Self, DnsSeedsError> {
71 Self::new(Query::Ip {
72 name: name.into(),
73 port,
74 })
75 }
76
77 /// Replace the resolver [DnsSeeds::new] builds, e.g. by one pointed at specific name servers.
78 /// That resolver keeps no response cache. A resolver which does returns the same addresses
79 /// until its cached answer expires, so bootstrap sees nodes come and go only that late.
80 pub fn with_resolver(mut self, resolver: TokioResolver) -> Self {
81 self.resolver = resolver;
82 self
83 }
84}
85
86impl SeedDiscovery for DnsSeeds {
87 type Error = DnsSeedsError;
88
89 async fn resolve(&mut self) -> Result<Vec<SocketAddr>, DnsSeedsError> {
90 match &self.query {
91 Query::Srv(name) => {
92 let records = self.resolver.srv_lookup(name.as_str()).await?;
93 let lookups = records
94 .answers()
95 .iter()
96 .filter_map(|record| match &record.data {
97 RData::SRV(srv) => Some(srv),
98 _ => None,
99 })
100 .map(|srv| {
101 let resolver = self.resolver.clone();
102 let target = srv.target.clone();
103 let port = srv.port;
104 async move { resolver.lookup_ip(target).await.map(|ips| (port, ips)) }
105 });
106 let mut addrs = Vec::new();
107 let mut first_error = None;
108 for result in join_all(lookups).await {
109 match result {
110 Ok((port, ips)) => {
111 addrs.extend(ips.iter().map(|ip| SocketAddr::new(ip, port)));
112 }
113
114 Err(error) => {
115 first_error.get_or_insert(error);
116 }
117 }
118 }
119
120 match first_error {
121 Some(error) if addrs.is_empty() => Err(error.into()),
122 _ => Ok(addrs),
123 }
124 }
125
126 Query::Ip { name, port } => {
127 let ips = self.resolver.lookup_ip(name.as_str()).await?;
128 let port = port.get();
129 Ok(ips.iter().map(|ip| SocketAddr::new(ip, port)).collect())
130 }
131 }
132 }
133}
134
135/// What [DnsSeeds] resolves, deserializable with the `serde` feature: a config file names either
136/// form as a single key, `srv: _tellus._udp.example` or `ip: { name: nodes.example, port: 7000 }`.
137///
138/// That is the normalized map form a loader like [config](https://crates.io/crates/config)
139/// produces; `serde_yaml` deserializes the same enum from a YAML tag (`!srv`) instead, as the
140/// tellus README's configuration section explains.
141#[derive(Debug, Clone, PartialEq, Eq)]
142#[cfg_attr(
143 feature = "serde",
144 derive(serde::Deserialize),
145 serde(rename_all = "snake_case", deny_unknown_fields)
146)]
147pub enum Query {
148 /// The SRV records at the given name, every record's port paired with each address its
149 /// target resolves to.
150 Srv(String),
151
152 /// The A/AAAA records at the given name, every address paired with the given port.
153 Ip {
154 /// The name to resolve.
155 name: String,
156
157 /// The port every resolved address is paired with.
158 port: NonZeroU16,
159 },
160}
161
162/// A DNS resolution failure; [bootstrap](tellus::cluster::bootstrap) logs and retries it.
163#[derive(Debug, Error)]
164#[error(transparent)]
165pub struct DnsSeedsError(#[from] NetError);
166
167fn system_resolver() -> Result<TokioResolver, DnsSeedsError> {
168 let mut builder = TokioResolver::builder_tokio()?;
169 // No response cache: bootstrap polls to see nodes come and go, which a cached answer hides.
170 builder.options_mut().cache_size = 0;
171 Ok(builder.build()?)
172}
173
174#[cfg(test)]
175mod tests {
176 #[cfg(feature = "serde")]
177 use crate::Query;
178 use crate::system_resolver;
179 #[cfg(feature = "serde")]
180 use std::num::NonZeroU16;
181
182 /// Bootstrap polls DNS to see nodes come and go, which a response cache would hide.
183 #[test]
184 fn the_system_resolver_keeps_no_response_cache() {
185 let resolver = system_resolver().expect("the system resolver is built");
186 assert_eq!(resolver.options().cache_size, 0);
187 }
188
189 /// The documented config form, which a config file provides: either query is one key, and a
190 /// zero port is unrepresentable rather than resolving to unusable addresses.
191 #[cfg(feature = "serde")]
192 #[test]
193 fn a_query_deserializes_from_its_documented_form() {
194 let query = serde_json::from_str::<Query>(r#"{ "srv": "_tellus._udp.example" }"#)
195 .expect("the SRV form deserializes");
196 assert_eq!(query, Query::Srv("_tellus._udp.example".to_string()));
197
198 let query =
199 serde_json::from_str::<Query>(r#"{ "ip": { "name": "nodes.example", "port": 7000 } }"#)
200 .expect("the A/AAAA form deserializes");
201 assert_eq!(
202 query,
203 Query::Ip {
204 name: "nodes.example".to_string(),
205 port: NonZeroU16::new(7_000).expect("7000 is not zero"),
206 }
207 );
208
209 assert!(
210 serde_json::from_str::<Query>(r#"{ "ip": { "name": "nodes.example", "port": 0 } }"#)
211 .is_err()
212 );
213 assert!(
214 serde_json::from_str::<Query>(r#"{ "ip": { "nmae": "nodes.example", "port": 7000 } }"#)
215 .is_err()
216 );
217 assert!(serde_json::from_str::<Query>(r#"{ "svr": "_tellus._udp.example" }"#).is_err());
218 }
219}