Add support for metadata API (#1137)

* Add support for metadata API

This change adds support for the (undocumented) metadata API. Through
this change, clients may request only the metadata of objects, reducing
I/O and storage pressure. To make it possible, the change adds new
request impl in kube-core, and new metadata methods on the Api type.

Additionally, this change also adds a new 'PartialObjectMeta' type that
is supposed to mirror the Go counterpart. A wrapper list has also been
added. Both of these support conversions to ObjectMeta (or
ObjectList<ObjectMeta>).

Metadata-only requests are made possible through the REST API by
including an extended Accept header. The header will notify the API
Server a JSON response is expected, and it will encode an additional
'as=PartialObjectMetadata' k=v pair to receive only the metadata of the
object. This is more efficient when only the metadata needs to be parsed
(as opposed to the whole spec).

Note: Older servers (k8s v1.14 and below) will retrieve the object and
then convert the metadata

Signed-off-by: Matei David <matei@buoyant.io>

* Run fmt

Signed-off-by: Matei David <matei@buoyant.io>

* Pass request unit tests

Signed-off-by: Matei David <matei@buoyant.io>

* Add docs for new types and methods

Signed-off-by: Matei David <matei@buoyant.io>

* Add list and patch integration tests

Signed-off-by: Matei David <matei@buoyant.io>

* Address feedback and fix doc tests

Signed-off-by: Matei David <matei@buoyant.io>

* Update kube-core/src/metadata.rs

Co-authored-by: Eirik A <sszynrae@gmail.com>
Signed-off-by: Matei David <matei.david.35@gmail.com>

* Add Resource impl for PartialObjectMeta

Signed-off-by: Matei David <matei@buoyant.io>

* Fix stale comment that referred to PartialObjectMetaList

Signed-off-by: Matei David <matei@buoyant.io>

* Run fmt

Signed-off-by: Matei David <matei@buoyant.io>

---------

Signed-off-by: Matei David <matei@buoyant.io>
Signed-off-by: Matei David <matei.david.35@gmail.com>
Co-authored-by: Eirik A <sszynrae@gmail.com>

kube-client/src/api/core_methods.rs

@@ -4,7 +4,9 @@ use serde::{de::DeserializeOwned, Serialize};
 use std::fmt::Debug;
 
 use crate::{api::Api, Error, Result};
-use kube_core::{object::ObjectList, params::*, response::Status, ErrorResponse, WatchEvent};
+use kube_core::{
+    metadata::PartialObjectMeta, object::ObjectList, params::*, response::Status, ErrorResponse, WatchEvent,
+};
 
 /// PUSH/PUT/POST/GET abstractions
 impl<K> Api<K>
@@ -35,6 +37,34 @@ where
         self.client.request::<K>(req).await
     }
 
+    ///  Get only the metadata for a named resource as
+    ///  [`kube_core::metadata::PartialObjectMeta`]
+    ///
+    ///
+    /// ```no_run
+    /// use kube::{Api, Client, core::metadata::PartialObjectMeta};
+    /// use k8s_openapi::api::core::v1::Pod;
+    /// #[tokio::main]
+    /// async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    ///     let client = Client::try_default().await?;
+    ///     let pods: Api<Pod> = Api::namespaced(client, "apps");
+    ///     let p: PartialObjectMeta = pods.get_metadata("blog").await?;
+    ///     Ok(())
+    /// }
+    /// ```
+    /// Note that the type may be converted to `ObjectMeta` through the usual
+    /// conversion traits.
+    ///
+    /// # Errors
+    ///
+    /// This function assumes that the object is expected to always exist, and returns [`Error`] if it does not.
+    /// Consider using [`Api::get_metadata_opt`] if you need to handle missing objects.
+    pub async fn get_metadata(&self, name: &str) -> Result<PartialObjectMeta> {
+        let mut req = self.request.get_metadata(name).map_err(Error::BuildRequest)?;
+        req.extensions_mut().insert("get");
+        self.client.request::<PartialObjectMeta>(req).await
+    }
+
     /// [Get](`Api::get`) a named resource if it exists, returns [`None`] if it doesn't exist
     ///
     /// ```no_run
@@ -60,6 +90,35 @@ where
         }
     }
 
