Type conversions and some more impls for observe host

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

Author: calebschoepp

Cargo.lock

@@ -2,6 +2,7 @@ use std::time::Duration;
 use std::time::SystemTime;
 use std::time::UNIX_EPOCH;
 
+use anyhow::anyhow;
 use anyhow::Result;
 use opentelemetry::global::ObjectSafeSpan;
 use opentelemetry::trace::TraceContextExt;
@@ -68,17 +69,20 @@ impl traces::HostSpan for InstanceState {
 
     async fn span_context(
         &mut self,
-        _resource: Resource<traces::Span>,
+        resource: Resource<traces::Span>,
     ) -> Result<traces::SpanContext> {
-        todo!()
+        if let Some(guest_span) = self.state.read().unwrap().guest_spans.get(resource.rep()) {
+            Ok(guest_span.inner.span_context().clone().into())
+        } else {
+            Err(anyhow!("BUG: cannot find resource in table"))
+        }
     }
 
     async fn is_recording(&mut self, resource: Resource<traces::Span>) -> Result<bool> {
         if let Some(guest_span) = self.state.read().unwrap().guest_spans.get(resource.rep()) {
             Ok(guest_span.inner.is_recording())
         } else {
-            tracing::debug!("can't find guest span to read");
-            Err(anyhow::anyhow!("can't find guest span to read"))
+            Err(anyhow!("BUG: cannot find resource in table"))
         }
     }
 
@@ -97,10 +101,10 @@ impl traces::HostSpan for InstanceState {
             for attribute in attributes {
                 guest_span.inner.set_attribute(attribute.into());
             }
+            Ok(())
         } else {
-            tracing::debug!("can't find guest span to set attributes on")
+            Err(anyhow!("BUG: cannot find resource in table"))
         }
-        Ok(())
     }
 
     async fn add_event(
@@ -126,26 +130,37 @@ impl traces::HostSpan for InstanceState {
             };
 
             let attributes = if let Some(attributes) = attributes {
-                attributes.into_iter().map(|a| a.into()).collect()
+                attributes.into_iter().map(Into::into).collect()
             } else {
                 vec![]
             };
 
             guest_span
                 .inner
                 .add_event_with_timestamp(name.into(), timestamp, attributes);
+
+            Ok(())
         } else {
-            tracing::debug!("can't find guest span to add events too")
+            Err(anyhow!("BUG: cannot find resource in table"))
         }
