Add stdout logging (#720)

This makes `dora run` output each node stdout as an info stdout.

This makes it easier to do realtime debugging. 

It also avoids having to have to add `flush=True` within python to see
logs.

binaries/daemon/src/lib.rs

@@ -1258,8 +1258,8 @@ impl Daemon {
                                 NodeErrorCause::Cascading { caused_by_node }
                             }
                             None if grace_duration_kill => NodeErrorCause::GraceDuration,
-                            None => NodeErrorCause::Other {
-                                stderr: dataflow
+                            None => {
+                                let cause = dataflow
                                     .and_then(|d| d.node_stderr_most_recent.get(&node_id))
                                     .map(|queue| {
                                         let mut s = if queue.is_full() {
@@ -1272,8 +1272,14 @@ impl Daemon {
                                         }
                                         s
                                     })
-                                    .unwrap_or_default(),
-                            },
+                                    .unwrap_or_default();
+
+                                tracing::error!("node {dataflow_id}/{node_id} failed with:");
+                                for line in cause.lines() {
+                                    tracing::error!("    {}", line);
+                                }
+                                NodeErrorCause::Other { stderr: cause }
+                            }
                         };
                         Err(NodeError {
                             timestamp: self.clock.new_timestamp(),

binaries/daemon/src/spawn.rs

@@ -290,7 +290,7 @@ pub async fn spawn_node(
         node_config,
     };
     let stdout_tx = tx.clone();
-
+    let node_id = node.id.clone();
     // Stdout listener stream
     tokio::spawn(async move {
         let mut buffer = String::new();
@@ -337,6 +337,11 @@ pub async fn spawn_node(
 
             // send the buffered lines
             let lines = std::mem::take(&mut buffer);
+            if std::env::var("DORA_QUIET").is_err() {
+                if lines.len() > 1 {
+                    tracing::info!("log_{}: {}", node_id, &lines[..lines.len() - 1]);
+                }
+            }
             let sent = stdout_tx.send(lines.clone()).await;
             if sent.is_err() {
                 println!("Could not log: {lines}");