adds watch-list implementation without breaking changes

README.md

@@ -2,7 +2,7 @@
 
 [![Crates.io](https://img.shields.io/crates/v/kube.svg)](https://crates.io/crates/kube)
 [![Rust 1.64](https://img.shields.io/badge/MSRV-1.64-dea584.svg)](https://github.com/rust-lang/rust/releases/tag/1.64.0)
-[![Tested against Kubernetes v1_21 and above](https://img.shields.io/badge/MK8SV-v1_21-326ce5.svg)](https://kube.rs/kubernetes-version)
+[![Tested against Kubernetes v1_22 and above](https://img.shields.io/badge/MK8SV-v1_22-326ce5.svg)](https://kube.rs/kubernetes-version)
 [![Best Practices](https://bestpractices.coreinfrastructure.org/projects/5413/badge)](https://bestpractices.coreinfrastructure.org/projects/5413)
 [![Discord chat](https://img.shields.io/discord/500028886025895936.svg?logo=discord&style=plastic)](https://discord.gg/tokio)
 
@@ -17,7 +17,7 @@ Select a version of `kube` along with the generated [k8s-openapi](https://github
 ```toml
 [dependencies]
 kube = { version = "0.84.0", features = ["runtime", "derive"] }
-k8s-openapi = { version = "0.18.0", features = ["v1_26"] }
+k8s-openapi = { version = "0.18.0", features = ["v1_27"] }
 ```
 
 [Features are available](https://github.com/kube-rs/kube/blob/main/kube/Cargo.toml#L18).
@@ -153,7 +153,7 @@ By default `openssl` is used for TLS, but [rustls](https://github.com/ctz/rustls
 ```toml
 [dependencies]
 kube = { version = "0.84.0", default-features = false, features = ["client", "rustls-tls"] }
-k8s-openapi = { version = "0.18.0", features = ["v1_26"] }
+k8s-openapi = { version = "0.18.0", features = ["v1_27"] }
 ```
 
 This will pull in `rustls` and `hyper-rustls`. If `default-features` is left enabled, you will pull in two TLS stacks, and the default will remain as `openssl`.

e2e/Cargo.toml

@@ -18,7 +18,7 @@ name = "boot"
 path = "boot.rs"
 
 [features]
-latest = ["k8s-openapi/v1_26"]
+latest = ["k8s-openapi/v1_27"]
 mk8sv = ["k8s-openapi/v1_21"]
 rustls = ["kube/rustls-tls"]
 openssl = ["kube/openssl-tls"]
@@ -29,6 +29,6 @@ tracing = "0.1.36"
 tracing-subscriber = "0.3.3"
 futures = "0.3.17"
 kube = { path = "../kube", version = "^0.84.0", default-features = false, features = ["client", "runtime", "ws", "admission", "gzip"] }
-k8s-openapi = { version = "0.18.0", default-features = false }
+k8s-openapi = { git = "https://github.com/Arnavion/k8s-openapi", branch = "master", default-features = false }
 serde_json = "1.0.68"
 tokio = { version = "1.14.0", features = ["full"] }

examples/Cargo.toml

@@ -21,7 +21,7 @@ rustls-tls = ["kube/client", "kube/rustls-tls"]
 runtime = ["kube/runtime", "kube/unstable-runtime"]
 refresh = ["kube/oauth", "kube/oidc"]
 ws = ["kube/ws"]
-latest = ["k8s-openapi/v1_26"]
+latest = ["k8s-openapi/v1_27"]
 
 [dev-dependencies]
 tokio-util = "0.7.0"
@@ -32,7 +32,7 @@ futures = "0.3.17"
 jsonpath_lib = "0.3.0"
 kube = { path = "../kube", version = "^0.84.0", default-features = false, features = ["admission"] }
 kube-derive = { path = "../kube-derive", version = "^0.84.0", default-features = false } # only needed to opt out of schema
-k8s-openapi = { version = "0.18.0", default-features = false }
+k8s-openapi = { git = "https://github.com/Arnavion/k8s-openapi", branch = "master", default-features = false }
 serde = { version = "1.0.130", features = ["derive"] }
 serde_json = "1.0.68"
 serde_yaml = "0.9.19"

justfile

@@ -15,7 +15,7 @@ fmt:
   rustfmt +nightly --edition 2021 $(find . -type f -iname *.rs)
 
 doc:
-  RUSTDOCFLAGS="--cfg docsrs" cargo +nightly doc --lib --workspace --features=derive,ws,oauth,oidc,jsonpatch,client,derive,runtime,admission,k8s-openapi/v1_26,unstable-runtime --open
+  RUSTDOCFLAGS="--cfg docsrs" cargo +nightly doc --lib --workspace --features=derive,ws,oauth,oidc,jsonpatch,client,derive,runtime,admission,k8s-openapi/v1_27,unstable-runtime --open
 
 deny:
   # might require rm Cargo.lock first to match CI

kube-client/Cargo.toml

@@ -32,7 +32,7 @@ config = ["__non_core", "pem", "home"]
 __non_core = ["tracing", "serde_yaml", "base64"]
 
 [package.metadata.docs.rs]
-features = ["client", "rustls-tls", "openssl-tls", "ws", "oauth", "oidc", "jsonpatch", "admission", "k8s-openapi/v1_26"]
+features = ["client", "rustls-tls", "openssl-tls", "ws", "oauth", "oidc", "jsonpatch", "admission", "k8s-openapi/v1_27"]
 # Define the configuration attribute `docsrs`. Used to enable `doc_cfg` feature.
 rustdoc-args = ["--cfg", "docsrs"]
 
@@ -72,7 +72,8 @@ hyper-openssl = { version = "0.9.2", optional = true }
 form_urlencoded = { version = "1.2.0", optional = true }
 
 [dependencies.k8s-openapi]
-version = "0.18.0"
+git = "https://github.com/Arnavion/k8s-openapi"
+branch = "master"
 default-features = false
 features = []
 
@@ -85,6 +86,7 @@ tokio-test = "0.4.0"
 tower-test = "0.4.0"
 
 [dev-dependencies.k8s-openapi]
-version = "0.18.0"
+git = "https://github.com/Arnavion/k8s-openapi"
+branch = "master"
 default-features = false
-features = ["v1_26"]
+features = ["v1_27"]

kube-core/Cargo.toml

@@ -15,7 +15,7 @@ repository = "https://github.com/kube-rs/kube"
 readme = "../README.md"
 
 [package.metadata.docs.rs]
-features = ["ws", "admission", "jsonpatch", "k8s-openapi/v1_26"]
+features = ["ws", "admission", "jsonpatch", "k8s-openapi/v1_27"]
 rustdoc-args = ["--cfg", "docsrs"]
 
 [features]
@@ -36,14 +36,16 @@ chrono = { version = "0.4.19", default-features = false, features = ["clock"] }
 schemars = { version = "0.8.6", optional = true }
 
 [dependencies.k8s-openapi]
-version = "0.18.0"
+git = "https://github.com/Arnavion/k8s-openapi"
+branch = "master"
 default-features = false
 features = []
 
 [dev-dependencies.k8s-openapi]
-version = "0.18.0"
+git = "https://github.com/Arnavion/k8s-openapi"
+branch = "master"
 default-features = false
-features = ["v1_26"]
+features = ["v1_27"]
 
 [dev-dependencies]
 assert-json-diff = "2.0.1"

kube-core/src/params.rs

@@ -305,6 +305,39 @@ pub struct WatchParams {
     /// If the feature gate WatchBookmarks is not enabled in apiserver,
     /// this field is ignored.
     pub bookmarks: bool,
+
+    /// `sendInitialEvents=true` may be set together with `watch=true`.
+    /// In that case, the watch stream will begin with synthetic events to
+    /// produce the current state of objects in the collection. Once all such
+    /// events have been sent, a synthetic "Bookmark" event  will be sent.
+    /// The bookmark will report the ResourceVersion (RV) corresponding to the
+    /// set of objects, and be marked with `"k8s.io/initial-events-end": "true"` annotation.
+    /// Afterwards, the watch stream will proceed as usual, sending watch events
+    /// corresponding to changes (subsequent to the RV) to objects watched.
+    ///
+    /// When `sendInitialEvents` option is set, we require `resourceVersionMatch`
+    /// option to also be set. The semantic of the watch request is as following:
+    /// - `resourceVersionMatch` = NotOlderThan
+    ///   is interpreted as "data at least as new as the provided `resourceVersion`"
+    ///   and the bookmark event is send when the state is synced
+    ///   to a `resourceVersion` at least as fresh as the one provided by the ListOptions.
+    ///   If `resourceVersion` is unset, this is interpreted as "consistent read" and the
+    ///   bookmark event is send when the state is synced at least to the moment
+    ///   when request started being processed.
+    /// - `resourceVersionMatch` set to any other value or unset
+    ///   Invalid error is returned.
+    ///
+    /// Defaults to true if `resourceVersion=""` or `resourceVersion="0"` (for backward
+    /// compatibility reasons) and to false otherwise.
+    pub send_initial_events: bool,
+
+    /// Determines how resourceVersion is matched applied to list calls.
+    pub version_match: Option<VersionMatch>,
+
+    /// An explicit resourceVersion using the given `VersionMatch` strategy
+    ///
+    /// See <https://kubernetes.io/docs/reference/using-api/api-concepts/#resource-versions> for details.
+    pub resource_version: Option<String>,
 }
 
 impl WatchParams {
@@ -315,6 +348,19 @@ impl WatchParams {
                 return Err(Error::Validation("WatchParams::timeout must be < 295s".into()));
             }
         }
+        if self.send_initial_events {
+            if self.version_match != Some(VersionMatch::NotOlderThan) {
+                return Err(Error::Validation(
+                    "WatchParams::version_match must be set to NotOlderThan when using send_initial_events"
+                        .into(),
+                ));
+            }
+            if !self.bookmarks {
+                return Err(Error::Validation(
+                    "WatchParams::bookmarks must be set when using send_initial_events".into(),
+                ));
+            }
+        }
         Ok(())
     }
 
@@ -334,6 +380,16 @@ impl WatchParams {
         if self.bookmarks {
             qp.append_pair("allowWatchBookmarks", "true");
         }
+        if let Some(resource_version_match) = &self.version_match {
+            match resource_version_match {
+                VersionMatch::NotOlderThan => {
+                    qp.append_pair("resourceVersionMatch", "NotOlderThan");
+                }
+                VersionMatch::Exact => {
+                    qp.append_pair("resourceVersionMatch", "Exact");
+                }
+            }
+        }
     }
 }
 
@@ -347,6 +403,9 @@ impl Default for WatchParams {
             label_selector: None,
             field_selector: None,
             timeout: None,
+            send_initial_events: false,
+            resource_version: None,
+            version_match: None,
         }
     }
 }
@@ -401,6 +460,53 @@ impl WatchParams {
         self.bookmarks = false;
         self
     }
+
+    /// `sendInitialEvents=true` may be set together with `watch=true`.
+    /// In that case, the watch stream will begin with synthetic events to
+    /// produce the current state of objects in the collection. Once all such
+    /// events have been sent, a synthetic "Bookmark" event  will be sent.
+    /// The bookmark will report the ResourceVersion (RV) corresponding to the
+    /// set of objects, and be marked with `"k8s.io/initial-events-end": "true"` annotation.
+    /// Afterwards, the watch stream will proceed as usual, sending watch events
+    /// corresponding to changes (subsequent to the RV) to objects watched.
+    ///
+    /// When `sendInitialEvents` option is set, we require `resourceVersionMatch`
+    /// option to also be set. The semantic of the watch request is as following:
+    /// - `resourceVersionMatch` = NotOlderThan
+    ///   is interpreted as "data at least as new as the provided `resourceVersion`"
+    ///   and the bookmark event is send when the state is synced
+    ///   to a `resourceVersion` at least as fresh as the one provided by the ListOptions.
+    ///   If `resourceVersion` is unset, this is interpreted as "consistent read" and the
+    ///   bookmark event is send when the state is synced at least to the moment
+    ///   when request started being processed.
+    /// - `resourceVersionMatch` set to any other value or unset
+    ///   Invalid error is returned.
+    ///
+    /// Defaults to true if `resourceVersion=""` or `resourceVersion="0"` (for backward
+    /// compatibility reasons) and to false otherwise.
+    #[must_use]
+    pub fn initial_events(mut self) -> Self {
+        self.send_initial_events = true;
+
+        self
+    }
+
+    /// Sets the resource version
+    #[must_use]
+    pub fn at(mut self, resource_version: &str) -> Self {
+        self.resource_version = Some(resource_version.into());
+        self
+    }
+
+    /// Sets an arbitary resource version match strategy
+    ///
+    /// A non-default strategy such as `VersionMatch::Exact` or `VersionMatch::NotGreaterThan`
+    /// requires an explicit `resource_version` set to pass request validation.
+    #[must_use]
+    pub fn matching(mut self, version_match: VersionMatch) -> Self {
+        self.version_match = Some(version_match);
+        self
+    }
 }
 
 /// Common query parameters for put/post calls

kube-core/src/watch.rs

@@ -59,4 +59,9 @@ pub struct Bookmark {
 pub struct BookmarkMeta {
     /// The only field we need from a Bookmark event.
     pub resource_version: String,
+
+    /// Kubernetes 1.27 Streaming Lists
+    /// The rest of the fields are optional and may be empty.
+    #[serde(default)]
+    pub annotations: std::collections::BTreeMap<String, String>,
 }

kube-derive/Cargo.toml

@@ -29,7 +29,7 @@ proc-macro = true
 serde = { version = "1.0.130", features = ["derive"] }
 serde_yaml = "0.9.19"
 kube = { path = "../kube", version = "<1.0.0, >=0.61.0", features = ["derive", "client"] }
-k8s-openapi = { version = "0.18.0", default-features = false, features = ["v1_26"] }
+k8s-openapi = { git = "https://github.com/Arnavion/k8s-openapi", branch = "master", default-features = false, features = ["v1_27"] }
 schemars = { version = "0.8.6", features = ["chrono"] }
 chrono = { version = "0.4.19", default-features = false }
 trybuild = "1.0.48"

kube-runtime/Cargo.toml

@@ -22,7 +22,7 @@ unstable-runtime-stream-control = []
 unstable-runtime-reconcile-on = []
 
 [package.metadata.docs.rs]
-features = ["k8s-openapi/v1_26", "unstable-runtime"]
+features = ["k8s-openapi/v1_27", "unstable-runtime"]
 # Define the configuration attribute `docsrs`. Used to enable `doc_cfg` feature.
 rustdoc-args = ["--cfg", "docsrs"]
 
@@ -46,7 +46,8 @@ async-trait = "0.1.64"
 hashbrown = "0.14.0"
 
 [dependencies.k8s-openapi]
-version = "0.18.0"
+git = "https://github.com/Arnavion/k8s-openapi"
+branch = "master"
 default-features = false
 
 [dev-dependencies]
@@ -58,6 +59,7 @@ schemars = "0.8.6"
 tracing-subscriber = "0.3.17"
 
 [dev-dependencies.k8s-openapi]
-version = "0.18.0"
+git = "https://github.com/Arnavion/k8s-openapi"
+branch = "master"
 default-features = false
-features = ["v1_26"]
+features = ["v1_27"]