-        Ok(())
     }
 
     async fn add_link(
         &mut self,
-        _resource: Resource<traces::Span>,
+        resource: Resource<traces::Span>,
         _link: traces::Link,
     ) -> Result<()> {
-        todo!()
+        if let Some(_guest_span) = self
+            .state
+            .write()
+            .unwrap()
+            .guest_spans
+            .get_mut(resource.rep())
+        {
+            todo!("update otel versions -> guest_span.inner.add_link(link.into());");
+        } else {
+            Err(anyhow!("BUG: cannot find resource in table"))
+        }
     }
 
     async fn set_status(
@@ -165,10 +180,10 @@ impl traces::HostSpan for InstanceState {
             .get_mut(resource.rep())
         {
             guest_span.inner.update_name(name.into());
+            Ok(())
         } else {
-            tracing::debug!("can't find guest span to set name on")
+            Err(anyhow!("BUG: cannot find resource in table"))
         }
-        Ok(())
     }
 
     async fn end(
@@ -183,11 +198,11 @@ impl traces::HostSpan for InstanceState {
             .guest_spans
             .get_mut(resource.rep())
         {
-            guest_span.inner.end()
+            guest_span.inner.end();
+            Ok(())
         } else {
-            tracing::debug!("can't find guest span to end")
+            Err(anyhow!("BUG: cannot find resource in table"))
         }
-        Ok(())
     }
 
     fn drop(&mut self, _resource: Resource<traces::Span>) -> Result<()> {
@@ -198,3 +213,5 @@ impl traces::HostSpan for InstanceState {
 }
 
 // TODO(Caleb): Improve debug tracing in failure cases
+// TODO(Caleb): Move the tests from integration.rs to here
+// TODO(Caleb): Write tests somewhere for all the finicky type conversion stuff

crates/factor-observe/src/host.rs

@@ -5,6 +5,7 @@ authors = { workspace = true }
 edition = { workspace = true }
 
 [dependencies]
+anyhow = { workspace = true }
 async-trait = "0.1"
-wasmtime = { workspace = true }
 opentelemetry = { version = "0.22.0", features = [ "metrics", "trace"] }
+wasmtime = { workspace = true }

crates/world/Cargo.toml

@@ -249,4 +249,66 @@ mod observe {
             opentelemetry::KeyValue::new(kv.key, kv.value)
         }
     }
+
+    impl From<traces::TraceFlags> for opentelemetry::trace::TraceFlags {
+        fn from(flags: traces::TraceFlags) -> Self {
+            Self::new(flags.as_array()[0] as u8)
+        }
+    }
+
+    impl TryFrom<traces::SpanContext> for opentelemetry::trace::SpanContext {
+        type Error = wasmtime::Error;
+
+        fn try_from(sc: traces::SpanContext) -> anyhow::Result<Self> {
+            // TODO: Endianess
+            let trace_id: [u8; 16] = {
+                let mut whole: [u8; 16] = [0; 16];
+                let (one, two) = whole.split_at_mut(8);
+                one.copy_from_slice(&sc.trace_id.0.to_le_bytes());
+                two.copy_from_slice(&sc.trace_id.1.to_le_bytes());
+                whole
+            };
+            Ok(Self::new(
+                opentelemetry::trace::TraceId::from_bytes(trace_id),
+                opentelemetry::trace::SpanId::from_bytes(sc.span_id.to_le_bytes()),
+                sc.trace_flags.into(),
+                sc.is_remote,
+                opentelemetry::trace::TraceState::from_key_value(sc.trace_state)?,
+            ))
+        }
+    }
+
+    impl From<opentelemetry::trace::SpanContext> for traces::SpanContext {
+        fn from(sc: opentelemetry::trace::SpanContext) -> Self {
+            Self {
+                trace_id: {
+                    let whole = u128::from_le_bytes(sc.trace_id().to_bytes());
+                    ((whole >> 64) as u64, whole as u64)
+                },
+                span_id: u64::from_le_bytes(sc.span_id().to_bytes()),
+                trace_flags: traces::TraceFlags::all(), // TODO(Caleb): This is broken
+                is_remote: sc.is_remote(),
+                trace_state: sc
+                    .trace_state()
+                    .header()
+                    .split(",")
+                    .map(|s| {
+                        let (key, value) = s.split_once('=').unwrap();
+                        (key.to_string(), value.to_string())
+                    })
+                    .collect(),
+            }
+        }
+    }
+
+    impl TryFrom<traces::Link> for opentelemetry::trace::Link {
+        type Error = wasmtime::Error;
+
+        fn try_from(link: traces::Link) -> anyhow::Result<Self> {
+            Ok(Self::new(
+                link.span_context.try_into()?,
+                link.attributes.into_iter().map(Into::into).collect(),
+            ))
+        }
+    }
 }

crates/world/src/conversions.rs

@@ -135,7 +135,7 @@ interface traces {
         /// Whether this `span-context` was propagated from a remote parent.
         is-remote: bool,
         /// The `trace-state` for this `span-context`.
-        trace-state: string,
+        trace-state: trace-state,
     }
 
     /// The trace that this `span-context` belongs to.
@@ -151,5 +151,5 @@ interface traces {
     }
 
     /// Carries system-specific configuration data, represented as a list of key-value pairs. `trace-state` allows multiple tracing systems to participate in the same trace.
-    type trace-state = option<list<tuple<string, string>>>;
+    type trace-state = list<tuple<string, string>>;
 }