[Tracing] Initial Honeycomb implementation (#23836)

This PR adds honeycomb tracing to mutations called from the sync worker. This will be expanded to include more things over time like queries + actions, but mutations were the easiest one to start with. Sampling is off by default, and we can turn this on in staging when it lands.

Current lifecycle of tracing a request:
- a collector is initialized on backend + funrun startup
- when a request comes in decide whether or not to trace the call at the beginning, controlled by a knob (head sampling)
-  enter the `run_mutation` loop with the actual span or no-op depending on if we sample this request
- use the `#[minitrace::trace]` macro to trace all relevant functions on this machine
- encode the current trace when we make a request to funrun
- funrun decodes the request and initializes a span with that root as its parent
- the parent is passed into isolate in the request
- isolate initializes its own root
- all this data is then sent to honeycomb and is aggregated there

GitOrigin-RevId: d220f0682b6defc30ee2ef77e16d4d9239ae4f71

Author: jordanhunt22

Cargo.lock

@@ -62,6 +62,7 @@ lru = "0.12.0"
 maplit = "1"
 mime = "0.3"
 mime2ext = "0.1.52"
+minitrace = { version = "0.6", features = [ "enable" ] }
 must-let = { git = "https://github.com/sujayakar/must-let", rev = "5b487d78db235e396e61dd03ce261ced0eafff9d" }
 num_cpus = "1.16.0"
 oauth2 = "4.4.2"
@@ -116,7 +117,7 @@ tokio-metrics-collector = { version = "0.2.0" }
 tokio-process-stream = { version = "0.4.0" }
 tokio-stream = { version = "^0.1.8", features = [ "io-util", "sync" ] }
 tokio-tungstenite = "0.20.0"
-tonic = { version = "0.10.0", features = [ "gzip" ] }
+tonic = { version = "0.10.2", features = [ "gzip" ] }
 tonic-build = "0.10.0"
 tonic-health = "0.10.0"
 tower = { version = "0.4", features = [ "limit", "timeout" ] }

Cargo.toml

@@ -39,6 +39,7 @@ maplit = { workspace = true }
 metrics = { path = "../metrics" }
 mime = { workspace = true }
 mime2ext = { workspace = true }
+minitrace = { workspace = true }
 model = { path = "../model" }
 node_executor = { path = "../../crates/node_executor" }
 num_cpus = { workspace = true }

crates/application/Cargo.toml

@@ -250,6 +250,7 @@ impl<RT: Runtime> FunctionRouter<RT> {
 }
 
 impl<RT: Runtime> FunctionRouter<RT> {
+    #[minitrace::trace]
     pub(crate) async fn execute_query_or_mutation(
         &self,
         tx: Transaction<RT>,
@@ -298,6 +299,7 @@ impl<RT: Runtime> FunctionRouter<RT> {
 
     // Execute using the function runner. Can be used for v8 udfs other than http
     // actions.
+    #[minitrace::trace]
     async fn function_runner_execute(
         &self,
         mut tx: Transaction<RT>,
@@ -644,6 +646,7 @@ impl<RT: Runtime> ApplicationFunctionRunner<RT> {
     }
 
     /// Runs a mutations and retries on OCC errors.
+    #[minitrace::trace]
     pub async fn retry_mutation(
         &self,
         request_id: RequestId,
@@ -678,6 +681,7 @@ impl<RT: Runtime> ApplicationFunctionRunner<RT> {
     }
 
     /// Runs a mutations and retries on OCC errors.
+    #[minitrace::trace]
     async fn _retry_mutation(
         &self,
         request_id: RequestId,
@@ -898,6 +902,7 @@ impl<RT: Runtime> ApplicationFunctionRunner<RT> {
     }
 
     /// Runs the mutation once without any logging.
+    #[minitrace::trace]
     async fn run_mutation_inner(
         &self,
         mut tx: Transaction<RT>,
@@ -1566,6 +1571,7 @@ impl<RT: Runtime> ApplicationFunctionRunner<RT> {
         Ok(result)
     }
 
+    #[minitrace::trace]
     async fn check_mutation_status(
         &self,
         tx: &mut Transaction<RT>,
@@ -1627,6 +1633,7 @@ impl<RT: Runtime> ApplicationFunctionRunner<RT> {
         Ok(())
     }
 
+    #[minitrace::trace]
     async fn write_mutation_status(
         &self,
         tx: &mut Transaction<RT>,

crates/application/src/application_function_runner/mod.rs

@@ -691,6 +691,11 @@ impl<RT: Runtime> Application<RT> {
         self.database.now_ts_for_reads()
     }
 
+    pub fn instance_name(&self) -> String {
+        self.instance_name.clone()
+    }
+
+    #[minitrace::trace]
     pub async fn begin(&self, identity: Identity) -> anyhow::Result<Transaction<RT>> {
         self.database.begin(identity).await
     }
@@ -700,6 +705,7 @@ impl<RT: Runtime> Application<RT> {
         self.commit(transaction, "test").await
     }
 
+    #[minitrace::trace]
     pub async fn commit(
         &self,
         transaction: Transaction<RT>,
@@ -908,6 +914,7 @@ impl<RT: Runtime> Application<RT> {
         }
     }
 
+    #[minitrace::trace]
     pub async fn mutation_udf(
         &self,
         request_id: RequestId,
@@ -1677,6 +1684,7 @@ impl<RT: Runtime> Application<RT> {
         Ok(auth_config)
     }
 
+    #[minitrace::trace]
     pub async fn apply_config_with_retries(
         &self,
         identity: Identity,

crates/application/src/lib.rs

@@ -37,6 +37,7 @@ hyper = { workspace = true }
 itertools = { workspace = true }
 maplit = { workspace = true }
 metrics = { path = "../metrics" }
+minitrace = { workspace = true }
 openidconnect = { workspace = true }
 packed_value = { path = "../packed_value" }
 parking_lot = { workspace = true }

crates/common/Cargo.toml

@@ -77,7 +77,7 @@ use serde::{
     Serialize,
 };
 use tokio::net::TcpSocket;
-use tonic::transport::NamedService;
+use tonic::server::NamedService;
 use tonic_health::server::health_reporter;
 use tower::{
     timeout::TimeoutLayer,

crates/common/src/http/mod.rs

@@ -989,3 +989,8 @@ pub static HTTP_SERVER_TIMEOUT_DURATION: LazyLock<Duration> =
 // Schema and code bundle pushes must be less than this.
 pub static MAX_PUSH_BYTES: LazyLock<usize> =
     LazyLock::new(|| env_config("MAX_PUSH_BYTES", 100_000_000));
+
+/// Percentage of request traces that should sampled
+pub static REQUEST_TRACE_SAMPLE_PERCENT: LazyLock<f64> = LazyLock::new(|| {
+    env_config("REQUEST_TRACE_SAMPLE_PERCENT", 0.0) // 0% by default
+});

crates/common/src/knobs.rs

@@ -49,6 +49,7 @@ pub mod knobs;
 pub mod log_lines;
 pub mod log_streaming;
 pub mod metrics;
+pub mod minitrace_helpers;
 pub mod numeric;
 pub mod paths;
 pub mod pause;

crates/common/src/lib.rs

@@ -0,0 +1,52 @@
+use std::collections::BTreeMap;
+
+use minitrace::{
+    collector::SpanContext,
+    Span,
+};
+use rand::Rng;
+
+use crate::{
+    knobs::REQUEST_TRACE_SAMPLE_PERCENT,
+    runtime::Runtime,
+};
+
+#[derive(Clone)]
+pub struct EncodedSpan(pub Option<String>);
+
+impl EncodedSpan {
+    pub fn empty() -> Self {
+        Self(None)
+    }
+
+    /// Encodes the passed in `SpanContext`
+    pub fn from_parent(parent: Option<SpanContext>) -> Self {
+        Self(parent.map(|ctx| ctx.encode_w3c_traceparent()))
+    }
+}
+
+/// Given an instance name returns a span with the sample percentage specified
+/// in `knobs.rs`
+pub fn get_sampled_span<RT: Runtime>(
+    request_name: String,
+    rt: RT,
+    properties: BTreeMap<String, String>,
+) -> Span {
+    let should_sample = rt
+        .clone()
+        .with_rng(|rng| rng.gen_bool(*REQUEST_TRACE_SAMPLE_PERCENT));
+    match should_sample {
+        true => Span::root(request_name, SpanContext::random()).with_properties(|| properties),
+        false => Span::noop(),
+    }
+}
+
+/// Creates a root span from an encoded parent trace
+pub fn initialize_root_from_parent(span_name: &'static str, encoded_parent: EncodedSpan) -> Span {
+    if let Some(p) = encoded_parent.0 {
+        if let Some(ctx) = SpanContext::decode_w3c_traceparent(p.as_str()) {
+            return Span::root(span_name, ctx);
+        }
+    }
+    Span::noop()
+}