+    /// [Get PartialObjectMeta](`Api::get_metadata`) for a named resource if it
+    /// exists, returns [`None`] if it doesn't exit
+    ///
+    /// ```no_run
+    /// use kube::{Api, Client};
+    /// use k8s_openapi::api::core::v1::Pod;
+    /// #[tokio::main]
+    /// async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    ///     let client = Client::try_default().await?;
+    ///     let pods: Api<Pod> = Api::namespaced(client, "apps");
+    ///     if let Some(pod) = pods.get_metadata_opt("blog").await? {
+    ///         // Pod was found
+    ///     } else {
+    ///         // Pod was not found
+    ///     }
+    ///     Ok(())
+    /// }
+    /// ```
+    ///
+    /// Note that [kube_core::metadata::PartialObjectMeta] may be converted to `ObjectMeta`
+    /// through the usual conversion traits.
+    pub async fn get_metadata_opt(&self, name: &str) -> Result<Option<PartialObjectMeta>> {
+        match self.get_metadata(name).await {
+            Ok(meta) => Ok(Some(meta)),
+            Err(Error::Api(ErrorResponse { reason, .. })) if &reason == "NotFound" => Ok(None),
+            Err(err) => Err(err),
+        }
+    }
+
     /// Get a list of resources
     ///
     /// You use this to get everything, or a subset matching fields/labels, say:
@@ -84,6 +143,33 @@ where
         self.client.request::<ObjectList<K>>(req).await
     }
 
+    /// Get a list of resources that contains only their metadata as
+    ///
+    /// Similar to [list](`Api::list`), you use this to get everything, or a
+    /// subset matching fields/labels. For example
+    ///
+    /// ```no_run
+    /// use kube::{core::{metadata::PartialObjectMeta, object::ObjectList}, api::{Api, ListParams, ResourceExt}, Client};
+    /// use k8s_openapi::{apimachinery::pkg::apis::meta::v1::ObjectMeta, api::core::v1::Pod};
+    /// #[tokio::main]
+    /// async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    ///     let client = Client::try_default().await?;
+    ///     let pods: Api<Pod> = Api::namespaced(client, "apps");
+    ///     let lp = ListParams::default().labels("app=blog"); // for this app only
+    ///     let list: ObjectList<PartialObjectMeta> = pods.list_metadata(&lp).await?;
+    ///     for p in list {
+    ///         let metadata = ObjectMeta::from(p);
+    ///         println!("Found Pod: {}", metadata.name.unwrap());
+    ///     }
+    ///     Ok(())
+    /// }
+    /// ```
+    pub async fn list_metadata(&self, lp: &ListParams) -> Result<ObjectList<PartialObjectMeta>> {
+        let mut req = self.request.list_metadata(lp).map_err(Error::BuildRequest)?;
+        req.extensions_mut().insert("list");
+        self.client.request::<ObjectList<PartialObjectMeta>>(req).await
+    }
+
     /// Create a resource
     ///
     /// This function requires a type that Serializes to `K`, which can be:
@@ -224,6 +310,57 @@ where
         self.client.request::<K>(req).await
     }
 
+    /// Patch a subset of a resource's properties and get back the resource
+    /// metadata as [`kube_core::metadata::PartialObjectMeta`]
+    ///
+    /// Takes a [`Patch`] along with [`PatchParams`] for the call.
+    ///
+    /// ```no_run
+    /// use kube::{api::{Api, PatchParams, Patch, Resource}, Client};
+    /// use k8s_openapi::api::core::v1::Pod;
+    /// #[tokio::main]
+    /// async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    ///     let client = Client::try_default().await?;
+    ///     let pods: Api<Pod> = Api::namespaced(client, "apps");
+    ///     let patch = serde_json::json!({
+    ///         "apiVersion": "v1",
+    ///         "kind": "Pod",
+    ///         "metadata": {
+    ///             "name": "blog",
+    ///             "labels": {
+    ///                 "key": "value"
+    ///             },
+    ///         },
+    ///         "spec": {
+    ///             "activeDeadlineSeconds": 5
+    ///         }
+    ///     });
+    ///     let params = PatchParams::apply("myapp");
+    ///     let patch = Patch::Apply(&patch);
+    ///     let o_patched = pods.patch_metadata("blog", &params, &patch).await?;
+    ///     println!("Patched {}", o_patched.metadata.name.unwrap());
+    ///     Ok(())
+    /// }
+    /// ```
+    /// [`Patch`]: super::Patch
+    /// [`PatchParams`]: super::PatchParams
+    ///
+    /// Note that this method cannot write to the status object (when it exists) of a resource.
+    /// To set status objects please see [`Api::replace_status`] or [`Api::patch_status`].
+    pub async fn patch_metadata<P: Serialize + Debug>(
+        &self,
+        name: &str,
+        pp: &PatchParams,
+        patch: &Patch<P>,
+    ) -> Result<PartialObjectMeta> {
+        let mut req = self
+            .request
+            .patch_metadata(name, pp, patch)
+            .map_err(Error::BuildRequest)?;
+        req.extensions_mut().insert("patch");
+        self.client.request::<PartialObjectMeta>(req).await
+    }
+
     /// Replace a resource entirely with a new one
     ///
     /// This is used just like [`Api::create`], but with one additional instruction:
