Split ListParams and WatchParams (#1162)

* Add resourceVersion to ListParams Signed-off-by: m.nabokikh <maksim.nabokikh@flant.com> * Apply rustfmt Signed-off-by: m.nabokikh <maksim.nabokikh@flant.com> * More validations for resource version options Signed-off-by: m.nabokikh <maksim.nabokikh@flant.com> * Add hack to allow the watcher to work Signed-off-by: m.nabokikh <maksim.nabokikh@flant.com> * Split ListParams and WatchParams Signed-off-by: m.nabokikh <maksim.nabokikh@flant.com> * Fix doc tests and rustfmt Signed-off-by: m.nabokikh <maksim.nabokikh@flant.com> * Remove continue_tooken and limit from watcher config Signed-off-by: m.nabokikh <maksim.nabokikh@flant.com> * Fix clippy errors, add more tests Signed-off-by: m.nabokikh <maksim.nabokikh@flant.com> * Migrate to 4 varients enum Signed-off-by: m.nabokikh <maksim.nabokikh@flant.com> * Add more tests and derive ListParams default Signed-off-by: m.nabokikh <maksim.nabokikh@flant.com> --------- Signed-off-by: m.nabokikh <maksim.nabokikh@flant.com>
kube-rs · Mar 27, 2023 · 30a0c39 · 30a0c39
1 parent 2e5e4de
commit 30a0c39
Show file tree

Hide file tree

Showing 26 changed files with 593 additions and 247 deletions.
diff --git a/examples/configmapgen_controller.rs b/examples/configmapgen_controller.rs
@@ -5,8 +5,11 @@ use anyhow::Result;
 use futures::StreamExt;
 use k8s_openapi::api::core::v1::ConfigMap;
 use kube::{
-    api::{Api, ListParams, ObjectMeta, Patch, PatchParams, Resource},
-    runtime::controller::{Action, Controller},
+    api::{Api, ObjectMeta, Patch, PatchParams, Resource},
+    runtime::{
+        controller::{Action, Controller},
+        watcher,
+    },
     Client, CustomResource,
 };
 use schemars::JsonSchema;
@@ -99,8 +102,8 @@ async fn main() -> Result<()> {
         }
     });
 
-    Controller::new(cmgs, ListParams::default())
-        .owns(cms, ListParams::default())
+    Controller::new(cmgs, watcher::Config::default())
+        .owns(cms, watcher::Config::default())
         .reconcile_all_on(reload_rx.map(|_| ()))
         .shutdown_on_signal()
         .run(reconcile, error_policy, Arc::new(Data { client }))

