Merge remote-tracking branch 'origin/main' into test/add-enum-tests

Author: NickLarsenNZ

Cargo.toml

@@ -46,7 +46,7 @@ educe = { version = "0.6.0", default-features = false }
 either = "1.6.1"
 form_urlencoded = "1.2.0"
 futures = { version = "0.3.17", default-features = false }
-hashbrown = "0.15.0"
+hashbrown = "0.16.0"
 home = "0.5.4"
 hostname = "0.4"
 http = "1.1.0"
@@ -81,7 +81,7 @@ tempfile = "3.1.0"
 thiserror = "2.0.3"
 tokio = "1.14.0"
 tokio-test = "0.4.0"
-tokio-tungstenite = "0.27.0"
+tokio-tungstenite = "0.28.0"
 tokio-util = "0.7.0"
 tower = "0.5.1"
 tower-http = "0.6.1"

examples/crd_derive.rs

@@ -3,7 +3,7 @@ use kube::{
     core::object::{HasSpec, HasStatus},
     CustomResource, CustomResourceExt, Resource,
 };
-use schemars::JsonSchema;
+use schemars::{json_schema, JsonSchema};
 use serde::{Deserialize, Serialize};
 
 /// Our spec for Foo
@@ -63,7 +63,7 @@ fn main() {
 }
 
 fn conditions(_: &mut schemars::generate::SchemaGenerator) -> schemars::Schema {
-    serde_json::from_value(serde_json::json!({
+    json_schema!({
         "type": "array",
         "x-kubernetes-list-type": "map",
         "x-kubernetes-list-map-keys": ["type"],
@@ -85,8 +85,7 @@ fn conditions(_: &mut schemars::generate::SchemaGenerator) -> schemars::Schema {
                 "type"
             ],
         },
-    }))
-    .unwrap()
+    })
 }
 
 // some tests

examples/crd_derive_schema.rs

@@ -9,6 +9,7 @@ use kube::{
     runtime::wait::{await_condition, conditions},
     Client, CustomResource, CustomResourceExt, KubeSchema,
 };
+use schemars::json_schema;
 use serde::{Deserialize, Serialize};
 
 // This example shows how the generated schema affects defaulting and validation.