@@ -330,4 +467,55 @@ where
         req.extensions_mut().insert("watch");
         self.client.request_events::<K>(req).await
     }
+
+    /// Watch a list of metadata for a given resources
+    ///
+    /// This returns a future that awaits the initial response,
+    /// then you can stream the remaining buffered `WatchEvent` objects.
+    ///
+    /// Note that a `watch_metadata` call can terminate for many reasons (even
+    /// before the specified [`ListParams::timeout`] is triggered), and will
+    /// have to be re-issued with the last seen resource version when or if it
+    /// closes.
+    ///
+    /// Consider using a managed [`watcher`] to deal with automatic re-watches and error cases.
+    ///
+    /// ```no_run
+    /// use kube::{api::{Api, ListParams, ResourceExt, WatchEvent}, Client};
+    /// use k8s_openapi::api::batch::v1::Job;
+    /// use futures::{StreamExt, TryStreamExt};
+    /// #[tokio::main]
+    /// async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    ///     let client = Client::try_default().await?;
+    ///     let jobs: Api<Job> = Api::namespaced(client, "apps");
+    ///     let lp = ListParams::default()
+    ///         .fields("metadata.name=my_job")
+    ///         .timeout(20); // upper bound of how long we watch for
+    ///     let mut stream = jobs.watch(&lp, "0").await?.boxed();
+    ///     while let Some(status) = stream.try_next().await? {
+    ///         match status {
+    ///             WatchEvent::Added(s) => println!("Added {}", s.metadata.name.unwrap()),
+    ///             WatchEvent::Modified(s) => println!("Modified: {}", s.metadata.name.unwrap()),
+    ///             WatchEvent::Deleted(s) => println!("Deleted {}", s.metadata.name.unwrap()),
+    ///             WatchEvent::Bookmark(s) => {},
+    ///             WatchEvent::Error(s) => println!("{}", s),
+    ///         }
+    ///     }
+    ///     Ok(())
+    /// }
+    /// ```
+    /// [`ListParams::timeout`]: super::ListParams::timeout
+    /// [`watcher`]: https://docs.rs/kube_runtime/*/kube_runtime/watcher/fn.watcher.html
+    pub async fn watch_metadata(
+        &self,
+        lp: &ListParams,
+        version: &str,
+    ) -> Result<impl Stream<Item = Result<WatchEvent<PartialObjectMeta>>>> {
+        let mut req = self
+            .request
+            .watch_metadata(lp, version)
+            .map_err(Error::BuildRequest)?;
+        req.extensions_mut().insert("watch");
+        self.client.request_events::<PartialObjectMeta>(req).await
+    }
 }

kube-client/src/lib.rs

@@ -469,6 +469,92 @@ mod test {
         Ok(())
     }
 
