From 610d89bbb9992deb8f9eef42af9b769af6f06083 Mon Sep 17 00:00:00 2001
From: clux <sszynrae@gmail.com>
Date: Tue, 7 Apr 2020 07:42:40 +0100
Subject: [PATCH] add watch bookmarks - fixes #54

also for #219
---
 kube/examples/configmap_informer.rs |  4 ++--
 kube/src/api/object.rs              |  6 ++++++
 kube/src/api/resource.rs            | 10 ++++++++++
 kube/src/runtime/informer.rs        |  5 ++++-
 kube/src/runtime/reflector.rs       |  3 +++
 5 files changed, 25 insertions(+), 3 deletions(-)

diff --git a/kube/examples/configmap_informer.rs b/kube/examples/configmap_informer.rs
index 62bd1de47..505d2d25c 100644
--- a/kube/examples/configmap_informer.rs
+++ b/kube/examples/configmap_informer.rs
@@ -10,13 +10,13 @@ use kube::{
 /// Example way to read secrets
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
-    std::env::set_var("RUST_LOG", "info,kube=debug");
+    std::env::set_var("RUST_LOG", "info,kube=trace");
     env_logger::init();
     let client = Client::try_default().await?;
     let namespace = std::env::var("NAMESPACE").unwrap_or("default".into());
 
     let cms: Api<ConfigMap> = Api::namespaced(client, &namespace);
-    let lp = ListParams::default().timeout(10); // short watch timeout in this example
+    let lp = ListParams::default().allow_bookmarks().timeout(10); // short watch timeout in this example
     let inf = Informer::new(cms).params(lp);
 
     loop {
diff --git a/kube/src/api/object.rs b/kube/src/api/object.rs
index faf3ae125..61f4f1a47 100644
--- a/kube/src/api/object.rs
+++ b/kube/src/api/object.rs
@@ -20,6 +20,11 @@ where
     Modified(K),
     /// Resource was deleted
     Deleted(K),
+    /// Resource bookmark
+    ///
+    /// From [Watch bookmarks](https://kubernetes.io/docs/reference/using-api/api-concepts/#watch-bookmarks)
+    /// NB: This became Beta first in Kubernetes 1.16
+    Bookmark(K),
     /// There was some kind of error
     Error(ErrorResponse),
 }
@@ -33,6 +38,7 @@ where
             WatchEvent::Added(_) => write!(f, "Added event"),
             WatchEvent::Modified(_) => write!(f, "Modified event"),
             WatchEvent::Deleted(_) => write!(f, "Deleted event"),
+            WatchEvent::Bookmark(_) => write!(f, "Bookmark event"),
             WatchEvent::Error(e) => write!(f, "Error event: {:?}", e),
         }
     }
diff --git a/kube/src/api/resource.rs b/kube/src/api/resource.rs
index a9bba1863..f3014e7dd 100644
--- a/kube/src/api/resource.rs
+++ b/kube/src/api/resource.rs
@@ -89,6 +89,7 @@ pub struct ListParams {
     pub include_uninitialized: bool,
     pub label_selector: Option<String>,
     pub timeout: Option<u32>,
+    pub allow_bookmarks: bool,
 }
 
 impl ListParams {
@@ -148,6 +149,12 @@ impl ListParams {
         self.include_uninitialized = true;
         self
     }
+
+    /// Enables watch bookmarks from the api server if supported
+    pub fn allow_bookmarks(mut self) -> Self {
+        self.allow_bookmarks = true;
+        self
+    }
 }
 
 // TODO: WatchParams (same as ListParams but with extra resource_version + allow_watch_bookmarks)
@@ -318,6 +325,9 @@ impl Resource {
         if let Some(labels) = &lp.label_selector {
             qp.append_pair("labelSelector", &labels);
         }
+        if lp.allow_bookmarks {
+            qp.append_pair("allowWatchBookmarks", "true");
+        }
 
         let urlstr = qp.finish();
         let req = http::Request::get(urlstr);
diff --git a/kube/src/runtime/informer.rs b/kube/src/runtime/informer.rs
index 2f2662fa4..4db298211 100644
--- a/kube/src/runtime/informer.rs
+++ b/kube/src/runtime/informer.rs
@@ -126,7 +126,10 @@ where
             async move {
                 // Check if we need to update our version based on the incoming events
                 match &event {
-                    Ok(WatchEvent::Added(o)) | Ok(WatchEvent::Modified(o)) | Ok(WatchEvent::Deleted(o)) => {
+                    Ok(WatchEvent::Added(o))
+                    | Ok(WatchEvent::Modified(o))
+                    | Ok(WatchEvent::Deleted(o))
+                    | Ok(WatchEvent::Bookmark(o)) => {
                         // always store the last seen resourceVersion
                         if let Some(nv) = Meta::resource_ver(o) {
                             *version.lock().await = nv.clone();
diff --git a/kube/src/runtime/reflector.rs b/kube/src/runtime/reflector.rs
index c0609f606..b394982a7 100644
--- a/kube/src/runtime/reflector.rs
+++ b/kube/src/runtime/reflector.rs
@@ -96,6 +96,9 @@ where
                     debug!("Removing {} from {}", Meta::name(&o), kind);
                     state.remove(&ObjectId::key_for(&o));
                 }
+                WatchEvent::Bookmark(o) => {
+                    debug!("Bookmarking {} from {}", Meta::name(&o), kind);
+                }
                 WatchEvent::Error(e) => {
                     warn!("Failed to watch {}: {:?}", kind, e);
                     return Err(Error::Api(e));