Unify API error types

examples/crd_api.rs

@@ -72,8 +72,8 @@ async fn main() -> anyhow::Result<()> {
             info!("Created {} ({:?})", Meta::name(&o), o.status.unwrap());
             debug!("Created CRD: {:?}", o.spec);
         }
-        Err(kube::Error::Api(ae)) => assert_eq!(ae.code, 409), // if you skipped delete, for instance
-        Err(e) => return Err(e.into()),                        // any other case is probably bad
+        Err(kube::Error::Api(ae)) => assert_eq!(ae.code, Some(409)), // if you skipped delete, for instance
+        Err(e) => return Err(e.into()),                              // any other case is probably bad
     }
     // Wait for the api to catch up
     sleep(Duration::from_secs(1)).await;

examples/crd_derive_schema.rs

@@ -203,12 +203,12 @@ async fn main() -> Result<()> {
     assert!(res.is_err());
     match res.err() {
         Some(kube::Error::Api(err)) => {
-            assert_eq!(err.code, 422);
-            assert_eq!(err.reason, "Invalid");
-            assert_eq!(err.status, "Failure");
+            assert_eq!(err.code, Some(422));
+            assert_eq!(err.reason.as_deref(), Some("Invalid"));
+            assert_eq!(err.status.as_deref(), Some("Failure"));
             assert_eq!(
-                err.message,
-                "Foo.clux.dev \"qux\" is invalid: spec.non_nullable: Required value"
+                err.message.as_deref(),
+                Some("Foo.clux.dev \"qux\" is invalid: spec.non_nullable: Required value")
             );
         }
         _ => assert!(false),

examples/job_api.rs

@@ -64,7 +64,7 @@ async fn main() -> anyhow::Result<()> {
                 }
             }
             WatchEvent::Deleted(s) => info!("Deleted {}", Meta::name(&s)),
-            WatchEvent::Error(s) => error!("{}", s),
+            WatchEvent::Error(s) => error!("{:?}", s),
             _ => {}
         }
     }

examples/pod_api.rs

@@ -41,8 +41,8 @@ async fn main() -> anyhow::Result<()> {
             // wait for it..
             std::thread::sleep(std::time::Duration::from_millis(5_000));
         }
-        Err(kube::Error::Api(ae)) => assert_eq!(ae.code, 409), // if you skipped delete, for instance
-        Err(e) => return Err(e.into()),                        // any other case is probably bad
+        Err(kube::Error::Api(ae)) => assert_eq!(ae.code, Some(409)), // if you skipped delete, for instance
+        Err(e) => return Err(e.into()),                              // any other case is probably bad
     }
 
     // Watch it phase for a few seconds
@@ -59,7 +59,7 @@ async fn main() -> anyhow::Result<()> {
                 info!("Modified: {} with phase: {}", Meta::name(&o), phase);
             }
             WatchEvent::Deleted(o) => info!("Deleted {}", Meta::name(&o)),
-            WatchEvent::Error(e) => error!("Error {}", e),
+            WatchEvent::Error(e) => error!("Error {:?}", e),
             _ => {}
         }
     }

kube-runtime/src/watcher.rs

@@ -2,6 +2,7 @@
 
 use derivative::Derivative;
 use futures::{stream::BoxStream, Stream, StreamExt};