+    #[tokio::test]
+    #[ignore] // requires a cluster
+    async fn can_operate_on_pod_metadata() -> Result<(), Box<dyn std::error::Error>> {
+        use crate::{
+            api::{DeleteParams, EvictParams, ListParams, Patch, PatchParams, WatchEvent},
+            core::subresource::LogParams,
+        };
+        use kube_core::{ObjectList, ObjectMeta};
+
+        let client = Client::try_default().await?;
+        let pods: Api<Pod> = Api::default_namespaced(client);
+
+        // create busybox pod that's alive for at most 30s
+        let p: Pod = serde_json::from_value(json!({
+            "apiVersion": "v1",
+            "kind": "Pod",
+            "metadata": {
+                "name": "busybox-kube-meta",
+                "labels": { "app": "kube-rs-test" },
+            },
+            "spec": {
+                "terminationGracePeriodSeconds": 1,
+                "restartPolicy": "Never",
+                "containers": [{
+                  "name": "busybox",
+                  "image": "busybox:1.34.1",
+                  "command": ["sh", "-c", "sleep 30s"],
+                }],
+            }
+        }))?;
+
+        match pods.create(&Default::default(), &p).await {
+            Ok(o) => assert_eq!(p.name_unchecked(), o.name_unchecked()),
+            Err(crate::Error::Api(ae)) => assert_eq!(ae.code, 409), // if we failed to clean-up
+            Err(e) => return Err(e.into()),                         // any other case if a failure
+        }
+
+        // Test we can get a pod as a PartialObjectMeta and convert to
+        // ObjectMeta
+        let pod_metadata = pods.get_metadata("busybox-kube-meta").await?;
+        assert_eq!("busybox-kube-meta", pod_metadata.name_any());
+        assert_eq!(
+            Some((&"app".to_string(), &"kube-rs-test".to_string())),
+            pod_metadata.labels().get_key_value("app")
+        );
+
+        // Test we can get a list of PartialObjectMeta for pods
+        let p_list = pods.list_metadata(&ListParams::default()).await?;
+
+        // Find only pod we are concerned with in this test and fail eagerly if
+        // name doesn't exist
+        let pod_metadata = p_list
+            .items
+            .into_iter()
+            .find(|p| p.name_any() == "busybox-kube-meta")
+            .unwrap();
+        assert_eq!(
+            pod_metadata.labels().get("app"),
+            Some(&"kube-rs-test".to_string())
+        );
+
+        // Attempt to patch pod
+        let patch = json!({
+            "metadata": {
+                "annotations": {
+                    "test": "123"
+                },
+            },
+            "spec": {
+                "activeDeadlineSeconds": 5
+            }
+        });
+        let patchparams = PatchParams::default();
+        let p_patched = pods
+            .patch_metadata("busybox-kube-meta", &patchparams, &Patch::Merge(&patch))
+            .await?;
+        assert_eq!(p_patched.annotations().get("test"), Some(&"123".to_string()));
+
+        // Clean-up
+        let dp = DeleteParams::default();
+        pods.delete("busybox-kube-meta", &dp).await?.map_left(|pdel| {
+            assert_eq!(pdel.name_any(), "busybox-kube-meta");
+        });
+
+        Ok(())
+    }
     #[tokio::test]
     #[ignore] // needs cluster (will create a CertificateSigningRequest)
     async fn csr_can_be_approved() -> Result<(), Box<dyn std::error::Error>> {

kube-core/src/metadata.rs

@@ -1,7 +1,11 @@
 //! Metadata structs used in traits, lists, and dynamic objects.
+use std::borrow::Cow;
+
 pub use k8s_openapi::apimachinery::pkg::apis::meta::v1::{ListMeta, ObjectMeta};
 use serde::{Deserialize, Serialize};
 
+use crate::{ApiResource, DynamicResourceScope, Resource};
+
 /// Type information that is flattened into every kubernetes object
 #[derive(Deserialize, Serialize, Clone, Default, Debug, Eq, PartialEq, Hash)]
 #[serde(rename_all = "camelCase")]
@@ -12,3 +16,53 @@ pub struct TypeMeta {
     /// The name of the API
     pub kind: String,
 }
+
+/// A generic representation of any object with `ObjectMeta`.
+///
+/// It allows clients to get access to a particular `ObjectMeta`
+/// schema without knowing the details of the version.
+#[derive(Deserialize, Serialize, Clone, Default, Debug)]
+#[serde(rename_all = "camelCase")]
+pub struct PartialObjectMeta {
+    /// The type fields, not always present
+    #[serde(flatten, default)]
+    pub types: Option<TypeMeta>,
+    /// Standard object's metadata
+    #[serde(default)]
+    pub metadata: ObjectMeta,
+}
+
+impl From<PartialObjectMeta> for ObjectMeta {
+    fn from(obj: PartialObjectMeta) -> Self {
+        ObjectMeta { ..obj.metadata }
+    }
+}
+
+impl Resource for PartialObjectMeta {
+    type DynamicType = ApiResource;
+    type Scope = DynamicResourceScope;
+
+    fn kind(dt: &ApiResource) -> Cow<'_, str> {
+        dt.kind.as_str().into()
+    }
+
+    fn group(dt: &ApiResource) -> Cow<'_, str> {
+        dt.group.as_str().into()
+    }
+
+    fn version(dt: &ApiResource) -> Cow<'_, str> {
+        dt.version.as_str().into()
+    }
+
+    fn plural(dt: &ApiResource) -> Cow<'_, str> {
+        dt.plural.as_str().into()
+    }
+
+    fn meta(&self) -> &ObjectMeta {
+        &self.metadata
+    }
+
+    fn meta_mut(&mut self) -> &mut ObjectMeta {
+        &mut self.metadata
+    }
+}