feat(wasi-observe): A WIP WASI Observe host component

Signed-off-by: Caleb Schoepp <[email protected]>

Author: calebschoepp

Cargo.lock

@@ -97,6 +97,7 @@ test-components = { path = "tests/test-components" }
 test-environment = { workspace = true }
 testing-framework = { path = "tests/testing-framework" }
 which = "4.2.5"
+fake-opentelemetry-collector = "0.19.0"
 
 [build-dependencies]
 cargo-target-dep = { git = "https://github.com/fermyon/cargo-target-dep", rev = "482f269eceb7b1a7e8fc618bf8c082dd24979cf1" }

Cargo.toml

@@ -0,0 +1,38 @@
+[package]
+name = "spin-factor-observe"
+version = { workspace = true }
+authors = { workspace = true }
+edition = { workspace = true }
+
+[dependencies]
+anyhow = "1.0"
+async-trait = "0.1"
+dotenvy = "0.15"
+futures-executor = "0.3"
+indexmap = "2.2.6"
+once_cell = "1"
+opentelemetry = { version = "0.22.0", features = [ "metrics", "trace"] }
+opentelemetry_sdk = { version = "0.22.1", features = ["rt-tokio"] }
+opentelemetry-otlp = { version = "0.15.0", default-features=false, features = ["http-proto", "trace", "http", "reqwest-client", "metrics", "grpc-tonic"] }
+pin-project-lite = "0.2"
+serde = "1.0.188"
+spin-app = { path = "../app" }
+spin-core = { path = "../core" }
+spin-expressions = { path = "../expressions" }
+spin-factors = { path = "../factors" }
+spin-telemetry = { path = "../telemetry" }
+spin-world = { path = "../world" }
+table = { path = "../table" }
+thiserror = "1"
+tokio = { version = "1", features = ["rt-multi-thread"] }
+tracing = "0.1.40"
+tracing-opentelemetry = "0.23.0"
+vaultrs = "0.6.2"
+
+[dev-dependencies]
+toml = "0.5"
+
+[lints]
+workspace = true
+
+# TODO(Caleb): Cleanup these dependencies

crates/factor-observe/Cargo.toml

@@ -0,0 +1,80 @@
+use pin_project_lite::pin_project;
+use std::{
+    future::Future,
+    sync::{Arc, RwLock},
+};
+
+use crate::State;
+
+pin_project! {
+    struct Instrumented<F> {
+        #[pin]
+        inner: F,
+        observe_context: ObserveContext,
+    }
+
+    impl<F> PinnedDrop for Instrumented<F> {
+        fn drop(this: Pin<&mut Self>) {
+            this.project().observe_context.drop_all();
+        }
+    }
+}
+
+pub trait FutureExt: Future + Sized {
+    /// Manage WASI Observe guest spans.
+    fn manage_wasi_observe_spans(
+        self,
+        observe_context: ObserveContext,
+    ) -> impl Future<Output = Self::Output>;
+}
+
+impl<F: Future> FutureExt for F {
+    fn manage_wasi_observe_spans(
+        self,
+        observe_context: ObserveContext,
+    ) -> impl Future<Output = Self::Output> {
+        Instrumented {
+            inner: self,
+            observe_context,
+        }
+    }
+}
+
+impl<F: Future> Future for Instrumented<F> {
+    type Output = F::Output;
+
+    /// Maintains the invariant that all active spans are entered before polling the inner future
+    /// and exited otherwise. If we don't do this then the timing (among many other things) of the
+    /// spans becomes wildly incorrect.
+    fn poll(
+        self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<Self::Output> {
+        let this = self.project();
+
+        // Enter the active spans before entering the inner poll
+        {
+            this.observe_context.state.write().unwrap().enter_all();
+        }
+
+        let ret = this.inner.poll(cx);
+
+        // Exit the active spans after exiting the inner poll
+        {
+            this.observe_context.state.write().unwrap().exit_all();
+        }
+
+        ret
+    }
+}
+
+/// The context necessary for the observe host component to function.
+pub struct ObserveContext {
+    pub(crate) state: Arc<RwLock<State>>,
+}
+
+impl ObserveContext {
+    fn drop_all(&self) {
+        self.state.write().unwrap().close_from_back_to(0);
+    }
+}