Make the backend send structured log lines (#24132)

This is to be landed once the dashboard and CLI have been updated to handle structured log lines.

GitOrigin-RevId: d4e5e0a268ed593712f4dee9ce5ca6f6cf588d1f

Author: sshader

crates/common/src/log_lines.rs

@@ -189,6 +189,41 @@ impl LogLine {
         }
     }
 
+    pub fn to_json(
+        self,
+        allow_structured: bool,
+        include_system_metadata: bool,
+    ) -> anyhow::Result<JsonValue> {
+        if !allow_structured {
+            Ok(JsonValue::String(self.to_pretty_string()))
+        } else {
+            match self {
+                LogLine::Unstructured(m) => Ok(JsonValue::String(m)),
+                LogLine::Structured {
+                    messages,
+                    level,
+                    is_truncated,
+                    timestamp,
+                    system_metadata,
+                } => {
+                    let system_metadata = if include_system_metadata {
+                        system_metadata
+                    } else {
+                        None
+                    };
+                    let log_line_json = LogLineJson {
+                        messages: messages.into(),
+                        is_truncated,
+                        timestamp: timestamp.as_ms_since_epoch()?,
+                        level: level.to_string(),
+                        system_metadata: system_metadata.map(SystemLogMetadataJson::from),
+                    };
+                    Ok(serde_json::to_value(log_line_json)?)
+                },
+            }
+        }
+    }
+
     pub fn new_developer_log_line(
         level: LogLevel,
         messages: Vec<String>,
@@ -208,7 +243,7 @@ impl LogLine {
         let mut total_length = 0;
         let mut truncated_messages: Vec<String> = vec![];
         for message in messages {
-            let remaining_space = MAX_LOG_LINE_LENGTH - TRUNCATED_LINE_SUFFIX.len() - total_length;
+            let remaining_space = MAX_LOG_LINE_LENGTH - total_length;
             if message.len() <= remaining_space {
                 total_length += message.len() + 1;
                 truncated_messages.push(message);
@@ -396,25 +431,7 @@ impl TryFrom<LogLine> for JsonValue {
     type Error = anyhow::Error;
 
     fn try_from(value: LogLine) -> Result<Self, Self::Error> {
-        match value {
-            LogLine::Unstructured(m) => Ok(JsonValue::String(m)),
-            LogLine::Structured {
-                messages,
-                level,
-                is_truncated,
-                timestamp,
-                system_metadata,
-            } => {
-                let log_line_json = LogLineJson {
-                    messages: messages.into(),
-                    is_truncated,
-                    timestamp: timestamp.as_ms_since_epoch()?,
-                    level: level.to_string(),
-                    system_metadata: system_metadata.map(SystemLogMetadataJson::from),
-                };
-                Ok(serde_json::to_value(log_line_json)?)
-            },
-        }
+        value.to_json(true, true)
     }
 }
 

crates/common/src/version.rs

@@ -88,6 +88,9 @@ pub enum ClientType {
     AirbyteExport,
     FivetranImport,
     FivetranExport,
+    // For HTTP requests from the dashboard. Requests from the dashboard via a
+    // Convex client will have an `NPM` version
+    Dashboard,
     Unrecognized(String),
 }
 
@@ -104,6 +107,7 @@ impl FromStr for ClientType {
             "streaming-import" => Self::StreamingImport,
             "airbyte-export" => Self::AirbyteExport,
             "fivetran-export" => Self::FivetranExport,
+            "dashboard" => Self::Dashboard,
             unrecognized => Self::Unrecognized(unrecognized.to_string()),
         };
         Ok(client_type)
@@ -122,6 +126,7 @@ impl Display for ClientType {
             Self::AirbyteExport => write!(f, "airbyte-export"),
             Self::FivetranImport => write!(f, "fivetran-import"),
             Self::FivetranExport => write!(f, "fivetran-export"),
+            Self::Dashboard => write!(f, "dashboard"),
             Self::Unrecognized(other_client) => write!(f, "{other_client}"),
         }
     }
@@ -138,6 +143,7 @@ impl ClientType {
             | Self::AirbyteExport
             | Self::FivetranImport
             | Self::FivetranExport
+            | Self::Dashboard
             | Self::Unrecognized(_) => None,
         }
     }
@@ -152,6 +158,7 @@ impl ClientType {
             | Self::AirbyteExport
             | Self::FivetranImport
             | Self::FivetranExport
+            | Self::Dashboard
             | Self::Unrecognized(_) => None,
         }
     }
@@ -172,6 +179,7 @@ impl ClientType {
             | Self::AirbyteExport
             | Self::FivetranImport
             | Self::FivetranExport
+            | Self::Dashboard
             | Self::Unrecognized(_) => "",
         }
     }
