PoC: Create an Api::dynamic and Api::cluster + scope constrain Api::all

examples/crd_api.rs

@@ -43,7 +43,7 @@ async fn main() -> Result<()> {
     let client = Client::try_default().await?;
 
     // Manage CRDs first
-    let crds: Api<CustomResourceDefinition> = Api::all(client.clone());
+    let crds: Api<CustomResourceDefinition> = Api::cluster(client.clone());
 
     // Delete any old versions of it first:
     let dp = DeleteParams::default();

examples/crd_apply.rs

@@ -40,7 +40,7 @@ async fn main() -> anyhow::Result<()> {
 
     // 0. Ensure the CRD is installed (you probably just want to do this on CI)
     // (crd file can be created by piping `Foo::crd`'s yaml ser to kubectl apply)
-    let crds: Api<CustomResourceDefinition> = Api::all(client.clone());
+    let crds: Api<CustomResourceDefinition> = Api::cluster(client.clone());
     info!("Creating crd: {}", serde_yaml::to_string(&Foo::crd())?);
     crds.patch("foos.clux.dev", &ssapply, &Patch::Apply(Foo::crd()))
         .await?;

examples/crd_derive_multi.rs

@@ -101,7 +101,7 @@ async fn main() -> anyhow::Result<()> {
 }
 
 async fn apply_crd(client: Client, crd: CustomResourceDefinition) -> anyhow::Result<()> {
-    let crds: Api<CustomResourceDefinition> = Api::all(client.clone());
+    let crds: Api<CustomResourceDefinition> = Api::cluster(client.clone());
     info!("Creating crd: {}", serde_yaml::to_string(&crd)?);
     let ssapply = PatchParams::apply("crd_derive_multi").force();
     crds.patch("manyderives.kube.rs", &ssapply, &Patch::Apply(&crd))
@@ -116,7 +116,7 @@ async fn apply_crd(client: Client, crd: CustomResourceDefinition) -> anyhow::Res
 }
 
 async fn cleanup(client: Client) -> anyhow::Result<()> {
-    let crds: Api<CustomResourceDefinition> = Api::all(client.clone());
+    let crds: Api<CustomResourceDefinition> = Api::cluster(client.clone());
     let obj = crds.delete("manyderives.kube.rs", &Default::default()).await?;
     if let either::Either::Left(o) = obj {
         let uid = o.uid().unwrap();

examples/crd_derive_schema.rs

@@ -220,7 +220,7 @@ async fn main() -> Result<()> {
 
 // Create CRD and wait for it to be ready.
 async fn create_crd(client: Client) -> Result<CustomResourceDefinition> {
-    let api = Api::<CustomResourceDefinition>::all(client);
+    let api = Api::<CustomResourceDefinition>::cluster(client);
     api.create(&PostParams::default(), &Foo::crd()).await?;
 
     // Wait until it's accepted and established by the api-server
@@ -235,7 +235,7 @@ async fn create_crd(client: Client) -> Result<CustomResourceDefinition> {
 
 // Delete the CRD if it exists and wait until it's deleted.
 async fn delete_crd(client: Client) -> Result<()> {
-    let api = Api::<CustomResourceDefinition>::all(client);
+    let api = Api::<CustomResourceDefinition>::cluster(client);
     if api.get("foos.clux.dev").await.is_ok() {
         api.delete("foos.clux.dev", &DeleteParams::default()).await?;
 

examples/crd_reflector.rs

@@ -25,7 +25,7 @@ async fn main() -> anyhow::Result<()> {
 
     // 0. Ensure the CRD is installed (you probably just want to do this on CI)
     // (crd file can be created by piping `Foo::crd`'s yaml ser to kubectl apply)
-    let crds: Api<CustomResourceDefinition> = Api::all(client.clone());
+    let crds: Api<CustomResourceDefinition> = Api::cluster(client.clone());
     info!("Creating crd: {}", serde_yaml::to_string(&Foo::crd())?);
     let ssapply = PatchParams::apply("crd_reflector_example").force();
     crds.patch("foos.clux.dev", &ssapply, &Patch::Apply(Foo::crd()))

examples/dynamic_api.rs

@@ -2,7 +2,7 @@
 
 use kube::{
     api::{Api, DynamicObject, ResourceExt},
-    discovery::{verbs, Discovery, Scope},
+    discovery::{verbs, Discovery},
     Client,
 };
 use tracing::*;
@@ -18,11 +18,7 @@ async fn main() -> anyhow::Result<()> {
             if !caps.supports_operation(verbs::LIST) {
                 continue;
             }
-            let api: Api<DynamicObject> = if caps.scope == Scope::Cluster {
-                Api::all_with(client.clone(), &ar)
-            } else {
-                Api::default_namespaced_with(client.clone(), &ar)
-            };
+            let api: Api<DynamicObject> = Api::dynamic(client.clone(), &ar, caps.scope);
 
             info!("{}/{} : {}", group.name(), ar.version, ar.kind);
 

examples/kubectl.rs

@@ -228,12 +228,17 @@ fn dynamic_api(
     ns: Option<&str>,
     all: bool,
 ) -> Api<DynamicObject> {
-    if caps.scope == Scope::Cluster || all {
-        Api::all_with(client, &ar)
-    } else if let Some(namespace) = ns {
-        Api::namespaced_with(client, namespace, &ar)
-    } else {
-        Api::default_namespaced_with(client, &ar)
+    match caps.scope {
+        Scope::Cluster => Api::cluster_with(client, &ar),
+        Scope::Namespaced => {
+            if all {
+                Api::all_with(client, &ar)
+            } else if let Some(namespace) = ns {
+                Api::namespaced_with(client, namespace, &ar)
+            } else {
+                Api::default_namespaced_with(client, &ar)
+            }
+        }
     }
 }
 

examples/node_reflector.rs

@@ -12,7 +12,7 @@ async fn main() -> anyhow::Result<()> {
     tracing_subscriber::fmt::init();
     let client = Client::try_default().await?;
 
-    let nodes: Api<Node> = Api::all(client.clone());
+    let nodes: Api<Node> = Api::cluster(client.clone());
     let wc = watcher::Config::default()
         .labels("kubernetes.io/arch=amd64") // filter instances by label
         .timeout(10); // short watch timeout in this example

examples/node_watcher.rs

@@ -12,7 +12,7 @@ async fn main() -> anyhow::Result<()> {
     tracing_subscriber::fmt::init();
     let client = Client::try_default().await?;
     let events: Api<Event> = Api::all(client.clone());
-    let nodes: Api<Node> = Api::all(client.clone());
+    let nodes: Api<Node> = Api::cluster(client.clone());
 
     let use_watchlist = std::env::var("WATCHLIST").map(|s| s == "1").unwrap_or(false);
     let wc = if use_watchlist {

kube-client/src/api/core_methods.rs

@@ -266,7 +266,7 @@ where
     /// # async fn wrapper() -> Result<(), Box<dyn std::error::Error>> {
     /// # let client: kube::Client = todo!();
 
-    /// let crds: Api<CustomResourceDefinition> = Api::all(client);
+    /// let crds: Api<CustomResourceDefinition> = Api::cluster(client);
     /// crds.delete("foos.clux.dev", &DeleteParams::default()).await?
     ///     .map_left(|o| println!("Deleting CRD: {:?}", o.status))
     ///     .map_right(|s| println!("Deleted CRD: {:?}", s));

kube-client/src/api/mod.rs

@@ -4,6 +4,7 @@ mod core_methods;
 #[cfg(feature = "ws")] mod remote_command;
 use std::fmt::Debug;
 
+use k8s_openapi::ClusterResourceScope;
 #[cfg(feature = "ws")] pub use remote_command::{AttachedProcess, TerminalSize};
 #[cfg(feature = "ws")] mod portforward;
 #[cfg(feature = "ws")] pub use portforward::Portforwarder;
@@ -29,6 +30,7 @@ pub mod entry;
 pub use kube_core::admission;
 pub(crate) use kube_core::params;
 pub use kube_core::{
+    discovery::Scope,
     dynamic::{ApiResource, DynamicObject},
     gvk::{GroupVersionKind, GroupVersionResource},
     metadata::{ListMeta, ObjectMeta, PartialObjectMeta, PartialObjectMetaExt, TypeMeta},
@@ -67,10 +69,51 @@ pub struct Api<K> {
 ///
 /// This generally means resources created via [`DynamicObject`](crate::api::DynamicObject).
 impl<K: Resource> Api<K> {
-    /// Cluster level resources, or resources viewed across all namespaces
+    /// Discovered scope in the default configuration
+    ///
+    /// This returns an `Api` scoped at the appropriate setting based on a [`discovery`](crate::discovery) result.
+    ///
+    /// ```no_run
+    /// # use kube::{Api, Client, discovery::{ApiResource, ApiCapabilities}, core::DynamicObject};
+    /// # let client: Client = todo!();
+    /// # let ar: ApiResource = todo!();
+    /// # let caps: ApiCapabilities = todo!();
+    /// let api: Api<DynamicObject> = Api::dynamic(client, &ar, caps.scope);
+    /// ```
+    pub fn dynamic(client: Client, dyntype: &K::DynamicType, scope: Scope) -> Self
+    where
+        K: Resource<Scope = DynamicResourceScope>,
+    {
+        match scope {
+            Scope::Cluster => Api::cluster_with(client, dyntype),
+            Scope::Namespaced => Api::default_namespaced_with(client, dyntype),
+        }
+    }
+
+    /// A list/watch only view into namespaced resources across all namespaces
     ///
     /// This function accepts `K::DynamicType` so it can be used with dynamic resources.
-    pub fn all_with(client: Client, dyntype: &K::DynamicType) -> Self {
+    pub fn all_with(client: Client, dyntype: &K::DynamicType) -> Self
+    where
+        K: Resource<Scope = DynamicResourceScope>,
+    {
+        let url = K::url_path(dyntype, None);
+        Self {
+            client,
+            request: Request::new(url),
+            namespace: None,
+            _phantom: std::iter::empty(),
+        }
+    }
+
+    /// Cluster level resources
+    ///
+    /// This function accepts `K::DynamicType` so it can be used with dynamic resources.
+    pub fn cluster_with(client: Client, dyntype: &K::DynamicType) -> Self
+    where
+        K: Resource<Scope = DynamicResourceScope>,
+    {
+        // TODO: inspect dyntype scope to verify somehow?
         let url = K::url_path(dyntype, None);
         Self {
             client,
@@ -130,27 +173,35 @@ impl<K: Resource> Api<K>
 where
     <K as Resource>::DynamicType: Default,
 {
-    /// Cluster level resources, or resources viewed across all namespaces
-    ///
-    /// Namespace scoped resource allowing querying across all namespaces:
+    /// Cluster scoped resource
     ///
     /// ```no_run
     /// # use kube::{Api, Client};
     /// # let client: Client = todo!();
-    /// use k8s_openapi::api::core::v1::Pod;
-    /// let api: Api<Pod> = Api::all(client);
+    /// use k8s_openapi::api::core::v1::Node;
+    /// let api: Api<Node> = Api::cluster(client);
     /// ```
     ///
-    /// Cluster scoped resources also use this entrypoint:
+    /// This will ONLY work on cluster level resources as set by `Scope`:
     ///
-    /// ```no_run
+    /// ```compile_fail
     /// # use kube::{Api, Client};
     /// # let client: Client = todo!();
-    /// use k8s_openapi::api::core::v1::Node;
-    /// let api: Api<Node> = Api::all(client);
+    /// use k8s_openapi::api::core::v1::Pod;
+    /// let api: Api<Pod> = Api::cluster(client); // resource not cluster-level!
     /// ```
-    pub fn all(client: Client) -> Self {
-        Self::all_with(client, &K::DynamicType::default())
+    pub fn cluster(client: Client) -> Self
+    where
+        K: Resource<Scope = ClusterResourceScope>,
+    {
+        let dyntype = K::DynamicType::default();
+        let url = K::url_path(&dyntype, None);
+        Self {
+            client,
+            request: Request::new(url),
+            namespace: None,
+            _phantom: std::iter::empty(),
+        }
     }
 
     /// Namespaced resource within a given namespace
@@ -214,6 +265,44 @@ where
         let ns = client.default_namespace().to_string();
         Self::namespaced(client, &ns)
     }
+
+    /// A list/watch only view into namespaced resources across all namespaces
+    ///
+    /// ```no_run
+    /// # use kube::{Api, Client};
+    /// # let client: Client = todo!();
+    /// use k8s_openapi::api::core::v1::Pod;
+    /// let api: Api<Pod> = Api::all(client.clone());
+    /// ```
+    ///
+    /// This will ONLY work on namespaced resources as set by `Scope`:
+    ///
+    /// ```compile_fail
+    /// # use kube::{Api, Client};
+    /// # let client: Client = todo!();
+    /// use k8s_openapi::api::core::v1::Node;
+    /// let api: Api<Node> = Api::all(client); // resource not namespaced!
+    /// ```
+    ///
+    /// See [`Api::cluster`] for cluster level resources.
+    ///
+    /// # Warning
+    ///
+    /// This variant **can only `list` and `watch` namespaced resources** and is commonly used with a [`watcher`].
+    /// If you need to create/patch/replace/get on a namespaced resource, you need a separate `Api::namespaced`.
+    pub fn all(client: Client) -> Self
+    where
+        K: Resource<Scope = NamespaceResourceScope>,
+    {
+        let dyntype = K::DynamicType::default();
+        let url = K::url_path(&dyntype, None);
+        Self {
+            client,
+            request: Request::new(url),
+            namespace: None,
+            _phantom: std::iter::empty(),
+        }
+    }
 }
 
 impl<K> From<Api<K>> for Client {
@@ -254,9 +343,10 @@ mod test {
         let (mock_service, _handle) = mock::pair::<Request<Body>, Response<Body>>();
         let client = Client::new(mock_service, "default");
 
-        let _: Api<corev1::Node> = Api::all(client.clone());
+        let _: Api<corev1::Node> = Api::cluster(client.clone());
         let _: Api<corev1::Pod> = Api::default_namespaced(client.clone());
-        let _: Api<corev1::PersistentVolume> = Api::all(client.clone());
+        let _: Api<corev1::Pod> = Api::all(client.clone());
+        let _: Api<corev1::PersistentVolume> = Api::cluster(client.clone());
         let _: Api<corev1::ConfigMap> = Api::namespaced(client, "default");
     }
 }

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

@@ -86,7 +86,7 @@ mod test {
             },
         }))?;
 
-        let nodes: Api<Node> = Api::all(client.clone());
+        let nodes: Api<Node> = Api::cluster(client.clone());
         nodes.create(&PostParams::default(), &fake_node).await?;
 
         let schedulables = ListParams::default().fields("spec.unschedulable==false");
@@ -114,7 +114,7 @@ mod test {
         let audiences = vec!["api".to_string()];
 
         let serviceaccounts: Api<ServiceAccount> = Api::namespaced(client.clone(), serviceaccount_namespace);
-        let tokenreviews: Api<TokenReview> = Api::all(client);
+        let tokenreviews: Api<TokenReview> = Api::cluster(client);
 
         // Create ServiceAccount
         let fake_sa = serde_json::from_value(json!({

kube-client/src/discovery/apigroup.rs

@@ -46,7 +46,7 @@ use std::{cmp::Reverse, collections::HashMap, iter::Iterator};
 ///     let client = Client::try_default().await?;
 ///     let apigroup = discovery::group(&client, "apiregistration.k8s.io").await?;
 ///     let (ar, caps) = apigroup.recommended_kind("APIService").unwrap();
-///     let api: Api<DynamicObject> = Api::all_with(client.clone(), &ar);
+///     let api: Api<DynamicObject> = Api::dynamic(client.clone(), &ar, caps.scope);
 ///     for service in api.list(&Default::default()).await? {
 ///         println!("Found APIService: {}", service.name_any());
 ///     }
@@ -238,7 +238,7 @@ impl ApiGroup {
     ///         if !caps.supports_operation(verbs::LIST) {
     ///             continue;
     ///         }
-    ///         let api: Api<DynamicObject> = Api::all_with(client.clone(), &ar);
+    ///         let api: Api<DynamicObject> = Api::dynamic(client.clone(), &ar, caps.scope);
     ///         for inst in api.list(&Default::default()).await? {
     ///             println!("Found {}: {}", ar.kind, inst.name_any());
     ///         }
@@ -267,7 +267,7 @@ impl ApiGroup {
     ///         if !caps.supports_operation(verbs::LIST) {
     ///             continue;
     ///         }
-    ///         let api: Api<DynamicObject> = Api::all_with(client.clone(), &ar);
+    ///         let api: Api<DynamicObject> = Api::dynamic(client.clone(), &ar, caps.scope);
     ///         for inst in api.list(&Default::default()).await? {
     ///             println!("Found {}: {}", ar.kind, inst.name_any());
     ///         }

kube-client/src/discovery/oneshot.rs

@@ -30,7 +30,7 @@ use kube_core::{
 ///     let client = Client::try_default().await?;
 ///     let apigroup = discovery::group(&client, "apiregistration.k8s.io").await?;
 ///     let (ar, caps) = apigroup.recommended_kind("APIService").unwrap();
-///     let api: Api<DynamicObject> = Api::all_with(client.clone(), &ar);
+///     let api: Api<DynamicObject> = Api::dynamic(client.clone(), &ar, caps.scope);
 ///     for service in api.list(&Default::default()).await? {
 ///         println!("Found APIService: {}", service.name_any());
 ///     }
@@ -67,7 +67,7 @@ pub async fn group(client: &Client, apigroup: &str) -> Result<ApiGroup> {
 ///     let gv = "apiregistration.k8s.io/v1".parse()?;
 ///     let apigroup = discovery::pinned_group(&client, &gv).await?;
 ///     let (ar, caps) = apigroup.recommended_kind("APIService").unwrap();
-///     let api: Api<DynamicObject> = Api::all_with(client.clone(), &ar);
+///     let api: Api<DynamicObject> = Api::dynamic(client.clone(), &ar, caps.scope);
 ///     for service in api.list(&Default::default()).await? {
 ///         println!("Found APIService: {}", service.name_any());
 ///     }
@@ -94,7 +94,7 @@ pub async fn pinned_group(client: &Client, gv: &GroupVersion) -> Result<ApiGroup
 ///     let client = Client::try_default().await?;
 ///     let gvk = GroupVersionKind::gvk("apiregistration.k8s.io", "v1", "APIService");
 ///     let (ar, caps) = discovery::pinned_kind(&client, &gvk).await?;
-///     let api: Api<DynamicObject> = Api::all_with(client.clone(), &ar);
+///     let api: Api<DynamicObject> = Api::dynamic(client.clone(), &ar, caps.scope);
 ///     for service in api.list(&Default::default()).await? {
 ///         println!("Found APIService: {}", service.name_any());
 ///     }

kube-client/src/lib.rs

@@ -574,7 +574,7 @@ mod test {
         }))?;
 
         let client = Client::try_default().await?;
-        let csr: Api<CertificateSigningRequest> = Api::all(client.clone());
+        let csr: Api<CertificateSigningRequest> = Api::cluster(client.clone());
         assert!(csr.create(&PostParams::default(), &dummy_csr).await.is_ok());
 
         // Patch the approval and approve the CSR