Only set span as errored in key-value when it might not be bad user input

Signed-off-by: Ryan Levick <ryan.levick@fermyon.com>

Author: rylev

Cargo.lock

@@ -11,11 +11,12 @@ spin-core = { path = "../core" }
 spin-factors = { path = "../factors" }
 spin-locked-app = { path = "../locked-app" }
 spin-resource-table = { path = "../table" }
+spin-telemetry = { path = "../telemetry" }
 spin-world = { path = "../world" }
+thiserror = { workspace = true }
 tokio = { workspace = true, features = ["macros", "sync", "rt"] }
 toml = { workspace = true }
 tracing = { workspace = true }
-thiserror = { workspace = true }
 
 [dev-dependencies]
 spin-factors-test = { path = "../factors-test" }

crates/factor-key-value/Cargo.toml

@@ -2,10 +2,11 @@ use super::{Cas, SwapError};
 use anyhow::{Context, Result};
 use spin_core::{async_trait, wasmtime::component::Resource};
 use spin_resource_table::Table;
+use spin_telemetry::OpenTelemetrySpanExt as _;
 use spin_world::v2::key_value;
 use spin_world::wasi::keyvalue as wasi_keyvalue;
 use std::{collections::HashSet, sync::Arc};
-use tracing::{instrument, Level};
+use tracing::instrument;
 
 const DEFAULT_STORE_TABLE_CAPACITY: u32 = 256;
 
@@ -69,7 +70,14 @@ impl KeyValueDispatch {
     }
 
     pub fn get_store<T: 'static>(&self, store: Resource<T>) -> anyhow::Result<&Arc<dyn Store>> {
-        self.stores.get(store.rep()).context("invalid store")
+        let res = self.stores.get(store.rep()).context("invalid store");
+        if let Err(err) = &res {
+            let current_span = tracing::Span::current();
+            current_span.set_status(spin_telemetry::opentelemetry::trace::Status::error(
+                err.to_string(),
+            ));
+        }
+        res
     }
 
     pub fn get_cas<T: 'static>(&self, cas: Resource<T>) -> Result<&Arc<dyn Cas>> {