@@ -274,6 +282,7 @@ impl ClientVersion {
             | ClientType::AirbyteExport
             | ClientType::FivetranImport
             | ClientType::FivetranExport
+            | ClientType::Dashboard
             | ClientType::Unrecognized(_) => true,
         }
     }

crates/isolate/src/tests/adversarial.rs

@@ -179,18 +179,18 @@ async fn test_console_line_too_long(rt: TestRuntime) -> anyhow::Result<()> {
         TRUNCATED_LINE_SUFFIX,
     );
     // The limit is 32768 MAX_LOG_LINE_LENGTH, but we don't count the [INFO]
-    // prefix, so just check that we're close
+    // prefix or the truncated line suffix, so just check that we're close
     assert!(log_lines[0].clone().to_pretty_string().len() > 32700);
-    assert!(log_lines[0].clone().to_pretty_string().len() < 32800);
+    assert!(log_lines[0].clone().to_pretty_string().len() < 32900);
 
     assert_contains(
         &log_lines[1].clone().to_pretty_string(),
         TRUNCATED_LINE_SUFFIX,
     );
     // The limit is 32768 MAX_LOG_LINE_LENGTH, but we don't count the [INFO]
-    // prefix, so just check that we're close
+    // prefix or the truncated line suffix, so just check that we're close
     assert!(log_lines[1].clone().to_pretty_string().len() > 32700);
-    assert!(log_lines[1].clone().to_pretty_string().len() < 32800);
+    assert!(log_lines[1].clone().to_pretty_string().len() < 32900);
     Ok(())
 }
 

crates/local_backend/src/logs.rs

@@ -16,8 +16,10 @@ use common::{
             Json,
             Query,
         },
+        ExtractClientVersion,
         HttpResponseError,
     },
+    version::ClientType,
     RequestId,
 };
 use errors::ErrorMetadata;
