diff --git a/easytier/src/dns/client.rs b/easytier/src/dns/client.rs index e8a8b4d5..36b0884a 100644 --- a/easytier/src/dns/client.rs +++ b/easytier/src/dns/client.rs @@ -1,154 +1,31 @@ -use super::config::{DnsGlobalCtxExt, DNS_SERVER_RPC_ADDR}; -use crate::common::config::ConfigLoader; -use crate::common::PeerId; -use crate::peer_center::instance::PeerCenterPeerManagerTrait; +use super::config::DNS_SERVER_RPC_ADDR; +use crate::dns::peer_mgr::DnsPeerMgr; use crate::peers::peer_manager::PeerManager; use crate::proto::dns::{ - DeterministicDigest, DnsPeerManagerRpc, DnsPeerManagerRpcClientFactory, - DnsPeerManagerRpcServer, DnsServerRpcClientFactory, DnsSnapshot, GetExportConfigRequest, - GetExportConfigResponse, HeartbeatRequest, + DeterministicDigest, DnsPeerManagerRpcServer, DnsServerRpcClientFactory, HeartbeatRequest, }; use crate::proto::peer_rpc::RoutePeerInfo; use crate::proto::rpc_impl::standalone::StandAloneClient; -use crate::proto::rpc_types; use crate::proto::rpc_types::controller::BaseController; use crate::tunnel::tcp::TcpTunnelConnector; -use anyhow::Context; use derivative::Derivative; -use derive_more::Deref; -use moka::future::Cache; -use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::atomic::Ordering; use std::sync::Arc; use std::time::Duration; use tokio::task::JoinSet; -use url::Url; use uuid::Uuid; -#[derive(Debug, Clone)] -pub struct DnsPeerInfo { - digest: Vec, - config: GetExportConfigResponse, -} - -impl DnsPeerInfo { - pub fn new(config: GetExportConfigResponse) -> Self { - Self { - digest: config.digest(), - config, - } - } -} - -#[derive(Derivative, Deref)] -#[derivative(Debug)] -pub struct DnsPeerManager { - #[deref] - mgr: Arc, - - cache: Cache, - dirty: AtomicBool, -} - -impl DnsPeerManager { - pub fn new(peer_mgr: Arc) -> Self { - Self { - mgr: peer_mgr.clone(), - cache: Cache::builder() - .time_to_live(Duration::from_secs(5)) - .build() - .into(), - dirty: AtomicBool::new(true), - } - } - - fn snapshot(&self) -> DnsSnapshot { - let global_ctx = self.get_global_ctx_ref(); - let config = global_ctx.config.get_dns(); - - let mut zones = Vec::new(); - - zones.extend(config.zones.iter().map(Into::into)); - zones.extend(global_ctx.dns_self_zone().as_ref().map(Into::into)); - - for (_, info) in self.cache.iter() { - zones.extend(info.config.zones.clone().into_iter()); - } - - DnsSnapshot { - zones, - addresses: config - .addresses - .clone() - .into_iter() - .map(Into::into) - .collect(), - listeners: config - .listeners - .iter() - .map(Url::from) - .map(Into::into) - .collect(), - } - } - - async fn refresh(&self, peer_id: PeerId, digest: Vec) { - if let Some(info) = self.cache.get(&peer_id).await { - if info.digest == *digest { - return; - } - }; - - match self.fetch(peer_id).await { - Ok(config) => { - self.cache.insert(peer_id, DnsPeerInfo::new(config)).await; - } - Err(e) => { - tracing::warn!("failed to fetch dns config from peer {}: {:?}", peer_id, e); - self.cache.invalidate(&peer_id).await; - } - } - - self.dirty.store(true, Ordering::Release); - } - - async fn fetch(&self, peer_id: PeerId) -> anyhow::Result { - self.get_peer_rpc_mgr() - .rpc_client() - .scoped_client::>( - self.mgr.my_peer_id(), - peer_id, - "".to_string(), - ) - .get_export_config(BaseController::default(), GetExportConfigRequest {}) - .await - .context("rpc call failed") - } -} - -#[async_trait::async_trait] -impl DnsPeerManagerRpc for DnsPeerManager { - type Controller = BaseController; - - async fn get_export_config( - &self, - _: Self::Controller, - _: GetExportConfigRequest, - ) -> rpc_types::error::Result { - Ok(self.get_global_ctx_ref().dns_export_config()) - } -} - #[derive(Derivative)] #[derivative(Debug)] pub struct DnsClient { - mgr: Arc, + mgr: Arc, tasks: JoinSet<()>, } impl DnsClient { pub fn new(peer_mgr: Arc) -> Self { - let mgr = Arc::new(DnsPeerManager::new(peer_mgr.clone())); + let mgr = Arc::new(DnsPeerMgr::new(peer_mgr.clone())); peer_mgr .get_peer_rpc_mgr() .rpc_server() diff --git a/easytier/src/dns/mod.rs b/easytier/src/dns/mod.rs index 7a2632e3..d543172e 100644 --- a/easytier/src/dns/mod.rs +++ b/easytier/src/dns/mod.rs @@ -2,3 +2,4 @@ pub mod client; pub mod config; mod utils; pub mod zone; +mod peer_mgr; diff --git a/easytier/src/dns/peer_mgr.rs b/easytier/src/dns/peer_mgr.rs new file mode 100644 index 00000000..e8c167e9 --- /dev/null +++ b/easytier/src/dns/peer_mgr.rs @@ -0,0 +1,130 @@ +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::time::Duration; +use derivative::Derivative; +use derive_more::Deref; +use moka::future::Cache; +use url::Url; +use anyhow::Context; +use crate::common::config::ConfigLoader; +use crate::common::PeerId; +use crate::dns::config::DnsGlobalCtxExt; +use crate::peer_center::instance::PeerCenterPeerManagerTrait; +use crate::peers::peer_manager::PeerManager; +use crate::proto::dns::{DeterministicDigest, DnsPeerManagerRpc, DnsPeerManagerRpcClientFactory, DnsSnapshot, GetExportConfigRequest, GetExportConfigResponse}; +use crate::proto::rpc_types; +use crate::proto::rpc_types::controller::BaseController; + +#[derive(Debug, Clone)] +pub struct DnsPeerInfo { + digest: Vec, + config: GetExportConfigResponse, +} + +impl DnsPeerInfo { + pub fn new(config: GetExportConfigResponse) -> Self { + Self { + digest: config.digest(), + config, + } + } +} + +#[derive(Derivative, Deref)] +#[derivative(Debug)] +pub struct DnsPeerMgr { + #[deref] + mgr: Arc, + + cache: Cache, + pub(super) dirty: AtomicBool, +} + +impl DnsPeerMgr { + pub fn new(peer_mgr: Arc) -> Self { + Self { + mgr: peer_mgr.clone(), + cache: Cache::builder() + .time_to_live(Duration::from_secs(5)) + .build() + .into(), + dirty: AtomicBool::new(true), + } + } + + pub fn snapshot(&self) -> DnsSnapshot { + let global_ctx = self.get_global_ctx_ref(); + let config = global_ctx.config.get_dns(); + + let mut zones = Vec::new(); + + zones.extend(config.zones.iter().map(Into::into)); + zones.extend(global_ctx.dns_self_zone().as_ref().map(Into::into)); + + for (_, info) in self.cache.iter() { + zones.extend(info.config.zones.clone().into_iter()); + } + + DnsSnapshot { + zones, + addresses: config + .addresses + .clone() + .into_iter() + .map(Into::into) + .collect(), + listeners: config + .listeners + .iter() + .map(Url::from) + .map(Into::into) + .collect(), + } + } + + pub(super) async fn refresh(&self, peer_id: PeerId, digest: Vec) { + if let Some(info) = self.cache.get(&peer_id).await { + if info.digest == *digest { + return; + } + }; + + match self.fetch(peer_id).await { + Ok(config) => { + self.cache.insert(peer_id, DnsPeerInfo::new(config)).await; + } + Err(e) => { + tracing::warn!("failed to fetch dns config from peer {}: {:?}", peer_id, e); + self.cache.invalidate(&peer_id).await; + } + } + + self.dirty.store(true, Ordering::Release); + } + + async fn fetch(&self, peer_id: PeerId) -> anyhow::Result { + self.get_peer_rpc_mgr() + .rpc_client() + .scoped_client::>( + self.mgr.my_peer_id(), + peer_id, + "".to_string(), + ) + .get_export_config(BaseController::default(), GetExportConfigRequest {}) + .await + .context("rpc call failed") + } +} + +#[async_trait::async_trait] +impl DnsPeerManagerRpc for DnsPeerMgr { + type Controller = BaseController; + + async fn get_export_config( + &self, + _: Self::Controller, + _: GetExportConfigRequest, + ) -> rpc_types::error::Result { + Ok(self.get_global_ctx_ref().dns_export_config()) + } +} \ No newline at end of file