Merge branch 'main' into rm-depr-stream-subscribe

README.md

@@ -20,18 +20,19 @@ kube = { version = "0.90.0", features = ["runtime", "derive"] }
 k8s-openapi = { version = "0.21.1", features = ["latest"] }
 ```
 
-[Features are available](https://github.com/kube-rs/kube/blob/main/kube/Cargo.toml#L18).
+See [features](https://kube.rs/features/) for a quick overview of default-enabled / opt-in functionality.
 
 ## Upgrading
 
-Please check the [CHANGELOG](https://github.com/kube-rs/kube/blob/main/CHANGELOG.md) when upgrading.
-All crates herein are versioned and [released](https://github.com/kube-rs/kube/blob/main/release.toml) together to guarantee [compatibility before 1.0](https://github.com/kube-rs/kube/issues/508).
+See [kube.rs/upgrading](https://kube.rs/upgrading/).
+Noteworthy changes are highlighted in [releases](https://github.com/kube-rs/kube/releases), and archived in the [changelog](https://kube.rs/changelog/).
 
 ## Usage
 
 See the **[examples directory](https://github.com/kube-rs/kube/blob/main/examples)** for how to use any of these crates.
 
 - **[kube API Docs](https://docs.rs/kube/)**
+- **[kube.rs](https://kube.rs)**
 
 Official examples:
 
@@ -42,14 +43,14 @@ For real world projects see [ADOPTERS](https://kube.rs/adopters/).
 
 ## Api
 
-The [`Api`](https://docs.rs/kube/*/kube/struct.Api.html) is what interacts with kubernetes resources, and is generic over [`Resource`](https://docs.rs/kube/*/kube/trait.Resource.html):
+The [`Api`](https://docs.rs/kube/*/kube/struct.Api.html) is what interacts with Kubernetes resources, and is generic over [`Resource`](https://docs.rs/kube/*/kube/trait.Resource.html):
 
 ```rust
 use k8s_openapi::api::core::v1::Pod;
 let pods: Api<Pod> = Api::default_namespaced(client);
 
