Configurable bind addrs

binaries/cli/src/main.rs

@@ -1,4 +1,7 @@
-use std::{net::Ipv4Addr, path::PathBuf};
+use std::{
+    net::{IpAddr, Ipv4Addr},
+    path::PathBuf,
+};
 
 use attach::attach_dataflow;
 use clap::Parser;
@@ -103,6 +106,10 @@ enum Command {
     Daemon {
         #[clap(long)]
         machine_id: Option<String>,
+        #[clap(long, default_value_t = SocketAddr::new(
+            IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 0)
+        )]
+        bind: SocketAddr,
         #[clap(long)]
         coordinator_addr: Option<SocketAddr>,
 
@@ -112,7 +119,12 @@ enum Command {
     /// Run runtime
     Runtime,
     /// Run coordinator
-    Coordinator { port: Option<u16> },
+    Coordinator {
+        #[clap(long, default_value_t = SocketAddr::new(
+            IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)), DORA_COORDINATOR_PORT_DEFAULT)
+        )]
+        bind: SocketAddr,
+    },
 }
 
 #[derive(Debug, clap::Args)]
@@ -266,20 +278,21 @@ fn run() -> eyre::Result<()> {
             }
         }
         Command::Destroy { config } => up::destroy(config.as_deref())?,
-        Command::Coordinator { port } => {
+        Command::Coordinator { bind } => {
             let rt = Builder::new_multi_thread()
                 .enable_all()
                 .build()
                 .context("tokio runtime failed")?;
             rt.block_on(async {
                 let (_port, task) =
-                    dora_coordinator::start(port, futures::stream::empty::<Event>()).await?;
+                    dora_coordinator::start(bind, futures::stream::empty::<Event>()).await?;
                 task.await
             })
             .context("failed to run dora-coordinator")?
         }
         Command::Daemon {
             coordinator_addr,
+            bind,
             machine_id,
             run_dataflow,
         } => {
@@ -306,7 +319,7 @@ fn run() -> eyre::Result<()> {
                             let localhost = Ipv4Addr::new(127, 0, 0, 1);
                             (localhost, DORA_COORDINATOR_PORT_DEFAULT).into()
                         });
-                        Daemon::run(addr, machine_id.unwrap_or_default()).await
+                        Daemon::run(addr, machine_id.unwrap_or_default(), bind).await
                     }
                 }
             })

binaries/coordinator/src/lib.rs

@@ -9,10 +9,7 @@ use dora_core::{
     daemon_messages::{DaemonCoordinatorEvent, DaemonCoordinatorReply, Timestamped},
     descriptor::{Descriptor, ResolvedNode},
     message::uhlc::{self, HLC},
-    topics::{
-        control_socket_addr, ControlRequest, ControlRequestReply, DataflowId,
-        DORA_COORDINATOR_PORT_DEFAULT,
-    },
+    topics::{control_socket_addr, ControlRequest, ControlRequestReply, DataflowId},
 };
 use eyre::{bail, eyre, ContextCompat, WrapErr};
 use futures::{stream::FuturesUnordered, Future, Stream, StreamExt};
@@ -39,15 +36,13 @@ mod run;
 mod tcp_utils;
 
 pub async fn start(
-    port: Option<u16>,
+    bind: SocketAddr,
     external_events: impl Stream<Item = Event> + Unpin,
-) -> Result<(u16, impl Future<Output = eyre::Result<()>>), eyre::ErrReport> {
-    let port = port.unwrap_or(DORA_COORDINATOR_PORT_DEFAULT);
-    let listener = listener::create_listener(port).await?;
-    let port = listener
+) -> Result<(SocketAddr, impl Future<Output = eyre::Result<()>>), eyre::ErrReport> {
+    let listener = listener::create_listener(bind).await?;
+    let bound_addr = listener
         .local_addr()
-        .wrap_err("failed to get local addr of listener")?
-        .port();
+        .wrap_err("failed to get local addr of listener")?;
     let mut tasks = FuturesUnordered::new();
 
     // Setup ctrl-c handler
@@ -65,7 +60,7 @@ pub async fn start(
         tracing::debug!("all spawned tasks finished, exiting..");
         Ok(())
     };
-    Ok((port, future))
+    Ok((bound_addr, future))
 }
 
 // Resolve the dataflow name.
