Merge branch 'main' into node-errors-before-init

binaries/cli/Cargo.toml

@@ -35,3 +35,4 @@ notify = "5.1.0"
 ctrlc = "3.2.5"
 tracing = "0.1.36"
 dora-tracing = { workspace = true, optional = true }
+bat = "0.23.0"

binaries/cli/src/logs.rs

@@ -0,0 +1,45 @@
+use communication_layer_request_reply::TcpRequestReplyConnection;
+use dora_core::topics::{ControlRequest, ControlRequestReply};
+use eyre::{bail, Context, Result};
+use uuid::Uuid;
+
+use bat::{Input, PrettyPrinter};
+
+pub fn logs(
+    session: &mut TcpRequestReplyConnection,
+    uuid: Option<Uuid>,
+    name: Option<String>,
+    node: String,
+) -> Result<()> {
+    let logs = {
+        let reply_raw = session
+            .request(
+                &serde_json::to_vec(&ControlRequest::Logs {
+                    uuid,
+                    name,
+                    node: node.clone(),
+                })
+                .wrap_err("")?,
+            )
+            .wrap_err("failed to send Logs request message")?;
+
+        let reply = serde_json::from_slice(&reply_raw).wrap_err("failed to parse reply")?;
+        match reply {
+            ControlRequestReply::Logs(logs) => logs,
+            other => bail!("unexpected reply to daemon logs: {other:?}"),
+        }
+    };
+
+    PrettyPrinter::new()
+        .header(true)
+        .grid(true)
+        .line_numbers(true)
+        .paging_mode(bat::PagingMode::QuitIfOneScreen)
+        .inputs(vec![Input::from_bytes(&logs)
+            .name("Logs")
+            .title(format!("Logs from {node}.").as_str())])
+        .print()
+        .wrap_err("Something went wrong with viewing log file")?;
+
+    Ok(())
+}

binaries/cli/src/main.rs

@@ -16,6 +16,7 @@ mod attach;
 mod build;
 mod check;
 mod graph;
+mod logs;
 mod template;
 mod up;
 
@@ -43,7 +44,9 @@ enum Command {
         open: bool,
     },
     /// Run build commands provided in the given dataflow.