-let p = pods.get("blog").await?;
-println!("Got blog pod with containers: {:?}", p.spec.unwrap().containers);
+let pod = pods.get("blog").await?;
+println!("Got pod: {pod:?}");
 
 let patch = json!({"spec": {
     "activeDeadlineSeconds": 5
@@ -67,7 +68,7 @@ See the examples ending in `_api` examples for more detail.
 
 Working with custom resources uses automatic code-generation via [proc_macros in kube-derive](https://docs.rs/kube/latest/kube/derive.CustomResource.html).
 
-You need to `#[derive(CustomResource)]` and some `#[kube(attrs..)]` on a spec struct:
+You need to add `#[derive(CustomResource, JsonSchema)]` and some `#[kube(attrs..)]` on a __spec__ struct:
 
 ```rust
 #[derive(CustomResource, Debug, Serialize, Deserialize, Default, Clone, JsonSchema)]
@@ -146,9 +147,11 @@ Controller::new(root_kind_api, Config::default())
 
 Here `reconcile` and `error_policy` refer to functions you define. The first will be called when the root or child elements change, and the second when the `reconciler` returns an `Err`.
 
+See the [controller guide](https://kube.rs/controllers/intro/) for how to write these.
+
 ## TLS
 
-By default [rustls](https://github.com/ctz/rustls) is used for TLS, but `openssl` is supported. To switch, turn off `default-features`, and enable the `openssl-tls` feature:
+By default [rustls](https://github.com/rustls/rustls) is used for TLS, but `openssl` is supported. To switch, turn off `default-features`, and enable the `openssl-tls` feature:
 
 ```toml
 [dependencies]

kube-client/src/error.rs

@@ -44,7 +44,7 @@ pub enum Error {
     HttpError(#[source] http::Error),
 
     /// Common error case when requesting parsing into own structs
-    #[error("Error deserializing response")]
+    #[error("Error deserializing response: {0}")]
     SerdeError(#[source] serde_json::Error),
 
     /// Failed to build request

kube-core/src/resource.rs

@@ -95,14 +95,60 @@ pub trait Resource {
     ///
     /// Note: this returns an `Option`, but for objects populated from the apiserver,
     /// this Option can be safely unwrapped.
+    ///
+    /// ```
+    /// use k8s_openapi::api::core::v1::ConfigMap;
+    /// use k8s_openapi::api::core::v1::Pod;
+    /// use k8s_openapi::apimachinery::pkg::apis::meta::v1::ObjectMeta;
+    /// use kube_core::Resource;
+    ///
+    /// let p = Pod::default();
+    /// let controller_ref = p.controller_owner_ref(&());
+    /// let cm = ConfigMap {
+    ///     metadata: ObjectMeta {
+    ///         name: Some("pod-configmap".to_string()),
+    ///         owner_references: Some(controller_ref.into_iter().collect()),
+    ///         ..ObjectMeta::default()
+    ///     },
+    ///     ..Default::default()
+    /// };
+    /// ```
     fn controller_owner_ref(&self, dt: &Self::DynamicType) -> Option<OwnerReference> {
+        Some(OwnerReference {
+            controller: Some(true),
+            ..self.owner_ref(dt)?
+        })
+    }
+
+    /// Generates an owner reference pointing to this resource
+    ///
+    /// Note: this returns an `Option`, but for objects populated from the apiserver,
+    /// this Option can be safely unwrapped.
+    ///
+    /// ```
+    /// use k8s_openapi::api::core::v1::ConfigMap;
+    /// use k8s_openapi::api::core::v1::Pod;
+    /// use k8s_openapi::apimachinery::pkg::apis::meta::v1::ObjectMeta;
+    /// use kube_core::Resource;
+    ///
+    /// let p = Pod::default();
+    /// let owner_ref = p.owner_ref(&());
+    /// let cm = ConfigMap {
+    ///     metadata: ObjectMeta {
+    ///         name: Some("pod-configmap".to_string()),
+    ///         owner_references: Some(owner_ref.into_iter().collect()),
+    ///         ..ObjectMeta::default()
+    ///     },
+    ///     ..Default::default()
+    /// };
+    /// ```
+    fn owner_ref(&self, dt: &Self::DynamicType) -> Option<OwnerReference> {
         let meta = self.meta();
         Some(OwnerReference {
             api_version: Self::api_version(dt).to_string(),
             kind: Self::kind(dt).to_string(),
             name: meta.name.clone()?,
             uid: meta.uid.clone()?,
-            controller: Some(true),
             ..OwnerReference::default()
         })
     }

kube-runtime/src/reflector/mod.rs

@@ -17,7 +17,7 @@ pub use store::{store, Store};
 
 /// Cache objects from a [`watcher()`] stream into a local [`Store`]
 ///
-/// Observes the raw [`Stream`] of [`watcher::Event`] objects, and modifies the cache.
+/// Observes the raw `Stream` of [`watcher::Event`] objects, and modifies the cache.
 /// It passes the raw [`watcher()`] stream through unmodified.
 ///
 /// ## Usage
@@ -98,6 +98,21 @@ pub use store::{store, Store};
 /// Additionally, only `labels`, `annotations` and `managed_fields` are safe to drop from `ObjectMeta`.
 ///
 /// For more information check out: <https://kube.rs/controllers/optimization/> for graphs and techniques.
+///
+/// ## Stream sharing
+///
+/// `reflector()` as an interface may optionally create a stream that can be
+/// shared with other components to help with resource usage.
+///
+/// To share a stream, the `Writer<K>` consumed by `reflector()` must be
+/// created through an interface that allows a store to be subscribed on, such
+/// as [`store_shared()`]. When the store supports being subscribed on, it will
+/// broadcast an event to all active listeners after caching any object
+/// contained in the event.
+///
+/// Creating subscribers requires an
+/// [`unstable`](https://github.com/kube-rs/kube/blob/main/kube-runtime/Cargo.toml#L17-L21)
+/// feature
 pub fn reflector<K, W>(mut writer: store::Writer<K>, stream: W) -> impl Stream<Item = W::Item>
 where
     K: Lookup + Clone,

kube-runtime/src/reflector/store.rs

@@ -83,6 +83,10 @@ where
     ///
     /// Multiple subscribe handles may be obtained, by either calling
     /// `subscribe` multiple times, or by calling `clone()`
+    ///
+    /// This function returns a `Some` when the [`Writer`] is constructed through
+    /// [`Writer::new_shared`] or [`store_shared`], and a `None` otherwise.
+    #[cfg(feature = "unstable-runtime-subscribe")]
     pub fn subscribe(&self) -> Option<ReflectHandle<K>> {
         self.dispatcher
             .as_ref()

kube-runtime/src/scheduler.rs

@@ -1,4 +1,4 @@
-//! Delays and deduplicates [`Stream`] items
+//! Delays and deduplicates [`Stream`](futures::stream::Stream) items
 
 use futures::{stream::Fuse, Stream, StreamExt};
 use hashbrown::{hash_map::Entry, HashMap};
@@ -251,7 +251,7 @@ where
     }
 }
 
-/// Stream transformer that delays and deduplicates [`Stream`] items.
+/// Stream transformer that delays and deduplicates items.
 ///
 /// Items are deduplicated: if an item is submitted multiple times before being emitted then it will only be
 /// emitted at the earliest `Instant`.

kube-runtime/src/utils/watch_ext.rs

@@ -209,6 +209,8 @@ pub trait WatchStreamExt: Stream {
     ///
     ///
     /// [`Store`]: crate::reflector::Store
+    /// [`subscribe()`]: crate::reflector::store::Writer::subscribe()
+    /// [`Stream`]: futures::stream::Stream
     /// [`ReflectHandle`]: crate::reflector::dispatcher::ReflectHandle
     /// ## Usage
     /// ```no_run