@@ -106,7 +114,7 @@ impl KeyValueDispatch {
 impl key_value::Host for KeyValueDispatch {}
 
 impl key_value::HostStore for KeyValueDispatch {
-    #[instrument(name = "spin_key_value.open", skip(self), err(level = Level::INFO), fields(otel.kind = "client", kv.backend=self.manager.summary(&name).unwrap_or("unknown".to_string())))]
+    #[instrument(name = "spin_key_value.open", skip(self), err, fields(otel.kind = "client", kv.backend=self.manager.summary(&name).unwrap_or("unknown".to_string())))]
     async fn open(&mut self, name: String) -> Result<Result<Resource<key_value::Store>, Error>> {
         Ok(async {
             if self.allowed_stores.contains(&name) {
@@ -124,54 +132,54 @@ impl key_value::HostStore for KeyValueDispatch {
         .await)
     }
 
-    #[instrument(name = "spin_key_value.get", skip(self, store, key), err(level = Level::INFO), fields(otel.kind = "client"))]
+    #[instrument(name = "spin_key_value.get", skip(self, store, key), fields(otel.kind = "client"))]
     async fn get(
         &mut self,
         store: Resource<key_value::Store>,
         key: String,
     ) -> Result<Result<Option<Vec<u8>>, Error>> {
         let store = self.get_store(store)?;
-        Ok(store.get(&key).await)
+        Ok(store.get(&key).await.map_err(track_error_on_span))
     }
 
-    #[instrument(name = "spin_key_value.set", skip(self, store, key, value), err(level = Level::INFO), fields(otel.kind = "client"))]
+    #[instrument(name = "spin_key_value.set", skip(self, store, key, value), fields(otel.kind = "client"))]
     async fn set(
         &mut self,
         store: Resource<key_value::Store>,
         key: String,
         value: Vec<u8>,
     ) -> Result<Result<(), Error>> {
         let store = self.get_store(store)?;
-        Ok(store.set(&key, &value).await)
+        Ok(store.set(&key, &value).await.map_err(track_error_on_span))
     }
 
-    #[instrument(name = "spin_key_value.delete", skip(self, store, key), err(level = Level::INFO), fields(otel.kind = "client"))]
+    #[instrument(name = "spin_key_value.delete", skip(self, store, key), fields(otel.kind = "client"))]
     async fn delete(
         &mut self,
         store: Resource<key_value::Store>,
         key: String,
     ) -> Result<Result<(), Error>> {
         let store = self.get_store(store)?;
-        Ok(store.delete(&key).await)
+        Ok(store.delete(&key).await.map_err(track_error_on_span))
     }
 
-    #[instrument(name = "spin_key_value.exists", skip(self, store, key), err(level = Level::INFO), fields(otel.kind = "client"))]
+    #[instrument(name = "spin_key_value.exists", skip(self, store, key), fields(otel.kind = "client"))]
     async fn exists(
         &mut self,
         store: Resource<key_value::Store>,
         key: String,
     ) -> Result<Result<bool, Error>> {
         let store = self.get_store(store)?;
-        Ok(store.exists(&key).await)
+        Ok(store.exists(&key).await.map_err(track_error_on_span))
     }
 
-    #[instrument(name = "spin_key_value.get_keys", skip(self, store), err(level = Level::INFO), fields(otel.kind = "client"))]
+    #[instrument(name = "spin_key_value.get_keys", skip(self, store), fields(otel.kind = "client"))]
     async fn get_keys(
         &mut self,
         store: Resource<key_value::Store>,
     ) -> Result<Result<Vec<String>, Error>> {
         let store = self.get_store(store)?;
-        Ok(store.get_keys().await)
+        Ok(store.get_keys().await.map_err(track_error_on_span))
     }
 
     async fn drop(&mut self, store: Resource<key_value::Store>) -> Result<()> {
@@ -180,8 +188,19 @@ impl key_value::HostStore for KeyValueDispatch {
     }
 }
 
+/// Make sure that infrastructure related errors are tracked in the current span.
+fn track_error_on_span(err: Error) -> Error {
+    if let Error::Other(_) | Error::StoreTableFull = &err {
+        let current_span = tracing::Span::current();
+        current_span.set_status(spin_telemetry::opentelemetry::trace::Status::error(
+            err.to_string(),
+        ));
+    }
+    err
+}
+
 fn to_wasi_err(e: Error) -> wasi_keyvalue::store::Error {
-    match e {
+    match track_error_on_span(e) {
         Error::AccessDenied => wasi_keyvalue::store::Error::AccessDenied,
         Error::NoSuchStore => wasi_keyvalue::store::Error::NoSuchStore,
         Error::StoreTableFull => wasi_keyvalue::store::Error::Other("store table full".to_string()),
@@ -190,6 +209,7 @@ fn to_wasi_err(e: Error) -> wasi_keyvalue::store::Error {
 }
 
 impl wasi_keyvalue::store::Host for KeyValueDispatch {
+    #[instrument(name = "wasi_key_value.open", skip(self, identifier), fields(otel.kind = "client"))]
     async fn open(
         &mut self,
         identifier: String,
@@ -217,6 +237,7 @@ impl wasi_keyvalue::store::Host for KeyValueDispatch {
 
 use wasi_keyvalue::store::Bucket;
 impl wasi_keyvalue::store::HostBucket for KeyValueDispatch {
+    #[instrument(name = "wasi_key_value.get", skip(self, self_, key), fields(otel.kind = "client"))]
     async fn get(
         &mut self,
         self_: Resource<Bucket>,
@@ -226,6 +247,7 @@ impl wasi_keyvalue::store::HostBucket for KeyValueDispatch {
         store.get(&key).await.map_err(to_wasi_err)
     }
 
+    #[instrument(name = "wasi_key_value.set", skip(self, self_, key, value), fields(otel.kind = "client"))]
     async fn set(
         &mut self,
         self_: Resource<Bucket>,
@@ -236,6 +258,7 @@ impl wasi_keyvalue::store::HostBucket for KeyValueDispatch {
         store.set(&key, &value).await.map_err(to_wasi_err)
     }
 
+    #[instrument(name = "wasi_key_value.delete", skip(self, self_, key), fields(otel.kind = "client"))]
     async fn delete(
         &mut self,
         self_: Resource<Bucket>,
@@ -245,6 +268,7 @@ impl wasi_keyvalue::store::HostBucket for KeyValueDispatch {
         store.delete(&key).await.map_err(to_wasi_err)
     }
 
+    #[instrument(name = "wasi_key_value.exists", skip(self, self_, key), fields(otel.kind = "client"))]
     async fn exists(
         &mut self,
         self_: Resource<Bucket>,
@@ -254,6 +278,7 @@ impl wasi_keyvalue::store::HostBucket for KeyValueDispatch {
         store.exists(&key).await.map_err(to_wasi_err)
     }
 
+    #[instrument(name = "wasi_key_value.list_keys", skip(self, self_, cursor), fields(otel.kind = "client"))]
     async fn list_keys(
         &mut self,
         self_: Resource<Bucket>,
@@ -278,7 +303,7 @@ impl wasi_keyvalue::store::HostBucket for KeyValueDispatch {
 }
 
 impl wasi_keyvalue::batch::Host for KeyValueDispatch {
-    #[instrument(name = "spin_key_value.get_many", skip(self, bucket, keys), err(level = Level::INFO), fields(otel.kind = "client"))]
+    #[instrument(name = "spin_key_value.get_many", skip(self, bucket, keys), fields(otel.kind = "client"))]
     #[allow(clippy::type_complexity)]
     async fn get_many(
         &mut self,
@@ -292,7 +317,7 @@ impl wasi_keyvalue::batch::Host for KeyValueDispatch {
         store.get_many(keys).await.map_err(to_wasi_err)
     }
 
-    #[instrument(name = "spin_key_value.set_many", skip(self, bucket, key_values), err(level = Level::INFO), fields(otel.kind = "client"))]
+    #[instrument(name = "spin_key_value.set_many", skip(self, bucket, key_values), fields(otel.kind = "client"))]
     async fn set_many(
         &mut self,
         bucket: Resource<wasi_keyvalue::batch::Bucket>,
@@ -305,7 +330,7 @@ impl wasi_keyvalue::batch::Host for KeyValueDispatch {
         store.set_many(key_values).await.map_err(to_wasi_err)
     }
 
-    #[instrument(name = "spin_key_value.delete_many", skip(self, bucket, keys), err(level = Level::INFO), fields(otel.kind = "client"))]
+    #[instrument(name = "spin_key_value.delete_many", skip(self, bucket, keys), fields(otel.kind = "client"))]
     async fn delete_many(
         &mut self,
         bucket: Resource<wasi_keyvalue::batch::Bucket>,
@@ -320,6 +345,7 @@ impl wasi_keyvalue::batch::Host for KeyValueDispatch {
 }
 
 impl wasi_keyvalue::atomics::HostCas for KeyValueDispatch {
+    #[instrument(name = "wasi_key_value_cas.new", skip(self, bucket, key), fields(otel.kind = "client"))]
     async fn new(
         &mut self,
         bucket: Resource<wasi_keyvalue::atomics::Bucket>,
@@ -342,6 +368,7 @@ impl wasi_keyvalue::atomics::HostCas for KeyValueDispatch {
             .map(Resource::new_own)
     }
 
+    #[instrument(name = "wasi_key_value_cas.current", skip(self, cas), fields(otel.kind = "client"))]
     async fn current(
         &mut self,
         cas: Resource<wasi_keyvalue::atomics::Cas>,
@@ -366,7 +393,7 @@ impl wasi_keyvalue::atomics::Host for KeyValueDispatch {
         Ok(error)
     }
 
-    #[instrument(name = "spin_key_value.increment", skip(self, bucket, key, delta), err(level = Level::INFO), fields(otel.kind = "client"))]
+    #[instrument(name = "spin_key_value.increment", skip(self, bucket, key, delta), fields(otel.kind = "client"))]
     async fn increment(
         &mut self,
         bucket: Resource<wasi_keyvalue::atomics::Bucket>,
@@ -377,7 +404,7 @@ impl wasi_keyvalue::atomics::Host for KeyValueDispatch {
         store.increment(key, delta).await.map_err(to_wasi_err)
     }
 
-    #[instrument(name = "spin_key_value.swap", skip(self, cas_res, value), err(level = Level::INFO), fields(otel.kind = "client"))]
+    #[instrument(name = "spin_key_value.swap", skip(self, cas_res, value), fields(otel.kind = "client"))]
     async fn swap(
         &mut self,
         cas_res: Resource<atomics::Cas>,
@@ -424,8 +451,8 @@ use spin_world::v1::key_value::Error as LegacyError;
 use spin_world::wasi::keyvalue::atomics;
 use spin_world::wasi::keyvalue::atomics::{CasError, HostCas};
 
-fn to_legacy_error(value: key_value::Error) -> LegacyError {
-    match value {
+fn to_legacy_error(err: Error) -> LegacyError {
+    match err {
         Error::StoreTableFull => LegacyError::StoreTableFull,
         Error::NoSuchStore => LegacyError::NoSuchStore,
         Error::AccessDenied => LegacyError::AccessDenied,