Set up tracing, OTel, and metrics (#16)

* initial rust initialization

* added pre-commit

* set up sqlx and rust

* added license files and initial migration;

* added initial impl of client

* tests compile

* fixed bugs with sqlx types

* added documentation

* added tests

* updated tasks

* added a test that mocks the example in README

* removed todos

* added convenience methods for uuid, rand, now

* added handling for spawning and joining subtasks from workflows

* added missing sql and test files

* merged migrations

* cleaned up bad code

* cleaned up json handling

* made process exit optional on too-long tasks

* fixed semaphore ordering

* fixed issues with clock skew

* improved handling of leases

* added comments on sql schema

* enforced that claim timeouts must be set

* cleaned up and documented sql

* added support for transactions that enqueue tasks

* added a bunch of tests

* documented and tested event semantics

* added benchmarks

* initial implementation of telemetry

* removed exporter setup from crate

* telemetry tests pass

* removed extra license file

* inject otel context as a string for key durable::otel_context

* protect durable:: headers for internal use

* addressed PR comment

Author: virajmehta

Cargo.lock

@@ -25,8 +25,27 @@ hostname = "0.4"
 rand = "0.9"
 futures = "0.3.31"
 
+# Optional telemetry dependencies
+tracing-opentelemetry = { version = "0.28", optional = true }
+opentelemetry = { version = "0.27", optional = true }
+opentelemetry_sdk = { version = "0.27", optional = true }
+metrics = { version = "0.24", optional = true }
+
+[features]
+default = []
+telemetry = [
+    "dep:tracing-opentelemetry",
+    "dep:opentelemetry",
+    "dep:opentelemetry_sdk",
+    "dep:metrics",
+]
+
 [dev-dependencies]
 criterion = { version = "0.5", features = ["async_tokio", "html_reports"] }
+tracing-fluent-assertions = "0.3"
+metrics-util = { version = "0.18", features = ["debugging"] }
+tracing-subscriber = { version = "0.3", features = ["registry"] }
+ordered-float = "4"
 
 [[bench]]
 name = "throughput"

Cargo.toml

@@ -48,6 +48,21 @@ impl CancellationPolicyDb {
 
 use crate::worker::Worker;
 
+/// Validates that user-provided headers don't use reserved prefixes.
+fn validate_headers(headers: &Option<HashMap<String, JsonValue>>) -> anyhow::Result<()> {
+    if let Some(headers) = headers {
+        for key in headers.keys() {
+            if key.starts_with("durable::") {
+                anyhow::bail!(
+                    "Header key '{}' uses reserved prefix 'durable::'. User headers cannot start with 'durable::'.",
+                    key
+                );
+            }
+        }
+    }
+    Ok(())
+}
+
 /// The main client for interacting with durable workflows.
 ///
 /// Use this client to:
@@ -355,6 +370,14 @@ where
     /// Spawn a task by name using a custom executor.
     ///
     /// The task must be registered before spawning.
+    #[cfg_attr(
+        feature = "telemetry",
+        tracing::instrument(
+            name = "durable.client.spawn",
+            skip(self, executor, params, options),
+            fields(queue, task_name = %task_name)
+        )
+    )]
     pub async fn spawn_by_name_with<'e, E>(
         &self,
         executor: E,
@@ -380,16 +403,30 @@ where
     }
 
     /// Internal spawn implementation without registry validation.
+    #[allow(unused_mut)] // mut is needed when telemetry feature is enabled
     async fn spawn_by_name_internal<'e, E>(
         &self,
         executor: E,
         task_name: &str,
         params: JsonValue,
