multi_watcher example - closes #302

Makefile

@@ -35,7 +35,9 @@ publish:
 
 minikube-create:
 	sudo rm -rf /tmp/juju-mk* /tmp/minikube*
-	sudo -E minikube start --driver=none --kubernetes-version v1.18.6 --extra-config kubeadm.ignore-preflight-errors=SystemVerification
+	minikube start --driver=docker \
+		--kubernetes-version v1.18.6 \
+		--extra-config kubeadm.ignore-preflight-errors=SystemVerification
 
 minikube:
 	kubectl config set-context --cluster=minikube --user=minikube --namespace=apps minikube

examples/Cargo.toml

@@ -76,6 +76,10 @@ path = "job_api.rs"
 name = "log_stream"
 path = "log_stream.rs"
 
+[[example]]
+name = "multi_watcher"
+path = "multi_watcher.rs"
+
 [[example]]
 name = "pod_api"
 path = "pod_api.rs"

examples/README.md

@@ -29,11 +29,13 @@ These example watch a single resource and does some basic filtering on the watch
 
 ```sh
 # watch all configmap events in a namespace
-cargo run --example configmap_watcher
+NAMESPACE=dev cargo run --example configmap_watcher
 # watch unready pods in a namespace
 NAMESPACE=dev cargo run --example pod_watcher
 # watch all event events
 cargo run --example event_watcher
+# watch deployments, configmaps, secrets in one namespace
+NAMESPACE=dev cargo run --example multi_watcher
 # watch broken nodes and cross reference with events api
 cargo run --example node_watcher
 ```

examples/multi_watcher.rs

@@ -0,0 +1,46 @@
+#[macro_use] extern crate log;
+use futures::{stream, StreamExt, TryStreamExt};
+use k8s_openapi::api::{
+    apps::v1::Deployment,
+    core::v1::{ConfigMap, Secret},
+};
+use kube::{
+    api::{Api, ListParams, Meta},
+    Client,
+};
+use kube_runtime::{utils::try_flatten_applied, watcher};
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    std::env::set_var("RUST_LOG", "info,node_watcher=debug,kube=debug");
+    env_logger::init();
+    let client = Client::try_default().await?;
+    let namespace = std::env::var("NAMESPACE").unwrap_or("default".into());
+
+    let deploys: Api<Deployment> = Api::namespaced(client.clone(), &namespace);
+    let cms: Api<ConfigMap> = Api::namespaced(client.clone(), &namespace);
+    let secret: Api<Secret> = Api::namespaced(client.clone(), &namespace);
+    let dep_watcher = watcher(deploys, ListParams::default());
+    let cm_watcher = watcher(cms, ListParams::default());
+    let sec_watcher = watcher(secret, ListParams::default());
+
+    // select on applied events from all watchers
+    enum Watched {
+        Config(ConfigMap),
+        Deploy(Deployment),
+        Secret(Secret),
+    }
+    let mut combo_stream = stream::select_all(vec![
+        try_flatten_applied(dep_watcher).map_ok(Watched::Deploy).boxed(),
+        try_flatten_applied(cm_watcher).map_ok(Watched::Config).boxed(),
+        try_flatten_applied(sec_watcher).map_ok(Watched::Secret).boxed(),
+    ]);
+    while let Some(o) = combo_stream.try_next().await? {
+        match o {
+            Watched::Config(cm) => info!("Got configmap: {}", Meta::name(&cm)),
+            Watched::Deploy(d) => info!("Got deployment: {}", Meta::name(&d)),
+            Watched::Secret(s) => info!("Got secret: {}", Meta::name(&s)),
+        }
+    }
+    Ok(())
+}