-    Build { dataflow: PathBuf },
+    Build {
+        dataflow: PathBuf,
+    },
     /// Generate a new project, node or operator. Choose the language between Rust, Python, C or C++.
     New {
         #[clap(flatten)]
@@ -85,7 +88,10 @@ enum Command {
     List,
     // Planned for future releases:
     // Dashboard,
-    // Logs,
+    Logs {
+        dataflow: String,
+        node: String,
+    },
     // Metrics,
     // Stats,
     // Get,
@@ -166,6 +172,13 @@ fn run() -> eyre::Result<()> {
             coordinator_path.as_deref(),
             daemon_path.as_deref(),
         )?,
+        Command::Logs { dataflow, node } => {
+            let uuid = Uuid::parse_str(&dataflow).ok();
+            let name = if uuid.is_some() { None } else { Some(dataflow) };
+            let mut session =
+                connect_to_coordinator().wrap_err("failed to connect to dora coordinator")?;
+            logs::logs(&mut *session, uuid, name, node)?
+        }
         Command::Start {
             dataflow,
             name,

binaries/coordinator/src/lib.rs

@@ -7,7 +7,7 @@ use dora_core::{
     config::{NodeId, OperatorId},
     coordinator_messages::RegisterResult,
     daemon_messages::{DaemonCoordinatorEvent, DaemonCoordinatorReply},
-    descriptor::Descriptor,
+    descriptor::{Descriptor, ResolvedNode},
     topics::{
         control_socket_addr, ControlRequest, ControlRequestReply, DataflowId,
         DORA_COORDINATOR_PORT_DEFAULT,
@@ -79,6 +79,46 @@ pub async fn start(
     Ok((port, future))
 }
 
+// Resolve the dataflow name.
+// Search for archived dataflows if they are provided.
+fn resolve_name(
+    name: String,
+    running_dataflows: &HashMap<Uuid, RunningDataflow>,
+    archived_dataflows: Option<&HashMap<Uuid, ArchivedDataflow>>,
+) -> eyre::Result<Uuid> {
+    let uuids: Vec<_> = running_dataflows
+        .iter()
+        .filter(|(_, v)| v.name.as_deref() == Some(name.as_str()))
+        .map(|(k, _)| k)
+        .copied()
+        .collect();
+    let archived_uuids: Vec<_> = if let Some(archived_dataflows) = archived_dataflows {
+        archived_dataflows
+            .iter()
+            .filter(|(_, v)| v.name.as_deref() == Some(name.as_str()))
+            .map(|(k, _)| k)
+            .copied()
+            .collect()
+    } else {
+        vec![]
+    };
+
+    if uuids.is_empty() {
+        if archived_uuids.is_empty() {
+            bail!("no dataflow with name `{name}`");
+        } else if let [uuid] = archived_uuids.as_slice() {
+            Ok(*uuid)
+        } else {
+            // TOOD: Index the archived dataflows in order to return logs based on the index.
+            bail!("multiple archived dataflows found with name `{name}`, Please provide the UUID instead.");
+        }
+    } else if let [uuid] = uuids.as_slice() {
+        Ok(*uuid)
+    } else {
+        bail!("multiple dataflows found with name `{name}`");
+    }
+}
+
 async fn start_inner(
     listener: TcpListener,
     tasks: &FuturesUnordered<JoinHandle<()>>,
@@ -116,6 +156,7 @@ async fn start_inner(
     let mut events = (abortable_events, daemon_events).merge();
 
     let mut running_dataflows: HashMap<Uuid, RunningDataflow> = HashMap::new();
+    let mut archived_dataflows: HashMap<Uuid, ArchivedDataflow> = HashMap::new();
     let mut daemon_connections: HashMap<_, DaemonConnection> = HashMap::new();
 
     while let Some(event) = events.next().await {
@@ -222,6 +263,11 @@ async fn start_inner(
                 DataflowEvent::DataflowFinishedOnMachine { machine_id, result } => {
                     match running_dataflows.entry(uuid) {
                         std::collections::hash_map::Entry::Occupied(mut entry) => {
+                            // Archive finished dataflow
+                            if archived_dataflows.get(&uuid).is_none() {
+                                archived_dataflows
+                                    .insert(uuid, ArchivedDataflow::from(entry.get()));
+                            }
                             entry.get_mut().machines.remove(&machine_id);
                             match result {
                                 Ok(()) => {
@@ -330,20 +376,7 @@ async fn start_inner(
                         }
                         ControlRequest::StopByName { name } => {
                             let stop = async {
-                                let uuids: Vec<_> = running_dataflows
-                                    .iter()
-                                    .filter(|(_, v)| v.name.as_deref() == Some(name.as_str()))
-                                    .map(|(k, _)| k)
-                                    .copied()
-                                    .collect();
-                                let dataflow_uuid = if uuids.is_empty() {
-                                    bail!("no running dataflow with name `{name}`");
-                                } else if let [uuid] = uuids.as_slice() {
-                                    *uuid
-                                } else {
-                                    bail!("multiple dataflows found with name `{name}`");
-                                };
-
+                                let dataflow_uuid = resolve_name(name, &running_dataflows, None)?;
                                 stop_dataflow(
                                     &running_dataflows,
                                     dataflow_uuid,
@@ -355,6 +388,25 @@ async fn start_inner(
                             stop.await
                                 .map(|uuid| ControlRequestReply::DataflowStopped { uuid })
                         }
+                        ControlRequest::Logs { uuid, name, node } => {
+                            let dataflow_uuid = if let Some(uuid) = uuid {
+                                uuid
+                            } else if let Some(name) = name {
+                                resolve_name(name, &running_dataflows, Some(&archived_dataflows))?
+                            } else {
+                                bail!("No uuid")
+                            };
+
+                            retrieve_logs(
+                                &running_dataflows,
+                                &archived_dataflows,
+                                dataflow_uuid,
+                                node.into(),
+                                &mut daemon_connections,
+                            )
+                            .await
+                            .map(|logs| ControlRequestReply::Logs(logs))
+                        }
                         ControlRequest::Destroy => {
                             tracing::info!("Received destroy command");
 
@@ -504,6 +556,21 @@ struct RunningDataflow {
     /// IDs of machines that are waiting until all nodes are started.
     pending_machines: BTreeSet<String>,
     init_success: bool,
+    nodes: Vec<ResolvedNode>,
+}
+
+struct ArchivedDataflow {
+    name: Option<String>,
+    nodes: Vec<ResolvedNode>,
+}
+
+impl From<&RunningDataflow> for ArchivedDataflow {
+    fn from(dataflow: &RunningDataflow) -> ArchivedDataflow {
+        ArchivedDataflow {
+            name: dataflow.name.clone(),
+            nodes: dataflow.nodes.clone(),
+        }
+    }
 }
 
 impl PartialEq for RunningDataflow {
@@ -592,14 +659,77 @@ async fn reload_dataflow(
     Ok(())
 }
 
+async fn retrieve_logs(
+    running_dataflows: &HashMap<Uuid, RunningDataflow>,
+    archived_dataflows: &HashMap<Uuid, ArchivedDataflow>,
+    dataflow_id: Uuid,
+    node_id: NodeId,
+    daemon_connections: &mut HashMap<String, DaemonConnection>,
+) -> eyre::Result<Vec<u8>> {
+    let nodes = if let Some(dataflow) = archived_dataflows.get(&dataflow_id) {
+        dataflow.nodes.clone()
+    } else if let Some(dataflow) = running_dataflows.get(&dataflow_id) {
+        dataflow.nodes.clone()
+    } else {
+        bail!("No dataflow found with UUID `{dataflow_id}`")
+    };
+
+    let message = serde_json::to_vec(&DaemonCoordinatorEvent::Logs {
+        dataflow_id,
+        node_id: node_id.clone(),
+    })?;
+
+    let machine_ids: Vec<String> = nodes
+        .iter()
+        .filter(|node| node.id == node_id)
+        .map(|node| node.deploy.machine.clone())
+        .collect();
+
+    let machine_id = if let [machine_id] = &machine_ids[..] {
+        machine_id
+    } else if machine_ids.is_empty() {
+        bail!("No machine contains {}/{}", dataflow_id, node_id)
+    } else {
+        bail!(
+            "More than one machine contains {}/{}. However, it should only be present on one.",
+            dataflow_id,
+            node_id
+        )
+    };
+
+    let daemon_connection = daemon_connections
+        .get_mut(machine_id.as_str())
+        .wrap_err("no daemon connection")?;
+    tcp_send(&mut daemon_connection.stream, &message)
+        .await
+        .wrap_err("failed to send logs message to daemon")?;
+
+    // wait for reply
+    let reply_raw = tcp_receive(&mut daemon_connection.stream)
+        .await
+        .wrap_err("failed to retrieve logs reply from daemon")?;
+    let reply_logs = match serde_json::from_slice(&reply_raw)
+        .wrap_err("failed to deserialize logs reply from daemon")?
+    {
+        DaemonCoordinatorReply::Logs(logs) => logs,
+        other => bail!("unexpected reply after sending logs: {other:?}"),
+    };
+    tracing::info!("successfully retrieved logs for `{dataflow_id}/{node_id}`");
+
+    reply_logs.map_err(|err| eyre!(err))
+}
+
 async fn start_dataflow(
     dataflow: Descriptor,
     working_dir: PathBuf,
     name: Option<String>,
     daemon_connections: &mut HashMap<String, DaemonConnection>,
 ) -> eyre::Result<RunningDataflow> {
-    let SpawnedDataflow { uuid, machines } =
-        spawn_dataflow(dataflow, working_dir, daemon_connections).await?;
+    let SpawnedDataflow {
+        uuid,
+        machines,
+        nodes,
+    } = spawn_dataflow(dataflow, working_dir, daemon_connections).await?;
     Ok(RunningDataflow {
         uuid,
         name,
@@ -610,6 +740,7 @@ async fn start_dataflow(
         },
         init_success: true,
         machines,
+        nodes,
     })
 }
 

binaries/coordinator/src/run/mod.rs

@@ -5,7 +5,7 @@ use crate::{
 
 use dora_core::{
     daemon_messages::{DaemonCoordinatorEvent, DaemonCoordinatorReply, SpawnDataflowNodes},
-    descriptor::Descriptor,
+    descriptor::{Descriptor, ResolvedNode},
 };
 use eyre::{bail, eyre, ContextCompat, WrapErr};
 use std::{
@@ -39,7 +39,7 @@ pub(super) async fn spawn_dataflow(
     let spawn_command = SpawnDataflowNodes {
         dataflow_id: uuid,
         working_dir,
-        nodes,
+        nodes: nodes.clone(),
         communication: dataflow.communication,
         machine_listen_ports,
     };
@@ -54,7 +54,11 @@ pub(super) async fn spawn_dataflow(
 
     tracing::info!("successfully spawned dataflow `{uuid}`");
 
-    Ok(SpawnedDataflow { uuid, machines })
+    Ok(SpawnedDataflow {
+        uuid,
+        machines,
+        nodes,
+    })
 }
 
 async fn spawn_dataflow_on_machine(
@@ -85,4 +89,5 @@ async fn spawn_dataflow_on_machine(
 pub struct SpawnedDataflow {
     pub uuid: Uuid,
     pub machines: BTreeSet<String>,
+    pub nodes: Vec<ResolvedNode>,
 }

binaries/daemon/src/lib.rs

@@ -19,6 +19,7 @@ use futures_concurrency::stream::Merge;
 use inter_daemon::InterDaemonConnection;
 use pending::PendingNodes;
 use shared_memory_server::ShmemConf;
+use std::env::temp_dir;
 use std::{
     borrow::Cow,
     collections::{BTreeMap, BTreeSet, HashMap},
@@ -28,14 +29,19 @@ use std::{
     time::Duration,
 };
 use tcp_utils::{tcp_receive, tcp_send};
+use tokio::fs::File;
+use tokio::io::AsyncReadExt;
 use tokio::net::TcpStream;
 use tokio::sync::mpsc::UnboundedSender;
+use tokio::sync::oneshot::Sender;
 use tokio::sync::{mpsc, oneshot};
 use tokio_stream::{wrappers::ReceiverStream, Stream, StreamExt};
+use tracing::error;
 use uuid::Uuid;
 
 mod coordinator;
 mod inter_daemon;
+mod log;
 mod node_communication;
 mod pending;
 mod spawn;
@@ -205,8 +211,7 @@ impl Daemon {
         while let Some(event) = events.next().await {
             match event {
                 Event::Coordinator(CoordinatorEvent { event, reply_tx }) => {
-                    let (reply, status) = self.handle_coordinator_event(event).await?;
-                    let _ = reply_tx.send(reply);
+                    let status = self.handle_coordinator_event(event, reply_tx).await?;
 
                     match status {
                         RunStatus::Continue => {}
@@ -257,8 +262,9 @@ impl Daemon {
     async fn handle_coordinator_event(
         &mut self,
         event: DaemonCoordinatorEvent,
-    ) -> eyre::Result<(Option<DaemonCoordinatorReply>, RunStatus)> {
-        let (reply, status) = match event {
+        reply_tx: Sender<Option<DaemonCoordinatorReply>>,
+    ) -> eyre::Result<RunStatus> {
+        let status = match event {
             DaemonCoordinatorEvent::Spawn(SpawnDataflowNodes {
                 dataflow_id,
                 working_dir,
@@ -290,7 +296,10 @@ impl Daemon {
                 }
                 let reply =
                     DaemonCoordinatorReply::SpawnResult(result.map_err(|err| format!("{err:?}")));
-                (Some(reply), RunStatus::Continue)
+                let _ = reply_tx.send(Some(reply)).map_err(|_| {
+                    error!("could not send `SpawnResult` reply from daemon to coordinator")
+                });
+                RunStatus::Continue
             }
             DaemonCoordinatorEvent::AllNodesReady {
                 dataflow_id,
@@ -313,7 +322,39 @@ impl Daemon {
                         );
                     }
                 }
-                (None, RunStatus::Continue)
+                let _ = reply_tx.send(None).map_err(|_| {
+                    error!("could not send `AllNodesReady` reply from daemon to coordinator")
+                });
+                RunStatus::Continue
+            }
+            DaemonCoordinatorEvent::Logs {
+                dataflow_id,
+                node_id,
+            } => {
+                tokio::spawn(async move {
+                    let logs = async {
+                        let log_dir = temp_dir();
+
+                        let mut file =
+                            File::open(log_dir.join(log::log_path(&dataflow_id, &node_id)))
+                                .await
+                                .wrap_err("Could not open log file")?;
+
+                        let mut contents = vec![];
+                        file.read_to_end(&mut contents)
+                            .await
+                            .wrap_err("Could not read content of log file")?;
+                        Result::<Vec<u8>, eyre::Report>::Ok(contents)
+                    }
+                    .await
+                    .map_err(|err| format!("{err:?}"));
+                    let _ = reply_tx
+                        .send(Some(DaemonCoordinatorReply::Logs(logs)))
+                        .map_err(|_| {
+                            error!("could not send logs reply from daemon to coordinator")
+                        });
+                });
+                RunStatus::Continue
             }
             DaemonCoordinatorEvent::ReloadDataflow {
                 dataflow_id,
@@ -323,7 +364,10 @@ impl Daemon {
                 let result = self.send_reload(dataflow_id, node_id, operator_id).await;
                 let reply =
                     DaemonCoordinatorReply::ReloadResult(result.map_err(|err| format!("{err:?}")));
-                (Some(reply), RunStatus::Continue)
+                let _ = reply_tx
+                    .send(Some(reply))
+                    .map_err(|_| error!("could not send reload reply from daemon to coordinator"));
+                RunStatus::Continue
             }
             DaemonCoordinatorEvent::StopDataflow { dataflow_id } => {
                 let stop = async {
@@ -337,19 +381,29 @@ impl Daemon {
                 let reply = DaemonCoordinatorReply::StopResult(
                     stop.await.map_err(|err| format!("{err:?}")),
                 );
-                (Some(reply), RunStatus::Continue)
+                let _ = reply_tx
+                    .send(Some(reply))
+                    .map_err(|_| error!("could not send stop reply from daemon to coordinator"));
+                RunStatus::Continue
             }
             DaemonCoordinatorEvent::Destroy => {
                 tracing::info!("received destroy command -> exiting");
                 let reply = DaemonCoordinatorReply::DestroyResult(Ok(()));
-                (Some(reply), RunStatus::Exit)
+                let _ = reply_tx
+                    .send(Some(reply))
+                    .map_err(|_| error!("could not send destroy reply from daemon to coordinator"));
+                RunStatus::Exit
+            }
+            DaemonCoordinatorEvent::Watchdog => {
+                let _ = reply_tx
+                    .send(Some(DaemonCoordinatorReply::WatchdogAck))
+                    .map_err(|_| {
+                        error!("could not send WatchdogAck reply from daemon to coordinator")
+                    });
+                RunStatus::Continue
             }
-            DaemonCoordinatorEvent::Watchdog => (
-                Some(DaemonCoordinatorReply::WatchdogAck),
-                RunStatus::Continue,
-            ),
         };
-        Ok((reply, status))
+        Ok(status)
     }
 
     async fn handle_inter_daemon_event(&mut self, event: InterDaemonEvent) -> eyre::Result<()> {
@@ -824,15 +878,24 @@ impl Daemon {
                     }
                     NodeExitStatus::IoError(err) => {
                         let err = eyre!(err).wrap_err(format!(
-                            "I/O error while waiting for node `{dataflow_id}/{node_id}`"
+                            "
+    I/O error while waiting for node `{dataflow_id}/{node_id}. 
+
+    Check logs using: dora logs {dataflow_id} {node_id}
+                            "
                         ));
                         tracing::error!("{err:?}");
                         Some(err)
                     }
                     NodeExitStatus::ExitCode(code) => {
-                        let err =
-                            eyre!("node {dataflow_id}/{node_id} finished with exit code {code}");
-                        tracing::warn!("{err}");
+                        let err = eyre!(
+                            "
+    {dataflow_id}/{node_id} failed with exit code {code}.
+
+    Check logs using: dora logs {dataflow_id} {node_id}
+                            "
+                        );
+                        tracing::error!("{err}");
                         Some(err)
                     }
                     NodeExitStatus::Signal(signal) => {
@@ -854,15 +917,24 @@ impl Daemon {
                             other => other.to_string().into(),
                         };
                         let err = eyre!(
-                            "node {dataflow_id}/{node_id} finished because of signal `{signal}`"
+                            "
+    {dataflow_id}/{node_id} failed with signal `{signal}`
+
+    Check logs using: dora logs {dataflow_id} {node_id}
+                            "
                         );
-                        tracing::warn!("{err}");
+                        tracing::error!("{err}");
                         Some(err)
                     }
                     NodeExitStatus::Unknown => {
-                        let err =
-                            eyre!("node {dataflow_id}/{node_id} finished with unknown exit code");
-                        tracing::warn!("{err}");
+                        let err = eyre!(
+                            "
+    {dataflow_id}/{node_id} failed with unknown exit code
+    
+    Check logs using: dora logs {dataflow_id} {node_id}
+                            "
+                        );
+                        tracing::error!("{err}");
                         Some(err)
                     }
                 };

binaries/daemon/src/log.rs

@@ -0,0 +1,8 @@
+use std::path::PathBuf;
+
+use dora_core::config::NodeId;
+use uuid::Uuid;
+
+pub fn log_path(dataflow_id: &Uuid, node_id: &NodeId) -> PathBuf {
+    PathBuf::from(format!("{dataflow_id}-{node_id}.txt"))
+}

binaries/daemon/src/main.rs

@@ -36,9 +36,6 @@ async fn main() -> eyre::Result<()> {
 }
 
 async fn run() -> eyre::Result<()> {
-    #[cfg(feature = "tracing")]
-    set_up_tracing("dora-daemon").wrap_err("failed to set up tracing subscriber")?;
-
     let Args {
         run_dataflow,
         machine_id,
@@ -50,6 +47,9 @@ async fn run() -> eyre::Result<()> {
         return tokio::task::block_in_place(dora_daemon::run_dora_runtime);
     }
 
+    #[cfg(feature = "tracing")]
+    set_up_tracing("dora-daemon").wrap_err("failed to set up tracing subscriber")?;
+
     let ctrl_c_events = {
         let (ctrl_c_tx, ctrl_c_rx) = mpsc::channel(1);
         let mut ctrlc_sent = false;

binaries/daemon/src/spawn.rs

@@ -1,5 +1,5 @@
 use crate::{
-    node_communication::spawn_listener_loop, node_inputs, runtime_node_inputs,
+    log, node_communication::spawn_listener_loop, node_inputs, runtime_node_inputs,
     runtime_node_outputs, DoraEvent, Event, NodeExitStatus,
 };
 use dora_core::{
@@ -9,8 +9,17 @@ use dora_core::{
 };
 use dora_download::download_file;
 use eyre::WrapErr;
-use std::{env::consts::EXE_EXTENSION, path::Path, process::Stdio};
-use tokio::sync::mpsc;
+use std::{
+    env::{consts::EXE_EXTENSION, temp_dir},
+    path::Path,
+    process::Stdio,
+};
+use tokio::{
+    fs::File,
+    io::{AsyncBufReadExt, AsyncWriteExt},
+    sync::{mpsc, oneshot},
+};
+use tracing::{debug, error};
 
 pub async fn spawn_node(
     dataflow_id: DataflowId,
@@ -88,13 +97,18 @@ pub async fn spawn_node(
                     command.env(key, value.to_string());
                 }
             }
-            command.spawn().wrap_err_with(move || {
-                format!(
-                    "failed to run `{}` with args `{}`",
-                    n.source,
-                    n.args.as_deref().unwrap_or_default()
-                )
-            })?
+            command
+                .stdin(Stdio::null())
+                .stdout(Stdio::piped())
+                .stderr(Stdio::piped())
+                .spawn()
+                .wrap_err_with(move || {
+                    format!(
+                        "failed to run `{}` with args `{}`",
+                        n.source,
+                        n.args.as_deref().unwrap_or_default()
+                    )
+                })?
         }
         dora_core::descriptor::CoreNodeKind::Runtime(n) => {
             let has_python_operator = n
@@ -125,7 +139,6 @@ pub async fn spawn_node(
                 eyre::bail!("Runtime can not mix Python Operator with other type of operator.");
             };
             command.current_dir(working_dir);
-            command.stdin(Stdio::null());
 
             let runtime_config = RuntimeConfig {
                 node: NodeConfig {
@@ -152,15 +165,58 @@ pub async fn spawn_node(
                 }
             }
 
-            command.spawn().wrap_err(format!(
-                "failed to run runtime {}/{}",
-                runtime_config.node.dataflow_id, runtime_config.node.node_id
-            ))?
+            command
+                .stdin(Stdio::null())
+                .stdout(Stdio::piped())
+                .stderr(Stdio::piped())
+                .spawn()
+                .wrap_err(format!(
+                    "failed to run runtime {}/{}",
+                    runtime_config.node.dataflow_id, runtime_config.node.node_id
+                ))?
         }
     };
 
+    let log_dir = temp_dir();
+
+    let (tx, mut rx) = mpsc::channel(10);
+    let mut file =
+        File::create(&log_dir.join(log::log_path(&dataflow_id, &node_id).with_extension("txt")))
+            .await
+            .expect("Failed to create log file");
+    let mut stdout_lines =
+        (tokio::io::BufReader::new(child.stdout.take().expect("failed to take stdout"))).lines();
+
+    let stdout_tx = tx.clone();
+
+    // Stdout listener stream
+    tokio::spawn(async move {
+        while let Ok(Some(line)) = stdout_lines.next_line().await {
+            let sent = stdout_tx.send(line.clone()).await;
+            if sent.is_err() {
+                println!("Could not log: {line}");
+            }
+        }
+    });
+
+    let mut stderr_lines =
+        (tokio::io::BufReader::new(child.stderr.take().expect("failed to take stderr"))).lines();
+
+    // Stderr listener stream
+    let stderr_tx = tx.clone();
+    tokio::spawn(async move {
+        while let Ok(Some(line)) = stderr_lines.next_line().await {
+            let sent = stderr_tx.send(line.clone()).await;
+            if sent.is_err() {
+                eprintln!("Could not log: {line}");
+            }
+        }
+    });
+
+    let (log_finish_tx, log_finish_rx) = oneshot::channel();
     tokio::spawn(async move {
         let exit_status = NodeExitStatus::from(child.wait().await);
+        let _ = log_finish_rx.await;
         let event = DoraEvent::SpawnedNodeResult {
             dataflow_id,
             node_id,
@@ -168,5 +224,28 @@ pub async fn spawn_node(
         };
         let _ = daemon_tx.send(event.into()).await;
     });
+
+    // Log to file stream.
+    tokio::spawn(async move {
+        while let Some(line) = rx.recv().await {
+            let _ = file
+                .write_all(line.as_bytes())
+                .await
+                .map_err(|err| error!("Could not log {line} to file due to {err}"));
+            let _ = file
+                .write(b"\n")
+                .await
+                .map_err(|err| error!("Could not add newline to log file due to {err}"));
+            debug!("{dataflow_id}/{} logged {line}", node.id.clone());
+            // Make sure that all data has been synced to disk.
+            let _ = file
+                .sync_all()
+                .await
+                .map_err(|err| error!("Could not sync logs to file due to {err}"));
+        }
+        let _ = log_finish_tx
+            .send(())
+            .map_err(|_| error!("Could not inform that log file thread finished"));
+    });
     Ok(())
 }

libraries/core/src/daemon_messages.rs

@@ -211,6 +211,10 @@ pub enum DaemonCoordinatorEvent {
         node_id: NodeId,
         operator_id: Option<OperatorId>,
     },
+    Logs {
+        dataflow_id: DataflowId,
+        node_id: NodeId,
+    },
     Destroy,
     Watchdog,
 }
@@ -237,6 +241,7 @@ pub enum DaemonCoordinatorReply {
     StopResult(Result<(), String>),
     DestroyResult(Result<(), String>),
     WatchdogAck,
+    Logs(Result<Vec<u8>, String>),
 }
 
 pub type DataflowId = Uuid;

libraries/core/src/topics.rs

@@ -42,6 +42,11 @@ pub enum ControlRequest {
     StopByName {
         name: String,
     },
+    Logs {
+        uuid: Option<Uuid>,
+        name: Option<String>,
+        node: String,
+    },
     Destroy,
     List,
     DaemonConnected,
@@ -59,6 +64,7 @@ pub enum ControlRequestReply {
     DestroyOk,
     DaemonConnected(bool),
     ConnectedMachines(BTreeSet<String>),
+    Logs(Vec<u8>),
 }
 
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]