Support for loading IP addresses from DiploNAT autodiscovery #4
7 changed files with 360 additions and 74 deletions
18
Cargo.lock
generated
18
Cargo.lock
generated
|
|
@ -2,6 +2,15 @@
|
|||
# It is not intended for manual editing.
|
||||
version = 3
|
||||
|
||||
[[package]]
|
||||
name = "aho-corasick"
|
||||
version = "0.7.20"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "cc936419f96fa211c1b9166887b38e5e40b19958e5b895be7c1f93adec7071ac"
|
||||
dependencies = [
|
||||
"memchr",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "ansi_term"
|
||||
version = "0.12.1"
|
||||
|
|
@ -104,8 +113,10 @@ dependencies = [
|
|||
"async-trait",
|
||||
"df-consul",
|
||||
"futures",
|
||||
"regex",
|
||||
"reqwest",
|
||||
"serde",
|
||||
"serde_json",
|
||||
"structopt",
|
||||
"tokio",
|
||||
"tracing",
|
||||
|
|
@ -114,11 +125,12 @@ dependencies = [
|
|||
|
||||
[[package]]
|
||||
name = "df-consul"
|
||||
version = "0.3.3"
|
||||
version = "0.3.4"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "0e38cfbab431b53dfd2d09f2a9902510c636d3d7397645bac5cf1959cfde2999"
|
||||
checksum = "565fcd7efcbdc3e3420e70bc38187a8dd6d5f22759858c32e6af14329bf27ff3"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"base64",
|
||||
"bytes",
|
||||
"futures",
|
||||
"log",
|
||||
|
|
@ -547,6 +559,8 @@ version = "1.7.0"
|
|||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "e076559ef8e241f2ae3479e36f97bd5741c0330689e217ad51ce2c76808b868a"
|
||||
dependencies = [
|
||||
"aho-corasick",
|
||||
"memchr",
|
||||
"regex-syntax",
|
||||
]
|
||||
|
||||
|
|
|
|||
47
Cargo.nix
47
Cargo.nix
|
|
@ -23,7 +23,7 @@ args@{
|
|||
ignoreLockHash,
|
||||
}:
|
||||
let
|
||||
nixifiedLockHash = "66706604b5c3f270cb3a6b72cb23bf3abba94b9d03fed4048584d92e89851d54";
|
||||
nixifiedLockHash = "d1a9c3cd406d87d45242e62c6697855d0e5e3c787d92d2fcd084b41469659da1";
|
||||
workspaceSrc = if args.workspaceSrc == null then ./. else args.workspaceSrc;
|
||||
currentLockHash = builtins.hashFile "sha256" (workspaceSrc + /Cargo.lock);
|
||||
lockHashIgnored = if ignoreLockHash
|
||||
|
|
@ -47,6 +47,20 @@ in
|
|||
workspace = {
|
||||
d53 = rustPackages.unknown.d53."0.1.0";
|
||||
};
|
||||
"registry+https://github.com/rust-lang/crates.io-index".aho-corasick."0.7.20" = overridableMkRustCrate (profileName: rec {
|
||||
name = "aho-corasick";
|
||||
version = "0.7.20";
|
||||
registry = "registry+https://github.com/rust-lang/crates.io-index";
|
||||
src = fetchCratesIo { inherit name version; sha256 = "cc936419f96fa211c1b9166887b38e5e40b19958e5b895be7c1f93adec7071ac"; };
|
||||
features = builtins.concatLists [
|
||||
[ "default" ]
|
||||
[ "std" ]
|
||||
];
|
||||
dependencies = {
|
||||
memchr = (rustPackages."registry+https://github.com/rust-lang/crates.io-index".memchr."2.5.0" { inherit profileName; }).out;
|
||||
};
|
||||
});
|
||||
|
||||
"registry+https://github.com/rust-lang/crates.io-index".ansi_term."0.12.1" = overridableMkRustCrate (profileName: rec {
|
||||
name = "ansi_term";
|
||||
version = "0.12.1";
|
||||
|
|
@ -188,10 +202,12 @@ in
|
|||
dependencies = {
|
||||
anyhow = (rustPackages."registry+https://github.com/rust-lang/crates.io-index".anyhow."1.0.66" { inherit profileName; }).out;
|
||||
async_trait = (buildRustPackages."registry+https://github.com/rust-lang/crates.io-index".async-trait."0.1.59" { profileName = "__noProfile"; }).out;
|
||||
df_consul = (rustPackages."registry+https://github.com/rust-lang/crates.io-index".df-consul."0.2.0" { inherit profileName; }).out;
|
||||
df_consul = (rustPackages."registry+https://github.com/rust-lang/crates.io-index".df-consul."0.3.4" { inherit profileName; }).out;
|
||||
futures = (rustPackages."registry+https://github.com/rust-lang/crates.io-index".futures."0.3.25" { inherit profileName; }).out;
|
||||
regex = (rustPackages."registry+https://github.com/rust-lang/crates.io-index".regex."1.7.0" { inherit profileName; }).out;
|
||||
reqwest = (rustPackages."registry+https://github.com/rust-lang/crates.io-index".reqwest."0.11.13" { inherit profileName; }).out;
|
||||
serde = (rustPackages."registry+https://github.com/rust-lang/crates.io-index".serde."1.0.149" { inherit profileName; }).out;
|
||||
serde_json = (rustPackages."registry+https://github.com/rust-lang/crates.io-index".serde_json."1.0.89" { inherit profileName; }).out;
|
||||
structopt = (rustPackages."registry+https://github.com/rust-lang/crates.io-index".structopt."0.3.26" { inherit profileName; }).out;
|
||||
tokio = (rustPackages."registry+https://github.com/rust-lang/crates.io-index".tokio."1.23.0" { inherit profileName; }).out;
|
||||
tracing = (rustPackages."registry+https://github.com/rust-lang/crates.io-index".tracing."0.1.37" { inherit profileName; }).out;
|
||||
|
|
@ -199,17 +215,20 @@ in
|
|||
};
|
||||
});
|
||||
|
||||
"registry+https://github.com/rust-lang/crates.io-index".df-consul."0.2.0" = overridableMkRustCrate (profileName: rec {
|
||||
"registry+https://github.com/rust-lang/crates.io-index".df-consul."0.3.4" = overridableMkRustCrate (profileName: rec {
|
||||
name = "df-consul";
|
||||
version = "0.2.0";
|
||||
version = "0.3.4";
|
||||
registry = "registry+https://github.com/rust-lang/crates.io-index";
|
||||
src = fetchCratesIo { inherit name version; sha256 = "a351d00f138e768845cdefb9ae27b79aeed97c698745c73bb2805cad1167aa81"; };
|
||||
src = fetchCratesIo { inherit name version; sha256 = "565fcd7efcbdc3e3420e70bc38187a8dd6d5f22759858c32e6af14329bf27ff3"; };
|
||||
dependencies = {
|
||||
anyhow = (rustPackages."registry+https://github.com/rust-lang/crates.io-index".anyhow."1.0.66" { inherit profileName; }).out;
|
||||
base64 = (rustPackages."registry+https://github.com/rust-lang/crates.io-index".base64."0.13.1" { inherit profileName; }).out;
|
||||
bytes = (rustPackages."registry+https://github.com/rust-lang/crates.io-index".bytes."1.3.0" { inherit profileName; }).out;
|
||||
futures = (rustPackages."registry+https://github.com/rust-lang/crates.io-index".futures."0.3.25" { inherit profileName; }).out;
|
||||
log = (rustPackages."registry+https://github.com/rust-lang/crates.io-index".log."0.4.17" { inherit profileName; }).out;
|
||||
reqwest = (rustPackages."registry+https://github.com/rust-lang/crates.io-index".reqwest."0.11.13" { inherit profileName; }).out;
|
||||
serde = (rustPackages."registry+https://github.com/rust-lang/crates.io-index".serde."1.0.149" { inherit profileName; }).out;
|
||||
tokio = (rustPackages."registry+https://github.com/rust-lang/crates.io-index".tokio."1.23.0" { inherit profileName; }).out;
|
||||
};
|
||||
});
|
||||
|
||||
|
|
@ -807,9 +826,27 @@ in
|
|||
registry = "registry+https://github.com/rust-lang/crates.io-index";
|
||||
src = fetchCratesIo { inherit name version; sha256 = "e076559ef8e241f2ae3479e36f97bd5741c0330689e217ad51ce2c76808b868a"; };
|
||||
features = builtins.concatLists [
|
||||
[ "aho-corasick" ]
|
||||
[ "default" ]
|
||||
[ "memchr" ]
|
||||
[ "perf" ]
|
||||
[ "perf-cache" ]
|
||||
[ "perf-dfa" ]
|
||||
[ "perf-inline" ]
|
||||
[ "perf-literal" ]
|
||||
[ "std" ]
|
||||
[ "unicode" ]
|
||||
[ "unicode-age" ]
|
||||
[ "unicode-bool" ]
|
||||
[ "unicode-case" ]
|
||||
[ "unicode-gencat" ]
|
||||
[ "unicode-perl" ]
|
||||
[ "unicode-script" ]
|
||||
[ "unicode-segment" ]
|
||||
];
|
||||
dependencies = {
|
||||
aho_corasick = (rustPackages."registry+https://github.com/rust-lang/crates.io-index".aho-corasick."0.7.20" { inherit profileName; }).out;
|
||||
memchr = (rustPackages."registry+https://github.com/rust-lang/crates.io-index".memchr."2.5.0" { inherit profileName; }).out;
|
||||
regex_syntax = (rustPackages."registry+https://github.com/rust-lang/crates.io-index".regex-syntax."0.6.28" { inherit profileName; }).out;
|
||||
};
|
||||
});
|
||||
|
|
|
|||
|
|
@ -15,6 +15,8 @@ tracing-subscriber = { version = "0.3", features = ["env-filter"] }
|
|||
structopt = "0.3"
|
||||
tokio = { version = "1.22", default-features = false, features = ["rt", "rt-multi-thread", "io-util", "net", "time", "macros", "sync", "signal", "fs"] }
|
||||
reqwest = { version = "0.11", default-features = false, features = ["json", "rustls-tls-webpki-roots" ] }
|
||||
regex = "1"
|
||||
serde = { version = "1.0.107", features = ["derive"] }
|
||||
serde_json = "1.0"
|
||||
|
||||
df-consul = "0.3.3"
|
||||
df-consul = "0.3.4"
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
#!/bin/sh
|
||||
|
||||
RUST_LOG=d53=info cargo run \
|
||||
RUST_LOG=d53=debug cargo run \
|
||||
-- \
|
||||
--consul-addr http://localhost:8500 \
|
||||
--providers deuxfleurs.org:gandi \
|
||||
|
|
|
|||
147
src/autodiscovery.rs
Normal file
147
src/autodiscovery.rs
Normal file
|
|
@ -0,0 +1,147 @@
|
|||
//! Fetch autodiscoverd IP addresses stored by Diplonat into Consul
|
||||
|
||||
use std::collections::HashMap;
|
||||
use std::net::{Ipv4Addr, Ipv6Addr};
|
||||
use std::sync::Arc;
|
||||
use std::time::{Duration, SystemTime};
|
||||
|
||||
use anyhow::{anyhow, bail, Result};
|
||||
use regex::Regex;
|
||||
use serde::{Deserialize, Serialize};
|
||||
use tokio::{select, sync::watch};
|
||||
use tracing::*;
|
||||
|
||||
use df_consul::*;
|
||||
|
||||
#[derive(Serialize, Deserialize, Debug, Eq, PartialEq)]
|
||||
pub struct DiplonatAutodiscoveryResult<A> {
|
||||
pub timestamp: u64,
|
||||
pub address: Option<A>,
|
||||
}
|
||||
|
||||
#[derive(Default, Debug, Eq, PartialEq)]
|
||||
pub struct AutodiscoveredAddresses {
|
||||
pub ipv4: HashMap<String, DiplonatAutodiscoveryResult<Ipv4Addr>>,
|
||||
pub ipv6: HashMap<String, DiplonatAutodiscoveryResult<Ipv6Addr>>,
|
||||
}
|
||||
|
||||
pub fn watch_autodiscovered_ips(
|
||||
consul: Consul,
|
||||
mut must_exit: watch::Receiver<bool>,
|
||||
) -> watch::Receiver<Arc<AutodiscoveredAddresses>> {
|
||||
let (tx, rx) = watch::channel(Arc::new(AutodiscoveredAddresses::default()));
|
||||
let rx2 = rx.clone();
|
||||
|
||||
tokio::spawn(async move {
|
||||
let mut last_index = None;
|
||||
let re = Regex::new(r".*autodiscovery/(ipv[46])/([^/]+)$").unwrap();
|
||||
|
||||
while !*must_exit.borrow() {
|
||||
let r = select! {
|
||||
_ = must_exit.changed() => continue,
|
||||
r = consul.kv_get_prefix("diplonat/autodiscovery/", last_index) => r,
|
||||
};
|
||||
|
||||
let entries = match r {
|
||||
Err(e) => {
|
||||
warn!("Error fetching diplonat autodiscovery consul prefix: {}", e);
|
||||
tokio::time::sleep(Duration::from_secs(30)).await;
|
||||
continue;
|
||||
}
|
||||
Ok(r) => {
|
||||
last_index = Some(r.index());
|
||||
r.into_inner()
|
||||
}
|
||||
};
|
||||
|
||||
let mut addresses = AutodiscoveredAddresses::default();
|
||||
|
||||
for (k, v) in entries {
|
||||
if let Err(e) = parse_autodiscovered_address(&re, &mut addresses, &k, &v) {
|
||||
warn!(
|
||||
"Invalid k/v pair in diplonat autodiscovery results: {} = {} ({})",
|
||||
k,
|
||||
std::str::from_utf8(&v).unwrap_or("<???>"),
|
||||
e
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
if addresses.strip_timestamps() != rx2.borrow().strip_timestamps() {
|
||||
addresses.dump();
|
||||
}
|
||||
|
||||
if tx.send(Arc::new(addresses)).is_err() {
|
||||
info!("Autodiscovered addresses watcher terminating");
|
||||
return;
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
rx
|
||||
}
|
||||
|
||||
fn parse_autodiscovered_address(
|
||||
re: &Regex,
|
||||
addresses: &mut AutodiscoveredAddresses,
|
||||
k: &str,
|
||||
v: &[u8],
|
||||
) -> Result<()> {
|
||||
let caps = re.captures(k).ok_or(anyhow!("key does not match regex"))?;
|
||||
|
||||
if let (Some(family), Some(node)) = (caps.get(1), caps.get(2)) {
|
||||
match family.as_str() {
|
||||
"ipv4" => {
|
||||
let r: DiplonatAutodiscoveryResult<Ipv4Addr> = serde_json::from_slice(v)?;
|
||||
addresses.ipv4.insert(node.as_str().to_string(), r);
|
||||
}
|
||||
"ipv6" => {
|
||||
let r: DiplonatAutodiscoveryResult<Ipv6Addr> = serde_json::from_slice(v)?;
|
||||
addresses.ipv6.insert(node.as_str().to_string(), r);
|
||||
}
|
||||
_ => bail!("invalid address family {}", family.as_str()),
|
||||
}
|
||||
} else {
|
||||
bail!("invalid regex captures {:?}", caps);
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
impl AutodiscoveredAddresses {
|
||||
fn strip_timestamps(
|
||||
&self,
|
||||
) -> (
|
||||
HashMap<&str, Option<Ipv4Addr>>,
|
||||
HashMap<&str, Option<Ipv6Addr>>,
|
||||
) {
|
||||
(
|
||||
self.ipv4
|
||||
.iter()
|
||||
.map(|(k, v)| (k.as_str(), v.address))
|
||||
.collect(),
|
||||
self.ipv6
|
||||
.iter()
|
||||
.map(|(k, v)| (k.as_str(), v.address))
|
||||
.collect(),
|
||||
)
|
||||
}
|
||||
|
||||
fn dump(&self) {
|
||||
println!("---- Autodiscovered addresses (fetched from DiploNAT): ----");
|
||||
for (k, v) in self.ipv4.iter() {
|
||||
println!(" IPv4 {} {} {:?}", k, v.timestamp, v.address);
|
||||
}
|
||||
for (k, v) in self.ipv6.iter() {
|
||||
println!(" IPv6 {} {} {:?}", k, v.timestamp, v.address);
|
||||
}
|
||||
println!("");
|
||||
}
|
||||
}
|
||||
|
||||
pub fn timestamp() -> u64 {
|
||||
SystemTime::now()
|
||||
.duration_since(SystemTime::UNIX_EPOCH)
|
||||
.expect("clock error")
|
||||
.as_secs()
|
||||
}
|
||||
|
|
@ -1,17 +1,23 @@
|
|||
use std::collections::{HashMap, HashSet};
|
||||
use std::fmt;
|
||||
use std::net::{Ipv4Addr, Ipv6Addr};
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
|
||||
use anyhow::Result;
|
||||
use serde::{Deserialize, Serialize};
|
||||
use tokio::{select, sync::watch};
|
||||
use tracing::*;
|
||||
|
||||
use df_consul::*;
|
||||
|
||||
const IPV4_TARGET_METADATA_TAG: &str = "public_ipv4";
|
||||
const IPV6_TARGET_METADATA_TAG: &str = "public_ipv6";
|
||||
use crate::autodiscovery::*;
|
||||
|
||||
const IP_TARGET_METADATA_TAG_PREFIX: &str = "public_";
|
||||
const CNAME_TARGET_METADATA_TAG: &str = "cname_target";
|
||||
|
||||
const AUTODISCOVERY_CACHE_DURATION: u64 = 600; // 10 minutes
|
||||
|
||||
// ---- Extract DNS config from Consul catalog ----
|
||||
|
||||
#[derive(Debug, Eq, PartialEq, Default)]
|
||||
|
|
@ -54,66 +60,60 @@ impl DnsConfig {
|
|||
}
|
||||
}
|
||||
|
||||
fn parse_d53_tag(tag: &str, node: &catalog::Node) -> Option<(DnsEntryKey, DnsEntryValue)> {
|
||||
let splits = tag.split(' ').collect::<Vec<_>>();
|
||||
if splits.len() != 2 {
|
||||
return None;
|
||||
}
|
||||
|
||||
let (record_type, targets) = match splits[0] {
|
||||
"d53-a" => match node.meta.get(IPV4_TARGET_METADATA_TAG) {
|
||||
Some(tgt) => (DnsRecordType::A, [tgt.to_string()].into_iter().collect()),
|
||||
None => {
|
||||
warn!("Got d53-a tag `{}` but node {} does not have a {} metadata value. Tag is ignored.", tag, node.node, IPV4_TARGET_METADATA_TAG);
|
||||
return None;
|
||||
}
|
||||
},
|
||||
"d53-aaaa" => match node.meta.get(IPV6_TARGET_METADATA_TAG) {
|
||||
Some(tgt) => (DnsRecordType::AAAA, [tgt.to_string()].into_iter().collect()),
|
||||
None => {
|
||||
warn!("Got d53-aaaa tag `{}` but node {} does not have a {} metadata value. Tag is ignored.", tag, node.node, IPV6_TARGET_METADATA_TAG);
|
||||
return None;
|
||||
}
|
||||
},
|
||||
"d53-cname" => match node.meta.get(CNAME_TARGET_METADATA_TAG) {
|
||||
Some(tgt) => (
|
||||
DnsRecordType::CNAME,
|
||||
[tgt.to_string()].into_iter().collect(),
|
||||
),
|
||||
None => {
|
||||
warn!("Got d53-cname tag `{}` but node {} does not have a {} metadata value. Tag is ignored.", tag, node.node, CNAME_TARGET_METADATA_TAG);
|
||||
return None;
|
||||
}
|
||||
},
|
||||
_ => return None,
|
||||
};
|
||||
|
||||
Some((
|
||||
DnsEntryKey {
|
||||
dns_path: splits[1].to_string(),
|
||||
record_type,
|
||||
},
|
||||
DnsEntryValue { targets },
|
||||
))
|
||||
}
|
||||
// ---- fetcher and autodiscovery cache ----
|
||||
|
||||
pub fn spawn_dns_config_task(
|
||||
consul: &Consul,
|
||||
mut must_exit: watch::Receiver<bool>,
|
||||
consul: Consul,
|
||||
must_exit: watch::Receiver<bool>,
|
||||
) -> watch::Receiver<Arc<DnsConfig>> {
|
||||
let (tx, rx) = watch::channel(Arc::new(DnsConfig::new()));
|
||||
|
||||
let mut catalog_rx = consul.watch_all_service_health(Duration::from_secs(60));
|
||||
let fetcher = DnsConfigTask { consul };
|
||||
|
||||
tokio::spawn(fetcher.task(tx, must_exit));
|
||||
|
||||
rx
|
||||
}
|
||||
|
||||
struct DnsConfigTask {
|
||||
consul: Consul,
|
||||
}
|
||||
|
||||
impl DnsConfigTask {
|
||||
async fn task(
|
||||
mut self,
|
||||
tx: watch::Sender<Arc<DnsConfig>>,
|
||||
mut must_exit: watch::Receiver<bool>,
|
||||
) {
|
||||
let mut autodiscovery_rx = watch_autodiscovered_ips(self.consul.clone(), must_exit.clone());
|
||||
|
||||
let mut catalog_rx = self
|
||||
.consul
|
||||
.watch_all_service_health(Duration::from_secs(60));
|
||||
|
||||
tokio::spawn(async move {
|
||||
while !*must_exit.borrow() {
|
||||
select! {
|
||||
_ = catalog_rx.changed() => (),
|
||||
_ = autodiscovery_rx.changed() => (),
|
||||
_ = must_exit.changed() => continue,
|
||||
};
|
||||
|
||||
let services = catalog_rx.borrow_and_update();
|
||||
let services = catalog_rx.borrow_and_update().clone();
|
||||
let autodiscovery = autodiscovery_rx.borrow_and_update().clone();
|
||||
match self.parse_catalog(&services, &autodiscovery) {
|
||||
Ok(dns_config) => tx.send(Arc::new(dns_config)).expect("Internal error"),
|
||||
Err(e) => {
|
||||
error!("Error when parsing tags: {}", e);
|
||||
}
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
fn parse_catalog(
|
||||
&mut self,
|
||||
services: &catalog::AllServiceHealth,
|
||||
autodiscovery: &AutodiscoveredAddresses,
|
||||
) -> Result<DnsConfig> {
|
||||
let mut dns_config = DnsConfig::new();
|
||||
for (_svc, nodes) in services.iter() {
|
||||
for node in nodes.iter() {
|
||||
|
|
@ -122,18 +122,103 @@ pub fn spawn_dns_config_task(
|
|||
continue;
|
||||
}
|
||||
for tag in node.service.tags.iter() {
|
||||
if let Some((k, v)) = parse_d53_tag(tag, &node.node) {
|
||||
if let Some((k, v)) = self.parse_d53_tag(tag, &node.node, autodiscovery)? {
|
||||
dns_config.add(k, v);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
tx.send(Arc::new(dns_config)).expect("Internal error");
|
||||
Ok(dns_config)
|
||||
}
|
||||
});
|
||||
|
||||
rx
|
||||
fn parse_d53_tag(
|
||||
&mut self,
|
||||
tag: &str,
|
||||
node: &catalog::Node,
|
||||
autodiscovery: &AutodiscoveredAddresses,
|
||||
) -> Result<Option<(DnsEntryKey, DnsEntryValue)>> {
|
||||
let splits = tag.split(' ').collect::<Vec<_>>();
|
||||
if splits.len() != 2 {
|
||||
return Ok(None);
|
||||
}
|
||||
|
||||
let (record_type, target) = match splits[0] {
|
||||
"d53-a" => match self.get_node_ipv4(&autodiscovery, &node)? {
|
||||
Some(tgt) => (DnsRecordType::A, tgt.to_string()),
|
||||
None => {
|
||||
warn!("Got d53-a tag `{}` but node {} does not appear to have a known public IPv4 address. Tag is ignored.", tag, node.node);
|
||||
return Ok(None);
|
||||
}
|
||||
},
|
||||
"d53-aaaa" => match self.get_node_ipv6(&autodiscovery, &node)? {
|
||||
Some(tgt) => (DnsRecordType::AAAA, tgt.to_string()),
|
||||
None => {
|
||||
warn!("Got d53-aaaa tag `{}` but node {} does not appear to have a known public IPv6 address. Tag is ignored.", tag, node.node);
|
||||
return Ok(None);
|
||||
}
|
||||
},
|
||||
"d53-cname" => match node.meta.get(CNAME_TARGET_METADATA_TAG) {
|
||||
Some(tgt) => (DnsRecordType::CNAME, tgt.to_string()),
|
||||
None => {
|
||||
warn!("Got d53-cname tag `{}` but node {} does not have a {} metadata value. Tag is ignored.", tag, node.node, CNAME_TARGET_METADATA_TAG);
|
||||
return Ok(None);
|
||||
}
|
||||
},
|
||||
_ => return Ok(None),
|
||||
};
|
||||
|
||||
Ok(Some((
|
||||
DnsEntryKey {
|
||||
dns_path: splits[1].to_string(),
|
||||
record_type,
|
||||
},
|
||||
DnsEntryValue {
|
||||
targets: [target].into_iter().collect(),
|
||||
},
|
||||
)))
|
||||
}
|
||||
|
||||
fn get_node_ipv4(
|
||||
&mut self,
|
||||
autodiscovery: &AutodiscoveredAddresses,
|
||||
node: &catalog::Node,
|
||||
) -> Result<Option<Ipv4Addr>> {
|
||||
Self::get_node_ip("ipv4", &autodiscovery.ipv4, node)
|
||||
}
|
||||
|
||||
fn get_node_ipv6(
|
||||
&mut self,
|
||||
autodiscovery: &AutodiscoveredAddresses,
|
||||
node: &catalog::Node,
|
||||
) -> Result<Option<Ipv6Addr>> {
|
||||
Self::get_node_ip("ipv6", &autodiscovery.ipv6, node)
|
||||
}
|
||||
|
||||
fn get_node_ip<A>(
|
||||
family: &'static str,
|
||||
autodiscovery: &HashMap<String, DiplonatAutodiscoveryResult<A>>,
|
||||
node: &catalog::Node,
|
||||
) -> Result<Option<A>>
|
||||
where
|
||||
A: Serialize + for<'de> Deserialize<'de> + std::fmt::Debug + std::str::FromStr + Copy + Eq,
|
||||
<A as std::str::FromStr>::Err: Send + Sync + std::error::Error + 'static,
|
||||
{
|
||||
match autodiscovery.get(&node.node) {
|
||||
Some(ar) if timestamp() <= ar.timestamp + AUTODISCOVERY_CACHE_DURATION => {
|
||||
Ok(ar.address)
|
||||
}
|
||||
x => {
|
||||
if let Some(ar) = x {
|
||||
warn!("{} address for {} from diplonat autodiscovery is outdated (value: {:?}), falling back on value from Consul node meta", family, node.node, ar.address);
|
||||
}
|
||||
|
||||
let meta_tag = format!("{}{}", IP_TARGET_METADATA_TAG_PREFIX, family);
|
||||
let addr = node.meta.get(&meta_tag).map(|x| x.parse()).transpose()?;
|
||||
Ok(addr)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ---- Display impls ----
|
||||
|
|
|
|||
|
|
@ -5,6 +5,7 @@ use tokio::select;
|
|||
use tokio::sync::watch;
|
||||
use tracing::*;
|
||||
|
||||
mod autodiscovery;
|
||||
mod dns_config;
|
||||
mod dns_updater;
|
||||
mod provider;
|
||||
|
|
@ -111,7 +112,7 @@ async fn main() {
|
|||
.map(ToString::to_string)
|
||||
.collect::<Vec<_>>();
|
||||
|
||||
let rx_dns_config = dns_config::spawn_dns_config_task(&consul, exit_signal.clone());
|
||||
let rx_dns_config = dns_config::spawn_dns_config_task(consul, exit_signal.clone());
|
||||
|
||||
let updater_task = tokio::spawn(dns_updater::dns_updater_task(
|
||||
rx_dns_config.clone(),
|
||||
|
|
|
|||
Loading…
Reference in a new issue