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