@@ -481,6 +476,12 @@ async fn start_inner(
                 let mut disconnected = BTreeSet::new();
                 for (machine_id, connection) in &mut daemon_connections {
                     if connection.last_heartbeat.elapsed() > Duration::from_secs(15) {
+                        tracing::warn!(
+                            "no heartbeat message from machine `{machine_id}` since {:?}",
+                            connection.last_heartbeat.elapsed()
+                        )
+                    }
+                    if connection.last_heartbeat.elapsed() > Duration::from_secs(30) {
                         disconnected.insert(machine_id.clone());
                         continue;
                     }
@@ -500,7 +501,7 @@ async fn start_inner(
                     }
                 }
                 if !disconnected.is_empty() {
-                    tracing::info!("Disconnecting daemons that failed watchdog: {disconnected:?}");
+                    tracing::error!("Disconnecting daemons that failed watchdog: {disconnected:?}");
                     for machine_id in disconnected {
                         daemon_connections.remove(&machine_id);
                     }

binaries/coordinator/src/listener.rs

@@ -1,15 +1,14 @@
 use crate::{tcp_utils::tcp_receive, DaemonEvent, DataflowEvent, Event};
 use dora_core::{coordinator_messages, daemon_messages::Timestamped, message::uhlc::HLC};
 use eyre::{eyre, Context};
-use std::{io::ErrorKind, net::Ipv4Addr, sync::Arc};
+use std::{io::ErrorKind, net::SocketAddr, sync::Arc};
 use tokio::{
     net::{TcpListener, TcpStream},
     sync::mpsc,
 };
 
-pub async fn create_listener(port: u16) -> eyre::Result<TcpListener> {
-    let localhost = Ipv4Addr::new(127, 0, 0, 1);
-    let socket = match TcpListener::bind((localhost, port)).await {
+pub async fn create_listener(bind: SocketAddr) -> eyre::Result<TcpListener> {
+    let socket = match TcpListener::bind(bind).await {
         Ok(socket) => socket,
         Err(err) => {
             return Err(eyre::Report::new(err).wrap_err("failed to create local TCP listener"))

binaries/daemon/src/inter_daemon.rs

@@ -1,11 +1,7 @@
 use crate::tcp_utils::{tcp_receive, tcp_send};
 use dora_core::daemon_messages::{InterDaemonEvent, Timestamped};
 use eyre::{Context, ContextCompat};
-use std::{
-    collections::BTreeMap,
-    io::ErrorKind,
-    net::{Ipv4Addr, SocketAddr},
-};
+use std::{collections::BTreeMap, io::ErrorKind, net::SocketAddr};
 use tokio::net::{TcpListener, TcpStream};
 
 pub struct InterDaemonConnection {
@@ -65,11 +61,11 @@ pub async fn send_inter_daemon_event(
 }
 
 pub async fn spawn_listener_loop(
+    bind: SocketAddr,
     machine_id: String,
     events_tx: flume::Sender<Timestamped<InterDaemonEvent>>,
 ) -> eyre::Result<SocketAddr> {
-    let localhost = Ipv4Addr::new(127, 0, 0, 1);
-    let socket = match TcpListener::bind((localhost, 0)).await {
+    let socket = match TcpListener::bind(bind).await {
         Ok(socket) => socket,
         Err(err) => {
             return Err(eyre::Report::new(err).wrap_err("failed to create local TCP listener"))
@@ -79,6 +75,7 @@ pub async fn spawn_listener_loop(
         .local_addr()
         .wrap_err("failed to get local addr of socket")?;
 
+    tracing::debug!("inter-daemon listener starting for machine `{machine_id}` on {socket_addr}");
     tokio::spawn(async move {
         listener_loop(socket, events_tx).await;
         tracing::debug!("inter-daemon listener loop finished for machine `{machine_id}`");

binaries/daemon/src/lib.rs

@@ -77,15 +77,19 @@ pub struct Daemon {
 }
 
 impl Daemon {
-    pub async fn run(coordinator_addr: SocketAddr, machine_id: String) -> eyre::Result<()> {
+    pub async fn run(
+        coordinator_addr: SocketAddr,
+        machine_id: String,
+        bind_addr: SocketAddr,
+    ) -> eyre::Result<()> {
         let clock = Arc::new(HLC::default());
 
         let ctrlc_events = set_up_ctrlc_handler(clock.clone())?;
 
         // spawn listen loop
         let (events_tx, events_rx) = flume::bounded(10);
         let listen_socket =
-            inter_daemon::spawn_listener_loop(machine_id.clone(), events_tx).await?;
+            inter_daemon::spawn_listener_loop(bind_addr, machine_id.clone(), events_tx).await?;
         let daemon_events = events_rx.into_stream().map(|e| Timestamped {
             inner: Event::Daemon(e.inner),
             timestamp: e.timestamp,

examples/multiple-daemons/run.rs

@@ -1,14 +1,14 @@
 use dora_coordinator::{ControlEvent, Event};
 use dora_core::{
     descriptor::Descriptor,
-    topics::{ControlRequest, ControlRequestReply, DataflowId},
+    topics::{ControlRequest, ControlRequestReply, DataflowId, DORA_COORDINATOR_PORT_DEFAULT},
 };
 use dora_tracing::set_up_tracing;
 use eyre::{bail, Context};
 
 use std::{
     collections::BTreeSet,
-    net::{Ipv4Addr, SocketAddr},
+    net::{IpAddr, Ipv4Addr, SocketAddr},
     path::Path,
     time::Duration,
 };
@@ -34,9 +34,13 @@ async fn main() -> eyre::Result<()> {
     build_dataflow(dataflow).await?;
 
     let (coordinator_events_tx, coordinator_events_rx) = mpsc::channel(1);
-    let (coordinator_port, coordinator) =
-        dora_coordinator::start(None, ReceiverStream::new(coordinator_events_rx)).await?;
-    let coordinator_addr = SocketAddr::new(Ipv4Addr::LOCALHOST.into(), coordinator_port);
+    let coordinator_bind = SocketAddr::new(
+        IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)),
+        DORA_COORDINATOR_PORT_DEFAULT,
+    );
+    let (coordinator_addr, coordinator) =
+        dora_coordinator::start(coordinator_bind, ReceiverStream::new(coordinator_events_rx))
+            .await?;
     let daemon_a = run_daemon(coordinator_addr.to_string(), "A".into());
     let daemon_b = run_daemon(coordinator_addr.to_string(), "B".into());