Merge branch 'master' into portforward-fix

CHANGELOG.md

@@ -7,8 +7,47 @@ UNRELEASED
 ===================
  * see https://github.com/kube-rs/kube-rs/compare/0.74.0...master
 
-0.74.0 / 2022-07-09
+[0.74.0](https://github.com/kube-rs/kube-rs/releases/tag/0.74.0) / 2022-07-09
 ===================
+<!-- Release notes generated using configuration in .github/release.yml at 0.74.0 -->
+
+## Highlights
+
+### Polish, bug fixes, guidelines, ci improvements, and new contributors
+This release features smaller improvements/additions/cleanups/fixes, many of which are from new first-time contributors! Thank you everyone!
+The listed [deadlock fix](https://github.com/kube-rs/kube-rs/pull/925) was backported to 0.73.1.
+
+We have also been trying to clarify and **prove** a lot more of our external-facing guarantees, and as a result:
+
+ - We have codified our [Kubernetes versioning policy](https://kube.rs/kubernetes-version/) 
+ - The [Rust version policy](https://kube.rs/rust-version/) has extended its support range
+ - Our [CI has been extended](https://github.com/kube-rs/kube-rs/pull/924)
+
+### [`ResourceExt::name` deprecation](https://github.com/kube-rs/kube-rs/pull/945)
+
+A consequence of all the policy writing and the improved clarity we have decided to deprecate the common [`ResourceExt::name`](https://docs.rs/kube/0.74.0/kube/trait.ResourceExt.html#tymethod.name) helper.
+
+This method could panic and it is unexpected for the users and bad for our consistency. To get the old functionality, you can replace any `.name()` call on a Kubernetes resources with [`.name_unchecked()`](https://docs.rs/kube/0.74.0/kube/trait.ResourceExt.html#tymethod.name_unchecked); but as the name implies, it can panic (in a local setting, or during admission). We recommend you replace it with the new [`ResourceExt::name_any`](https://docs.rs/kube/0.74.0/kube/trait.ResourceExt.html#tymethod.name_any) for a general identifier:
+
+```diff
+-pod.name()
++pod.name_any()
+```
+
+## What's Changed
+### Added
+* Add support for passing the `fieldValidation` query parameter on patch by @phroggyy in https://github.com/kube-rs/kube-rs/pull/929
+* Add `conditions::is_job_completed` by @clux in https://github.com/kube-rs/kube-rs/pull/935
+### Changed
+* Deprecate `ResourceExt::name` in favour of safe name_* alternatives by @clux in https://github.com/kube-rs/kube-rs/pull/945
+### Removed
+* Remove `#[kube(apiextensions)]` flag from `kube-derive` by @clux in https://github.com/kube-rs/kube-rs/pull/920
+### Fixed
+* Document every public derived fn from kube-derive by @clux in https://github.com/kube-rs/kube-rs/pull/919
+* fix applier hangs which can happen with many watched objects by @moustafab in https://github.com/kube-rs/kube-rs/pull/925
+* Applier: Improve reconciler reschedule context to avoid deadlocking on full channel by @teozkr in https://github.com/kube-rs/kube-rs/pull/932
+* Fix deserialization issue in AdmissionResponse by @clux in https://github.com/kube-rs/kube-rs/pull/939
+* Admission controller example fixes by @Alibirb in https://github.com/kube-rs/kube-rs/pull/950
 
 [0.73.1](https://github.com/kube-rs/kube-rs/releases/tag/0.73.1) / 2022-06-03
 ===================

examples/Cargo.toml

@@ -138,6 +138,10 @@ path = "pod_attach.rs"
 name = "pod_exec"
 path = "pod_exec.rs"
 
+[[example]]
+name = "pod_paged"
+path = "pod_paged.rs"
+
 [[example]]
 name = "pod_evict"
 path = "pod_evict.rs"

examples/configmapgen_controller.rs

@@ -67,7 +67,7 @@ async fn reconcile(generator: Arc<ConfigMapGenerator>, ctx: Arc<Data>) -> Result
 }
 
 /// The controller triggers this on reconcile errors
-fn error_policy(_error: &Error, _ctx: Arc<Data>) -> Action {
+fn error_policy(_object: Arc<ConfigMapGenerator>, _error: &Error, _ctx: Arc<Data>) -> Action {
     Action::requeue(Duration::from_secs(1))
 }
 

examples/crd_api.rs

@@ -120,7 +120,7 @@ async fn main() -> Result<()> {
 
     // Create Foo qux with status
     info!("Create Foo instance qux");
-    let mut f2 = Foo::new("qux", FooSpec {
+    let f2 = Foo::new("qux", FooSpec {
         name: "qux".into(),
         replicas: 0,
         info: "unpatched qux".into(),

examples/pod_paged.rs

@@ -0,0 +1,43 @@
+use k8s_openapi::api::core::v1::Pod;
+use kube::{
+    api::{Api, ListParams, ResourceExt},
+    Client,
+};
+use tracing::*;
+
+const PAGE_SIZE: u32 = 5;
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    tracing_subscriber::fmt::init();
+    let client = Client::try_default().await?;
+    let api = Api::<Pod>::default_namespaced(client);
+
+    let mut continue_token: Option<String> = None;
+    for page in 1.. {
+        info!("Fetching Page #{page}");
+        continue_token = fetch_page(&api, continue_token).await?;
+
+        if continue_token.is_none() {
+            info!("End of list");
+            break;
+        }
+    }
+
+    Ok(())
+}
+
+async fn fetch_page(api: &Api<Pod>, continue_token: Option<String>) -> anyhow::Result<Option<String>> {
+    let mut lp = ListParams::default().limit(PAGE_SIZE);
+    if let Some(token) = continue_token {
+        lp = lp.continue_token(&token);
+    }
+
+    let pods = api.list(&lp).await?;
+    let continue_token = pods.metadata.continue_.clone();
+    for p in pods {
+        info!("Found Pod: {}", p.name_any());
+    }
+
+    Ok(continue_token)
+}

examples/secret_syncer.rs

@@ -100,7 +100,7 @@ async fn main() -> anyhow::Result<()> {
                 .await
             }
         },
-        |_err, _| Action::requeue(Duration::from_secs(2)),
+        |_obj, _err, _| Action::requeue(Duration::from_secs(2)),
         Arc::new(()),
     )
     .for_each(|msg| async move { info!("Reconciled: {:?}", msg) })

kube-client/src/api/subresource.rs

@@ -76,6 +76,25 @@ where
         self.client.request::<K>(req).await
     }
 
+    /// Create an instance of the subresource
+    pub async fn create_subresource<T>(
+        &self,
+        subresource_name: &str,
+        name: &str,
+        pp: &PostParams,
+        data: Vec<u8>,
+    ) -> Result<T>
+    where
+        T: DeserializeOwned,
+    {
+        let mut req = self
+            .request
+            .create_subresource(subresource_name, name, pp, data)
+            .map_err(Error::BuildRequest)?;
+        req.extensions_mut().insert("create_subresource");
+        self.client.request::<T>(req).await
+    }
+
     /// Patch an instance of the subresource
     pub async fn patch_subresource<P: serde::Serialize + Debug>(
         &self,

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

@@ -2,8 +2,11 @@ use crate::{
     api::{Api, Resource},
     Error, Result,
 };
-use k8s_openapi::api::core::v1::Node;
-use kube_core::util::Restart;
+use k8s_openapi::api::{
+    authentication::v1::TokenRequest,
+    core::v1::{Node, ServiceAccount},
+};
+use kube_core::{params::PostParams, util::Restart};
 use serde::de::DeserializeOwned;
 
 k8s_openapi::k8s_if_ge_1_19! {
@@ -38,6 +41,24 @@ impl Api<Node> {
     }
 }
 
+impl Api<ServiceAccount> {
+    /// Create a TokenRequest of a ServiceAccount
+    pub async fn create_token_request(
+        &self,
+        name: &str,
+        pp: &PostParams,
+        token_request: &TokenRequest,
+    ) -> Result<TokenRequest> {
+        let bytes = serde_json::to_vec(token_request).map_err(Error::SerdeError)?;
+        let mut req = self
+            .request
+            .create_subresource("token", name, pp, bytes)
+            .map_err(Error::BuildRequest)?;
+        req.extensions_mut().insert("create_token_request");
+        self.client.request::<TokenRequest>(req).await
+    }
+}
+
 // Tests that require a cluster and the complete feature set
 // Can be run with `cargo test -p kube-client --lib -- --ignored`
 #[cfg(test)]
@@ -47,7 +68,10 @@ mod test {
         api::{Api, DeleteParams, ListParams, PostParams},
         Client,
     };
-    use k8s_openapi::api::core::v1::Node;
+    use k8s_openapi::api::{
+        authentication::v1::{TokenRequest, TokenRequestSpec, TokenReview, TokenReviewSpec},
+        core::v1::{Node, ServiceAccount},
+    };
     use serde_json::json;
 
     #[tokio::test]
@@ -81,4 +105,72 @@ mod test {
         nodes.delete(node_name, &DeleteParams::default()).await?;
         Ok(())
     }
+
+    #[tokio::test]
+    #[ignore] // requires a cluster
+    async fn create_token_request() -> Result<(), Box<dyn std::error::Error>> {
+        let client = Client::try_default().await?;
+
+        let serviceaccount_name = "fakesa";
+        let serviceaccount_namespace = "default";
+        let audiences = vec!["api".to_string()];
+
+        let serviceaccounts: Api<ServiceAccount> = Api::namespaced(client.clone(), serviceaccount_namespace);
+        let tokenreviews: Api<TokenReview> = Api::all(client);
+
+        // Create ServiceAccount
+        let fake_sa = serde_json::from_value(json!({
+            "apiVersion": "v1",
+            "kind": "ServiceAccount",
+            "metadata": {
+                "name": serviceaccount_name,
+            },
+        }))?;
+        serviceaccounts.create(&PostParams::default(), &fake_sa).await?;
+
+        // Create TokenRequest
+        let tokenrequest = serviceaccounts
+            .create_token_request(serviceaccount_name, &PostParams::default(), &TokenRequest {
+                metadata: Default::default(),
+                spec: TokenRequestSpec {
+                    audiences: audiences.clone(),
+                    bound_object_ref: None,
+                    expiration_seconds: None,
+                },
+                status: None,
+            })
+            .await?;
+        let token = tokenrequest.status.unwrap().token;
+        assert!(!token.is_empty());
+
+        // Check created token is valid with TokenReview
+        let tokenreview = tokenreviews
+            .create(&PostParams::default(), &TokenReview {
+                metadata: Default::default(),
+                spec: TokenReviewSpec {
+                    audiences: Some(audiences.clone()),
+                    token: Some(token),
+                },
+                status: None,
+            })
+            .await?;
+        let tokenreviewstatus = tokenreview.status.unwrap();
+        assert_eq!(tokenreviewstatus.audiences, Some(audiences));
+        assert_eq!(tokenreviewstatus.authenticated, Some(true));
+        assert_eq!(tokenreviewstatus.error, None);
+        assert_eq!(
+            tokenreviewstatus.user.unwrap().username,
+            Some(format!(
+                "system:serviceaccount:{}:{}",
+                serviceaccount_namespace, serviceaccount_name
+            ))
+        );
+
+        // Cleanup ServiceAccount
+        serviceaccounts
+            .delete(serviceaccount_name, &DeleteParams::default())
+            .await?;
+
+        Ok(())
+    }
 }

kube-client/src/client/builder.rs

@@ -71,7 +71,6 @@ impl TryFrom<Config> for ClientBuilder<BoxService<Request<hyper::Body>, Response
         use http::header::HeaderMap;
         use tracing::Span;
 
-        let timeout = config.timeout;
         let default_ns = config.default_namespace.clone();
 
         let client: hyper::Client<_, hyper::Body> = {
@@ -101,8 +100,14 @@ impl TryFrom<Config> for ClientBuilder<BoxService<Request<hyper::Body>, Response
             ));
 
             let mut connector = TimeoutConnector::new(connector);
-            connector.set_connect_timeout(timeout);
-            connector.set_read_timeout(timeout);
+
+            // Set the timeout for the client and fallback to default deprecated timeout until it's removed
+            #[allow(deprecated)]
+            {
+                connector.set_connect_timeout(config.connect_timeout.or(config.timeout));
+                connector.set_read_timeout(config.read_timeout.or(config.timeout));
+                connector.set_write_timeout(config.write_timeout);
+            }
 
             hyper::Client::builder().build(connector)
         };

kube-client/src/client/mod.rs

@@ -132,7 +132,10 @@ impl Client {
         &self.default_ns
     }
 
-    async fn send(&self, request: Request<Body>) -> Result<Response<Body>> {
+    /// Perform a raw HTTP request against the API and return the raw response back.
+    /// This method can be used to get raw access to the API which may be used to, for example,
+    /// create a proxy server or application-level gateway between localhost and the API server.
+    pub async fn send(&self, request: Request<Body>) -> Result<Response<Body>> {
         let mut svc = self.inner.clone();
         let res = svc
             .ready()

kube-client/src/config/mod.rs

@@ -128,9 +128,25 @@ pub struct Config {
     pub default_namespace: String,
     /// The configured root certificate
     pub root_cert: Option<Vec<Vec<u8>>>,
+    /// Set the timeout for connecting to the Kubernetes API.
+    ///
+    /// A value of `None` means no timeout
+    pub connect_timeout: Option<std::time::Duration>,
+    /// Set the timeout for the Kubernetes API response.
+    ///
+    /// A value of `None` means no timeout
+    pub read_timeout: Option<std::time::Duration>,
+    /// Set the timeout for the Kubernetes API request.
+    ///
+    /// A value of `None` means no timeout
+    pub write_timeout: Option<std::time::Duration>,
     /// Timeout for calls to the Kubernetes API.
     ///
     /// A value of `None` means no timeout
+    #[deprecated(
+        since = "0.75.0",
+        note = "replaced by more granular members `connect_timeout`, `read_timeout` and `write_timeout`. This member will be removed in 0.78.0."
+    )]
     pub timeout: Option<std::time::Duration>,
     /// Whether to accept invalid certificates
     pub accept_invalid_certs: bool,
@@ -148,10 +164,14 @@ impl Config {
     /// Most likely you want to use [`Config::infer`] to infer the config from
     /// the environment.
     pub fn new(cluster_url: http::Uri) -> Self {
+        #[allow(deprecated)]
         Self {
             cluster_url,
             default_namespace: String::from("default"),
             root_cert: None,
+            connect_timeout: Some(DEFAULT_CONNECT_TIMEOUT),
+            read_timeout: Some(DEFAULT_READ_TIMEOUT),
+            write_timeout: None,
             timeout: Some(DEFAULT_TIMEOUT),
             accept_invalid_certs: false,
             auth_info: AuthInfo::default(),
@@ -196,10 +216,14 @@ impl Config {
         let default_namespace = incluster_config::load_default_ns()?;
         let root_cert = incluster_config::load_cert()?;
 
+        #[allow(deprecated)]
         Ok(Self {
             cluster_url,
             default_namespace,
             root_cert: Some(root_cert),
+            connect_timeout: Some(DEFAULT_CONNECT_TIMEOUT),
+            read_timeout: Some(DEFAULT_READ_TIMEOUT),
+            write_timeout: None,
             timeout: Some(DEFAULT_TIMEOUT),
             accept_invalid_certs: false,
             auth_info: AuthInfo {
@@ -254,10 +278,14 @@ impl Config {
             root_cert = Some(ca_bundle);
         }
 
+        #[allow(deprecated)]
         Ok(Self {
             cluster_url,
             default_namespace,
             root_cert,
+            connect_timeout: Some(DEFAULT_CONNECT_TIMEOUT),
+            read_timeout: Some(DEFAULT_READ_TIMEOUT),
+            write_timeout: None,
             timeout: Some(DEFAULT_TIMEOUT),
             accept_invalid_certs,
             proxy_url: loader.proxy_url()?,
@@ -325,6 +353,8 @@ fn certs(data: &[u8]) -> Result<Vec<Vec<u8>>, pem::PemError> {
 // https://github.com/kube-rs/kube-rs/issues/146#issuecomment-590924397
 /// Default Timeout
 const DEFAULT_TIMEOUT: Duration = Duration::from_secs(295);
+const DEFAULT_CONNECT_TIMEOUT: Duration = Duration::from_secs(30);
+const DEFAULT_READ_TIMEOUT: Duration = Duration::from_secs(295);
 
 // temporary catalina hack for openssl only
 #[cfg(all(target_os = "macos", feature = "native-tls"))]