-        options: SpawnOptions,
+        mut options: SpawnOptions,
     ) -> anyhow::Result<SpawnResult>
     where
         E: Executor<'e, Database = Postgres>,
     {
+        // Validate user headers don't use reserved prefix
+        validate_headers(&options.headers)?;
+
+        // Inject trace context into headers for distributed tracing
+        #[cfg(feature = "telemetry")]
+        {
+            let headers = options.headers.get_or_insert_with(HashMap::new);
+            crate::telemetry::inject_trace_context(headers);
+        }
+
+        #[cfg(feature = "telemetry")]
+        tracing::Span::current().record("queue", &self.queue_name);
+
         let max_attempts = options.max_attempts.unwrap_or(self.default_max_attempts);
 
         let db_options = Self::serialize_spawn_options(&options, max_attempts)?;
@@ -405,6 +442,9 @@ where
             .fetch_one(executor)
             .await?;
 
+        #[cfg(feature = "telemetry")]
+        crate::telemetry::record_task_spawned(&self.queue_name, task_name);
+
         Ok(SpawnResult {
             task_id: row.task_id,
             run_id: row.run_id,
@@ -452,6 +492,14 @@ where
     }
 
     /// Emit an event to a queue (defaults to this client's queue)
+    #[cfg_attr(
+        feature = "telemetry",
+        tracing::instrument(
+            name = "durable.client.emit_event",
+            skip(self, payload),
+            fields(queue, event_name = %event_name)
+        )
+    )]
     pub async fn emit_event<T: Serialize>(
         &self,
         event_name: &str,
@@ -461,6 +509,10 @@ where
         anyhow::ensure!(!event_name.is_empty(), "event_name must be non-empty");
 
         let queue = queue_name.unwrap_or(&self.queue_name);
+
+        #[cfg(feature = "telemetry")]
+        tracing::Span::current().record("queue", queue);
+
         let payload_json = serde_json::to_value(payload)?;
 
         let query = "SELECT durable.emit_event($1, $2, $3)";
@@ -471,6 +523,9 @@ where
             .execute(&self.pool)
             .await?;
 
+        #[cfg(feature = "telemetry")]
+        crate::telemetry::record_event_emitted(queue, event_name);
+
         Ok(())
     }
 

src/client.rs

@@ -156,6 +156,14 @@ where
     ///     stripe::charge(amount, &idempotency_key).await
     /// }).await?;
     /// ```
+    #[cfg_attr(
+        feature = "telemetry",
+        tracing::instrument(
+            name = "durable.task.step",
+            skip(self, f),
+            fields(task_id = %self.task_id, step_name = %name)
+        )
+    )]
     pub async fn step<T, F, Fut>(&mut self, name: &str, f: F) -> TaskResult<T>
     where
         T: Serialize + DeserializeOwned + Send,
@@ -174,8 +182,21 @@ where
         let result = f().await?;
 
         // Persist checkpoint (also extends claim lease)
+        #[cfg(feature = "telemetry")]
+        let checkpoint_start = std::time::Instant::now();
+
         self.persist_checkpoint(&checkpoint_name, &result).await?;
 
+        #[cfg(feature = "telemetry")]
+        {
+            let duration = checkpoint_start.elapsed().as_secs_f64();
+            crate::telemetry::record_checkpoint_duration(
+                &self.queue_name,
+                &self.task.task_name,
+                duration,
+            );
+        }
+
         Ok(result)
     }
 
@@ -225,6 +246,14 @@ where
     ///
     /// Wake time is computed using the database clock to ensure consistency
     /// with the scheduler and enable deterministic testing via `durable.fake_now`.
