feat: add opentelemetry tracing and metrics

Cargo.toml

@@ -23,17 +23,6 @@ path = "src/main.rs"
 name = "cargo-lunatic"
 path = "src/cargo_lunatic.rs"
 
-[features]
-default = ["metrics"]
-metrics = [
-    "lunatic-process-api/metrics",
-    "lunatic-process/metrics",
-    "lunatic-registry-api/metrics",
-    "lunatic-timer-api/metrics",
-    "dep:lunatic-metrics-api",
-]
-prometheus = ["dep:metrics-exporter-prometheus", "metrics"]
-
 [dependencies]
 hash-map-id = { workspace = true }
 lunatic-control = { workspace = true }
@@ -42,29 +31,33 @@ lunatic-distributed = { workspace = true }
 lunatic-distributed-api = { workspace = true }
 lunatic-error-api = { workspace = true }
 lunatic-messaging-api = { workspace = true }
+lunatic-metrics-api = { workspace = true }
 lunatic-networking-api = { workspace = true }
 lunatic-process = { workspace = true }
 lunatic-process-api = { workspace = true }
 lunatic-registry-api = { workspace = true }
+lunatic-sqlite-api = { workspace = true }
 lunatic-stdout-capture = { workspace = true }
 lunatic-timer-api = { workspace = true }
+lunatic-trap-api = { workspace = true }
 lunatic-version-api = { workspace = true }
-lunatic-metrics-api = { workspace = true, optional = true }
 lunatic-wasi-api = { workspace = true }
-lunatic-trap-api = { workspace = true }
-lunatic-sqlite-api = { workspace = true }
 
 anyhow = { workspace = true }
 async-ctrlc = "1.2.0"
 clap = { version = "4.0", features = ["cargo", "derive"] }
 dashmap = { workspace = true }
 env_logger = "0.9"
+hyper = { version = "0.14", features = ["full"] }
 log = { workspace = true }
-metrics-exporter-prometheus = { version = "0.11.0", optional = true }
+opentelemetry = { workspace = true, features = ["rt-tokio"] }
+opentelemetry-jaeger = { version = "0.18.0", git = "https://github.com/tqwewe/opentelemetry-rust", branch = "cow", features = ["rt-tokio"] }
+opentelemetry-prometheus = { version = "0.12.0", git = "https://github.com/tqwewe/opentelemetry-rust", branch = "cow" }
+prometheus = "0.13"
 regex = "1.7"
 reqwest = { workspace = true }
 serde = { workspace = true, features = ["derive"] }
-tokio = { workspace = true, features = ["macros", "rt-multi-thread", "net"] }
+tokio = { workspace = true, features = ["macros", "rt-multi-thread", "net", "signal"] }
 toml = "0.5"
 uuid = { workspace = true }
 wasmtime = { workspace = true }
@@ -127,7 +120,8 @@ anyhow = "1.0"
 bincode = "1.3"
 dashmap = "5.4"
 log = "0.4"
-metrics = "0.20.1"
+once_cell = "1.17"
+opentelemetry = { version = "0.19", git = "https://github.com/tqwewe/opentelemetry-rust", branch = "cow", features = ["metrics"] }
 reqwest = "0.11"
 rustls-pemfile = "1.0"
 serde = "1.0"

benches/benchmark.rs

@@ -7,6 +7,12 @@ use lunatic_process::{
     runtimes::wasmtime::{default_config, WasmtimeRuntime},
 };
 use lunatic_runtime::{state::DefaultProcessState, DefaultProcessConfig};
