[Tracing] Add support for V8 actions (#24441)

Adds support for `V8` actions + relevant `syscalls`

This doesn't include tracing for other `async_ops` like fetches or storage gets

GitOrigin-RevId: 508ad4db88053c90d9f995eb5aa002828cad9630

Author: jordanhunt22

crates/application/src/application_function_runner/mod.rs

@@ -270,6 +270,7 @@ impl<RT: Runtime> FunctionRouter<RT> {
         Ok((tx, outcome))
     }
 
+    #[minitrace::trace]
     pub(crate) async fn execute_action(
         &self,
         tx: Transaction<RT>,
@@ -955,6 +956,7 @@ impl<RT: Runtime> ApplicationFunctionRunner<RT> {
         Ok((tx, mutation_outcome))
     }
 
+    #[minitrace::trace]
     pub async fn run_action(
         &self,
         request_id: RequestId,
@@ -1014,6 +1016,7 @@ impl<RT: Runtime> ApplicationFunctionRunner<RT> {
 
     /// Runs the actions without logging to the UDF log. It is the caller
     /// responsibility to log to the UDF log.
+    #[minitrace::trace]
     pub async fn run_action_no_udf_log(
         &self,
         name: CanonicalizedUdfPath,
@@ -1055,6 +1058,7 @@ impl<RT: Runtime> ApplicationFunctionRunner<RT> {
     }
 
     /// Runs the action without any logging.
+    #[minitrace::trace]
     async fn run_action_inner(
         &self,
         name: CanonicalizedUdfPath,
@@ -1665,6 +1669,7 @@ impl<RT: Runtime> ApplicationFunctionRunner<RT> {
 
 #[async_trait]
 impl<RT: Runtime> ActionCallbacks for ApplicationFunctionRunner<RT> {
+    #[minitrace::trace]
     async fn execute_query(
         &self,
         identity: Identity,
@@ -1693,6 +1698,7 @@ impl<RT: Runtime> ActionCallbacks for ApplicationFunctionRunner<RT> {
         Ok(FunctionResult { result })
     }
 
+    #[minitrace::trace]
     async fn execute_mutation(
         &self,
         identity: Identity,
@@ -1723,6 +1729,7 @@ impl<RT: Runtime> ActionCallbacks for ApplicationFunctionRunner<RT> {
         Ok(FunctionResult { result })
     }
 
+    #[minitrace::trace]
     async fn execute_action(
         &self,
         identity: Identity,

crates/application/src/lib.rs

@@ -964,6 +964,7 @@ impl<RT: Runtime> Application<RT> {
         }
     }
 
+    #[minitrace::trace]
     pub async fn action_udf(
         &self,
         request_id: RequestId,

crates/isolate/src/environment/action/async_syscall.rs

@@ -36,6 +36,7 @@ use crate::{
 };
 
 impl<RT: Runtime> TaskExecutor<RT> {
+    #[minitrace::trace]
     pub async fn run_async_syscall(&self, name: String, args: JsonValue) -> anyhow::Result<String> {
         let start = self.rt.monotonic_now();
         let timer = async_syscall_timer(&name);

crates/isolate/src/environment/action/mod.rs

@@ -35,6 +35,7 @@ use common::{
         LogLine,
         SystemLogMetadata,
     },
+    minitrace_helpers::EncodedSpan,
     runtime::{
         Runtime,
         SpawnHandle,
@@ -61,6 +62,7 @@ use humansize::{
 };
 use itertools::Itertools;
 use keybroker::Identity;
+use minitrace::collector::SpanContext;
 use model::{
     environment_variables::types::{
         EnvVarName,
@@ -453,10 +455,12 @@ impl<RT: Runtime> ActionEnvironment<RT> {
             .unbounded_send(TaskRequest {
                 task_id,
                 variant: TaskRequestEnum::AsyncOp(AsyncOpRequest::SendStream { stream, stream_id }),
+                parent_trace: EncodedSpan::from_parent(SpanContext::current_local_parent()),
             })
             .expect("TaskExecutor went away?");
     }
 
+    #[minitrace::trace]
     pub async fn run_action(
         mut self,
         isolate: &mut Isolate<RT>,
@@ -516,6 +520,7 @@ impl<RT: Runtime> ActionEnvironment<RT> {
         Ok(outcome)
     }
 
+    #[minitrace::trace]
     async fn run_action_inner(
         isolate: &mut RequestScope<'_, '_, RT, Self>,
         request_params: ActionRequestParams,
@@ -712,6 +717,7 @@ impl<RT: Runtime> ActionEnvironment<RT> {
         }
     }
 
+    #[minitrace::trace]
     async fn run_inner<'a, 'b: 'a, T, Fut>(
         scope: &mut ExecutionScope<'a, 'b, RT, Self>,
         handle: IsolateHandle,
@@ -1010,6 +1016,7 @@ impl<RT: Runtime> ActionEnvironment<RT> {
             .unbounded_send(TaskRequest {
                 task_id,
                 variant: request,
+                parent_trace: EncodedSpan::from_parent(SpanContext::current_local_parent()),
             })
             .expect("TaskExecutor went away?");
         Ok(())

crates/isolate/src/environment/action/task.rs

@@ -1,9 +1,12 @@
 //! Our action runtime runs "tasks" asynchronously, which either be
 //! async syscalls or async ops.
 
-use common::runtime::{
-    Runtime,
-    UnixTimestamp,
+use common::{
+    minitrace_helpers::EncodedSpan,
+    runtime::{
+        Runtime,
+        UnixTimestamp,
+    },
 };
 use deno_core::{
     serde_v8,
@@ -30,6 +33,7 @@ use crate::{
 pub struct TaskRequest {
     pub task_id: TaskId,
     pub variant: TaskRequestEnum,
+    pub parent_trace: EncodedSpan,
 }
 
 #[derive(Copy, Clone, Eq, PartialEq, Ord, PartialOrd, Debug, derive_more::Display)]

crates/isolate/src/environment/action/task_executor.rs

@@ -8,6 +8,7 @@ use common::{
     execution_context::ExecutionContext,
     http::fetch::FetchClient,
     knobs::MAX_CONCURRENT_ACTION_OPS,
+    minitrace_helpers::initialize_root_from_parent,
     runtime::{
         Runtime,
         UnixTimestamp,
@@ -27,6 +28,7 @@ use keybroker::{
     Identity,
     KeyBroker,
 };
+use minitrace::future::FutureExt;
 use parking_lot::Mutex;
 use serde_json::Value as JsonValue;
 use usage_tracking::FunctionUsageTracker;
@@ -95,8 +97,9 @@ impl<RT: Runtime> TaskExecutor<RT> {
                 },
                 task_request = pending_tasks.next() => {
                     if let Some(task_request) = task_request {
+                        let root = initialize_root_from_parent("TaskExecutor::execute_task", task_request.parent_trace.clone());
                         self.task_order.push_running_task(&task_request);
-                        running_tasks.push(self.clone().run_async_task(task_request));
+                        running_tasks.push(self.clone().run_async_task(task_request).in_span(root));
                     } else {
                         requests_closed = true;
                     }
@@ -105,6 +108,7 @@ impl<RT: Runtime> TaskExecutor<RT> {
         }
     }
 
+    #[minitrace::trace]
     async fn run_async_task(self, task_request: TaskRequest) -> TaskId {
         let task_id = task_request.task_id;
         let variant = match task_request.variant {

crates/pb/protos/backend.proto

@@ -73,6 +73,9 @@ message ExecuteQueryRequest {
     common.PathAndArgs path_and_args = 3;
     optional bool block_logging = 5;
     optional common.ExecutionContext execution_context = 6;
+
+    // Encoded parent trace
+    optional string encoded_parent_trace = 7;
 }
 
 message ExecuteQueryResponse {
@@ -88,6 +91,9 @@ message ExecuteMutationRequest {
     common.PathAndArgs path_and_args = 3;
     optional bool block_logging = 5;
     optional common.ExecutionContext execution_context = 6;
+
+    // Encoded parent trace
+    optional string encoded_parent_trace = 7;
 }
 
 message ExecuteMutationResponse {
@@ -103,6 +109,9 @@ message ExecuteActionRequest {
     common.PathAndArgs path_and_args = 3;
     optional bool block_logging = 5;
     optional common.ExecutionContext execution_context = 6;
+
+    // Encoded parent trace
+    optional string encoded_parent_trace = 7;
 }
 
 message ExecuteActionResponse {

crates/sync/src/worker.rs

@@ -499,6 +499,16 @@ impl<RT: Runtime> SyncWorker<RT> {
                     Some(id) => RequestId::new_for_ws_session(id, request_id),
                     None => RequestId::new(),
                 };
+                let root = self.rt.with_rng(|rng| {
+                    get_sampled_span(
+                        "sync-worker/action",
+                        rng,
+                        btreemap! {
+                           "udf_type".into() => UdfType::Action.to_lowercase_string().into(),
+                           "udf_path".into() => udf_path.clone().into(),
+                        },
+                    )
+                });
                 let future = async move {
                     let result = application
                         .action_udf(
@@ -509,6 +519,7 @@ impl<RT: Runtime> SyncWorker<RT> {
                             AllowedVisibility::PublicOnly,
                             FunctionCaller::SyncWorker(client_version),
                         )
+                        .in_span(root)
                         .await?;
                     let response = match result {
                         Ok(udf_return) => ServerMessage::ActionResponse {