Merge pull request #1556 from nightkr/feature/error-boundary

Add error boundary wrapper type

Cargo.toml

@@ -74,6 +74,7 @@ schemars = "0.8.6"
 secrecy = "0.8.0"
 serde = "1.0.130"
 serde_json = "1.0.68"
+serde-value = "0.7.0"
 serde_yaml = "0.9.19"
 syn = "2.0.38"
 tame-oauth = "0.10.0"

examples/Cargo.toml

@@ -183,6 +183,10 @@ path = "pod_reflector.rs"
 name = "pod_watcher"
 path = "pod_watcher.rs"
 
+[[example]]
+name = "errorbounded_configmap_watcher"
+path = "errorbounded_configmap_watcher.rs"
+
 [[example]]
 name = "request_raw"
 path = "request_raw.rs"

examples/errorbounded_configmap_watcher.rs

@@ -0,0 +1,84 @@
+use std::borrow::Cow;
+
+use futures::prelude::*;
+use k8s_openapi::{api::core::v1::Pod, NamespaceResourceScope};
+use kube::{
+    api::{Api, ObjectMeta, ResourceExt},
+    core::DeserializeGuard,
+    runtime::{reflector::ObjectRef, watcher, WatchStreamExt},
+    Client, Resource,
+};
+use serde::Deserialize;
+use tracing::*;
+
+// Variant of ConfigMap that only accepts ConfigMaps with a CA certificate
+// to demonstrate parsing failure
+#[derive(Deserialize, Debug, Clone)]
+struct CaConfigMap {
+    metadata: ObjectMeta,
+    data: CaConfigMapData,
+}
+
+#[derive(Deserialize, Debug, Clone)]
+struct CaConfigMapData {
+    #[serde(rename = "ca.crt")]
+    ca_crt: String,
+}
+
+// Normally you would derive this, but ConfigMap doesn't follow the standard spec/status pattern
+impl Resource for CaConfigMap {
+    type DynamicType = ();
+    type Scope = NamespaceResourceScope;
+
+    fn kind(&(): &Self::DynamicType) -> Cow<'_, str> {
+        Cow::Borrowed("ConfigMap")
+    }
+
+    fn group(&(): &Self::DynamicType) -> Cow<'_, str> {
+        Cow::Borrowed("")
+    }
+
+    fn version(&(): &Self::DynamicType) -> Cow<'_, str> {
+        Cow::Borrowed("v1")
+    }
+
+    fn plural(&(): &Self::DynamicType) -> Cow<'_, str> {
+        Cow::Borrowed("configmaps")
+    }
+
+    fn meta(&self) -> &ObjectMeta {
+        &self.metadata
+    }
+
+    fn meta_mut(&mut self) -> &mut ObjectMeta {
+        &mut self.metadata
+    }
+}
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    tracing_subscriber::fmt::init();
+    let client = Client::try_default().await?;
+    let api = Api::<DeserializeGuard<CaConfigMap>>::default_namespaced(client);
+    let use_watchlist = std::env::var("WATCHLIST").map(|s| s == "1").unwrap_or(false);
+    let wc = if use_watchlist {
+        // requires WatchList feature gate on 1.27 or later
+        watcher::Config::default().streaming_lists()
+    } else {
+        watcher::Config::default()
+    };
+
+    watcher(api, wc)
+        .applied_objects()
+        .default_backoff()
+        .try_for_each(|cm| async move {
+            info!("saw {}", ObjectRef::from_obj(&cm));
+            match cm.0 {
+                Ok(cm) => info!("contents: {cm:?}"),
+                Err(err) => warn!("failed to parse: {err}"),
+            }
+            Ok(())
+        })
+        .await?;
+    Ok(())
+}

kube-core/Cargo.toml

@@ -35,6 +35,7 @@ json-patch = { workspace = true, optional = true }
 chrono = { workspace = true, features = ["now"] }
 schemars = { workspace = true, optional = true }
 k8s-openapi.workspace = true
+serde-value.workspace = true
 
 [dev-dependencies]
 k8s-openapi = { workspace = true, features = ["latest"] }

kube-core/src/error_boundary.rs