diff --git a/examples/crd_derive_schema.rs b/examples/crd_derive_schema.rs
@@ -3,8 +3,8 @@ use futures::{StreamExt, TryStreamExt};
 use k8s_openapi::apiextensions_apiserver::pkg::apis::apiextensions::v1::CustomResourceDefinition;
 use kube::{
     api::{
-        Api, ApiResource, DeleteParams, DynamicObject, GroupVersionKind, ListParams, Patch, PatchParams,
-        PostParams, WatchEvent,
+        Api, ApiResource, DeleteParams, DynamicObject, GroupVersionKind, Patch, PatchParams, PostParams,
+        WatchEvent, WatchParams,
     },
     runtime::wait::{await_condition, conditions},
     Client, CustomResource, CustomResourceExt,
@@ -241,10 +241,10 @@ async fn delete_crd(client: Client) -> Result<()> {
 
         // Wait until deleted
         let timeout_secs = 15;
-        let lp = ListParams::default()
+        let wp = WatchParams::default()
             .fields("metadata.name=foos.clux.dev")
             .timeout(timeout_secs);
-        let mut stream = api.watch(&lp, "0").await?.boxed_local();
+        let mut stream = api.watch(&wp, "0").await?.boxed_local();
         while let Some(status) = stream.try_next().await? {
             if let WatchEvent::Deleted(_) = status {
                 return Ok(());

diff --git a/examples/crd_reflector.rs b/examples/crd_reflector.rs
@@ -3,7 +3,7 @@ use k8s_openapi::apiextensions_apiserver::pkg::apis::apiextensions::v1::CustomRe
 use tracing::*;
 
 use kube::{
-    api::{Api, ListParams, Patch, PatchParams, ResourceExt},
+    api::{Api, Patch, PatchParams, ResourceExt},
     runtime::{reflector, watcher, WatchStreamExt},
     Client, CustomResource, CustomResourceExt,
 };
@@ -36,8 +36,8 @@ async fn main() -> anyhow::Result<()> {
     let (reader, writer) = reflector::store::<Foo>();
 
     let foos: Api<Foo> = Api::default_namespaced(client);
-    let lp = ListParams::default().timeout(20); // low timeout in this example
-    let rf = reflector(writer, watcher(foos, lp));
+    let wc = watcher::Config::default().timeout(20); // low timeout in this example
+    let rf = reflector(writer, watcher(foos, wc));
 
     tokio::spawn(async move {
         loop {

diff --git a/examples/dynamic_watcher.rs b/examples/dynamic_watcher.rs
@@ -1,6 +1,6 @@
 use futures::{Stream, StreamExt, TryStreamExt};
 use kube::{
-    api::{Api, DynamicObject, GroupVersionKind, ListParams, Resource, ResourceExt},
+    api::{Api, DynamicObject, GroupVersionKind, Resource, ResourceExt},
     runtime::{metadata_watcher, watcher, watcher::Event, WatchStreamExt},
 };
 use serde::de::DeserializeOwned;
@@ -28,13 +28,13 @@ async fn main() -> anyhow::Result<()> {
 
     // Use the full resource info to create an Api with the ApiResource as its DynamicType
     let api = Api::<DynamicObject>::all_with(client, &ar);
-    let lp = ListParams::default();
+    let wc = watcher::Config::default();
 
     // Start a metadata or a full resource watch
     if watch_metadata {
-        handle_events(metadata_watcher(api, lp)).await
+        handle_events(metadata_watcher(api, wc)).await
     } else {
-        handle_events(watcher(api, lp)).await
+        handle_events(watcher(api, wc)).await
     }
 }
 

diff --git a/examples/event_watcher.rs b/examples/event_watcher.rs
@@ -1,7 +1,7 @@
 use futures::{pin_mut, TryStreamExt};
 use k8s_openapi::api::core::v1::Event;
 use kube::{
-    api::{Api, ListParams},
+    api::Api,
     runtime::{watcher, WatchStreamExt},
     Client,
 };
@@ -13,9 +13,9 @@ async fn main() -> anyhow::Result<()> {
     let client = Client::try_default().await?;
 
     let events: Api<Event> = Api::all(client);
-    let lp = ListParams::default();
+    let wc = watcher::Config::default();
 
-    let ew = watcher(events, lp).applied_objects();
+    let ew = watcher(events, wc).applied_objects();
 
     pin_mut!(ew);
     while let Some(event) = ew.try_next().await? {

diff --git a/examples/kubectl.rs b/examples/kubectl.rs
@@ -123,12 +123,12 @@ impl App {
         Ok(())
     }
 
-    async fn watch(&self, api: Api<DynamicObject>, mut lp: ListParams) -> Result<()> {
+    async fn watch(&self, api: Api<DynamicObject>, mut wc: watcher::Config) -> Result<()> {
         if let Some(n) = &self.name {
-            lp = lp.fields(&format!("metadata.name={n}"));
+            wc = wc.fields(&format!("metadata.name={n}"));
         }
         // present a dumb table for it for now. kubectl does not do this anymore.
-        let mut stream = watcher(api, lp).applied_objects().boxed();
+        let mut stream = watcher(api, wc).applied_objects().boxed();
         println!("{0:<width$} {1:<20}", "NAME", "AGE", width = 63);
         while let Some(inst) = stream.try_next().await? {
             let age = format_creation_since(inst.creation_timestamp());
@@ -201,14 +201,20 @@ async fn main() -> Result<()> {
         if let Some(label) = &app.selector {
             lp = lp.labels(label);
         }
+
+        let mut wc = watcher::Config::default();
+        if let Some(label) = &app.selector {
+            wc = wc.labels(label);
+        }
+
         let api = dynamic_api(ar, caps, client, &app.namespace, app.all);
 
         tracing::info!(?app.verb, ?resource, name = ?app.name.clone().unwrap_or_default(), "requested objects");
         match app.verb {
             Verb::Edit => app.edit(api).await?,
             Verb::Get => app.get(api, lp).await?,
             Verb::Delete => app.delete(api, lp).await?,
-            Verb::Watch => app.watch(api, lp).await?,
+            Verb::Watch => app.watch(api, wc).await?,
             Verb::Apply => bail!("verb {:?} cannot act on an explicit resource", app.verb),
         }
     } else if app.verb == Verb::Apply {

diff --git a/examples/multi_watcher.rs b/examples/multi_watcher.rs
@@ -4,7 +4,7 @@ use k8s_openapi::api::{
     core::v1::{ConfigMap, Secret},
 };
 use kube::{
-    api::{Api, ListParams, ResourceExt},
+    api::{Api, ResourceExt},
     runtime::{watcher, WatchStreamExt},
     Client,
 };
@@ -18,9 +18,9 @@ async fn main() -> anyhow::Result<()> {
     let deploys: Api<Deployment> = Api::default_namespaced(client.clone());
     let cms: Api<ConfigMap> = Api::default_namespaced(client.clone());
     let secret: Api<Secret> = Api::default_namespaced(client.clone());
-    let dep_watcher = watcher(deploys, ListParams::default());
-    let cm_watcher = watcher(cms, ListParams::default());
-    let sec_watcher = watcher(secret, ListParams::default());
+    let dep_watcher = watcher(deploys, watcher::Config::default());
+    let cm_watcher = watcher(cms, watcher::Config::default());
+    let sec_watcher = watcher(secret, watcher::Config::default());
 
     // select on applied events from all watchers
     let mut combo_stream = stream::select_all(vec![

diff --git a/examples/node_reflector.rs b/examples/node_reflector.rs
@@ -1,7 +1,7 @@
 use futures::{StreamExt, TryStreamExt};
 use k8s_openapi::api::core::v1::Node;
 use kube::{
-    api::{Api, ListParams, ResourceExt},
+    api::{Api, ResourceExt},
     runtime::{reflector, watcher, WatchStreamExt},
     Client,
 };
@@ -13,12 +13,12 @@ async fn main() -> anyhow::Result<()> {
     let client = Client::try_default().await?;
 
     let nodes: Api<Node> = Api::all(client.clone());
-    let lp = ListParams::default()
+    let wc = watcher::Config::default()
         .labels("kubernetes.io/arch=amd64") // filter instances by label
         .timeout(10); // short watch timeout in this example
 
     let (reader, writer) = reflector::store();
-    let rf = reflector(writer, watcher(nodes, lp));
+    let rf = reflector(writer, watcher(nodes, wc));
 
     // Periodically read our state in the background
     tokio::spawn(async move {

diff --git a/examples/node_watcher.rs b/examples/node_watcher.rs
@@ -15,8 +15,8 @@ async fn main() -> anyhow::Result<()> {
     let events: Api<Event> = Api::all(client.clone());
     let nodes: Api<Node> = Api::all(client.clone());
 
-    let lp = ListParams::default().labels("beta.kubernetes.io/arch=amd64");
-    let obs = watcher(nodes, lp)
+    let wc = watcher::Config::default().labels("beta.kubernetes.io/arch=amd64");
+    let obs = watcher(nodes, wc)
         .backoff(ExponentialBackoff::default())
         .applied_objects();
 

diff --git a/examples/pod_attach.rs b/examples/pod_attach.rs
@@ -6,7 +6,7 @@ use k8s_openapi::api::core::v1::Pod;
 
 use kube::{
     api::{
-        Api, AttachParams, AttachedProcess, DeleteParams, ListParams, PostParams, ResourceExt, WatchEvent,
+        Api, AttachParams, AttachedProcess, DeleteParams, PostParams, ResourceExt, WatchEvent, WatchParams,
     },
     Client,
 };
@@ -35,8 +35,8 @@ async fn main() -> anyhow::Result<()> {
     pods.create(&PostParams::default(), &p).await?;
 
     // Wait until the pod is running, otherwise we get 500 error.
-    let lp = ListParams::default().fields("metadata.name=example").timeout(10);
-    let mut stream = pods.watch(&lp, "0").await?.boxed();
+    let wp = WatchParams::default().fields("metadata.name=example").timeout(10);
+    let mut stream = pods.watch(&wp, "0").await?.boxed();
     while let Some(status) = stream.try_next().await? {
         match status {
             WatchEvent::Added(o) => {

diff --git a/examples/pod_cp.rs b/examples/pod_cp.rs
@@ -3,7 +3,7 @@ use k8s_openapi::api::core::v1::Pod;
 use tracing::*;
 
 use kube::{
-    api::{Api, AttachParams, DeleteParams, ListParams, PostParams, ResourceExt, WatchEvent},
+    api::{Api, AttachParams, DeleteParams, PostParams, ResourceExt, WatchEvent, WatchParams},
     Client,
 };
 use tokio::io::AsyncWriteExt;
@@ -34,8 +34,8 @@ async fn main() -> anyhow::Result<()> {
     pods.create(&PostParams::default(), &p).await?;
 
     // Wait until the pod is running, otherwise we get 500 error.
-    let lp = ListParams::default().fields("metadata.name=example").timeout(10);
-    let mut stream = pods.watch(&lp, "0").await?.boxed();
+    let wp = WatchParams::default().fields("metadata.name=example").timeout(10);
+    let mut stream = pods.watch(&wp, "0").await?.boxed();
     while let Some(status) = stream.try_next().await? {
         match status {
             WatchEvent::Added(o) => {

diff --git a/examples/pod_evict.rs b/examples/pod_evict.rs
@@ -4,7 +4,7 @@ use serde_json::json;
 use tracing::*;
 
 use kube::{
-    api::{Api, EvictParams, ListParams, PostParams, ResourceExt, WatchEvent},
+    api::{Api, EvictParams, PostParams, ResourceExt, WatchEvent, WatchParams},
     Client,
 };
 
@@ -35,10 +35,10 @@ async fn main() -> anyhow::Result<()> {
     pods.create(&pp, &empty_pod).await?;
 
     // Wait until the pod is running, although it's not necessary
-    let lp = ListParams::default()
+    let wp = WatchParams::default()
         .fields("metadata.name=empty-pod")
         .timeout(10);
-    let mut stream = pods.watch(&lp, "0").await?.boxed();
+    let mut stream = pods.watch(&wp, "0").await?.boxed();
     while let Some(status) = stream.try_next().await? {
         match status {
             WatchEvent::Added(o) => {

diff --git a/examples/pod_exec.rs b/examples/pod_exec.rs
@@ -4,7 +4,7 @@ use tracing::*;
 
 use kube::{
     api::{
-        Api, AttachParams, AttachedProcess, DeleteParams, ListParams, PostParams, ResourceExt, WatchEvent,
+        Api, AttachParams, AttachedProcess, DeleteParams, PostParams, ResourceExt, WatchEvent, WatchParams,
     },
     Client,
 };
@@ -34,8 +34,8 @@ async fn main() -> anyhow::Result<()> {
     pods.create(&PostParams::default(), &p).await?;
 
     // Wait until the pod is running, otherwise we get 500 error.
-    let lp = ListParams::default().fields("metadata.name=example").timeout(10);
-    let mut stream = pods.watch(&lp, "0").await?.boxed();
+    let wp = WatchParams::default().fields("metadata.name=example").timeout(10);
+    let mut stream = pods.watch(&wp, "0").await?.boxed();
     while let Some(status) = stream.try_next().await? {
         match status {
             WatchEvent::Added(o) => {

diff --git a/examples/pod_reflector.rs b/examples/pod_reflector.rs
@@ -1,7 +1,7 @@
 use futures::TryStreamExt;
 use k8s_openapi::api::core::v1::Pod;
 use kube::{
-    api::{Api, ListParams},
+    api::Api,
     runtime::{reflector, watcher, WatchStreamExt},
     Client, ResourceExt,
 };
@@ -28,7 +28,7 @@ async fn main() -> anyhow::Result<()> {
         }
     });
 
-    let stream = watcher(api, ListParams::default()).map_ok(|ev| {
+    let stream = watcher(api, watcher::Config::default()).map_ok(|ev| {
         ev.modify(|pod| {
             // memory optimization for our store - we don't care about fields/annotations/status
             pod.managed_fields_mut().clear();

diff --git a/examples/pod_shell.rs b/examples/pod_shell.rs
@@ -3,7 +3,7 @@ use k8s_openapi::api::core::v1::Pod;
 use tracing::*;
 
 use kube::{
-    api::{Api, AttachParams, DeleteParams, ListParams, PostParams, ResourceExt, WatchEvent},
+    api::{Api, AttachParams, DeleteParams, PostParams, ResourceExt, WatchEvent, WatchParams},
     Client,
 };
 
@@ -31,8 +31,8 @@ async fn main() -> anyhow::Result<()> {
     pods.create(&PostParams::default(), &p).await?;
 
     // Wait until the pod is running, otherwise we get 500 error.
-    let lp = ListParams::default().fields("metadata.name=example").timeout(10);
-    let mut stream = pods.watch(&lp, "0").await?.boxed();
+    let wp = WatchParams::default().fields("metadata.name=example").timeout(10);
+    let mut stream = pods.watch(&wp, "0").await?.boxed();
     while let Some(status) = stream.try_next().await? {
         match status {
             WatchEvent::Added(o) => {

diff --git a/examples/pod_watcher.rs b/examples/pod_watcher.rs
@@ -1,7 +1,7 @@
 use futures::prelude::*;
 use k8s_openapi::api::core::v1::Pod;
 use kube::{
-    api::{Api, ListParams, ResourceExt},
+    api::{Api, ResourceExt},
     runtime::{watcher, WatchStreamExt},
     Client,
 };
@@ -13,7 +13,7 @@ async fn main() -> anyhow::Result<()> {
     let client = Client::try_default().await?;
     let api = Api::<Pod>::default_namespaced(client);
 
-    watcher(api, ListParams::default())
+    watcher(api, watcher::Config::default())
         .applied_objects()
         .try_for_each(|p| async move {
             info!("saw {}", p.name_any());

diff --git a/examples/secret_reflector.rs b/examples/secret_reflector.rs
@@ -1,7 +1,7 @@
 use futures::TryStreamExt;
 use k8s_openapi::api::core::v1::Secret;
 use kube::{
-    api::{Api, ListParams, ResourceExt},
+    api::{Api, ResourceExt},
     runtime::{reflector, reflector::Store, watcher, WatchStreamExt},
     Client,
 };
@@ -53,10 +53,10 @@ async fn main() -> anyhow::Result<()> {
     let client = Client::try_default().await?;
 
     let secrets: Api<Secret> = Api::default_namespaced(client);
-    let lp = ListParams::default().timeout(10); // short watch timeout in this example
+    let wc = watcher::Config::default().timeout(10); // short watch timeout in this example
 
     let (reader, writer) = reflector::store::<Secret>();
-    let rf = reflector(writer, watcher(secrets, lp));
+    let rf = reflector(writer, watcher(secrets, wc));
 
     spawn_periodic_reader(reader); // read from a reader in the background
 

diff --git a/examples/secret_syncer.rs b/examples/secret_syncer.rs
@@ -6,11 +6,12 @@
 use futures::StreamExt;
 use k8s_openapi::api::core::v1::{ConfigMap, Secret};
 use kube::{
-    api::{Api, DeleteParams, ListParams, ObjectMeta, Patch, PatchParams, Resource},
+    api::{Api, DeleteParams, ObjectMeta, Patch, PatchParams, Resource},
     error::ErrorResponse,
     runtime::{
         controller::{Action, Controller},
         finalizer::{finalizer, Event},
+        watcher,
     },
 };
 use std::{sync::Arc, time::Duration};
@@ -78,7 +79,7 @@ async fn main() -> anyhow::Result<()> {
     let client = kube::Client::try_default().await?;
     Controller::new(
         Api::<ConfigMap>::all(client.clone()),
-        ListParams::default().labels("configmap-secret-syncer.nullable.se/sync=true"),
+        watcher::Config::default().labels("configmap-secret-syncer.nullable.se/sync=true"),
     )
     .run(
         |cm, _| {