@@ -45,7 +47,7 @@ pub enum FunctionExecutionJson {
     Completion {
         udf_type: String,
         identifier: String,
-        log_lines: Vec<String>,
+        log_lines: Vec<JsonValue>,
         timestamp: f64,
         cached_result: bool,
         execution_time: f64,
@@ -59,7 +61,7 @@ pub enum FunctionExecutionJson {
         udf_type: String,
         identifier: String,
         timestamp: f64,
-        log_lines: Vec<String>,
+        log_lines: Vec<JsonValue>,
         request_id: String,
         execution_id: String,
     },
@@ -86,8 +88,8 @@ pub async fn stream_udf_execution(
             let (log_entries, new_cursor) = entries_future_r?;
             let entries = log_entries
                 .into_iter()
-                .map(execution_to_json)
-                .collect::<anyhow::Result<_>>()?;
+                .map(|e| execution_to_json(e, false))
+                .try_collect()?;
             let response = StreamUdfExecutionResponse {
                 entries,
                 new_cursor,
@@ -131,6 +133,7 @@ pub struct StreamFunctionLogs {
 pub async fn stream_function_logs(
     State(st): State<LocalAppState>,
     ExtractIdentity(identity): ExtractIdentity,
+    ExtractClientVersion(client_version): ExtractClientVersion,
     Query(query_args): Query<StreamFunctionLogs>,
 ) -> Result<impl IntoResponse, HttpResponseError> {
     let entries_future = st
@@ -144,6 +147,22 @@ pub async fn stream_function_logs(
         )),
         _ => None,
     };
+    // As of writing, this endpoint is only used by the CLI and dashboard, both of
+    // which support either unstructured `string` log lines or structured log
+    // lines.
+    let supports_structured_log_lines = match client_version.client() {
+        ClientType::CLI => true,
+        ClientType::Dashboard => true,
+        ClientType::NPM
+        | ClientType::Actions
+        | ClientType::Python
+        | ClientType::Rust
+        | ClientType::StreamingImport
+        | ClientType::AirbyteExport
+        | ClientType::FivetranImport
+        | ClientType::FivetranExport
+        | ClientType::Unrecognized(_) => false,
+    };
     futures::select_biased! {
         entries_future_r = entries_future.fuse() => {
             let (log_entries, new_cursor) = entries_future_r?;
@@ -166,15 +185,17 @@ pub async fn stream_function_logs(
                 .map(|e| {
                     let json = match e {
                         FunctionExecutionPart::Completion(c) => {
-                            execution_to_json(c)?
+                            execution_to_json(c, supports_structured_log_lines)?
                         },
                         FunctionExecutionPart::Progress(c) => {
                             FunctionExecutionJson::Progress {
                                 udf_type: c.event_source.udf_type.to_string(),
                                 identifier: c.event_source.path,
                                 timestamp: c.function_start_timestamp.as_secs_f64(),
                                 log_lines: c.log_lines
-                                    .into_iter().map(|l| l.to_pretty_string()).collect(),
+                                    .into_iter()
+                                    .map(|l| l.to_json(supports_structured_log_lines, false))
+                                    .try_collect()?,
                                 request_id: c.event_source.context.request_id.to_string(),
                                 execution_id: c.event_source.context.execution_id.to_string()
                             }
@@ -204,7 +225,10 @@ pub async fn stream_function_logs(
     }
 }
 
-fn execution_to_json(execution: FunctionExecution) -> anyhow::Result<FunctionExecutionJson> {
+fn execution_to_json(
+    execution: FunctionExecution,
+    supports_structured_log_lines: bool,
+) -> anyhow::Result<FunctionExecutionJson> {
     let json = match execution.params {
         UdfParams::Function { error, identifier } => {
             let identifier: String = identifier.strip().into();
@@ -214,8 +238,8 @@ fn execution_to_json(execution: FunctionExecution) -> anyhow::Result<FunctionExe
                 log_lines: execution
                     .log_lines
                     .into_iter()
-                    .map(|l| l.to_pretty_string())
-                    .collect(),
+                    .map(|l| l.to_json(supports_structured_log_lines, false))
+                    .try_collect()?,
                 timestamp: execution.unix_timestamp.as_secs_f64(),
                 cached_result: execution.cached_result,
                 execution_time: execution.execution_time,
@@ -237,8 +261,8 @@ fn execution_to_json(execution: FunctionExecution) -> anyhow::Result<FunctionExe
                 log_lines: execution
                     .log_lines
                     .into_iter()
-                    .map(|l| l.to_pretty_string())
-                    .collect(),
+                    .map(|l| l.to_json(supports_structured_log_lines, false))
+                    .try_collect()?,
                 timestamp: execution.unix_timestamp.as_secs_f64(),
                 cached_result: execution.cached_result,
                 execution_time: execution.execution_time,

crates/local_backend/src/subs/mod.rs

@@ -321,6 +321,7 @@ fn new_sync_worker_config(client_version: ClientVersion) -> anyhow::Result<SyncW
         | ClientType::AirbyteExport
         | ClientType::FivetranImport
         | ClientType::FivetranExport
+        | ClientType::Dashboard
         | ClientType::Actions => Err(anyhow::anyhow!(
             "No websocket support for client: {}",
             client_version.client()