+    #[cfg_attr(
+        feature = "telemetry",
+        tracing::instrument(
+            name = "durable.task.sleep_for",
+            skip(self),
+            fields(task_id = %self.task_id, duration_ms = duration.as_millis() as u64)
+        )
+    )]
     pub async fn sleep_for(&mut self, name: &str, duration: std::time::Duration) -> TaskResult<()> {
         validate_user_name(name)?;
         let checkpoint_name = self.get_checkpoint_name(name);
@@ -268,6 +297,14 @@ where
     ///     Some(Duration::from_secs(7 * 24 * 3600)),
     /// ).await?;
     /// ```
+    #[cfg_attr(
+        feature = "telemetry",
+        tracing::instrument(
+            name = "durable.task.await_event",
+            skip(self, timeout),
+            fields(task_id = %self.task_id, event_name = %event_name)
+        )
+    )]
     pub async fn await_event<T: DeserializeOwned>(
         &mut self,
         event_name: &str,
@@ -323,6 +360,14 @@ where
     /// updates the payload (last write wins). Tasks waiting for this event
     /// are woken with the payload at the time of the write that woke them;
     /// subsequent writes do not propagate to already-woken tasks.
+    #[cfg_attr(
+        feature = "telemetry",
+        tracing::instrument(
+            name = "durable.task.emit_event",
+            skip(self, payload),
+            fields(task_id = %self.task_id, event_name = %event_name)
+        )
+    )]
     pub async fn emit_event<T: Serialize>(&self, event_name: &str, payload: &T) -> TaskResult<()> {
         if event_name.is_empty() {
             return Err(TaskError::Failed(anyhow::anyhow!(
@@ -352,6 +397,14 @@ where
     ///
     /// # Errors
     /// Returns `TaskError::Control(Cancelled)` if the task was cancelled.
+    #[cfg_attr(
+        feature = "telemetry",
+        tracing::instrument(
+            name = "durable.task.heartbeat",
+            skip(self),
+            fields(task_id = %self.task_id)
+        )
+    )]
     pub async fn heartbeat(&self, duration: Option<std::time::Duration>) -> TaskResult<()> {
         let extend_by = duration
             .map(|d| d.as_secs() as i32)
@@ -454,6 +507,14 @@ where
     /// let r1: ItemResult = ctx.join("item-1", h1).await?;
     /// let r2: ItemResult = ctx.join("item-2", h2).await?;
     /// ```
+    #[cfg_attr(
+        feature = "telemetry",
+        tracing::instrument(
+            name = "durable.task.spawn",
+            skip(self, params, options),
+            fields(task_id = %self.task_id, subtask_name = T::NAME)
+        )
+    )]
     pub async fn spawn<T>(
         &mut self,
         name: &str,
@@ -515,6 +576,19 @@ where
         options: SpawnOptions,
     ) -> TaskResult<TaskHandle<T>> {
         validate_user_name(name)?;
+
+        // Validate headers don't use reserved prefix
+        if let Some(ref headers) = options.headers {
+            for key in headers.keys() {
+                if key.starts_with("durable::") {
+                    return Err(TaskError::Failed(anyhow::anyhow!(
+                        "Header key '{}' uses reserved prefix 'durable::'. User headers cannot start with 'durable::'.",
+                        key
+                    )));
+                }
+            }
+        }
+
         let checkpoint_name = self.get_checkpoint_name(&format!("$spawn:{name}"));
 
         // Return cached task_id if already spawned
@@ -574,8 +648,7 @@ where
     ///
     /// # Arguments
     ///
-    /// * `name` - Unique name for this join operation (used for checkpointing).
-    ///   Uniqueness is constrained just within this task, not globally or for child tasks.
+    /// * `name` - Unique name for this join operation (used for checkpointing)
     /// * `handle` - The [`TaskHandle`] returned by [`spawn`](Self::spawn)
     ///
     /// # Errors
@@ -590,6 +663,14 @@ where
     /// // ... do other work ...
     /// let result: ComputeResult = ctx.join("compute", handle).await?;
     /// ```
+    #[cfg_attr(
+        feature = "telemetry",
+        tracing::instrument(
+            name = "durable.task.join",
+            skip(self, handle),
+            fields(task_id = %self.task_id, child_task_id = %handle.task_id)
+        )
+    )]
     pub async fn join<T: DeserializeOwned>(
         &mut self,
         name: &str,

src/context.rs

@@ -96,6 +96,8 @@ mod client;
 mod context;
 mod error;
 mod task;
+#[cfg(feature = "telemetry")]
+pub mod telemetry;
 mod types;
 mod worker;