+use k8s_openapi::apimachinery::pkg::apis::meta::v1::Status;
 use kube::{
     api::{ListParams, Meta, WatchEvent},
     Api,
@@ -23,11 +24,8 @@ pub enum Error {
         source: kube::Error,
         backtrace: Backtrace,
     },
-    #[snafu(display("error returned by apiserver during watch: {}", source))]
-    WatchError {
-        source: kube::error::ErrorResponse,
-        backtrace: Backtrace,
-    },
+    #[snafu(display("error returned by apiserver during watch: {:?}", status))]
+    WatchError { status: Status, backtrace: Backtrace },
     #[snafu(display("watch stream failed: {}", source))]
     WatchFailed {
         source: kube::Error,
@@ -153,15 +151,15 @@ async fn step_trampolined<K: Meta + Clone + DeserializeOwned + Send + 'static>(
             }),
             Some(Ok(WatchEvent::Error(err))) => {
                 // HTTP GONE, means we have desynced and need to start over and re-list :(
-                let new_state = if err.code == 410 {
+                let new_state = if err.code == Some(410) {
                     State::Empty
                 } else {
                     State::Watching {
                         resource_version,
                         stream,
                     }
                 };
-                (Some(Err(err).context(WatchError)), new_state)
+                (Some(WatchError { status: err }.fail()), new_state)
             }
             Some(Err(err)) => (Some(Err(err).context(WatchFailed)), State::Watching {
                 resource_version,

kube/src/api/object.rs

@@ -1,7 +1,5 @@
-use crate::{
-    api::metadata::{ListMeta, Meta, ObjectMeta, TypeMeta},
-    error::ErrorResponse,
-};
+use crate::api::metadata::{ListMeta, Meta, ObjectMeta, TypeMeta};
+use k8s_openapi::apimachinery::pkg::apis::meta::v1::Status;
 use serde::{Deserialize, Serialize};
 use std::fmt::Debug;
 
@@ -27,7 +25,7 @@ where
     /// NB: This became Beta first in Kubernetes 1.16.
     Bookmark(Bookmark),
     /// There was some kind of error
-    Error(ErrorResponse),
+    Error(Status),
 }
 
 impl<K> Debug for WatchEvent<K>

kube/src/api/subresource.rs

@@ -1,10 +1,10 @@
 use bytes::Bytes;
 use futures::Stream;
+use k8s_openapi::apimachinery::pkg::apis::meta::v1::Status;
 use serde::de::DeserializeOwned;
 
 use crate::{
     api::{Api, DeleteParams, Patch, PatchParams, PostParams, Resource},
-    client::Status,
     Error, Result,
 };
 

kube/src/api/typed.rs

@@ -1,11 +1,12 @@
 use either::Either;
 use futures::Stream;
+use k8s_openapi::apimachinery::pkg::apis::meta::v1::Status;
 use serde::{de::DeserializeOwned, Serialize};
 use std::iter;
 
 use crate::{
     api::{DeleteParams, ListParams, Meta, ObjectList, Patch, PatchParams, PostParams, Resource, WatchEvent},
-    client::{Client, Status},
+    client::Client,
     Result,
 };
 
@@ -309,7 +310,7 @@ where
     ///             WatchEvent::Modified(s) => println!("Modified: {}", Meta::name(&s)),
     ///             WatchEvent::Deleted(s) => println!("Deleted {}", Meta::name(&s)),
     ///             WatchEvent::Bookmark(s) => {},
-    ///             WatchEvent::Error(s) => println!("{}", s),
+    ///             WatchEvent::Error(s) => println!("{:?}", s),
     ///         }
     ///     }
     ///     Ok(())

kube/src/client/mod.rs

@@ -9,7 +9,6 @@
 use crate::{
     api::{Meta, WatchEvent},
     config::Config,
-    error::ErrorResponse,
     service::Service,
     Error, Result,
 };
@@ -22,8 +21,8 @@ use either::{Either, Left, Right};
 use futures::{self, Stream, StreamExt, TryStream, TryStreamExt};
 use http::{self, Request, Response, StatusCode};
 use hyper::Body;
-use k8s_openapi::apimachinery::pkg::apis::meta::v1 as k8s_meta_v1;
-use serde::{de::DeserializeOwned, Deserialize};
+use k8s_openapi::apimachinery::pkg::apis::meta::v1::{self as k8s_meta_v1, Status};
+use serde::de::DeserializeOwned;
 use serde_json::{self, Value};
 use tokio_util::{
     codec::{FramedRead, LinesCodec, LinesCodecError},
@@ -239,7 +238,7 @@ impl Client {
                         }
 
                         // Got general error response
-                        if let Ok(e_resp) = serde_json::from_str::<ErrorResponse>(&line) {
+                        if let Ok(e_resp) = serde_json::from_str::<Status>(&line) {
                             return Some(Err(Error::Api(e_resp)));
                         }
                         // Parsing error
@@ -328,17 +327,19 @@ fn handle_api_errors(text: &str, s: StatusCode) -> Result<()> {
     if s.is_client_error() || s.is_server_error() {
         // Print better debug when things do fail
         // trace!("Parsing error: {}", text);
-        if let Ok(errdata) = serde_json::from_str::<ErrorResponse>(text) {
+        if let Ok(errdata) = serde_json::from_str::<Status>(text) {
             debug!("Unsuccessful: {:?}", errdata);
             Err(Error::Api(errdata))
         } else {
             warn!("Unsuccessful data error parse: {}", text);
             // Propagate errors properly via reqwest
-            let ae = ErrorResponse {
-                status: s.to_string(),
-                code: s.as_u16(),
-                message: format!("{:?}", text),
-                reason: "Failed to parse error data".into(),
+            let ae = Status {
+                metadata: Default::default(),
+                status: Some(s.to_string()),
+                code: Some(s.as_u16().into()),
+                message: Some(format!("{:?}", text)),
+                reason: Some("Failed to parse error data".into()),
+                details: None,
             };
             debug!("Unsuccessful: {:?} (reconstruct)", ae);
             Err(Error::Api(ae))
@@ -357,74 +358,6 @@ impl TryFrom<Config> for Client {
     }
 }
 
-// TODO: replace with Status in k8s openapi?
-
-/// A Kubernetes status object
-#[allow(missing_docs)]
-#[derive(Deserialize, Debug)]
-pub struct Status {
-    // TODO: typemeta
-    // TODO: metadata that can be completely empty (listmeta...)
-    #[serde(default, skip_serializing_if = "String::is_empty")]
-    pub status: String,
-    #[serde(default, skip_serializing_if = "String::is_empty")]
-    pub message: String,
-    #[serde(default, skip_serializing_if = "String::is_empty")]
-    pub reason: String,
-    #[serde(default, skip_serializing_if = "Option::is_none")]
-    pub details: Option<StatusDetails>,
-    #[serde(default, skip_serializing_if = "num::Zero::is_zero")]
-    pub code: u16,
-}
-
-/// Status details object on the [`Status`] object
-#[derive(Deserialize, Debug)]
-#[serde(rename_all = "camelCase")]
-#[allow(missing_docs)]
-pub struct StatusDetails {
-    #[serde(default, skip_serializing_if = "String::is_empty")]
-    pub name: String,
-    #[serde(default, skip_serializing_if = "String::is_empty")]
-    pub group: String,
-    #[serde(default, skip_serializing_if = "String::is_empty")]
-    pub kind: String,
-    #[serde(default, skip_serializing_if = "String::is_empty")]
-    pub uid: String,
-    #[serde(default, skip_serializing_if = "Vec::is_empty")]
-    pub causes: Vec<StatusCause>,
-    #[serde(default, skip_serializing_if = "num::Zero::is_zero")]
-    pub retry_after_seconds: u32,
-}
-
-/// Status cause object on the [`StatusDetails`] object
-#[derive(Deserialize, Debug)]
-#[allow(missing_docs)]
-pub struct StatusCause {
-    #[serde(default, skip_serializing_if = "String::is_empty")]
-    pub reason: String,
-    #[serde(default, skip_serializing_if = "String::is_empty")]
-    pub message: String,
-    #[serde(default, skip_serializing_if = "String::is_empty")]
-    pub field: String,
-}
-
-#[cfg(test)]
-mod test {
-    use super::Status;
-
-    // ensure our status schema is sensible
-    #[test]
-    fn delete_deserialize_test() {
-        let statusresp = r#"{"kind":"Status","apiVersion":"v1","metadata":{},"status":"Success","details":{"name":"some-app","group":"clux.dev","kind":"foos","uid":"1234-some-uid"}}"#;
-        let s: Status = serde_json::from_str::<Status>(statusresp).unwrap();
-        assert_eq!(s.details.unwrap().name, "some-app");
-
-        let statusnoname = r#"{"kind":"Status","apiVersion":"v1","metadata":{},"status":"Success","details":{"group":"clux.dev","kind":"foos","uid":"1234-some-uid"}}"#;
-        let s2: Status = serde_json::from_str::<Status>(statusnoname).unwrap();
-        assert_eq!(s2.details.unwrap().name, ""); // optional probably better..
-    }
-}
-
 #[cfg(feature = "ws")]
 // Verify upgrade response according to RFC6455.
 // Based on `tungstenite` and added subprotocol verification.

kube/src/error.rs

@@ -1,7 +1,7 @@
 //! Error handling in [`kube`][crate]
 
 use http::header::InvalidHeaderValue;
-use serde::{Deserialize, Serialize};
+use k8s_openapi::apimachinery::pkg::apis::meta::v1::Status;
 use std::path::PathBuf;
 use thiserror::Error;
 
@@ -14,8 +14,8 @@ pub enum Error {
     /// It's also used in `WatchEvent` from watch calls.
     ///
     /// It's quite common to get a `410 Gone` when the `resourceVersion` is too old.
-    #[error("ApiError: {0} ({0:?})")]
-    Api(#[source] ErrorResponse),
+    #[error("ApiError: {0:?}")]
+    Api(Status),
 
     /// ConnectionError for when TcpStream fails to connect.
     #[error("ConnectionError: {0}")]
@@ -261,19 +261,3 @@ impl From<OAuthError> for Error {
         ConfigError::OAuth(e).into()
     }
 }
-
-/// An error response from the API.
-#[derive(Error, Deserialize, Serialize, Debug, Clone, Eq, PartialEq)]
-#[error("{message}: {reason}")]
-pub struct ErrorResponse {
-    /// The status
-    pub status: String,
-    /// A message about the error
-    #[serde(default)]
-    pub message: String,
-    /// The reason for the error
-    #[serde(default)]
-    pub reason: String,
-    /// The error code
-    pub code: u16,
-}

kube/src/lib.rs

@@ -61,7 +61,7 @@
 //!                 println!("Modified: {} with phase: {}", Meta::name(&o), phase);
 //!             }
 //!             WatchEvent::Deleted(o) => println!("Deleted {}", Meta::name(&o)),
-//!             WatchEvent::Error(e) => println!("Error {}", e),
+//!             WatchEvent::Error(e) => println!("Error {:?}", e),
 //!             _ => {}
 //!         }
 //!     }

kube/src/runtime/informer.rs

@@ -142,7 +142,7 @@ where
                     }
                     Ok(WatchEvent::Error(e)) => {
                         // 410 Gone => we need to restart from latest next call
-                        if e.code == 410 {
+                        if e.code == Some(410) {
                             warn!("Stream desynced: {:?}", e);
                             *needs_resync.lock().await = true;
                         }

tests/dapp.rs

@@ -64,7 +64,7 @@ async fn main() -> anyhow::Result<()> {
                 }
             }
             WatchEvent::Deleted(s) => info!("Deleted {}", Meta::name(&s)),
-            WatchEvent::Error(s) => error!("{}", s),
+            WatchEvent::Error(s) => error!("{:?}", s),
             _ => {}
         }
     }