+use opentelemetry::{
+    global::{BoxedTracer, GlobalMeterProvider},
+    metrics::noop::NoopMeterProvider,
+    trace::noop::NoopTracer,
+    Context,
+};
 use tokio::sync::RwLock;
 
 fn criterion_benchmark(c: &mut Criterion) {
@@ -23,6 +29,15 @@ fn criterion_benchmark(c: &mut Criterion) {
             .unwrap(),
     );
 
+    let tracer = Arc::new(BoxedTracer::new(Box::new(NoopTracer::new())));
+    let tracer_context = Arc::new(Context::new());
+    let meter_provider = GlobalMeterProvider::new(NoopMeterProvider::new());
+    let logger = Arc::new(
+        env_logger::Builder::new()
+            .filter_level(log::LevelFilter::Off)
+            .build(),
+    );
+
     let env = Arc::new(LunaticEnvironment::new(0));
     c.bench_function("spawn process", |b| {
         b.to_async(&rt).iter(|| async {
@@ -34,6 +49,10 @@ fn criterion_benchmark(c: &mut Criterion) {
                 module.clone(),
                 config.clone(),
                 registry,
+                tracer.clone(),
+                tracer_context.clone(),
+                meter_provider.clone(),
+                logger.clone(),
             )
             .unwrap();
             lunatic_process::wasm::spawn_wasm(

crates/hash-map-id/src/lib.rs

@@ -35,6 +35,13 @@ where
     pub fn get(&self, id: u64) -> Option<&T> {
         self.store.get(&id)
     }
+
+    pub fn get_last(&self) -> Option<&T> {
+        self.store
+            .iter()
+            .max_by_key(|(k, _)| **k)
+            .map(|(_, value)| value)
+    }
 }
 
 impl<T> Default for HashMapId<T>

crates/lunatic-common-api/Cargo.toml

@@ -9,4 +9,6 @@ license = "Apache-2.0/MIT"
 
 [dependencies]
 anyhow = { workspace = true }
+once_cell = { workspace = true }
+opentelemetry = { workspace = true }
 wasmtime = { workspace = true }

crates/lunatic-common-api/src/lib.rs

@@ -1,5 +1,7 @@
-use anyhow::{anyhow, Context, Result};
 use std::{fmt::Display, future::Future, io::Write, pin::Pin};
+
+use anyhow::{anyhow, Context, Result};
+use once_cell::sync::OnceCell;
 use wasmtime::{Caller, Memory, Val};
 
 const ALLOCATOR_FUNCTION_NAME: &str = "lunatic_alloc";
@@ -78,6 +80,23 @@ pub async fn write_to_guest_vec<T: Send>(
     Ok(alloc_ptr)
 }
 
+pub trait MetricsExt<T> {
+    fn with_current_context<F>(&self, f: F)
+    where
+        F: Fn(&T, opentelemetry::Context);
+}
+
+impl<T> MetricsExt<T> for OnceCell<T> {
+    fn with_current_context<F>(&self, f: F)
+    where
+        F: Fn(&T, opentelemetry::Context),
+    {
+        if let Some(v) = self.get() {
+            f(v, opentelemetry::Context::current());
+        }
+    }
+}
+
 pub trait IntoTrap<T> {
     fn or_trap<S: Display>(self, info: S) -> Result<T>;
 }

crates/lunatic-messaging-api/Cargo.toml

@@ -14,5 +14,6 @@ lunatic-process = { workspace = true }
 lunatic-process-api = { workspace = true }
 
 anyhow = { workspace = true }
+opentelemetry = { workspace = true }
 tokio = { workspace = true, features = ["time"] }
 wasmtime = { workspace = true }

crates/lunatic-messaging-api/src/lib.rs

@@ -5,17 +5,17 @@ use std::{
 };
 
 use anyhow::{anyhow, Result};
-use lunatic_common_api::{get_memory, IntoTrap};
+use lunatic_common_api::{get_memory, IntoTrap, MetricsExt};
 use lunatic_networking_api::NetworkingCtx;
-use lunatic_process_api::ProcessCtx;
-use tokio::time::{timeout, Duration};
-use wasmtime::{Caller, Linker};
-
 use lunatic_process::{
     message::{DataMessage, Message},
     state::ProcessState,
-    Signal,
+    Signal, MESSAGES_METRICS,
 };
+use lunatic_process_api::ProcessCtx;
+use opentelemetry::KeyValue;
+use tokio::time::{timeout, Duration};
+use wasmtime::{Caller, Linker};
 
 // Register the mailbox APIs to the linker
 pub fn register<T: ProcessState + ProcessCtx<T> + NetworkingCtx + Send + 'static>(
@@ -306,7 +306,7 @@ fn take_module<T: ProcessState + ProcessCtx<T> + NetworkingCtx + 'static>(
         .or_trap("lunatic::message::take_module")?;
     let module = match message {
         Message::Data(data) => data
-            .take_module(index as usize)
+            .take_resource(index as usize)
             .or_trap("lunatic::message::take_module")?,
         Message::LinkDied(_) => {
             return Err(anyhow!("Unexpected `Message::LinkDied` in scratch area"))
@@ -316,10 +316,10 @@ fn take_module<T: ProcessState + ProcessCtx<T> + NetworkingCtx + 'static>(
 }
 
 // Adds a tcp stream resource to the message that is currently in the scratch area and returns
-// the new location of it. This will remove the tcp stream from  the current process' resources.
+// the new location of it. This will remove the tcp stream from the current process' resources.
 //
 // Traps:
-// * If TCP stream ID doesn't exist
+// * If TCP stream ID doesn't exist.
 // * If no data message is in the scratch area.
 fn push_tcp_stream<T: ProcessState + ProcessCtx<T> + NetworkingCtx>(
     mut caller: Caller<T>,
@@ -361,7 +361,7 @@ fn take_tcp_stream<T: ProcessState + ProcessCtx<T> + NetworkingCtx>(
         .or_trap("lunatic::message::take_tcp_stream")?;
     let tcp_stream = match message {
         Message::Data(data) => data
-            .take_tcp_stream(index as usize)
+            .take_resource(index as usize)
             .or_trap("lunatic::message::take_tcp_stream")?,
         Message::LinkDied(_) => {
             return Err(anyhow!("Unexpected `Message::LinkDied` in scratch area"))
@@ -417,7 +417,7 @@ fn take_tls_stream<T: ProcessState + ProcessCtx<T> + NetworkingCtx>(
         .or_trap("lunatic::message::take_tls_stream")?;
     let tls_stream = match message {
         Message::Data(data) => data
-            .take_tls_stream(index as usize)
+            .take_resource(index as usize)
             .or_trap("lunatic::message::take_tls_stream")?,
         Message::LinkDied(_) => {
             return Err(anyhow!("Unexpected `Message::LinkDied` in scratch area"))
@@ -526,6 +526,8 @@ fn receive<T: ProcessState + ProcessCtx<T> + Send>(
     timeout_duration: u64,
 ) -> Box<dyn Future<Output = Result<u32>> + Send + '_> {
     Box::new(async move {
+        let id = caller.data().id();
+
         let tags = if tag_len > 0 {
             let memory = get_memory(&mut caller)?;
             let buffer = memory
@@ -544,6 +546,13 @@ fn receive<T: ProcessState + ProcessCtx<T> + Send>(
         };
 
         let pop = caller.data_mut().mailbox().pop(tags.as_deref());
+
+        MESSAGES_METRICS.with_current_context(|metrics, cx| {
+            metrics
+                .outstanding
+                .add(&cx, -1, &[KeyValue::new("process_id", id as i64)]);
+        });
+
         if let Ok(message) = match timeout_duration {
             // Without timeout
             u64::MAX => Ok(pop.await),
@@ -608,7 +617,7 @@ fn take_udp_socket<T: ProcessState + ProcessCtx<T> + NetworkingCtx>(
         .or_trap("lunatic::message::take_udp_socket")?;
     let udp_socket = match message {
         Message::Data(data) => data
-            .take_udp_socket(index as usize)
+            .take_resource(index as usize)
             .or_trap("lunatic::message::take_udp_socket")?,
         Message::LinkDied(_) => {
             return Err(anyhow!("Unexpected `Message::LinkDied` in scratch area"))

crates/lunatic-metrics-api/Cargo.toml

@@ -8,8 +8,13 @@ repository = "https://github.com/lunatic-solutions/lunatic/tree/main/crates/luna
 license = "Apache-2.0/MIT"
 
 [dependencies]
-anyhow = { workspace = true }
-wasmtime = { workspace = true }
-metrics = { workspace = true }
+hash-map-id = { workspace = true }
 lunatic-common-api = { workspace = true }
+lunatic-process = { workspace = true }
+lunatic-process-api = { workspace = true }
+
+anyhow = { workspace = true }
 log = { workspace = true }
+opentelemetry = { workspace = true }
+serde_json = "1.0"
+wasmtime = { workspace = true }