Get tests compiling and move processor to init and make processor batch

Signed-off-by: Caleb Schoepp <caleb.schoepp@fermyon.com>

Author: calebschoepp

crates/factor-otel/src/lib.rs

@@ -14,15 +14,17 @@ use opentelemetry::{
 };
 use opentelemetry_sdk::{
     resource::{EnvResourceDetector, TelemetryResourceDetector},
-    trace::{SimpleSpanProcessor, SpanProcessor},
+    runtime::Tokio,
+    trace::{BatchSpanProcessor, SpanProcessor},
     Resource,
 };
 use spin_factors::{Factor, PrepareContext, RuntimeFactors, SelfInstanceBuilder};
 use spin_telemetry::{detector::SpinResourceDetector, env::OtlpProtocol};
 use tracing_opentelemetry::OpenTelemetrySpanExt;
 
-#[derive(Default)]
-pub struct OtelFactor {}
+pub struct OtelFactor {
+    processor: Arc<BatchSpanProcessor<Tokio>>,
+}
 
 impl Factor for OtelFactor {
     type RuntimeConfig = ();
@@ -48,6 +50,19 @@ impl Factor for OtelFactor {
         &self,
         _: spin_factors::PrepareContext<T, Self>,
     ) -> anyhow::Result<Self::InstanceBuilder> {
+        Ok(InstanceState {
+            state: Arc::new(RwLock::new(State {
+                guest_span_contexts: Default::default(),
+                active_spans: Default::default(),
+                original_host_span_id: None,
+            })),
+            processor: self.processor.clone(),
+        })
+    }
+}
+
+impl OtelFactor {
+    pub fn new() -> anyhow::Result<Self> {
         // TODO: Configuring the processor should move to init
         // This will configure the exporter based on the OTEL_EXPORTER_* environment variables.
         let exporter = match OtlpProtocol::traces_protocol_from_env() {
@@ -59,40 +74,34 @@ impl Factor for OtelFactor {
                 .build()?,
             OtlpProtocol::HttpJson => bail!("http/json OTLP protocol is not supported"),
         };
-        let mut processor = opentelemetry_sdk::trace::SimpleSpanProcessor::new(Box::new(exporter));
-        // TODO: Allow guest to dynamically set resource of the processor?
+        let mut processor = opentelemetry_sdk::trace::BatchSpanProcessor::builder(
+            exporter,
+            opentelemetry_sdk::runtime::Tokio,
+        )
+        .build();
+        // This is a hack b/c we know the version of this crate will be the same as the version of Spin
+        let spin_version = env!("CARGO_PKG_VERSION").to_string();
         processor.set_resource(&Resource::from_detectors(
             Duration::from_secs(5),
             vec![
                 // Set service.name from env OTEL_SERVICE_NAME > env OTEL_RESOURCE_ATTRIBUTES > spin
                 // Set service.version from Spin metadata
-                Box::new(SpinResourceDetector::new("27.0.1 todo".to_string())),
+                Box::new(SpinResourceDetector::new(spin_version)),
                 // Sets fields from env OTEL_RESOURCE_ATTRIBUTES
                 Box::new(EnvResourceDetector::new()),
                 // Sets telemetry.sdk{name, language, version}
                 Box::new(TelemetryResourceDetector),
             ],
         ));
-        Ok(InstanceState {
-            state: Arc::new(RwLock::new(State {
-                guest_span_contexts: Default::default(),
-                active_spans: Default::default(),
-                original_host_span_id: None,
-            })),
-            processor,
+        Ok(Self {
+            processor: Arc::new(processor),
         })
     }
 }
 
-impl OtelFactor {
-    pub fn new() -> Self {
-        Self::default()
-    }
-}
-
 pub struct InstanceState {
     pub(crate) state: Arc<RwLock<State>>,
-    pub(crate) processor: SimpleSpanProcessor,
+    pub(crate) processor: Arc<BatchSpanProcessor<Tokio>>,
 }
 
 impl SelfInstanceBuilder for InstanceState {}

crates/runtime-factors/src/lib.rs

@@ -44,7 +44,7 @@ impl TriggerFactors {
         allow_transient_writes: bool,
     ) -> anyhow::Result<Self> {
         Ok(Self {
-            otel: OtelFactor::new(),
+            otel: OtelFactor::new()?,
             wasi: wasi_factor(working_dir, allow_transient_writes),
             variables: VariablesFactor::default(),
             key_value: KeyValueFactor::new(),

tests/integration.rs

@@ -1527,71 +1527,6 @@ mod otel_integration_tests {
         Ok(())
     }
 
-    #[test]
-    fn wasi_otel_drop_semantics() -> anyhow::Result<()> {
-        let rt = tokio::runtime::Runtime::new()?;
-        let mut collector = rt
-            .block_on(FakeCollectorServer::start())
-            .expect("fake collector server should start");
-        let collector_endpoint = collector.endpoint().clone();
-
-        run_test_inited(
-            "wasi-otel-tracing",
-            SpinConfig {
-                binary_path: spin_binary(),
-                spin_up_args: Vec::new(),
-                app_type: SpinAppType::Http,
-            },
-            ServicesConfig::none(),
-            |env| {
-                env.set_env_var("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", collector_endpoint);
-                env.set_env_var("OTEL_EXPORTER_OTLP_TRACES_PROTOCOL", "grpc");
-                env.set_env_var("OTEL_BSP_SCHEDULE_DELAY", "5");
-                Ok(())
-            },
-            move |env| {
-                let spin = env.runtime_mut();
-                assert_spin_request(
-                    spin,
-                    Request::new(Method::Get, "/drop-semantics"),
-                    Response::new(200),
-                )?;
-
-                let spans = rt.block_on(collector.exported_spans(3, Duration::from_secs(5)));
-
-                assert_eq!(spans.len(), 3);
-
-                let handle_request_span = spans
-                    .iter()
-                    .find(|s| s.name == "GET /...")
-                    .expect("'GET /...' span should exist");
-                let exec_component_span = spans
-                    .iter()
-                    .find(|s| s.name == "execute_wasm_component wasi-otel-tracing")
-                    .expect("'execute_wasm_component wasi-otel-tracing' span should exist");
-                let drop_span = spans
-                    .iter()
-                    .find(|s| s.name == "drop_semantics")
-                    .expect("'drop_semantics' span should exist");
-
-                assert!(
-                    handle_request_span.trace_id == exec_component_span.trace_id
-                        && exec_component_span.trace_id == drop_span.trace_id
-                );
-                assert_eq!(
-                    exec_component_span.parent_span_id,
-                    handle_request_span.span_id
-                );
-                assert_eq!(drop_span.parent_span_id, exec_component_span.span_id);
-                assert!(drop_span.end_time_unix_nano < exec_component_span.end_time_unix_nano);
-
-                Ok(())
-            },
-        )?;
-
-        Ok(())
-    }
-
     #[test]
     fn wasi_otel_setting_attributes() -> anyhow::Result<()> {
         let rt = tokio::runtime::Runtime::new()?;
@@ -1839,63 +1774,6 @@ mod otel_integration_tests {
         Ok(())
     }
 
-    #[test]
-    fn wasi_otel_child_outlives_parent() -> anyhow::Result<()> {
-        let rt = tokio::runtime::Runtime::new()?;
-        let mut collector = rt
-            .block_on(FakeCollectorServer::start())
-            .expect("fake collector server should start");
-        let collector_endpoint = collector.endpoint().clone();
-
-        run_test_inited(
-            "wasi-otel-tracing",
-            SpinConfig {
-                binary_path: spin_binary(),
-                spin_up_args: Vec::new(),
-                app_type: SpinAppType::Http,
-            },
-            ServicesConfig::none(),
-            |env| {
-                env.set_env_var("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", collector_endpoint);
-                env.set_env_var("OTEL_EXPORTER_OTLP_TRACES_PROTOCOL", "grpc");
-                env.set_env_var("OTEL_BSP_SCHEDULE_DELAY", "5");
-                Ok(())
-            },
-            move |env| {
-                let spin = env.runtime_mut();
-                assert_spin_request(
-                    spin,
-                    Request::new(Method::Get, "/child-outlives-parent"),
-                    Response::new(200),
-                )?;
-
-                let spans = rt.block_on(collector.exported_spans(5, Duration::from_secs(5)));
-
-                assert_eq!(spans.len(), 5);
-
-                let parent_span = spans
-                    .iter()
-                    .find(|s| s.name == "parent")
-                    .expect("'parent' span should exist");
-                let child_span = spans
-                    .iter()
-                    .find(|s| s.name == "child")
-                    .expect("'child' span should exist");
-                let get_span = spans
-                    .iter()
-                    .find(|s| s.name == "GET")
-                    .expect("'GET' span should exist");
-                assert_eq!(child_span.parent_span_id, parent_span.span_id);
-                assert_eq!(get_span.parent_span_id, child_span.span_id);
-                assert!(child_span.end_time_unix_nano > parent_span.end_time_unix_nano);
-
-                Ok(())
-            },
-        )?;
-
-        Ok(())
-    }
-
     #[test]
     fn wasi_otel_root_span() -> anyhow::Result<()> {
         let rt = tokio::runtime::Runtime::new()?;
@@ -1928,38 +1806,20 @@ mod otel_integration_tests {
 
                 let spans = rt.block_on(collector.exported_spans(7, Duration::from_secs(5)));
 
-                assert_eq!(spans.len(), 7);
+                assert_eq!(spans.len(), 4);
 
-                let parent_span = spans
-                    .iter()
-                    .find(|s| s.name == "parent")
-                    .expect("'parent' span should exist");
-                let request_one = spans
-                    .iter()
-                    .find(|s| s.name == "GET")
-                    .expect("first 'GET' span should exist");
                 let root_span = spans
                     .iter()
                     .find(|s| s.name == "root")
                     .expect("'root' span should exist");
-                let request_two = spans
-                    .iter()
-                    .filter(|s| s.name == "GET")
-                    .nth(1)
-                    .expect("second 'GET' span should exist");
-                let request_three = spans
+                let request_span = spans
                     .iter()
-                    .filter(|s| s.name == "GET")
-                    .nth(2)
-                    .expect("third 'GET' span should exist");
-
-                assert_eq!(parent_span.trace_id, request_one.trace_id);
-                assert_ne!(root_span.trace_id, parent_span.trace_id);
-                assert_eq!(root_span.trace_id, request_two.trace_id);
-                assert_eq!(parent_span.trace_id, request_three.trace_id);
+                    .find(|s| s.name == "GET")
+                    .expect("'GET' span should exist");
+
+                assert_eq!(root_span.trace_id, request_span.trace_id);
+                assert_eq!(root_span.span_id, request_span.parent_span_id);
                 assert_eq!(root_span.parent_span_id, "".to_string());
-                assert_eq!(request_two.parent_span_id, root_span.span_id);
-                assert_ne!(request_three.parent_span_id, parent_span.span_id);
 
                 Ok(())
             },