@@ -121,16 +122,15 @@ impl FooSpec {
 
 // https://kubernetes.io/docs/reference/using-api/server-side-apply/#merge-strategy
 fn set_listable_schema(_: &mut schemars::generate::SchemaGenerator) -> schemars::Schema {
-    serde_json::from_value(serde_json::json!({
+    json_schema!({
         "type": "array",
         "items": {
             "format": "u32",
             "minium": 0,
             "type": "integer"
         },
         "x-kubernetes-list-type": "set"
-    }))
-    .unwrap()
+    })
 }
 
 fn default_value() -> String {

kube-client/src/api/util/mod.rs

@@ -79,10 +79,10 @@ mod test {
 
         let node_name = "fakenode";
         let fake_node = serde_json::from_value(json!({
-        "apiVersion": "v1",
-        "kind": "Node",
-        "metadata": {
-            "name": node_name,
+            "apiVersion": "v1",
+            "kind": "Node",
+            "metadata": {
+                "name": node_name,
             },
         }))?;
 

kube-core/src/duration.rs

@@ -283,12 +283,10 @@ impl schemars::JsonSchema for Duration {
     }
 
     fn json_schema(_: &mut schemars::generate::SchemaGenerator) -> schemars::Schema {
-        use schemars::json_schema;
-
         // the format should *not* be "duration", because "duration" means
         // the duration is formatted in ISO 8601, as described here:
         // https://datatracker.ietf.org/doc/html/draft-handrews-json-schema-validation-02#section-7.3.1
-        json_schema!({
+        schemars::json_schema!({
             "type": "string",
         })
     }

kube-runtime/src/utils/predicate.rs

@@ -1,14 +1,15 @@
-use crate::{reflector::ObjectRef, watcher::Error};
+use crate::watcher::Error;
 use core::{
     pin::Pin,
     task::{ready, Context, Poll},
 };
 use futures::Stream;
-use kube_client::Resource;
+use kube_client::{api::ObjectMeta, Resource};
 use pin_project::pin_project;
 use std::{
     collections::{hash_map::DefaultHasher, HashMap},
     hash::{Hash, Hasher},
+    marker::PhantomData,
 };
 
 fn hash<T: Hash + ?Sized>(t: &T) -> u64 {
@@ -17,6 +18,24 @@ fn hash<T: Hash + ?Sized>(t: &T) -> u64 {
     hasher.finish()
 }
 
+/// Private cache key that includes UID in equality for predicate filtering
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+struct PredicateCacheKey {
+    name: String,
+    namespace: Option<String>,
+    uid: Option<String>,
+}
+
+impl From<&ObjectMeta> for PredicateCacheKey {
+    fn from(meta: &ObjectMeta) -> Self {
+        Self {
+            name: meta.name.clone().unwrap_or_default(),
+            namespace: meta.namespace.clone(),
+            uid: meta.uid.clone(),
+        }
+    }
+}
+
 /// A predicate is a hasher of Kubernetes objects stream filtering
 pub trait Predicate<K> {
     /// A predicate only needs to implement optional hashing when keys exist
@@ -103,7 +122,9 @@ pub struct PredicateFilter<St, K: Resource, P: Predicate<K>> {
     #[pin]
     stream: St,
     predicate: P,
-    cache: HashMap<ObjectRef<K>, u64>,
+    cache: HashMap<PredicateCacheKey, u64>,
+    // K: Resource necessary to get .meta() of an object during polling
+    _phantom: PhantomData<K>,
 }
 impl<St, K, P> PredicateFilter<St, K, P>
 where
@@ -116,6 +137,7 @@ where
             stream,
             predicate,
             cache: HashMap::new(),
+            _phantom: PhantomData,
         }
     }
 }
@@ -134,17 +156,9 @@ where
             break match ready!(me.stream.as_mut().poll_next(cx)) {
                 Some(Ok(obj)) => {
                     if let Some(val) = me.predicate.hash_property(&obj) {
-                        let key = ObjectRef::from_obj(&obj);
-                        let changed = if let Some(old) = me.cache.get(&key) {
-                            *old != val
-                        } else {
-                            true
-                        };
-                        if let Some(old) = me.cache.get_mut(&key) {
-                            *old = val;
-                        } else {
-                            me.cache.insert(key, val);
-                        }
+                        let key = PredicateCacheKey::from(obj.meta());
+                        let changed = me.cache.get(&key) != Some(&val);
+                        me.cache.insert(key, val);
                         if changed {
                             Some(Ok(obj))
                         } else {
@@ -251,4 +265,58 @@ pub(crate) mod tests {
         assert_eq!(second.meta().generation, Some(2));
         assert!(matches!(poll!(rx.next()), Poll::Ready(None)));
     }
+
+    #[tokio::test]
+    async fn predicate_filtering_should_handle_recreated_resources_with_same_generation() {
+        use k8s_openapi::api::core::v1::Pod;
+
+        let mkobj = |g: i32, uid: &str| {
+            let p: Pod = serde_json::from_value(json!({
+                "apiVersion": "v1",
+                "kind": "Pod",
+                "metadata": {
+                    "name": "blog",
+                    "namespace": "default",
+                    "generation": Some(g),
+                    "uid": uid,
+                },
+                "spec": {
+                    "containers": [{
+                      "name": "blog",
+                      "image": "clux/blog:0.1.0"
+                    }],
+                }
+            }))
+            .unwrap();
+            p
+        };
+
+        // Simulate: create (gen=1, uid=1) -> update (gen=1, uid=1) -> delete ->
+        // create (gen=1, uid=2) -> delete -> create (gen=2, uid=3)
+        let data = stream::iter([
+            Ok(mkobj(1, "uid-1")),
+            Ok(mkobj(1, "uid-1")),
+            Ok(mkobj(1, "uid-2")),
+            Ok(mkobj(2, "uid-3")),
+        ]);
+        let mut rx = pin!(PredicateFilter::new(data, predicates::generation));
+
+        // mkobj(1, uid-1) passed through
+        let first = rx.next().now_or_never().unwrap().unwrap().unwrap();
+        assert_eq!(first.meta().generation, Some(1));
+        assert_eq!(first.meta().uid.as_deref(), Some("uid-1"));
+
+        // (no repeat mkobj(1, uid-1) - same UID and generation)
+        // mkobj(1, uid-2) next - different UID detected
+        let second = rx.next().now_or_never().unwrap().unwrap().unwrap();
+        assert_eq!(second.meta().generation, Some(1));
+        assert_eq!(second.meta().uid.as_deref(), Some("uid-2"));
+
+        // mkobj(2, uid-3) next
+        let third = rx.next().now_or_never().unwrap().unwrap().unwrap();
+        assert_eq!(third.meta().generation, Some(2));
+        assert_eq!(third.meta().uid.as_deref(), Some("uid-3"));
+
+        assert!(matches!(poll!(rx.next()), Poll::Ready(None)));
+    }
 }

kube-runtime/src/watcher.rs

@@ -719,6 +719,7 @@ where
 /// that we have seen on the stream. If this is successful then the stream is simply resumed from where it left off.
 /// If this fails because the resource version is no longer valid then we start over with a new stream, starting with
 /// an [`Event::Init`]. The internals mechanics of recovery should be considered an implementation detail.
+#[doc(alias = "informer")]
 pub fn watcher<K: Resource + Clone + DeserializeOwned + Debug + Send + 'static>(
     api: Api<K>,
     watcher_config: Config,

kube/Cargo.toml

@@ -13,32 +13,50 @@ categories = ["network-programming", "caching", "api-bindings", "encoding"]
 
 [features]
 default = ["client", "rustls-tls", "ring"]
+#! See [kube.rs/features](https://kube.rs/features/) for more details
 
-# default features
+## default client
 client = ["kube-client/client", "config"]
+## default config
 config = ["kube-client/config"]
+## default tls stack
 rustls-tls = ["kube-client/rustls-tls", "client"]
 
-# alternative features
+## legacy tls stack (must disable rustls-tls to use)
 openssl-tls = ["kube-client/openssl-tls", "client"]
+## aws rustls provider
 aws-lc-rs = ["kube-client?/aws-lc-rs"]
+## ring rustls provider
 ring = ["kube-client?/ring"]
+## enable webpki roots in rustls
+webpki-roots = ["kube-client/webpki-roots", "client"]
 
-# auxiliary features
+## enable kube-derive and proc macros
+derive = ["kube-derive", "kube-core/schema"]
+## enable runtime for controllers/watchers/reflectors
+runtime = ["kube-runtime"]
+## enable websocket client support for portforward/exec/attach
 ws = ["kube-client/ws", "kube-core/ws"]
-kubelet-debug = ["kube-client/kubelet-debug", "kube-core/kubelet-debug"]
+## enable client oauth support
 oauth = ["kube-client/oauth", "client"]
+## enable client oidc support
 oidc = ["kube-client/oidc", "client"]
+## enable client socks5 support
+socks5 = ["kube-client/socks5", "client"]
+## enable client http proxy support
+http-proxy = ["kube-client/http-proxy", "client"]
+## enable client gzip usage
 gzip = ["kube-client/gzip", "client"]
+## enable support for jsonpatch style patch parameters
 jsonpatch = ["kube-core/jsonpatch"]
+## enable the admission module
 admission = ["kube-core/admission"]
-derive = ["kube-derive", "kube-core/schema"]
-runtime = ["kube-runtime"]
+## enable unstable runtime features
 unstable-runtime = ["kube-runtime/unstable-runtime", "runtime"]
+## enable unstable client features
 unstable-client = ["kube-client/unstable-client", "client"]
-socks5 = ["kube-client/socks5", "client"]
-http-proxy = ["kube-client/http-proxy", "client"]
-webpki-roots = ["kube-client/webpki-roots", "client"]
+## enable the kubelet debug interface
+kubelet-debug = ["kube-client/kubelet-debug", "kube-core/kubelet-debug"]
 
 [package.metadata.docs.rs]
 features = ["client", "rustls-tls", "openssl-tls", "derive", "ws", "oauth", "jsonpatch", "admission", "runtime", "k8s-openapi/latest", "unstable-runtime", "socks5", "http-proxy"]

kube/src/lib.rs

@@ -1,3 +1,6 @@
+#![doc(
+    html_logo_url = "https://user-images.githubusercontent.com/639336/155115130-758a8ba9-e209-42de-bf6d-cde7be3ed86f.svg#only-light"
+)]
 //! Kube is an umbrella-crate for interacting with [Kubernetes](http://kubernetes.io) in Rust.
 //!
 //! # Overview
@@ -103,6 +106,9 @@
 //!
 //! # Examples
 //! A large list of complete, runnable examples with explainations are available in the [examples folder](https://github.com/kube-rs/kube/tree/main/examples).
+//!
+//! # Features
+//! Documented at [kube.rs/features](https://kube.rs/features/).
 #![cfg_attr(docsrs, feature(doc_cfg))]
 
 macro_rules! cfg_client {
@@ -390,8 +396,9 @@ mod test {
         Ok(())
     }
 
-    #[tokio::test]
-    #[ignore = "needs cluster (fetches api resources, and lists all)"]
+    // #[tokio::test]
+    // #[ignore = "needs cluster (fetches api resources, and lists all)"]
+    // TODO: fixup. gets rate limited in default k3s on CI now.
     #[cfg(feature = "derive")]
     async fn derived_resources_discoverable() -> Result<(), Box<dyn std::error::Error>> {
         use crate::{
@@ -415,7 +422,7 @@ mod test {
         let establish = await_condition(crds.clone(), "testcrs.kube.rs", conditions::is_crd_established());
         let crd = tokio::time::timeout(std::time::Duration::from_secs(10), establish).await??;
         assert!(conditions::is_crd_established().matches_object(crd.as_ref()));
-        tokio::time::sleep(std::time::Duration::from_secs(2)).await; // Established condition is actually not enough for api discovery :(
+        tokio::time::sleep(std::time::Duration::from_secs(5)).await; // Established condition is actually not enough for api discovery :(
 
         // create partial information for it to discover
         let gvk = GroupVersionKind::gvk("kube.rs", "v1", "TestCr");