kube/
lib.rs

Help
1//! Kube is an umbrella-crate for interacting with [Kubernetes](http://kubernetes.io) in Rust.
2//!
3//! # Overview
4//!
5//! Kube contains a Kubernetes client, a controller runtime, a custom resource derive, and various tooling
6//! required for building applications or controllers that interact with Kubernetes.
7//!
8//! The main modules are:
9//!
10//! - [`client`] with the Kubernetes [`Client`] and its layers
11//! - [`config`] for cluster [`Config`]
12//! - [`api`] with the generic Kubernetes [`Api`]
13//! - [`derive`](kube_derive) with the [`CustomResource`] derive for building controllers types
14//! - [`runtime`] with a [`Controller`](crate::runtime::Controller) / [`watcher`](crate::runtime::watcher()) / [`reflector`](crate::runtime::reflector::reflector) / [`Store`](crate::runtime::reflector::Store)
15//! - [`core`] with generics from `apimachinery`
16//!
17//! You can use each of these as you need with the help of the [exported features](https://github.com/kube-rs/kube/blob/main/kube/Cargo.toml#L18).
18//!
19//! # Using the Client
20//! ```no_run
21//! use futures::{StreamExt, TryStreamExt};
22//! use kube::{Client, api::{Api, ResourceExt, ListParams, PostParams}};
23//! use k8s_openapi::api::core::v1::Pod;
24//!
25//! #[tokio::main]
26//! async fn main() -> Result<(), Box<dyn std::error::Error>> {
27//!     // Infer the runtime environment and try to create a Kubernetes Client
28//!     let client = Client::try_default().await?;
29//!
30//!     // Read pods in the configured namespace into the typed interface from k8s-openapi
31//!     let pods: Api<Pod> = Api::default_namespaced(client);
32//!     for p in pods.list(&ListParams::default()).await? {
33//!         println!("found pod {}", p.name_any());
34//!     }
35//!     Ok(())
36//! }
37//! ```
38//!
39//! For details, see:
40//!
41//! - [`Client`](crate::client) for the extensible Kubernetes client
42//! - [`Api`] for the generic api methods available on Kubernetes resources
43//! - [k8s-openapi](https://docs.rs/k8s-openapi/*/k8s_openapi/) for documentation about the generated Kubernetes types
44//!
45//! # Using the Runtime with the Derive macro
46//!
47//! ```no_run
48//! use schemars::JsonSchema;
49//! use serde::{Deserialize, Serialize};
50//! use serde_json::json;
51//! use futures::{StreamExt, TryStreamExt};
52//! use k8s_openapi::apiextensions_apiserver::pkg::apis::apiextensions::v1::CustomResourceDefinition;
53//! use kube::{
54//!     api::{Api, DeleteParams, PatchParams, Patch, ResourceExt},
55//!     core::CustomResourceExt,
56//!     Client, CustomResource,
57//!     runtime::{watcher, WatchStreamExt, wait::{conditions, await_condition}},
58//! };
59//!
60//! // Our custom resource
61//! #[derive(CustomResource, Deserialize, Serialize, Clone, Debug, JsonSchema)]
62//! #[kube(group = "clux.dev", version = "v1", kind = "Foo", namespaced)]
63//! pub struct FooSpec {
64//!     info: String,
65//!     #[schemars(length(min = 3))]
66//!     name: String,
67//!     replicas: i32,
68//! }
69//!
70//! #[tokio::main]
71//! async fn main() -> Result<(), Box<dyn std::error::Error>> {
72//!     let client = Client::try_default().await?;
73//!     let crds: Api<CustomResourceDefinition> = Api::all(client.clone());
74//!
75//!     // Apply the CRD so users can create Foo instances in Kubernetes
76//!     crds.patch("foos.clux.dev",
77//!         &PatchParams::apply("my_manager"),
78//!         &Patch::Apply(Foo::crd())
79//!     ).await?;
80//!
81//!     // Wait for the CRD to be ready
82//!     tokio::time::timeout(
83//!         std::time::Duration::from_secs(10),
84//!         await_condition(crds, "foos.clux.dev", conditions::is_crd_established())
85//!     ).await?;
86//!
87//!     // Watch for changes to foos in the configured namespace
88//!     let foos: Api<Foo> = Api::default_namespaced(client.clone());
89//!     let wc = watcher::Config::default();
90//!     let mut apply_stream = watcher(foos, wc).applied_objects().boxed();
91//!     while let Some(f) = apply_stream.try_next().await? {
92//!         println!("saw apply to {}", f.name_any());
93//!     }
94//!     Ok(())
95//! }
96//! ```
97//!
98//! For details, see:
99//!
100//! - [`CustomResource`] for documentation how to configure custom resources
101//! - [`runtime::watcher`](crate::runtime::watcher()) for how to long-running watches work and why you want to use this over [`Api::watch`](crate::Api::watch)
102//! - [`runtime`] for abstractions that help with more complicated Kubernetes application
103//!
104//! # Examples
105//! A large list of complete, runnable examples with explainations are available in the [examples folder](https://github.com/kube-rs/kube/tree/main/examples).
106#![cfg_attr(docsrs, feature(doc_cfg))]
107
108macro_rules! cfg_client {
109    ($($item:item)*) => {
110        $(
111            #[cfg_attr(docsrs, doc(cfg(feature = "client")))]
112            #[cfg(feature = "client")]
113            $item
114        )*
115    }
116}
117macro_rules! cfg_config {
118    ($($item:item)*) => {
119        $(
120            #[cfg_attr(docsrs, doc(cfg(feature = "config")))]
121            #[cfg(feature = "config")]
122            $item
123        )*
124    }
125}
126
127macro_rules! cfg_error {
128    ($($item:item)*) => {
129        $(
130            #[cfg_attr(docsrs, doc(cfg(any(feature = "config", feature = "client"))))]
131            #[cfg(any(feature = "config", feature = "client"))]
132            $item
133        )*
134    }
135}
136
137cfg_client! {
138    pub use kube_client::api;
139    pub use kube_client::discovery;
140    pub use kube_client::client;
141
142    #[doc(inline)]
143    pub use api::Api;
144    #[doc(inline)]
145    pub use client::Client;
146    #[doc(inline)]
147    pub use discovery::Discovery;
148}
149
150cfg_config! {
151    pub use kube_client::config;
152    #[doc(inline)]
153    pub use config::Config;
154}
155
156cfg_error! {
157    pub use kube_client::error;
158    #[doc(inline)] pub use error::Error;
159    /// Convient alias for `Result<T, Error>`
160    pub type Result<T, E = Error> = std::result::Result<T, E>;
161}
162
163/// Re-exports from [`kube-derive`](kube_derive)
164#[cfg(feature = "derive")]
165#[cfg_attr(docsrs, doc(cfg(feature = "derive")))]
166pub use kube_derive::CustomResource;
167
168/// Re-exports from `kube-runtime`
169#[cfg(feature = "runtime")]
170#[cfg_attr(docsrs, doc(cfg(feature = "runtime")))]
171#[doc(inline)]
172pub use kube_runtime as runtime;
173
174pub use crate::core::{CustomResourceExt, Resource, ResourceExt};
175/// Re-exports from `kube_core`
176#[doc(inline)]
177pub use kube_core as core;
178
179// Mock tests for the runtime
180#[cfg(test)]
181#[cfg(all(feature = "derive", feature = "runtime"))]
182mod mock_tests;
183
184// Tests that require a cluster and the complete feature set
185// Can be run with `cargo test -p kube --lib --features=runtime,derive -- --ignored`
186#[cfg(test)]
187#[cfg(all(feature = "derive", feature = "client"))]
188mod test {
189    use crate::{
190        api::{DeleteParams, Patch, PatchParams},
191        Api, Client, CustomResourceExt, Resource, ResourceExt,
192    };
193    use kube_derive::CustomResource;
194    use schemars::JsonSchema;
195    use serde::{Deserialize, Serialize};
196
197    #[derive(CustomResource, Deserialize, Serialize, Clone, Debug, JsonSchema)]
198    #[kube(group = "clux.dev", version = "v1", kind = "Foo", namespaced)]
199    #[kube(status = "FooStatus")]
200    #[kube(scale = r#"{"specReplicasPath":".spec.replicas", "statusReplicasPath":".status.replicas"}"#)]
201    #[kube(crates(kube_core = "crate::core"))] // for dev-dep test structure
202    pub struct FooSpec {
203        name: String,
204        info: Option<String>,
205        replicas: isize,
206    }
207
208    #[derive(Deserialize, Serialize, Clone, Debug, Default, JsonSchema)]
209    pub struct FooStatus {
210        is_bad: bool,
211        replicas: isize,
212    }
213
214    #[tokio::test]
215    #[ignore = "needs kubeconfig"]
216    async fn custom_resource_generates_correct_core_structs() {
217        use crate::core::{ApiResource, DynamicObject, GroupVersionKind};
218        let client = Client::try_default().await.unwrap();
219
220        let gvk = GroupVersionKind::gvk("clux.dev", "v1", "Foo");
221        let api_resource = ApiResource::from_gvk(&gvk);
222        let a1: Api<DynamicObject> = Api::namespaced_with(client.clone(), "myns", &api_resource);
223        let a2: Api<Foo> = Api::namespaced(client, "myns");
224
225        // make sure they return the same url_path through their impls
226        assert_eq!(a1.resource_url(), a2.resource_url());
227    }
228
229    use k8s_openapi::{
230        api::core::v1::ConfigMap,
231        apiextensions_apiserver::pkg::apis::apiextensions::v1::CustomResourceDefinition,
232    };
233    #[tokio::test]
234    #[ignore = "needs cluster (creates + patches foo crd)"]
235    #[cfg(all(feature = "derive", feature = "runtime"))]
236    async fn derived_resource_queriable_and_has_subresources() -> Result<(), Box<dyn std::error::Error>> {
237        use crate::runtime::wait::{await_condition, conditions};
238
239        use serde_json::json;
240        let client = Client::try_default().await?;
241        let ssapply = PatchParams::apply("kube").force();
242        let crds: Api<CustomResourceDefinition> = Api::all(client.clone());
243        // Server-side apply CRD and wait for it to get ready
244        crds.patch("foos.clux.dev", &ssapply, &Patch::Apply(Foo::crd()))
245            .await?;
246        let establish = await_condition(crds.clone(), "foos.clux.dev", conditions::is_crd_established());
247        let _ = tokio::time::timeout(std::time::Duration::from_secs(10), establish).await?;
248        // Use it
249        let foos: Api<Foo> = Api::default_namespaced(client.clone());
250        // Apply from generated struct
251        {
252            let foo = Foo::new("baz", FooSpec {
253                name: "baz".into(),
254                info: Some("old baz".into()),
255                replicas: 1,
256            });
257            let o = foos.patch("baz", &ssapply, &Patch::Apply(&foo)).await?;
258            assert_eq!(o.spec.name, "baz");
259            let oref = o.object_ref(&());
260            assert_eq!(oref.name.unwrap(), "baz");
261            assert_eq!(oref.uid, o.uid());
262        }
263        // Apply from partial json!
264        {
265            let patch = json!({
266                "apiVersion": "clux.dev/v1",
267                "kind": "Foo",
268                "spec": {
269                    "name": "foo",
270                    "replicas": 2
271                }
272            });
273            let o = foos.patch("baz", &ssapply, &Patch::Apply(patch)).await?;
274            assert_eq!(o.spec.replicas, 2, "patching spec updated spec.replicas");
275        }
276        // check subresource
277        {
278            let scale = foos.get_scale("baz").await?;
279            assert_eq!(scale.spec.unwrap().replicas, Some(2));
280            let status = foos.get_status("baz").await?;
281            assert!(status.status.is_none(), "nothing has set status");
282        }
283        // set status subresource
284        {
285            let fs = serde_json::json!({"status": FooStatus { is_bad: false, replicas: 1 }});
286            let o = foos
287                .patch_status("baz", &Default::default(), &Patch::Merge(&fs))
288                .await?;
289            assert!(o.status.is_some(), "status set after patch_status");
290        }
291        // set scale subresource
292        {
293            let fs = serde_json::json!({"spec": { "replicas": 3 }});
294            let o = foos
295                .patch_scale("baz", &Default::default(), &Patch::Merge(&fs))
296                .await?;
297            assert_eq!(o.status.unwrap().replicas, 1, "scale replicas got patched");
298            let linked_replicas = o.spec.unwrap().replicas.unwrap();
299            assert_eq!(linked_replicas, 3, "patch_scale updates linked spec.replicas");
300        }
301
302        // cleanup
303        foos.delete_collection(&DeleteParams::default(), &Default::default())
304            .await?;
305        crds.delete("foos.clux.dev", &DeleteParams::default()).await?;
306        Ok(())
307    }
308
309    #[tokio::test]
310    #[ignore = "needs cluster (lists pods)"]
311    async fn custom_serialized_objects_are_queryable_and_iterable() -> Result<(), Box<dyn std::error::Error>>
312    {
313        use crate::core::{
314            object::{HasSpec, HasStatus, NotUsed, Object},
315            ApiResource,
316        };
317        use k8s_openapi::api::core::v1::Pod;
318        #[derive(Clone, Deserialize, Debug)]
319        struct PodSpecSimple {
320            containers: Vec<ContainerSimple>,
321        }
322        #[derive(Clone, Deserialize, Debug)]
323        struct ContainerSimple {
324            #[allow(dead_code)]
325            image: String,
326        }
327        type PodSimple = Object<PodSpecSimple, NotUsed>;
328
329        // use known type information from pod (can also use discovery for this)
330        let ar = ApiResource::erase::<Pod>(&());
331
332        let client = Client::try_default().await?;
333        let api: Api<PodSimple> = Api::default_namespaced_with(client, &ar);
334        let mut list = api.list(&Default::default()).await?;
335        // check we can mutably iterate over ObjectList
336        for pod in &mut list {
337            pod.spec_mut().containers = vec![];
338            *pod.status_mut() = None;
339            pod.annotations_mut()
340                .entry("kube-seen".to_string())
341                .or_insert_with(|| "yes".to_string());
342            pod.labels_mut()
343                .entry("kube.rs".to_string())
344                .or_insert_with(|| "hello".to_string());
345            pod.finalizers_mut().push("kube-finalizer".to_string());
346            pod.managed_fields_mut().clear();
347            // NB: we are **not** pushing these back upstream - (Api::apply or Api::replace needed for it)
348        }
349        // check we can iterate over ObjectList normally - and check the mutations worked
350        for pod in list {
351            assert!(pod.annotations().get("kube-seen").is_some());
352            assert!(pod.labels().get("kube.rs").is_some());
353            assert!(pod.finalizers().contains(&"kube-finalizer".to_string()));
354            assert!(pod.spec().containers.is_empty());
355            assert!(pod.managed_fields().is_empty());
356        }
357        Ok(())
358    }
359
360    #[tokio::test]
361    #[ignore = "needs cluster (fetches api resources, and lists all)"]
362    #[cfg(feature = "derive")]
363    async fn derived_resources_discoverable() -> Result<(), Box<dyn std::error::Error>> {
364        use crate::{
365            core::{DynamicObject, GroupVersion, GroupVersionKind},
366            discovery::{self, verbs, ApiGroup, Discovery, Scope},
367            runtime::wait::{await_condition, conditions, Condition},
368        };
369
370        #[derive(CustomResource, Deserialize, Serialize, Clone, Debug, JsonSchema)]
371        #[kube(group = "kube.rs", version = "v1", kind = "TestCr", namespaced)]
372        #[kube(crates(kube_core = "crate::core"))] // for dev-dep test structure
373        struct TestCrSpec {}
374
375        let client = Client::try_default().await?;
376
377        // install crd is installed
378        let crds: Api<CustomResourceDefinition> = Api::all(client.clone());
379        let ssapply = PatchParams::apply("kube").force();
380        crds.patch("testcrs.kube.rs", &ssapply, &Patch::Apply(TestCr::crd()))
381            .await?;
382        let establish = await_condition(crds.clone(), "testcrs.kube.rs", conditions::is_crd_established());
383        let crd = tokio::time::timeout(std::time::Duration::from_secs(10), establish).await??;
384        assert!(conditions::is_crd_established().matches_object(crd.as_ref()));
385        tokio::time::sleep(std::time::Duration::from_secs(2)).await; // Established condition is actually not enough for api discovery :(
386
387        // create partial information for it to discover
388        let gvk = GroupVersionKind::gvk("kube.rs", "v1", "TestCr");
389        let gv = GroupVersion::gv("kube.rs", "v1");
390
391        // discover by both (recommended kind on groupversion) and (pinned gvk) and they should equal
392        let apigroup = discovery::oneshot::pinned_group(&client, &gv).await?;
393        let (ar1, caps1) = apigroup.recommended_kind("TestCr").unwrap();
394        let (ar2, caps2) = discovery::pinned_kind(&client, &gvk).await?;
395        assert_eq!(caps1.operations.len(), caps2.operations.len(), "unequal caps");
396        assert_eq!(ar1, ar2, "unequal apiresource");
397        assert_eq!(DynamicObject::api_version(&ar2), "kube.rs/v1", "unequal dynver");
398
399        // run (almost) full discovery
400        let discovery = Discovery::new(client.clone())
401            // skip something in discovery (clux.dev crd being mutated in other tests)
402            .exclude(&["rbac.authorization.k8s.io", "clux.dev"])
403            .run()
404            .await?;
405
406        // check our custom resource first by resolving within groups
407        assert!(discovery.has_group("kube.rs"), "missing group kube.rs");
408        let (ar, _caps) = discovery.resolve_gvk(&gvk).unwrap();
409        assert_eq!(ar.group, gvk.group, "unexpected discovered group");
410        assert_eq!(ar.version, gvk.version, "unexcepted discovered ver");
411        assert_eq!(ar.kind, gvk.kind, "unexpected discovered kind");
412
413        // check all non-excluded groups that are iterable
414        let mut groups = discovery.groups_alphabetical().into_iter();
415        let firstgroup = groups.next().unwrap();
416        assert_eq!(firstgroup.name(), ApiGroup::CORE_GROUP, "core not first");
417        for group in groups {
418            for (ar, caps) in group.recommended_resources() {
419                if !caps.supports_operation(verbs::LIST) {
420                    continue;
421                }
422                let api: Api<DynamicObject> = if caps.scope == Scope::Namespaced {
423                    Api::default_namespaced_with(client.clone(), &ar)
424                } else {
425                    Api::all_with(client.clone(), &ar)
426                };
427                api.list(&Default::default()).await?;
428            }
429        }
430
431        // cleanup
432        crds.delete("testcrs.kube.rs", &DeleteParams::default()).await?;
433        Ok(())
434    }
435
436    #[tokio::test]
437    #[ignore = "needs cluster (will create await a pod)"]
438    #[cfg(feature = "runtime")]
439    async fn pod_can_await_conditions() -> Result<(), Box<dyn std::error::Error>> {
440        use crate::{
441            api::{DeleteParams, PostParams},
442            runtime::wait::{await_condition, conditions, delete::delete_and_finalize, Condition},
443            Api, Client,
444        };
445        use k8s_openapi::api::core::v1::Pod;
446        use std::time::Duration;
447        use tokio::time::timeout;
448
449        let client = Client::try_default().await?;
450        let pods: Api<Pod> = Api::default_namespaced(client);
451
452        // create busybox pod that's alive for at most 20s
453        let data: Pod = serde_json::from_value(serde_json::json!({
454            "apiVersion": "v1",
455            "kind": "Pod",
456            "metadata": {
457                "name": "busybox-kube4",
458                "labels": { "app": "kube-rs-test" },
459            },
460            "spec": {
461                "terminationGracePeriodSeconds": 1,
462                "restartPolicy": "Never",
463                "containers": [{
464                  "name": "busybox",
465                  "image": "busybox:1.34.1",
466                  "command": ["sh", "-c", "sleep 20"],
467                }],
468            }
469        }))?;
470
471        let pp = PostParams::default();
472        assert_eq!(
473            data.name_unchecked(),
474            pods.create(&pp, &data).await?.name_unchecked()
475        );
476
477        // Watch it phase for a few seconds
478        let is_running = await_condition(pods.clone(), "busybox-kube4", conditions::is_pod_running());
479        let _ = timeout(Duration::from_secs(15), is_running).await?;
480
481        // Verify we can get it
482        let pod = pods.get("busybox-kube4").await?;
483        assert_eq!(pod.spec.as_ref().unwrap().containers[0].name, "busybox");
484
485        // Wait for a more complicated condition: ContainersReady AND Initialized
486        // TODO: remove these once we can write these functions generically
487        fn is_each_container_ready() -> impl Condition<Pod> {
488            |obj: Option<&Pod>| {
489                if let Some(o) = obj {
490                    if let Some(s) = &o.status {
491                        if let Some(conds) = &s.conditions {
492                            if let Some(pcond) = conds.iter().find(|c| c.type_ == "ContainersReady") {
493                                return pcond.status == "True";
494                            }
495                        }
496                    }
497                }
498                false
499            }
500        }
501        let is_fully_ready = await_condition(
502            pods.clone(),
503            "busybox-kube4",
504            conditions::is_pod_running().and(is_each_container_ready()),
505        );
506        let _ = timeout(Duration::from_secs(10), is_fully_ready).await?;
507
508        // Delete it - and wait for deletion to complete
509        let dp = DeleteParams::default();
510        delete_and_finalize(pods.clone(), "busybox-kube4", &dp).await?;
511
512        // verify it is properly gone
513        assert!(pods.get("busybox-kube4").await.is_err());
514
515        Ok(())
516    }
517
518    #[tokio::test]
519    #[ignore = "needs cluster (lists cms)"]
520    async fn api_get_opt_handles_404() -> Result<(), Box<dyn std::error::Error>> {
521        let client = Client::try_default().await?;
522        let api = Api::<ConfigMap>::default_namespaced(client);
523        assert_eq!(
524            api.get_opt("this-cm-does-not-exist-ajklisdhfqkljwhreq").await?,
525            None
526        );
527        Ok(())
528    }
529}
kube/lib.rs

kube/
lib.rs
