Improve ssh remote error handling and logging (#15035)

Release Notes:

- N/A

crates/remote/Cargo.toml

@@ -27,6 +27,8 @@ log.workspace = true
 parking_lot.workspace = true
 prost.workspace = true
 rpc.workspace = true
+serde.workspace = true
+serde_json.workspace = true
 smol.workspace = true
 tempfile.workspace = true
 util.workspace = true

crates/remote/src/json_log.rs

@@ -0,0 +1,59 @@
+use log::{Level, Log, Record};
+use serde::{Deserialize, Serialize};
+
+#[derive(Deserialize, Debug, Serialize)]
+pub struct LogRecord<'a> {
+    pub level: usize,
+    pub module_path: Option<&'a str>,
+    pub file: Option<&'a str>,
+    pub line: Option<u32>,
+    pub message: String,
+}
+
+impl<'a> LogRecord<'a> {
+    pub fn new(record: &'a Record<'a>) -> Self {
+        Self {
+            level: serialize_level(record.level()),
+            module_path: record.module_path(),
+            file: record.file(),
+            line: record.line(),
+            message: record.args().to_string(),
+        }
+    }
+
+    pub fn log(&'a self, logger: &dyn Log) {
+        if let Some(level) = deserialize_level(self.level) {
+            logger.log(
+                &log::Record::builder()
+                    .module_path(self.module_path)
+                    .target("remote_server")
+                    .args(format_args!("{}", self.message))
+                    .file(self.file)
+                    .line(self.line)
+                    .level(level)
+                    .build(),
+            )
+        }
+    }
+}
+
+fn serialize_level(level: Level) -> usize {
+    match level {
+        Level::Error => 1,
+        Level::Warn => 2,
+        Level::Info => 3,
+        Level::Debug => 4,
+        Level::Trace => 5,
+    }
+}
+
+fn deserialize_level(level: usize) -> Option<Level> {
+    match level {
+        1 => Some(Level::Error),
+        2 => Some(Level::Warn),
+        3 => Some(Level::Info),
+        4 => Some(Level::Debug),
+        5 => Some(Level::Trace),
+        _ => None,
+    }
+}

crates/remote/src/remote.rs

@@ -1,3 +1,4 @@
+pub mod json_log;
 pub mod protocol;
 pub mod ssh_session;
 

crates/remote/src/ssh_session.rs

@@ -1,5 +1,8 @@
-use crate::protocol::{
-    message_len_from_buffer, read_message_with_len, write_message, MessageId, MESSAGE_LEN_SIZE,
+use crate::{
+    json_log::LogRecord,
+    protocol::{
+        message_len_from_buffer, read_message_with_len, write_message, MessageId, MESSAGE_LEN_SIZE,
+    },
 };
 use anyhow::{anyhow, Context as _, Result};
 use collections::HashMap;
@@ -117,9 +120,14 @@ impl SshSession {
         let (outgoing_tx, mut outgoing_rx) = mpsc::unbounded::<Envelope>();
         let (incoming_tx, incoming_rx) = mpsc::unbounded::<Envelope>();
 
+        run_cmd(client_state.ssh_command(&remote_binary_path).arg("version")).await?;
+
         let mut remote_server_child = client_state
-            .ssh_command(&remote_binary_path)
-            .arg("run")
+            .ssh_command(&format!(
+                "RUST_LOG={} {:?} run",
+                std::env::var("RUST_LOG").unwrap_or(String::new()),
+                remote_binary_path,
+            ))
             .spawn()
             .context("failed to spawn remote server")?;
         let mut child_stderr = remote_server_child.stderr.take().unwrap();
@@ -198,9 +206,13 @@ impl SshSession {
                                 let mut start_ix = 0;
                                 while let Some(ix) = stderr_buffer[start_ix..stderr_offset].iter().position(|b| b == &b'\n') {
                                     let line_ix = start_ix + ix;
-                                    let content = String::from_utf8_lossy(&stderr_buffer[start_ix..line_ix]);
+                                    let content = &stderr_buffer[start_ix..line_ix];
                                     start_ix = line_ix + 1;
-                                    eprintln!("(remote) {}", content);
+                                    if let Ok(record) = serde_json::from_slice::<LogRecord>(&content) {
+                                        record.log(log::logger())
+                                    } else {
+                                        eprintln!("(remote) {}", String::from_utf8_lossy(content));
+                                    }
                                 }
                                 stderr_buffer.drain(0..start_ix);
                                 stderr_offset -= start_ix;