node & server: add log

server & node: log

node & server: instrument
This commit is contained in:
Luna Yao
2026-03-31 15:09:02 +02:00
parent 8cb9856f09
commit 0e5d2cbf08
2 changed files with 31 additions and 9 deletions
+13 -3
View File
@@ -18,6 +18,7 @@ use tokio::sync::{broadcast, Notify};
use tokio::task::JoinSet;
use tokio::time::{sleep, sleep_until, Instant};
use tokio_util::sync::CancellationToken;
use tracing::instrument;
use uuid::Uuid;
#[derive(Debug)]
@@ -73,7 +74,9 @@ impl DnsNode {
pub fn start(&self) {
let this = self.clone();
self.runtime.start(None, |token| async move {
tokio::join!(this.run_election(token.clone()), this.run_node(token));
tracing::info!("starting DnsNode");
this.elect.notify_one();
tokio::join!(this.run_election(token.clone()), this.run(token));
});
}
@@ -81,6 +84,7 @@ impl DnsNode {
self.runtime.stop().await.unwrap_or(Ok(()))
}
#[instrument(skip_all, name = "DnsNode election loop")]
async fn run_election(&self, token: CancellationToken) {
loop {
tokio::select! {
@@ -93,11 +97,16 @@ impl DnsNode {
_ = sleep(DNS_SERVER_ELECTION_INTERVAL) => {}
}
tracing::info!("trying to become DNS server");
let mut rpc =
StandAloneServer::new(TcpTunnelListener::new(DNS_SERVER_RPC_ADDR.clone()));
if rpc.serve().await.is_err() {
// Another instance already owns the address — that's fine.
// Another node already owns the address — that's fine.
tracing::info!(
"failed to bind RPC server, another node might have won the election"
);
continue;
}
@@ -129,7 +138,8 @@ impl DnsNode {
}
}
async fn run_node(&self, token: CancellationToken) {
#[instrument(skip_all, name = "DnsNode main loop")]
async fn run(&self, token: CancellationToken) {
let mut rpc = StandAloneClient::new(TcpTunnelConnector::new(DNS_SERVER_RPC_ADDR.clone()));
let mut heartbeat = HeartbeatRequest {
id: Some(self.id().into()),
+18 -6
View File
@@ -34,6 +34,7 @@ use std::io;
use std::net::{IpAddr, Ipv4Addr, SocketAddr, SocketAddrV4};
use std::{sync::Arc, time::Duration};
use tokio_util::sync::CancellationToken;
use tracing::{instrument, Instrument};
#[derive(Clone)]
pub struct DynamicCatalog {
@@ -187,11 +188,14 @@ impl DnsServer {
}
}
runtime.start(Some(server.shutdown_token().clone()), |_| async move {
server
.block_until_done()
.await
.unwrap_or_else(|e| tracing::error!("DNS server exited with error: {:?}", e));
runtime.start(Some(server.shutdown_token().clone()), |_| {
async move {
server
.block_until_done()
.await
.unwrap_or_else(|e| tracing::error!("DNS server exited with error: {:?}", e));
}
.instrument(tracing::info_span!("DNS server backend runtime"))
});
Ok(())
@@ -241,6 +245,7 @@ impl DnsServer {
Ok(())
}
#[instrument(skip_all, name = "DnsServer main loop")]
pub async fn run(&self, token: CancellationToken) {
let dirty = &self.mgr.dirty;
let mut runtime = None;
@@ -412,6 +417,8 @@ impl DnsServer {
return None;
}
tracing::warn!("HIJACKING PACKET");
let response_payload = {
let response = ResponseHandle::new(512);
@@ -526,7 +533,12 @@ mod tests {
async fn create_test_server() -> Arc<DnsServer> {
let peer_mgr = create_mock_peer_manager().await;
let global_ctx = peer_mgr.get_global_ctx();
Arc::new(DnsServer::new(peer_mgr, global_ctx, #[cfg(feature = "tun")] ArcNicCtx::default()))
Arc::new(DnsServer::new(
peer_mgr,
global_ctx,
#[cfg(feature = "tun")]
ArcNicCtx::default(),
))
}
/// Build a raw IPv4 packet (as `Vec<u8>`) carrying the given L4 payload bytes.