Add more data to the UDF log stream (#38877)

I want to the logs page to have more data, let's add the following:
- Usage stats (to help customers debug usage in real time without axiom/datadog integrations set up)
- `execution_timestamp` - if the result is not cached, this may be useful in the future for rendering a request tree instead of a sequential list of logs
- `occ_info` - for showing insights in real time
- execution environment - useful to show for actions
- identity type (e.g. user, system, instance_admin, team_acting_user -- useful for showing who ran the function)

In the future, I'd also like to add a `caller_execution_id` so we can render the logs in a tree view

GitOrigin-RevId: 41cab817e6b7548e8e0247b0efc46495bdcc73b2

Author: atrakh

crates/local_backend/src/logs.rs

@@ -56,6 +56,11 @@ pub enum FunctionExecutionJson {
         error: Option<String>,
         request_id: String,
         execution_id: String,
+        usage_stats: JsonValue,
+        occ_info: Option<JsonValue>,
+        execution_timestamp: f64,
+        identity_type: String,
+        environment: String,
     },
     #[serde(rename_all = "camelCase")]
     Progress {
@@ -230,10 +235,44 @@ pub async fn stream_function_logs(
     }
 }
 
+fn usage_stats_to_json(
+    stats: &usage_tracking::AggregatedFunctionUsageStats,
+    action_memory_used_mb: Option<u64>,
+) -> JsonValue {
+    serde_json::json!({
+        "databaseReadBytes": stats.database_read_bytes,
+        "databaseWriteBytes": stats.database_write_bytes,
+        "databaseReadDocuments": stats.database_read_documents,
+        "storageReadBytes": stats.storage_read_bytes,
+        "storageWriteBytes": stats.storage_write_bytes,
+        "vectorIndexReadBytes": stats.vector_index_read_bytes,
+        "vectorIndexWriteBytes": stats.vector_index_write_bytes,
+        "actionMemoryUsedMb": action_memory_used_mb,
+    })
+}
+
 fn execution_to_json(
     execution: FunctionExecution,
     supports_structured_log_lines: bool,
 ) -> anyhow::Result<FunctionExecutionJson> {
+    let usage_stats_json =
+        usage_stats_to_json(&execution.usage_stats, execution.action_memory_used_mb);
+    let occ_info_json = execution
+        .occ_info
+        .as_ref()
+        .map(|occ| {
+            let log_occ_info = common::log_streaming::OccInfo {
+                table_name: occ.table_name.clone(),
+                document_id: occ.document_id.clone(),
+                write_source: occ.write_source.clone(),
+                retry_count: occ.retry_count,
+            };
+            serde_json::to_value(log_occ_info)
+        })
+        .transpose()?;
+    let identity_type = execution.identity.tag().value.to_string();
+    let environment = execution.environment.to_string();
+    let execution_timestamp = execution.execution_timestamp.as_secs_f64();
     let json = match execution.params {
         UdfParams::Function { error, identifier } => {
             let component_path = identifier.component.serialize();
@@ -252,6 +291,11 @@ fn execution_to_json(
                 error: error.map(|e| e.to_string()),
                 request_id: execution.context.request_id.to_string(),
                 execution_id: execution.context.execution_id.to_string(),
+                usage_stats: usage_stats_json,
+                occ_info: occ_info_json,
+                execution_timestamp,
+                identity_type,
+                environment,
             }
         },
         UdfParams::Http { result, identifier } => {
@@ -274,6 +318,11 @@ fn execution_to_json(
                 error: error.map(|e| e.to_string()),
                 request_id: execution.context.request_id.to_string(),
                 execution_id: execution.context.execution_id.to_string(),
+                usage_stats: usage_stats_json,
+                occ_info: occ_info_json,
+                execution_timestamp,
+                identity_type,
+                environment,
             }
         },
     };