@@ -0,0 +1,149 @@
+//! Types for isolating deserialization failures. See [`DeserializeGuard`].
+
+use std::{borrow::Cow, fmt::Display};
+
+use k8s_openapi::apimachinery::pkg::apis::meta::v1::ObjectMeta;
+use serde::Deserialize;
+use serde_value::DeserializerError;
+
+use crate::{PartialObjectMeta, Resource};
+
+/// A wrapper type for K that lets deserializing the parent object succeed, even if the K is invalid.
+///
+/// For example, this can be used to still access valid objects from an `Api::list` call or `watcher`.
+// We can't implement Deserialize on Result<K, InvalidObject> directly, both because of the orphan rule and because
+// it would conflict with serde's blanket impl on Result<K, E>, even if E isn't Deserialize.
+#[derive(Debug, Clone)]
+pub struct DeserializeGuard<K>(pub Result<K, InvalidObject>);
+
+/// An object that failed to be deserialized by the [`DeserializeGuard`].
+#[derive(Debug, Clone)]
+pub struct InvalidObject {
+    // Should ideally be D::Error, but we don't know what type it has outside of Deserialize::deserialize()
+    // It *could* be Box<std::error::Error>, but we don't know that it is Send+Sync
+    /// The error message from deserializing the object.
+    pub error: String,
+    /// The metadata of the invalid object.
+    pub metadata: ObjectMeta,
+}
+
+impl Display for InvalidObject {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        self.error.fmt(f)
+    }
+}
+
+impl<'de, K> Deserialize<'de> for DeserializeGuard<K>
+where
+    K: Deserialize<'de>,
+    // Not actually used, but we assume that K is a Kubernetes-style resource with a `metadata` section
+    K: Resource,
+{
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: serde::Deserializer<'de>,
+    {
+        // Deserialize::deserialize consumes the deserializer, and we want to retry parsing as an ObjectMetaContainer
+        // if the initial parse fails, so that we can still implement Resource for the error case
+        let buffer = serde_value::Value::deserialize(deserializer)?;
+
+        // FIXME: can we avoid cloning the whole object? metadata should be enough, and even then we could prune managedFields
+        K::deserialize(buffer.clone())
+            .map(Ok)
+            .or_else(|err| {
+                let PartialObjectMeta { metadata, .. } =
+                    PartialObjectMeta::<K>::deserialize(buffer).map_err(DeserializerError::into_error)?;
+                Ok(Err(InvalidObject {
+                    error: err.to_string(),
+                    metadata,
+                }))
+            })
+            .map(DeserializeGuard)
+    }
+}
+
+impl<K: Resource> Resource for DeserializeGuard<K> {
+    type DynamicType = K::DynamicType;
+    type Scope = K::Scope;
+
+    fn kind(dt: &Self::DynamicType) -> Cow<str> {
+        K::kind(dt)
+    }
+
+    fn group(dt: &Self::DynamicType) -> Cow<str> {
+        K::group(dt)
+    }
+
+    fn version(dt: &Self::DynamicType) -> Cow<str> {
+        K::version(dt)
+    }
+
+    fn plural(dt: &Self::DynamicType) -> Cow<str> {
+        K::plural(dt)
+    }
+
+    fn meta(&self) -> &ObjectMeta {
+        self.0.as_ref().map_or_else(|err| &err.metadata, K::meta)
+    }
+
+    fn meta_mut(&mut self) -> &mut ObjectMeta {
+        self.0.as_mut().map_or_else(|err| &mut err.metadata, K::meta_mut)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use k8s_openapi::api::core::v1::{ConfigMap, Pod};
+    use serde_json::json;
+
+    use crate::{DeserializeGuard, Resource};
+
+    #[test]
+    fn should_parse_meta_of_invalid_objects() {
+        let pod_error = serde_json::from_value::<DeserializeGuard<Pod>>(json!({
+            "metadata": {
+                "name": "the-name",
+                "namespace": "the-namespace",
+            },
+            "spec": {
+                "containers": "not-a-list",
+            },
+        }))
+        .unwrap();
+        assert_eq!(pod_error.meta().name.as_deref(), Some("the-name"));
+        assert_eq!(pod_error.meta().namespace.as_deref(), Some("the-namespace"));
+        pod_error.0.unwrap_err();
+    }
+
+    #[test]
+    fn should_allow_valid_objects() {
+        let configmap = serde_json::from_value::<DeserializeGuard<ConfigMap>>(json!({
+            "metadata": {
+                "name": "the-name",
+                "namespace": "the-namespace",
+            },
+            "data": {
+                "foo": "bar",
+            },
+        }))
+        .unwrap();
+        assert_eq!(configmap.meta().name.as_deref(), Some("the-name"));
+        assert_eq!(configmap.meta().namespace.as_deref(), Some("the-namespace"));
+        assert_eq!(
+            configmap.0.unwrap().data,
+            Some([("foo".to_string(), "bar".to_string())].into())
+        )
+    }
+
+    #[test]
+    fn should_catch_invalid_objects() {
+        serde_json::from_value::<DeserializeGuard<Pod>>(json!({
+            "spec": {
+                "containers": "not-a-list"
+            }
+        }))
+        .unwrap()
+        .0
+        .unwrap_err();
+    }
+}

kube-core/src/lib.rs

@@ -70,3 +70,6 @@ pub use error::ErrorResponse;
 
 mod version;
 pub use version::Version;
+
+pub mod error_boundary;
+pub use error_boundary::DeserializeGuard;