kube_client/
lib.rs

1//! Crate for interacting with the Kubernetes API
2//!
3//! This crate includes the tools for manipulating Kubernetes resources as
4//! well as keeping track of those resources as they change over time
5//!
6//! # Example
7//!
8//! The following example will create a [`Pod`](k8s_openapi::api::core::v1::Pod)
9//! and then watch for it to become available using a manual [`Api::watch`] call.
10//!
11//! ```rust,no_run
12//! use futures::{StreamExt, TryStreamExt};
13//! use kube_client::api::{Api, ResourceExt, ListParams, PatchParams, Patch};
14//! use kube_client::Client;
15//! use k8s_openapi::api::core::v1::Pod;
16//!
17//! #[tokio::main]
18//! async fn main() -> Result<(), Box<dyn std::error::Error>> {
19//!     // Read the environment to find config for kube client.
20//!     // Note that this tries an in-cluster configuration first,
21//!     // then falls back on a kubeconfig file.
22//!     let client = Client::try_default().await?;
23//!
24//!     // Interact with pods in the configured namespace with the typed interface from k8s-openapi
25//!     let pods: Api<Pod> = Api::default_namespaced(client);
26//!
27//!     // Create a Pod (cheating here with json, but it has to validate against the type):
28//!     let patch: Pod = serde_json::from_value(serde_json::json!({
29//!         "apiVersion": "v1",
30//!         "kind": "Pod",
31//!         "metadata": {
32//!             "name": "my-pod"
33//!         },
34//!         "spec": {
35//!             "containers": [
36//!                 {
37//!                     "name": "my-container",
38//!                     "image": "myregistry.azurecr.io/hello-world:v1",
39//!                 },
40//!             ],
41//!         }
42//!     }))?;
43//!
44//!     // Apply the Pod via server-side apply
45//!     let params = PatchParams::apply("myapp");
46//!     let result = pods.patch("my-pod", &params, &Patch::Apply(&patch)).await?;
47//!
48//!     // List pods in the configured namespace
49//!     for p in pods.list(&ListParams::default()).await? {
50//!         println!("found pod {}", p.name_any());
51//!     }
52//!
53//!     Ok(())
54//! }
55//! ```
56//!
57//! For more details, see:
58//!
59//! - [`Client`](crate::client) for the extensible Kubernetes client
60//! - [`Config`](crate::config) for the Kubernetes config abstraction
61//! - [`Api`](crate::Api) for the generic api methods available on Kubernetes resources
62//! - [k8s-openapi](https://docs.rs/k8s-openapi) for how to create typed kubernetes objects directly
63#![cfg_attr(docsrs, feature(doc_cfg))]
64// Nightly clippy (0.1.64) considers Drop a side effect, see https://github.com/rust-lang/rust-clippy/issues/9608
65#![allow(clippy::unnecessary_lazy_evaluations)]
66
67macro_rules! cfg_client {
68    ($($item:item)*) => {
69        $(
70            #[cfg_attr(docsrs, doc(cfg(feature = "client")))]
71            #[cfg(feature = "client")]
72            $item
73        )*
74    }
75}
76macro_rules! cfg_config {
77    ($($item:item)*) => {
78        $(
79            #[cfg_attr(docsrs, doc(cfg(feature = "config")))]
80            #[cfg(feature = "config")]
81            $item
82        )*
83    }
84}
85
86macro_rules! cfg_error {
87    ($($item:item)*) => {
88        $(
89            #[cfg_attr(docsrs, doc(cfg(any(feature = "config", feature = "client"))))]
90            #[cfg(any(feature = "config", feature = "client"))]
91            $item
92        )*
93    }
94}
95
96cfg_client! {
97    pub mod api;
98    pub mod discovery;
99    pub mod client;
100
101    #[doc(inline)]
102    pub use api::Api;
103    #[doc(inline)]
104    pub use client::Client;
105    #[doc(inline)]
106    pub use discovery::Discovery;
107}
108
109cfg_config! {
110    pub mod config;
111    #[doc(inline)]
112    pub use config::Config;
113}
114
115cfg_error! {
116    pub mod error;
117    #[doc(inline)] pub use error::Error;
118    /// Convenient alias for `Result<T, Error>`
119    pub type Result<T, E = Error> = std::result::Result<T, E>;
120}
121
122// Private, non-gated modules
123mod util;
124
125pub use crate::core::{CustomResourceExt, Resource, ResourceExt};
126/// Re-exports from kube_core
127pub use kube_core as core;
128
129// Tests that require a cluster and the complete feature set
130// Can be run with `cargo test -p kube-client --lib features=rustls-tls,ws -- --ignored`
131#[cfg(all(feature = "client", feature = "config"))]
132#[cfg(test)]
133#[allow(unused_imports)] // varying test imports depending on feature
134mod test {
135    use crate::{
136        Api, Client, Config, ResourceExt,
137        api::{AttachParams, AttachedProcess},
138        client::ConfigExt,
139    };
140    use futures::{AsyncBufRead, AsyncBufReadExt, StreamExt, TryStreamExt};
141    use k8s_openapi::api::core::v1::{EphemeralContainer, Pod, PodSpec};
142    use kube_core::{
143        params::{DeleteParams, Patch, PatchParams, PostParams, WatchParams},
144        response::StatusSummary,
145    };
146    use serde_json::json;
147    use tower::ServiceBuilder;
148
149    // hard disabled test atm due to k3d rustls issues: https://github.com/kube-rs/kube/issues?q=is%3Aopen+is%3Aissue+label%3Arustls
150    #[allow(dead_code)]
151    // #[tokio::test]
152    #[ignore = "needs cluster (lists pods)"]
153    #[cfg(feature = "rustls-tls")]
154    async fn custom_client_rustls_configuration() -> Result<(), Box<dyn std::error::Error>> {
155        use hyper_util::rt::TokioExecutor;
156
157        let config = Config::infer().await?;
158        let https = config.rustls_https_connector()?;
159        let service = ServiceBuilder::new()
160            .layer(config.base_uri_layer())
161            .service(hyper_util::client::legacy::Client::builder(TokioExecutor::new()).build(https));
162        let client = Client::new(service, config.default_namespace);
163        let pods: Api<Pod> = Api::default_namespaced(client);
164        pods.list(&Default::default()).await?;
165        Ok(())
166    }
167
168    #[tokio::test]
169    #[ignore = "needs cluster (lists pods)"]
170    #[cfg(feature = "openssl-tls")]
171    async fn custom_client_openssl_tls_configuration() -> Result<(), Box<dyn std::error::Error>> {
172        use hyper_util::rt::TokioExecutor;
173
174        let config = Config::infer().await?;
175        let https = config.openssl_https_connector()?;
176        let service = ServiceBuilder::new()
177            .layer(config.base_uri_layer())
178            .service(hyper_util::client::legacy::Client::builder(TokioExecutor::new()).build(https));
179        let client = Client::new(service, config.default_namespace);
180        let pods: Api<Pod> = Api::default_namespaced(client);
181        pods.list(&Default::default()).await?;
182        Ok(())
183    }
184
185    #[tokio::test]
186    #[ignore = "needs cluster (lists api resources)"]
187    #[cfg(feature = "client")]
188    async fn group_discovery_oneshot() -> Result<(), Box<dyn std::error::Error>> {
189        use crate::{core::DynamicObject, discovery};
190        let client = Client::try_default().await?;
191        let apigroup = discovery::group(&client, "apiregistration.k8s.io").await?;
192        let (ar, _caps) = apigroup.recommended_kind("APIService").unwrap();
193        let api: Api<DynamicObject> = Api::all_with(client.clone(), &ar);
194        api.list(&Default::default()).await?;
195
196        Ok(())
197    }
198
199    #[tokio::test]
200    #[ignore = "needs cluster (uses aggregated discovery, requires k8s 1.26+)"]
201    #[cfg(feature = "client")]
202    async fn aggregated_discovery_apis() -> Result<(), Box<dyn std::error::Error>> {
203        let client = Client::try_default().await?;
204
205        // Test /apis aggregated discovery
206        let apis_discovery = client.list_api_groups_aggregated().await?;
207        assert!(!apis_discovery.items.is_empty(), "should have API groups");
208
209        // Find the apps group
210        let apps_group = apis_discovery
211            .items
212            .iter()
213            .find(|g| g.metadata.as_ref().and_then(|m| m.name.as_ref()) == Some(&"apps".to_string()));
214        assert!(apps_group.is_some(), "should have apps group");
215
216        let apps = apps_group.unwrap();
217        assert!(!apps.versions.is_empty(), "apps should have versions");
218
219        // Check that deployments resource exists in apps/v1
220        let v1 = apps.versions.iter().find(|v| v.version == Some("v1".to_string()));
221        assert!(v1.is_some(), "apps should have v1");
222
223        let deployments = v1
224            .unwrap()
225            .resources
226            .iter()
227            .find(|r| r.resource == Some("deployments".to_string()));
228        assert!(deployments.is_some(), "apps/v1 should have deployments");
229
230        Ok(())
231    }
232
233    #[tokio::test]
234    #[ignore = "needs cluster (uses aggregated discovery, requires k8s 1.26+)"]
235    #[cfg(feature = "client")]
236    async fn aggregated_discovery_core() -> Result<(), Box<dyn std::error::Error>> {
237        let client = Client::try_default().await?;
238
239        // Test /api aggregated discovery (core group)
240        let core_discovery = client.list_core_api_versions_aggregated().await?;
241        assert!(!core_discovery.items.is_empty(), "should have core group");
242
243        let core = &core_discovery.items[0];
244        let v1 = core.versions.iter().find(|v| v.version == Some("v1".to_string()));
245        assert!(v1.is_some(), "core should have v1");
246
247        // Check that pods resource exists
248        let pods = v1
249            .unwrap()
250            .resources
251            .iter()
252            .find(|r| r.resource == Some("pods".to_string()));
253        assert!(pods.is_some(), "core/v1 should have pods");
254
255        let pods_resource = pods.unwrap();
256        assert_eq!(pods_resource.scope, Some("Namespaced".to_string()));
257        assert!(pods_resource.verbs.contains(&"list".to_string()));
258
259        Ok(())
260    }
261
262    #[tokio::test]
263    #[ignore = "needs cluster (uses aggregated discovery, requires k8s 1.26+)"]
264    #[cfg(feature = "client")]
265    async fn discovery_run_aggregated() -> Result<(), Box<dyn std::error::Error>> {
266        use crate::discovery::{Discovery, verbs};
267
268        let client = Client::try_default().await?;
269
270        // Test Discovery::run_aggregated()
271        let discovery = Discovery::new(client.clone()).run_aggregated().await?;
272
273        // Should have discovered groups
274        assert!(discovery.groups().count() > 0, "should have discovered groups");
275
276        // Should have core group
277        assert!(discovery.has_group(""), "should have core group");
278
279        // Should have apps group
280        assert!(discovery.has_group("apps"), "should have apps group");
281
282        // Check that we can find deployments in apps group
283        let apps = discovery.get("apps").expect("apps group");
284        let (ar, caps) = apps.recommended_kind("Deployment").expect("Deployment kind");
285        assert_eq!(ar.kind, "Deployment");
286        assert!(caps.supports_operation(verbs::LIST));
287
288        // Check that we can find pods in core group
289        let core = discovery.get("").expect("core group");
290        let (ar, caps) = core.recommended_kind("Pod").expect("Pod kind");
291        assert_eq!(ar.kind, "Pod");
292        assert!(caps.supports_operation(verbs::GET));
293
294        Ok(())
295    }
296
297    #[tokio::test]
298    #[ignore = "needs cluster (will create and edit a pod)"]
299    async fn pod_can_use_core_apis() -> Result<(), Box<dyn std::error::Error>> {
300        use kube::api::{DeleteParams, ListParams, Patch, PatchParams, PostParams, WatchEvent};
301
302        let client = Client::try_default().await?;
303        let pods: Api<Pod> = Api::default_namespaced(client);
304
305        // create busybox pod that's alive for at most 30s
306        let p: Pod = serde_json::from_value(json!({
307            "apiVersion": "v1",
308            "kind": "Pod",
309            "metadata": {
310                "name": "busybox-kube1",
311                "labels": { "app": "kube-rs-test" },
312            },
313            "spec": {
314                "terminationGracePeriodSeconds": 1,
315                "restartPolicy": "Never",
316                "containers": [{
317                  "name": "busybox",
318                  "image": "busybox:stable",
319                  "command": ["sh", "-c", "sleep 30"],
320                }],
321            }
322        }))?;
323
324        let pp = PostParams::default();
325        match pods.create(&pp, &p).await {
326            Ok(o) => assert_eq!(p.name_unchecked(), o.name_unchecked()),
327            Err(crate::Error::Api(ae)) => assert_eq!(ae.code, 409), // if we failed to clean-up
328            Err(e) => return Err(e.into()),                         // any other case if a failure
329        }
330
331        // Manual watch-api for it to become ready
332        // NB: don't do this; using conditions (see pod_api example) is easier and less error prone
333        let wp = WatchParams::default()
334            .fields(&format!("metadata.name={}", "busybox-kube1"))
335            .timeout(15);
336        let mut stream = pods.watch(&wp, "0").await?.boxed();
337        while let Some(ev) = stream.try_next().await? {
338            // can debug format watch event
339            let _ = format!("we: {ev:?}");
340            match ev {
341                WatchEvent::Modified(o) => {
342                    let s = o.status.as_ref().expect("status exists on pod");
343                    let phase = s.phase.clone().unwrap_or_default();
344                    if phase == "Running" {
345                        break;
346                    }
347                }
348                WatchEvent::Error(e) => panic!("watch error: {e}"),
349                _ => {}
350            }
351        }
352
353        // Verify we can get it
354        let mut pod = pods.get("busybox-kube1").await?;
355        assert_eq!(p.spec.as_ref().unwrap().containers[0].name, "busybox");
356
357        // verify replace with explicit resource version
358        // NB: don't do this; use server side apply
359        {
360            assert!(pod.resource_version().is_some());
361            pod.spec.as_mut().unwrap().active_deadline_seconds = Some(5);
362
363            let pp = PostParams::default();
364            let patched_pod = pods.replace("busybox-kube1", &pp, &pod).await?;
365            assert_eq!(patched_pod.spec.unwrap().active_deadline_seconds, Some(5));
366        }
367
368        // Delete it
369        let dp = DeleteParams::default();
370        pods.delete("busybox-kube1", &dp).await?.map_left(|pdel| {
371            assert_eq!(pdel.name_unchecked(), "busybox-kube1");
372        });
373
374        Ok(())
375    }
376
377    #[tokio::test]
378    #[ignore = "needs cluster (will create and attach to a pod)"]
379    #[cfg(feature = "ws")]
380    async fn pod_can_exec_and_write_to_stdin() -> Result<(), Box<dyn std::error::Error>> {
381        use crate::api::{DeleteParams, ListParams, Patch, PatchParams, WatchEvent};
382        use tokio::io::AsyncWriteExt;
383
384        let client = Client::try_default().await?;
385        let pods: Api<Pod> = Api::default_namespaced(client);
386
387        // create busybox pod that's alive for at most 30s
388        let p: Pod = serde_json::from_value(json!({
389            "apiVersion": "v1",
390            "kind": "Pod",
391            "metadata": {
392                "name": "busybox-kube5",
393                "labels": { "app": "kube-rs-test" },
394            },
395            "spec": {
396                "terminationGracePeriodSeconds": 1,
397                "restartPolicy": "Never",
398                "containers": [{
399                  "name": "busybox",
400                  "image": "busybox:stable",
401                  "command": ["sh", "-c", "sleep 30"],
402                }],
403            }
404        }))?;
405
406        match pods.create(&Default::default(), &p).await {
407            Ok(o) => assert_eq!(p.name_unchecked(), o.name_unchecked()),
408            Err(crate::Error::Api(ae)) => assert_eq!(ae.code, 409), // if we failed to clean-up
409            Err(e) => return Err(e.into()),                         // any other case if a failure
410        }
411
412        // Manual watch-api for it to become ready
413        // NB: don't do this; using conditions (see pod_api example) is easier and less error prone
414        let wp = WatchParams::default()
415            .fields(&format!("metadata.name={}", "busybox-kube5"))
416            .timeout(15);
417        let mut stream = pods.watch(&wp, "0").await?.boxed();
418        while let Some(ev) = stream.try_next().await? {
419            match ev {
420                WatchEvent::Modified(o) => {
421                    let s = o.status.as_ref().expect("status exists on pod");
422                    let phase = s.phase.clone().unwrap_or_default();
423                    if phase != "Running" {
424                        continue;
425                    }
426                    let Some(statuses) = s.container_statuses.as_ref() else {
427                        continue;
428                    };
429                    if statuses.iter().all(|s| s.ready) {
430                        break;
431                    }
432                }
433                WatchEvent::Error(e) => panic!("watch error: {e}"),
434                _ => {}
435            }
436        }
437
438        // Verify exec works and we can get the output
439        {
440            let mut attached = pods
441                .exec(
442                    "busybox-kube5",
443                    vec!["sh", "-c", "for i in $(seq 1 3); do echo $i; done"],
444                    &AttachParams::default().stderr(false),
445                )
446                .await?;
447            let stdout = tokio_util::io::ReaderStream::new(attached.stdout().unwrap());
448            let out = stdout
449                .filter_map(|r| async { r.ok().and_then(|v| String::from_utf8(v.to_vec()).ok()) })
450                .collect::<Vec<_>>()
451                .await
452                .join("");
453            attached.join().await.unwrap();
454            assert_eq!(out.lines().count(), 3);
455            assert_eq!(out, "1\n2\n3\n");
456        }
457
458        // Verify we read from stdout after stdin is closed.
459        {
460            let name = "busybox-kube5";
461            let command = vec!["sh", "-c", "sleep 2; echo test string 2"];
462            let ap = AttachParams::default().stdin(true).stderr(false);
463
464            // Make a connection so we can determine if the K8s cluster supports stream closing.
465            let mut req = pods.request.exec(name, command.clone(), &ap)?;
466            req.extensions_mut().insert("exec");
467            let stream = pods.client.connect(req).await?;
468
469            // This only works is the cluster supports protocol version v5.channel.k8s.io
470            // Skip for older protocols.
471            if stream.supports_stream_close() {
472                let mut attached = pods.exec(name, command, &ap).await?;
473                let mut stdin_writer = attached.stdin().unwrap();
474                let mut stdout_stream = tokio_util::io::ReaderStream::new(attached.stdout().unwrap());
475
476                stdin_writer.write_all(b"this will be ignored\n").await?;
477                _ = stdin_writer.shutdown().await;
478
479                // AttachedProcess resolves with status object.
480                let status = attached.take_status().unwrap();
481                if let Some(status) = status.await {
482                    assert_eq!(status.reason, None);
483                    assert_eq!(status.status, Some("Success".to_owned()));
484                }
485
486                let next_stdout = stdout_stream.next();
487                let stdout = String::from_utf8(next_stdout.await.unwrap().unwrap().to_vec()).unwrap();
488                assert_eq!(stdout, "test string 2\n");
489            }
490        }
491
492        // Verify we can write to Stdin
493        {
494            let mut attached = pods
495                .exec(
496                    "busybox-kube5",
497                    vec!["sh"],
498                    &AttachParams::default().stdin(true).stderr(false),
499                )
500                .await?;
501            let mut stdin_writer = attached.stdin().unwrap();
502            let mut stdout_stream = tokio_util::io::ReaderStream::new(attached.stdout().unwrap());
503            let next_stdout = stdout_stream.next();
504            stdin_writer.write_all(b"echo test string 1\n").await?;
505            let stdout = String::from_utf8(next_stdout.await.unwrap().unwrap().to_vec()).unwrap();
506            println!("{stdout}");
507            assert_eq!(stdout, "test string 1\n");
508
509            // AttachedProcess resolves with status object.
510            // Send `exit 1` to get a failure status.
511            stdin_writer.write_all(b"exit 1\n").await?;
512            let status = attached.take_status().unwrap();
513            if let Some(status) = status.await {
514                println!("{status:?}");
515                assert_eq!(status.status, Some("Failure".to_owned()));
516                assert_eq!(status.reason, Some("NonZeroExitCode".to_owned()));
517            }
518        }
519
520        // Delete it
521        let dp = DeleteParams::default();
522        pods.delete("busybox-kube5", &dp).await?.map_left(|pdel| {
523            assert_eq!(pdel.name_unchecked(), "busybox-kube5");
524        });
525
526        Ok(())
527    }
528
529    #[tokio::test]
530    #[ignore = "needs cluster (will create and tail logs from a pod)"]
531    async fn can_get_pod_logs_and_evict() -> Result<(), Box<dyn std::error::Error>> {
532        use crate::{
533            api::{DeleteParams, EvictParams, ListParams, Patch, PatchParams, WatchEvent},
534            core::subresource::LogParams,
535        };
536
537        let client = Client::try_default().await?;
538        let pods: Api<Pod> = Api::default_namespaced(client);
539
540        // create busybox pod that's alive for at most 30s
541        let p: Pod = serde_json::from_value(json!({
542            "apiVersion": "v1",
543            "kind": "Pod",
544            "metadata": {
545                "name": "busybox-kube3",
546                "labels": { "app": "kube-rs-test" },
547            },
548            "spec": {
549                "terminationGracePeriodSeconds": 1,
550                "restartPolicy": "Never",
551                "containers": [{
552                  "name": "busybox",
553                  "image": "busybox:stable",
554                  "command": ["sh", "-c", "for i in $(seq 1 5); do echo kube $i; sleep 0.1; done"],
555                }],
556            }
557        }))?;
558
559        match pods.create(&Default::default(), &p).await {
560            Ok(o) => assert_eq!(p.name_unchecked(), o.name_unchecked()),
561            Err(crate::Error::Api(ae)) => assert_eq!(ae.code, 409), // if we failed to clean-up
562            Err(e) => return Err(e.into()),                         // any other case if a failure
563        }
564
565        // Manual watch-api for it to become ready
566        // NB: don't do this; using conditions (see pod_api example) is easier and less error prone
567        let wp = WatchParams::default()
568            .fields(&format!("metadata.name={}", "busybox-kube3"))
569            .timeout(15);
570        let mut stream = pods.watch(&wp, "0").await?.boxed();
571        while let Some(ev) = stream.try_next().await? {
572            match ev {
573                WatchEvent::Modified(o) => {
574                    let s = o.status.as_ref().expect("status exists on pod");
575                    let phase = s.phase.clone().unwrap_or_default();
576                    if phase == "Running" {
577                        break;
578                    }
579                }
580                WatchEvent::Error(e) => panic!("watch error: {e}"),
581                _ => {}
582            }
583        }
584
585        // Get current list of logs
586        let lp = LogParams {
587            follow: true,
588            ..LogParams::default()
589        };
590        let mut logs_stream = pods.log_stream("busybox-kube3", &lp).await?.lines();
591
592        // wait for container to finish
593        tokio::time::sleep(std::time::Duration::from_secs(2)).await;
594
595        let all_logs = pods.logs("busybox-kube3", &Default::default()).await?;
596        assert_eq!(all_logs, "kube 1\nkube 2\nkube 3\nkube 4\nkube 5\n");
597
598        // individual logs may or may not buffer
599        let mut output = vec![];
600        while let Some(line) = logs_stream.try_next().await? {
601            output.push(line);
602        }
603        assert_eq!(output, vec!["kube 1", "kube 2", "kube 3", "kube 4", "kube 5"]);
604
605        // evict the pod
606        let ep = EvictParams::default();
607        let eres = pods.evict("busybox-kube3", &ep).await?;
608        assert_eq!(eres.code, 201); // created
609        assert!(eres.is_success());
610
611        Ok(())
612    }
613
614    #[tokio::test]
615    #[ignore = "requires a cluster"]
616    async fn can_operate_on_pod_metadata() -> Result<(), Box<dyn std::error::Error>> {
617        use crate::{
618            api::{DeleteParams, EvictParams, ListParams, Patch, PatchParams, WatchEvent},
619            core::subresource::LogParams,
620        };
621        use kube_core::{ObjectList, ObjectMeta, PartialObjectMeta, PartialObjectMetaExt};
622
623        let client = Client::try_default().await?;
624        let pods: Api<Pod> = Api::default_namespaced(client);
625
626        // create busybox pod that's alive for at most 30s
627        let p: Pod = serde_json::from_value(json!({
628            "apiVersion": "v1",
629            "kind": "Pod",
630            "metadata": {
631                "name": "busybox-kube-meta",
632                "labels": { "app": "kube-rs-test" },
633            },
634            "spec": {
635                "terminationGracePeriodSeconds": 1,
636                "restartPolicy": "Never",
637                "containers": [{
638                  "name": "busybox",
639                  "image": "busybox:stable",
640                  "command": ["sh", "-c", "sleep 30s"],
641                }],
642            }
643        }))?;
644
645        match pods.create(&Default::default(), &p).await {
646            Ok(o) => assert_eq!(p.name_unchecked(), o.name_unchecked()),
647            Err(crate::Error::Api(ae)) => assert_eq!(ae.code, 409), // if we failed to clean-up
648            Err(e) => return Err(e.into()),                         // any other case if a failure
649        }
650
651        // Test we can get a pod as a PartialObjectMeta and convert to
652        // ObjectMeta
653        let pod_metadata = pods.get_metadata("busybox-kube-meta").await?;
654        assert_eq!("busybox-kube-meta", pod_metadata.name_any());
655        assert_eq!(
656            Some((&"app".to_string(), &"kube-rs-test".to_string())),
657            pod_metadata.labels().get_key_value("app")
658        );
659
660        // Test we can get a list of PartialObjectMeta for pods
661        let p_list = pods.list_metadata(&ListParams::default()).await?;
662
663        // Find only pod we are concerned with in this test and fail eagerly if
664        // name doesn't exist
665        let pod_metadata = p_list
666            .items
667            .into_iter()
668            .find(|p| p.name_any() == "busybox-kube-meta")
669            .unwrap();
670        assert_eq!(
671            pod_metadata.labels().get("app"),
672            Some(&"kube-rs-test".to_string())
673        );
674
675        // Attempt to patch pod metadata
676        let patch = ObjectMeta {
677            annotations: Some([("test".to_string(), "123".to_string())].into()),
678            ..Default::default()
679        }
680        .into_request_partial::<Pod>();
681
682        let patchparams = PatchParams::default();
683        let p_patched = pods
684            .patch_metadata("busybox-kube-meta", &patchparams, &Patch::Merge(&patch))
685            .await?;
686        assert_eq!(p_patched.annotations().get("test"), Some(&"123".to_string()));
687        assert_eq!(p_patched.types.as_ref().unwrap().kind, "PartialObjectMetadata");
688        assert_eq!(p_patched.types.as_ref().unwrap().api_version, "meta.k8s.io/v1");
689
690        // Clean-up
691        let dp = DeleteParams::default();
692        pods.delete("busybox-kube-meta", &dp).await?.map_left(|pdel| {
693            assert_eq!(pdel.name_any(), "busybox-kube-meta");
694        });
695
696        Ok(())
697    }
698    #[tokio::test]
699    #[ignore = "needs cluster (will create a CertificateSigningRequest)"]
700    async fn csr_can_be_approved() -> Result<(), Box<dyn std::error::Error>> {
701        use crate::api::PostParams;
702        use k8s_openapi::api::certificates::v1::{
703            CertificateSigningRequest, CertificateSigningRequestCondition, CertificateSigningRequestStatus,
704        };
705
706        let csr_name = "fake";
707        let dummy_csr: CertificateSigningRequest = serde_json::from_value(json!({
708            "apiVersion": "certificates.k8s.io/v1",
709            "kind": "CertificateSigningRequest",
710            "metadata": { "name": csr_name },
711            "spec": {
712                "request": "LS0tLS1CRUdJTiBDRVJUSUZJQ0FURSBSRVFVRVNULS0tLS0KTUlJQ1ZqQ0NBVDRDQVFBd0VURVBNQTBHQTFVRUF3d0dZVzVuWld4aE1JSUJJakFOQmdrcWhraUc5dzBCQVFFRgpBQU9DQVE4QU1JSUJDZ0tDQVFFQTByczhJTHRHdTYxakx2dHhWTTJSVlRWMDNHWlJTWWw0dWluVWo4RElaWjBOCnR2MUZtRVFSd3VoaUZsOFEzcWl0Qm0wMUFSMkNJVXBGd2ZzSjZ4MXF3ckJzVkhZbGlBNVhwRVpZM3ExcGswSDQKM3Z3aGJlK1o2MVNrVHF5SVBYUUwrTWM5T1Nsbm0xb0R2N0NtSkZNMUlMRVI3QTVGZnZKOEdFRjJ6dHBoaUlFMwpub1dtdHNZb3JuT2wzc2lHQ2ZGZzR4Zmd4eW8ybmlneFNVekl1bXNnVm9PM2ttT0x1RVF6cXpkakJ3TFJXbWlECklmMXBMWnoyalVnald4UkhCM1gyWnVVV1d1T09PZnpXM01LaE8ybHEvZi9DdS8wYk83c0x0MCt3U2ZMSU91TFcKcW90blZtRmxMMytqTy82WDNDKzBERHk5aUtwbXJjVDBnWGZLemE1dHJRSURBUUFCb0FBd0RRWUpLb1pJaHZjTgpBUUVMQlFBRGdnRUJBR05WdmVIOGR4ZzNvK21VeVRkbmFjVmQ1N24zSkExdnZEU1JWREkyQTZ1eXN3ZFp1L1BVCkkwZXpZWFV0RVNnSk1IRmQycVVNMjNuNVJsSXJ3R0xuUXFISUh5VStWWHhsdnZsRnpNOVpEWllSTmU3QlJvYXgKQVlEdUI5STZXT3FYbkFvczFqRmxNUG5NbFpqdU5kSGxpT1BjTU1oNndLaTZzZFhpVStHYTJ2RUVLY01jSVUyRgpvU2djUWdMYTk0aEpacGk3ZnNMdm1OQUxoT045UHdNMGM1dVJVejV4T0dGMUtCbWRSeEgvbUNOS2JKYjFRQm1HCkkwYitEUEdaTktXTU0xMzhIQXdoV0tkNjVoVHdYOWl4V3ZHMkh4TG1WQzg0L1BHT0tWQW9FNkpsYWFHdTlQVmkKdjlOSjVaZlZrcXdCd0hKbzZXdk9xVlA3SVFjZmg3d0drWm89Ci0tLS0tRU5EIENFUlRJRklDQVRFIFJFUVVFU1QtLS0tLQo=",
713                "signerName": "kubernetes.io/kube-apiserver-client",
714                "expirationSeconds": 86400,
715                "usages": ["client auth"]
716            }
717        }))?;
718
719        let client = Client::try_default().await?;
720        let csr: Api<CertificateSigningRequest> = Api::all(client.clone());
721        assert!(csr.create(&PostParams::default(), &dummy_csr).await.is_ok());
722
723        // Patch the approval and approve the CSR
724        let approval_type = "ApprovedFake";
725        let csr_status: CertificateSigningRequestStatus = CertificateSigningRequestStatus {
726            certificate: None,
727            conditions: Some(vec![CertificateSigningRequestCondition {
728                type_: approval_type.to_string(),
729                last_update_time: None,
730                last_transition_time: None,
731                message: Some(format!("{} {}", approval_type, "by kube-rs client")),
732                reason: Some("kube-rsClient".to_string()),
733                status: "True".to_string(),
734            }]),
735        };
736        let csr_status_patch = Patch::Merge(serde_json::json!({ "status": csr_status }));
737        let _ = csr
738            .patch_approval(csr_name, &Default::default(), &csr_status_patch)
739            .await?;
740        let csr_after_approval = csr.get_approval(csr_name).await?;
741
742        assert_eq!(
743            csr_after_approval
744                .status
745                .as_ref()
746                .unwrap()
747                .conditions
748                .as_ref()
749                .unwrap()[0]
750                .type_,
751            approval_type.to_string()
752        );
753        csr.delete(csr_name, &DeleteParams::default()).await?;
754        Ok(())
755    }
756
757    #[tokio::test]
758    #[ignore = "needs cluster for ephemeral containers operations"]
759    async fn can_operate_on_ephemeral_containers() -> Result<(), Box<dyn std::error::Error>> {
760        let client = Client::try_default().await?;
761
762        // Ephemeral containers were stabilized in Kubernetes v1.25.
763        // This test therefore exits early if the current cluster version is older than v1.25.
764        let api_version = client.apiserver_version().await?;
765        if api_version.major.parse::<i32>()? < 1 || api_version.minor.parse::<i32>()? < 25 {
766            return Ok(());
767        }
768
769        let pod: Pod = serde_json::from_value(serde_json::json!({
770            "apiVersion": "v1",
771            "kind": "Pod",
772            "metadata": {
773                "name": "ephemeral-container-test",
774                "labels": { "app": "kube-rs-test" },
775            },
776            "spec": {
777                "restartPolicy": "Never",
778                "containers": [{
779                  "name": "busybox",
780                  "image": "busybox:stable",
781                  "command": ["sh", "-c", "sleep 2"],
782                }],
783            }
784        }))?;
785
786        let pod_name = pod.name_any();
787        let pods = Api::<Pod>::default_namespaced(client);
788
789        // If cleanup failed and a pod already exists, we attempt to remove it
790        // before proceeding. This is important as ephemeral containers can't
791        // be removed from a Pod's spec. Therefore this test must start with a fresh
792        // Pod every time.
793        let _ = pods
794            .delete(&pod.name_any(), &DeleteParams::default())
795            .await
796            .map(|v| v.map_left(|pdel| assert_eq!(pdel.name_any(), pod.name_any())));
797
798        // Ephemeral containers can only be applied to a running pod, so one must
799        // be created before any operations are tested.
800        match pods.create(&Default::default(), &pod).await {
801            Ok(o) => assert_eq!(pod.name_unchecked(), o.name_unchecked()),
802            Err(e) => return Err(e.into()), // any other case if a failure
803        }
804
805        let current_ephemeral_containers = pods
806            .get_ephemeral_containers(&pod.name_any())
807            .await?
808            .spec
809            .unwrap()
810            .ephemeral_containers;
811
812        // We expect no ephemeral containers initially, get_ephemeral_containers should
813        // reflect that.
814        assert_eq!(current_ephemeral_containers, None);
815
816        let mut busybox_eph: EphemeralContainer = serde_json::from_value(json!(
817            {
818                "name": "myephemeralcontainer1",
819                "image": "busybox:stable",
820                "command": ["sh", "-c", "sleep 2"],
821            }
822        ))?;
823
824        // Attempt to replace ephemeral containers.
825
826        let patch: Pod = serde_json::from_value(json!({
827            "metadata": { "name": pod_name },
828            "spec":{ "ephemeralContainers": [ busybox_eph ] }
829        }))?;
830
831        let current_containers = pods
832            .replace_ephemeral_containers(&pod_name, &PostParams::default(), &patch)
833            .await?
834            .spec
835            .unwrap()
836            .ephemeral_containers
837            .expect("could find ephemeral container");
838
839        // Note that we can't compare the whole ephemeral containers object, as some fields
840        // are set by the cluster. We therefore compare the fields specified in the patch.
841        assert_eq!(current_containers.len(), 1);
842        assert_eq!(current_containers[0].name, busybox_eph.name);
843        assert_eq!(current_containers[0].image, busybox_eph.image);
844        assert_eq!(current_containers[0].command, busybox_eph.command);
845
846        // Attempt to patch ephemeral containers.
847
848        // The new ephemeral container will have different values from the
849        // first to ensure we can test for its presence.
850        busybox_eph = serde_json::from_value(json!(
851            {
852                "name": "myephemeralcontainer2",
853                "image": "busybox:stable",
854                "command": ["sh", "-c", "sleep 1"],
855            }
856        ))?;
857
858        let patch: Pod =
859            serde_json::from_value(json!({ "spec": { "ephemeralContainers": [ busybox_eph ] }}))?;
860
861        let current_containers = pods
862            .patch_ephemeral_containers(&pod_name, &PatchParams::default(), &Patch::Strategic(patch))
863            .await?
864            .spec
865            .unwrap()
866            .ephemeral_containers
867            .expect("could find ephemeral container");
868
869        // There should only be 2 ephemeral containers at this point,
870        // one from each patch
871        assert_eq!(current_containers.len(), 2);
872
873        let new_container = current_containers
874            .iter()
875            .find(|c| c.name == busybox_eph.name)
876            .expect("could find myephemeralcontainer2");
877
878        // Note that we can't compare the whole ephemeral container object, as some fields
879        // get set in the cluster. We therefore compare the fields specified in the patch.
880        assert_eq!(new_container.image, busybox_eph.image);
881        assert_eq!(new_container.command, busybox_eph.command);
882
883        // Attempt to get ephemeral containers.
884
885        let expected_containers = current_containers;
886
887        let current_containers = pods
888            .get_ephemeral_containers(&pod.name_any())
889            .await?
890            .spec
891            .unwrap()
892            .ephemeral_containers
893            .unwrap();
894
895        assert_eq!(current_containers, expected_containers);
896
897        pods.delete(&pod.name_any(), &DeleteParams::default())
898            .await?
899            .map_left(|pdel| {
900                assert_eq!(pdel.name_any(), pod.name_any());
901            });
902
903        Ok(())
904    }
905
906    #[tokio::test]
907    #[ignore = "needs kubelet debug methods"]
908    #[cfg(feature = "kubelet-debug")]
909    async fn pod_can_exec_and_write_to_stdin_from_node_proxy() -> Result<(), Box<dyn std::error::Error>> {
910        use crate::{
911            api::{DeleteParams, ListParams, Patch, PatchParams, WatchEvent},
912            core::kubelet_debug::KubeletDebugParams,
913        };
914
915        let client = Client::try_default().await?;
916        let pods: Api<Pod> = Api::default_namespaced(client);
917
918        // create busybox pod that's alive for at most 30s
919        let p: Pod = serde_json::from_value(json!({
920            "apiVersion": "v1",
921            "kind": "Pod",
922            "metadata": {
923                "name": "busybox-kube2",
924                "labels": { "app": "kube-rs-test" },
925            },
926            "spec": {
927                "terminationGracePeriodSeconds": 1,
928                "restartPolicy": "Never",
929                "containers": [{
930                  "name": "busybox",
931                  "image": "busybox:stable",
932                  "command": ["sh", "-c", "sleep 30"],
933                }],
934            }
935        }))?;
936
937        match pods.create(&Default::default(), &p).await {
938            Ok(o) => assert_eq!(p.name_unchecked(), o.name_unchecked()),
939            Err(crate::Error::Api(ae)) => assert_eq!(ae.code, 409), // if we failed to clean-up
940            Err(e) => return Err(e.into()),                         // any other case if a failure
941        }
942
943        // Manual watch-api for it to become ready
944        // NB: don't do this; using conditions (see pod_api example) is easier and less error prone
945        let wp = WatchParams::default()
946            .fields(&format!("metadata.name={}", "busybox-kube2"))
947            .timeout(15);
948        let mut stream = pods.watch(&wp, "0").await?.boxed();
949        while let Some(ev) = stream.try_next().await? {
950            match ev {
951                WatchEvent::Modified(o) => {
952                    let s = o.status.as_ref().expect("status exists on pod");
953                    let phase = s.phase.clone().unwrap_or_default();
954                    if phase == "Running" {
955                        break;
956                    }
957                }
958                WatchEvent::Error(e) => panic!("watch error: {e}"),
959                _ => {}
960            }
961        }
962
963        let mut config = Config::infer().await?;
964        config.accept_invalid_certs = true;
965        config.cluster_url = "https://localhost:10250".parse().unwrap();
966        let kubelet_client: Client = config.try_into()?;
967
968        // Verify exec works and we can get the output
969        {
970            let mut attached = kubelet_client
971                .kubelet_node_exec(
972                    &KubeletDebugParams {
973                        name: "busybox-kube2",
974                        namespace: "default",
975                        ..Default::default()
976                    },
977                    "busybox",
978                    vec!["sh", "-c", "for i in $(seq 1 3); do echo $i; done"],
979                    &AttachParams::default().stderr(false),
980                )
981                .await?;
982            let stdout = tokio_util::io::ReaderStream::new(attached.stdout().unwrap());
983            let out = stdout
984                .filter_map(|r| async { r.ok().and_then(|v| String::from_utf8(v.to_vec()).ok()) })
985                .collect::<Vec<_>>()
986                .await
987                .join("");
988            attached.join().await.unwrap();
989            assert_eq!(out.lines().count(), 3);
990            assert_eq!(out, "1\n2\n3\n");
991        }
992
993        // Verify we can write to Stdin
994        {
995            use tokio::io::AsyncWriteExt;
996            let mut attached = kubelet_client
997                .kubelet_node_exec(
998                    &KubeletDebugParams {
999                        name: "busybox-kube2",
1000                        namespace: "default",
1001                        ..Default::default()
1002                    },
1003                    "busybox",
1004                    vec!["sh"],
1005                    &AttachParams::default().stdin(true).stderr(false),
1006                )
1007                .await?;
1008            let mut stdin_writer = attached.stdin().unwrap();
1009            let mut stdout_stream = tokio_util::io::ReaderStream::new(attached.stdout().unwrap());
1010            let next_stdout = stdout_stream.next();
1011            stdin_writer.write_all(b"echo test string 1\n").await?;
1012            let stdout = String::from_utf8(next_stdout.await.unwrap().unwrap().to_vec()).unwrap();
1013            println!("{stdout}");
1014            assert_eq!(stdout, "test string 1\n");
1015
1016            // AttachedProcess resolves with status object.
1017            // Send `exit 1` to get a failure status.
1018            stdin_writer.write_all(b"exit 1\n").await?;
1019            let status = attached.take_status().unwrap();
1020            if let Some(status) = status.await {
1021                println!("{status:?}");
1022                assert_eq!(status.status, Some("Failure".to_owned()));
1023                assert_eq!(status.reason, Some("NonZeroExitCode".to_owned()));
1024            }
1025        }
1026
1027        // Delete it
1028        let dp = DeleteParams::default();
1029        pods.delete("busybox-kube2", &dp).await?.map_left(|pdel| {
1030            assert_eq!(pdel.name_unchecked(), "busybox-kube2");
1031        });
1032
1033        Ok(())
1034    }
1035}
kube_client/lib.rs

kube_client/
lib.rs
