kube_runtime/
wait.rs

1//! Waits for objects to reach desired states
2use std::{future, pin::pin};
3
4use futures::TryStreamExt;
5use kube_client::{Api, Resource};
6use serde::de::DeserializeOwned;
7use std::fmt::Debug;
8use thiserror::Error;
9
10use crate::watcher::{self, watch_object};
11
12/// Errors from `await_condition`
13#[derive(Debug, Error)]
14pub enum Error {
15    /// The underlying watcher failed to probe the stream
16    #[error("failed to probe for whether the condition is fulfilled yet: {0}")]
17    ProbeFailed(#[source] watcher::Error),
18}
19
20/// Watch an object, and wait for some condition `cond` to return `true`.
21///
22/// `cond` is passed `Some` if the object is found, otherwise `None`.
23///
24/// The object is returned when the condition is fulfilled.
25///
26/// # Caveats
27///
28/// Keep in mind that the condition is typically fulfilled by an external service, which might not even be available. `await_condition`
29/// does *not* automatically add a timeout. If this is desired, wrap it in [`tokio::time::timeout`].
30///
31/// # Errors
32///
33/// Fails if the type is not known to the Kubernetes API, or if the [`Api`] does not have
34/// permission to `watch` and `list` it.
35///
36/// Does *not* fail if the object is not found.
37///
38/// # Usage
39///
40/// ```
41/// use k8s_openapi::apiextensions_apiserver::pkg::apis::apiextensions::v1::CustomResourceDefinition;
42/// use kube::{Api, runtime::wait::{await_condition, conditions}};
43/// # async fn wrapper() -> Result<(), Box<dyn std::error::Error>> {
44/// # let client: kube::Client = todo!();
45///
46/// let crds: Api<CustomResourceDefinition> = Api::all(client);
47/// // .. create or apply a crd here ..
48/// let establish = await_condition(crds, "foos.clux.dev", conditions::is_crd_established());
49/// let _ = tokio::time::timeout(std::time::Duration::from_secs(10), establish).await?;
50/// # Ok(())
51/// # }
52/// ```
53#[allow(clippy::missing_panics_doc)] // watch never actually terminates, expect cannot fail
54pub async fn await_condition<K>(api: Api<K>, name: &str, cond: impl Condition<K>) -> Result<Option<K>, Error>
55where
56    K: Clone + Debug + Send + DeserializeOwned + Resource + 'static,
57{
58    // Skip updates until the condition is satisfied.
59    let mut stream = pin!(watch_object(api, name).try_skip_while(|obj| {
60        let matches = cond.matches_object(obj.as_ref());
61        future::ready(Ok(!matches))
62    }));
63
64    // Then take the first update that satisfies the condition.
65    let obj = stream
66        .try_next()
67        .await
68        .map_err(Error::ProbeFailed)?
69        .expect("stream must not terminate");
70    Ok(obj)
71}
72
73/// A trait for condition functions to be used by [`await_condition`]
74///
75/// Note that this is auto-implemented for functions of type `fn(Option<&K>) -> bool`.
76///
77/// # Usage
78///
79/// ```
80/// use kube::runtime::wait::Condition;
81/// use k8s_openapi::api::core::v1::Pod;
82/// fn my_custom_condition(my_cond: &str) -> impl Condition<Pod> + '_ {
83///     move |obj: Option<&Pod>| {
84///         if let Some(pod) = &obj
85///             && let Some(status) = &pod.status
86///             && let Some(conds) = &status.conditions
87///             && let Some(pcond) = conds.iter().find(|c| c.type_ == my_cond)
88///         {
89///             return pcond.status == "True";
90///         }
91///         false
92///     }
93/// }
94/// ```
95pub trait Condition<K> {
96    /// The main trait condition that must be implemented
97    fn matches_object(&self, obj: Option<&K>) -> bool;
98
99    /// Returns a `Condition` that holds if `self` does not
100    ///
101    /// # Usage
102    ///
103    /// ```
104    /// # use kube_runtime::wait::Condition;
105    /// let condition: fn(Option<&()>) -> bool = |_| true;
106    /// assert!(condition.matches_object(None));
107    /// assert!(!condition.not().matches_object(None));
108    /// ```
109    fn not(self) -> conditions::Not<Self>
110    where
111        Self: Sized,
112    {
113        conditions::Not(self)
114    }
115
116    /// Returns a `Condition` that holds if `self` and `other` both do
117    ///
118    /// # Usage
119    ///
120    /// ```
121    /// # use kube_runtime::wait::Condition;
122    /// let cond_false: fn(Option<&()>) -> bool = |_| false;
123    /// let cond_true: fn(Option<&()>) -> bool = |_| true;
124    /// assert!(!cond_false.and(cond_false).matches_object(None));
125    /// assert!(!cond_false.and(cond_true).matches_object(None));
126    /// assert!(!cond_true.and(cond_false).matches_object(None));
127    /// assert!(cond_true.and(cond_true).matches_object(None));
128    /// ```
129    fn and<Other: Condition<K>>(self, other: Other) -> conditions::And<Self, Other>
130    where
131        Self: Sized,
132    {
133        conditions::And(self, other)
134    }
135
136    /// Returns a `Condition` that holds if either `self` or `other` does
137    ///
138    /// # Usage
139    ///
140    /// ```
141    /// # use kube_runtime::wait::Condition;
142    /// let cond_false: fn(Option<&()>) -> bool = |_| false;
143    /// let cond_true: fn(Option<&()>) -> bool = |_| true;
144    /// assert!(!cond_false.or(cond_false).matches_object(None));
145    /// assert!(cond_false.or(cond_true).matches_object(None));
146    /// assert!(cond_true.or(cond_false).matches_object(None));
147    /// assert!(cond_true.or(cond_true).matches_object(None));
148    /// ```
149    fn or<Other: Condition<K>>(self, other: Other) -> conditions::Or<Self, Other>
150    where
151        Self: Sized,
152    {
153        conditions::Or(self, other)
154    }
155}
156
157impl<K, F: Fn(Option<&K>) -> bool> Condition<K> for F {
158    fn matches_object(&self, obj: Option<&K>) -> bool {
159        (self)(obj)
160    }
161}
162
163/// Common conditions to wait for
164pub mod conditions {
165    pub use super::Condition;
166    use k8s_openapi::{
167        api::{
168            apps::v1::Deployment,
169            batch::v1::Job,
170            core::v1::{Pod, Service},
171            networking::v1::Ingress,
172        },
173        apiextensions_apiserver::pkg::apis::apiextensions::v1::CustomResourceDefinition,
174    };
175    use kube_client::Resource;
176
177    /// An await condition that returns `true` once the object has been deleted.
178    ///
179    /// An object is considered to be deleted if the object can no longer be found, or if its
180    /// [`uid`](kube_client::api::ObjectMeta#structfield.uid) changes. This means that an object is considered to be deleted even if we miss
181    /// the deletion event and the object is recreated in the meantime.
182    #[must_use]
183    pub fn is_deleted<K: Resource>(uid: &str) -> impl Condition<K> + '_ {
184        move |obj: Option<&K>| {
185            // NB: Object not found implies success.
186            obj.is_none_or(
187                // Object is found, but a changed uid would mean that it was deleted and recreated
188                |obj| obj.meta().uid.as_deref() != Some(uid),
189            )
190        }
191    }
192
193    /// An await condition that returns `true` once the object has been created.
194    ///
195    /// This is the counterpart to [`is_deleted`](super::conditions::is_deleted).
196    #[must_use]
197    pub fn is_created<K: Resource>() -> impl Condition<K> {
198        |obj: Option<&K>| obj.is_some()
199    }
200
201    /// An await condition for `CustomResourceDefinition` that returns `true` once it has been accepted and established
202    ///
203    /// Note that this condition only guarantees you that you can use `Api<CustomResourceDefinition>` when it is ready.
204    /// It usually takes extra time for Discovery to notice the custom resource, and there is no condition for this.
205    #[must_use]
206    pub fn is_crd_established() -> impl Condition<CustomResourceDefinition> {
207        |obj: Option<&CustomResourceDefinition>| {
208            if let Some(o) = obj
209                && let Some(s) = &o.status
210                && let Some(conds) = &s.conditions
211                && let Some(pcond) = conds.iter().find(|c| c.type_ == "Established")
212            {
213                return pcond.status == "True";
214            }
215            false
216        }
217    }
218
219    /// An await condition for `Pod` that returns `true` once it is running
220    #[must_use]
221    pub fn is_pod_running() -> impl Condition<Pod> {
222        |obj: Option<&Pod>| {
223            if let Some(pod) = &obj
224                && let Some(status) = &pod.status
225                && let Some(phase) = &status.phase
226            {
227                return phase == "Running";
228            }
229            false
230        }
231    }
232
233    /// An await condition for `Job` that returns `true` once it is completed
234    #[must_use]
235    pub fn is_job_completed() -> impl Condition<Job> {
236        |obj: Option<&Job>| {
237            if let Some(job) = &obj
238                && let Some(s) = &job.status
239                && let Some(conds) = &s.conditions
240                && let Some(pcond) = conds.iter().find(|c| c.type_ == "Complete")
241            {
242                return pcond.status == "True";
243            }
244            false
245        }
246    }
247
248    /// An await condition for `Deployment` that returns `true` once the latest deployment has completed
249    ///
250    /// This looks for the condition that Kubernetes sets for completed deployments:
251    /// <https://kubernetes.io/docs/concepts/workloads/controllers/deployment/#complete-deployment>
252    #[must_use]
253    pub fn is_deployment_completed() -> impl Condition<Deployment> {
254        |obj: Option<&Deployment>| {
255            if let Some(depl) = &obj
256                && let Some(s) = &depl.status
257                && let Some(conds) = &s.conditions
258                && let Some(dcond) = conds.iter().find(|c| {
259                    k8s_openapi::k8s_if_ge_1_37! {
260                        c.type_.as_deref() == Some("Progressing")
261                            && c.reason.as_deref() == Some("NewReplicaSetAvailable")
262                    }
263                    k8s_openapi::k8s_if_le_1_36! {
264                        c.type_ == "Progressing" && c.reason == Some("NewReplicaSetAvailable".to_string())
265                    }
266                })
267            {
268                k8s_openapi::k8s_if_ge_1_37! {
269                  return dcond.status.as_deref() == Some("True");
270                }
271                k8s_openapi::k8s_if_le_1_36! {
272                  return dcond.status == "True";
273                }
274            }
275            false
276        }
277    }
278
279    /// An await condition for `Service`s of type `LoadBalancer` that returns `true` once the backing load balancer has an external IP or hostname
280    #[must_use]
281    pub fn is_service_loadbalancer_provisioned() -> impl Condition<Service> {
282        |obj: Option<&Service>| {
283            if let Some(svc) = &obj
284                && let Some(spec) = &svc.spec
285            {
286                // ignore services that are not type LoadBalancer (return true immediately)
287                if spec.type_ != Some("LoadBalancer".to_string()) {
288                    return true;
289                }
290                // carry on if this is a LoadBalancer service
291                if let Some(s) = &svc.status
292                    && let Some(lbs) = &s.load_balancer
293                    && let Some(ings) = &lbs.ingress
294                {
295                    return ings.iter().all(|ip| ip.ip.is_some() || ip.hostname.is_some());
296                }
297            }
298            false
299        }
300    }
301
302    /// An await condition for `Ingress` that returns `true` once the backing load balancer has an external IP or hostname
303    #[must_use]
304    pub fn is_ingress_provisioned() -> impl Condition<Ingress> {
305        |obj: Option<&Ingress>| {
306            if let Some(ing) = &obj
307                && let Some(s) = &ing.status
308                && let Some(lbs) = &s.load_balancer
309                && let Some(ings) = &lbs.ingress
310            {
311                return ings.iter().all(|ip| ip.ip.is_some() || ip.hostname.is_some());
312            }
313            false
314        }
315    }
316
317    /// See [`Condition::not`]
318    #[derive(Copy, Clone, Debug, PartialEq, Eq)]
319    pub struct Not<A>(pub(super) A);
320    impl<A: Condition<K>, K> Condition<K> for Not<A> {
321        fn matches_object(&self, obj: Option<&K>) -> bool {
322            !self.0.matches_object(obj)
323        }
324    }
325
326    /// See [`Condition::and`]
327    #[derive(Copy, Clone, Debug, PartialEq, Eq)]
328    pub struct And<A, B>(pub(super) A, pub(super) B);
329    impl<A, B, K> Condition<K> for And<A, B>
330    where
331        A: Condition<K>,
332        B: Condition<K>,
333    {
334        fn matches_object(&self, obj: Option<&K>) -> bool {
335            self.0.matches_object(obj) && self.1.matches_object(obj)
336        }
337    }
338
339    /// See [`Condition::or`]
340    #[derive(Copy, Clone, Debug, PartialEq, Eq)]
341    pub struct Or<A, B>(pub(super) A, pub(super) B);
342    impl<A, B, K> Condition<K> for Or<A, B>
343    where
344        A: Condition<K>,
345        B: Condition<K>,
346    {
347        fn matches_object(&self, obj: Option<&K>) -> bool {
348            self.0.matches_object(obj) || self.1.matches_object(obj)
349        }
350    }
351
352    mod tests {
353        #[test]
354        /// pass when CRD is established
355        fn crd_established_ok() {
356            use super::{Condition, is_crd_established};
357
358            let crd = r#"
359                apiVersion: apiextensions.k8s.io/v1
360                kind: CustomResourceDefinition
361                metadata:
362                  name: testthings.kube.rs
363                spec:
364                  group: kube.rs
365                  names:
366                    categories: []
367                    kind: TestThing
368                    plural: testthings
369                    shortNames: []
370                    singular: testthing
371                  scope: Namespaced
372                  versions:
373                    - additionalPrinterColumns: []
374                      name: v1
375                      schema:
376                        openAPIV3Schema:
377                          type: object
378                          x-kubernetes-preserve-unknown-fields: true
379                      served: true
380                      storage: true
381                status:
382                  acceptedNames:
383                    kind: TestThing
384                    listKind: TestThingList
385                    plural: testthings
386                    singular: testthing
387                  conditions:
388                    - lastTransitionTime: "2025-03-06T03:10:03Z"
389                      message: no conflicts found
390                      reason: NoConflicts
391                      status: "True"
392                      type: NamesAccepted
393                    - lastTransitionTime: "2025-03-06T03:10:03Z"
394                      message: the initial names have been accepted
395                      reason: InitialNamesAccepted
396                      status: "True"
397                      type: Established
398                storedVersions:
399                  - v1
400            "#;
401
402            let c = serde_saphyr::from_str(crd).unwrap();
403            assert!(is_crd_established().matches_object(Some(&c)))
404        }
405
406        #[test]
407        /// fail when CRD is not yet ready
408        fn crd_established_fail() {
409            use super::{Condition, is_crd_established};
410
411            let crd = r#"
412                apiVersion: apiextensions.k8s.io/v1
413                kind: CustomResourceDefinition
414                metadata:
415                  name: testthings.kube.rs
416                spec:
417                  group: kube.rs
418                  names:
419                    categories: []
420                    kind: TestThing
421                    plural: testthings
422                    shortNames: []
423                    singular: testthing
424                  scope: Namespaced
425                  versions:
426                    - additionalPrinterColumns: []
427                      name: v1
428                      schema:
429                        openAPIV3Schema:
430                          type: object
431                          x-kubernetes-preserve-unknown-fields: true
432                      served: true
433                      storage: true
434                status:
435                  acceptedNames:
436                    kind: TestThing
437                    listKind: TestThingList
438                    plural: testthings
439                    singular: testthing
440                  conditions:
441                    - lastTransitionTime: "2025-03-06T03:10:03Z"
442                      message: no conflicts found
443                      reason: NoConflicts
444                      status: "True"
445                      type: NamesAccepted
446                    - lastTransitionTime: "2025-03-06T03:10:03Z"
447                      message: the initial names have been accepted
448                      reason: InitialNamesAccepted
449                      status: "False"
450                      type: Established
451                storedVersions:
452                  - v1
453            "#;
454
455            let c = serde_saphyr::from_str(crd).unwrap();
456            assert!(!is_crd_established().matches_object(Some(&c)))
457        }
458
459        #[test]
460        /// fail when CRD does not exist
461        fn crd_established_missing() {
462            use super::{Condition, is_crd_established};
463
464            assert!(!is_crd_established().matches_object(None))
465        }
466
467        #[test]
468        /// pass when pod is running
469        fn pod_running_ok() {
470            use super::{Condition, is_pod_running};
471
472            let pod = r#"
473                apiVersion: v1
474                kind: Pod
475                metadata:
476                  namespace: default
477                  name: testpod
478                spec:
479                  containers:
480                    - name: testcontainer
481                      image: alpine
482                      command: [ sleep ]
483                      args: [ "100000" ]
484                status:
485                  conditions:
486                    - lastProbeTime: null
487                      lastTransitionTime: "2025-03-06T03:53:07Z"
488                      status: "True"
489                      type: PodReadyToStartContainers
490                    - lastProbeTime: null
491                      lastTransitionTime: "2025-03-06T03:52:58Z"
492                      status: "True"
493                      type: Initialized
494                    - lastProbeTime: null
495                      lastTransitionTime: "2025-03-06T03:53:24Z"
496                      status: "True"
497                      type: Ready
498                    - lastProbeTime: null
499                      lastTransitionTime: "2025-03-06T03:53:24Z"
500                      status: "True"
501                      type: ContainersReady
502                    - lastProbeTime: null
503                      lastTransitionTime: "2025-03-06T03:52:58Z"
504                      status: "True"
505                      type: PodScheduled
506                  containerStatuses:
507                    - containerID: containerd://598323380ae59d60c1ab98f9091c94659137a976d52136a8083775d47fea5875
508                      image: docker.io/library/alpine:latest
509                      imageID: docker.io/library/alpine@sha256:a8560b36e8b8210634f77d9f7f9efd7ffa463e380b75e2e74aff4511df3ef88c
510                      lastState: {}
511                      name: testcontainer
512                      ready: true
513                      restartCount: 0
514                      started: true
515                      state:
516                        running:
517                          startedAt: "2025-03-06T03:59:20Z"
518                  phase: Running
519                  qosClass: Burstable
520            "#;
521
522            let p = serde_saphyr::from_str(pod).unwrap();
523            assert!(is_pod_running().matches_object(Some(&p)))
524        }
525
526        #[test]
527        /// fail if pod is unschedulable
528        fn pod_running_unschedulable() {
529            use super::{Condition, is_pod_running};
530
531            let pod = r#"
532                apiVersion: v1
533                kind: Pod
534                metadata:
535                  namespace: default
536                  name: testpod
537                spec:
538                  containers:
539                    - name: testcontainer
540                      image: alpine
541                      command: [ sleep ]
542                      args: [ "100000" ]
543                status:
544                  conditions:
545                    - lastProbeTime: null
546                      lastTransitionTime: "2025-03-06T03:52:25Z"
547                      message: '0/1 nodes are available: 1 node(s) were unschedulable. preemption: 0/1 nodes are available: 1 Preemption is not helpful for scheduling.'
548                      reason: Unschedulable
549                      status: "False"
550                      type: PodScheduled
551                  phase: Pending
552                  qosClass: Burstable
553            "#;
554
555            let p = serde_saphyr::from_str(pod).unwrap();
556            assert!(!is_pod_running().matches_object(Some(&p)))
557        }
558
559        #[test]
560        /// fail if pod does not exist
561        fn pod_running_missing() {
562            use super::{Condition, is_pod_running};
563
564            assert!(!is_pod_running().matches_object(None))
565        }
566
567        #[test]
568        /// pass if job completed
569        fn job_completed_ok() {
570            use super::{Condition, is_job_completed};
571
572            let job = r#"
573                apiVersion: batch/v1
574                kind: Job
575                metadata:
576                  name: pi
577                  namespace: default
578                spec:
579                  template:
580                    spec:
581                      containers:
582                      - name: pi
583                        command:
584                        - perl
585                        - -Mbignum=bpi
586                        - -wle
587                        - print bpi(2000)
588                        image: perl:5.34.0
589                        imagePullPolicy: IfNotPresent
590                status:
591                  completionTime: "2025-03-06T05:27:56Z"
592                  conditions:
593                  - lastProbeTime: "2025-03-06T05:27:56Z"
594                    lastTransitionTime: "2025-03-06T05:27:56Z"
595                    message: Reached expected number of succeeded pods
596                    reason: CompletionsReached
597                    status: "True"
598                    type: SuccessCriteriaMet
599                  - lastProbeTime: "2025-03-06T05:27:56Z"
600                    lastTransitionTime: "2025-03-06T05:27:56Z"
601                    message: Reached expected number of succeeded pods
602                    reason: CompletionsReached
603                    status: "True"
604                    type: Complete
605                  ready: 0
606                  startTime: "2025-03-06T05:27:27Z"
607                  succeeded: 1
608                  terminating: 0
609                  uncountedTerminatedPods: {}
610            "#;
611
612            let j = serde_saphyr::from_str(job).unwrap();
613            assert!(is_job_completed().matches_object(Some(&j)))
614        }
615
616        #[test]
617        /// fail if job is still in progress
618        fn job_completed_running() {
619            use super::{Condition, is_job_completed};
620
621            let job = r#"
622                apiVersion: batch/v1
623                kind: Job
624                metadata:
625                  name: pi
626                  namespace: default
627                spec:
628                  backoffLimit: 4
629                  completionMode: NonIndexed
630                  completions: 1
631                  manualSelector: false
632                  parallelism: 1
633                  template:
634                    spec:
635                      containers:
636                      - name: pi
637                        command:
638                        - perl
639                        - -Mbignum=bpi
640                        - -wle
641                        - print bpi(2000)
642                        image: perl:5.34.0
643                        imagePullPolicy: IfNotPresent
644                status:
645                  active: 1
646                  ready: 0
647                  startTime: "2025-03-06T05:27:27Z"
648                  terminating: 0
649                  uncountedTerminatedPods: {}
650            "#;
651
652            let j = serde_saphyr::from_str(job).unwrap();
653            assert!(!is_job_completed().matches_object(Some(&j)))
654        }
655
656        #[test]
657        /// fail if job does not exist
658        fn job_completed_missing() {
659            use super::{Condition, is_job_completed};
660
661            assert!(!is_job_completed().matches_object(None))
662        }
663
664        #[test]
665        /// pass when deployment has been fully rolled out
666        fn deployment_completed_ok() {
667            use super::{Condition, is_deployment_completed};
668
669            let depl = r#"
670                apiVersion: apps/v1
671                kind: Deployment
672                metadata:
673                  name: testapp
674                  namespace: default
675                spec:
676                  progressDeadlineSeconds: 600
677                  replicas: 3
678                  revisionHistoryLimit: 10
679                  selector:
680                    matchLabels:
681                      app: test
682                  strategy:
683                    rollingUpdate:
684                      maxSurge: 25%
685                      maxUnavailable: 25%
686                    type: RollingUpdate
687                  template:
688                    metadata:
689                      creationTimestamp: null
690                      labels:
691                        app: test
692                    spec:
693                      containers:
694                      - image: postgres
695                        imagePullPolicy: Always
696                        name: postgres
697                        ports:
698                        - containerPort: 5432
699                          protocol: TCP
700                        env:
701                        - name: POSTGRES_PASSWORD
702                          value: foobar
703                status:
704                  availableReplicas: 3
705                  conditions:
706                  - lastTransitionTime: "2025-03-06T06:06:57Z"
707                    lastUpdateTime: "2025-03-06T06:06:57Z"
708                    message: Deployment has minimum availability.
709                    reason: MinimumReplicasAvailable
710                    status: "True"
711                    type: Available
712                  - lastTransitionTime: "2025-03-06T06:03:20Z"
713                    lastUpdateTime: "2025-03-06T06:06:57Z"
714                    message: ReplicaSet "testapp-7fcd4b58c9" has successfully progressed.
715                    reason: NewReplicaSetAvailable
716                    status: "True"
717                    type: Progressing
718                  observedGeneration: 2
719                  readyReplicas: 3
720                  replicas: 3
721                  updatedReplicas: 3
722            "#;
723
724            let d = serde_saphyr::from_str(depl).unwrap();
725            assert!(is_deployment_completed().matches_object(Some(&d)))
726        }
727
728        #[test]
729        /// fail if deployment update is still rolling out
730        fn deployment_completed_pending() {
731            use super::{Condition, is_deployment_completed};
732
733            let depl = r#"
734                apiVersion: apps/v1
735                kind: Deployment
736                metadata:
737                  name: testapp
738                  namespace: default
739                spec:
740                  progressDeadlineSeconds: 600
741                  replicas: 3
742                  revisionHistoryLimit: 10
743                  selector:
744                    matchLabels:
745                      app: test
746                  strategy:
747                    rollingUpdate:
748                      maxSurge: 25%
749                      maxUnavailable: 25%
750                    type: RollingUpdate
751                  template:
752                    metadata:
753                      creationTimestamp: null
754                      labels:
755                        app: test
756                    spec:
757                      containers:
758                      - image: postgres
759                        imagePullPolicy: Always
760                        name: postgres
761                        ports:
762                        - containerPort: 5432
763                          protocol: TCP
764                        env:
765                        - name: POSTGRES_PASSWORD
766                          value: foobar
767                status:
768                  conditions:
769                  - lastTransitionTime: "2025-03-06T06:03:20Z"
770                    lastUpdateTime: "2025-03-06T06:03:20Z"
771                    message: Deployment does not have minimum availability.
772                    reason: MinimumReplicasUnavailable
773                    status: "False"
774                    type: Available
775                  - lastTransitionTime: "2025-03-06T06:03:20Z"
776                    lastUpdateTime: "2025-03-06T06:03:20Z"
777                    message: ReplicaSet "testapp-77789cd7d4" is progressing.
778                    reason: ReplicaSetUpdated
779                    status: "True"
780                    type: Progressing
781                  observedGeneration: 1
782                  replicas: 3
783                  unavailableReplicas: 3
784                  updatedReplicas: 3
785            "#;
786
787            let d = serde_saphyr::from_str(depl).unwrap();
788            assert!(!is_deployment_completed().matches_object(Some(&d)))
789        }
790
791        #[test]
792        /// fail if deployment does not exist
793        fn deployment_completed_missing() {
794            use super::{Condition, is_deployment_completed};
795
796            assert!(!is_deployment_completed().matches_object(None))
797        }
798
799        #[test]
800        /// pass if loadbalancer service has received a loadbalancer IP
801        fn service_lb_provisioned_ok_ip() {
802            use super::{Condition, is_service_loadbalancer_provisioned};
803
804            let service = r"
805                apiVersion: v1
806                kind: Service
807                metadata:
808                  name: test
809                spec:
810                  selector:
811                    app.kubernetes.io/name: test
812                  type: LoadBalancer
813                  ports:
814                    - protocol: TCP
815                      port: 80
816                      targetPort: 9376
817                  clusterIP: 10.0.171.239
818                status:
819                  loadBalancer:
820                    ingress:
821                      - ip: 192.0.2.127
822            ";
823
824            let s = serde_saphyr::from_str(service).unwrap();
825            assert!(is_service_loadbalancer_provisioned().matches_object(Some(&s)))
826        }
827
828        #[test]
829        /// pass if loadbalancer service has received a loadbalancer hostname
830        fn service_lb_provisioned_ok_hostname() {
831            use super::{Condition, is_service_loadbalancer_provisioned};
832
833            let service = r"
834                apiVersion: v1
835                kind: Service
836                metadata:
837                  name: test
838                spec:
839                  selector:
840                    app.kubernetes.io/name: test
841                  type: LoadBalancer
842                  ports:
843                    - protocol: TCP
844                      port: 80
845                      targetPort: 9376
846                  clusterIP: 10.0.171.239
847                status:
848                  loadBalancer:
849                    ingress:
850                      - hostname: example.exposed.service
851            ";
852
853            let s = serde_saphyr::from_str(service).unwrap();
854            assert!(is_service_loadbalancer_provisioned().matches_object(Some(&s)))
855        }
856
857        #[test]
858        /// fail if loadbalancer service is still waiting for a LB
859        fn service_lb_provisioned_pending() {
860            use super::{Condition, is_service_loadbalancer_provisioned};
861
862            let service = r"
863                apiVersion: v1
864                kind: Service
865                metadata:
866                  name: test
867                spec:
868                  selector:
869                    app.kubernetes.io/name: test
870                  type: LoadBalancer
871                  ports:
872                    - protocol: TCP
873                      port: 80
874                      targetPort: 9376
875                  clusterIP: 10.0.171.239
876                status:
877                  loadBalancer: {}
878            ";
879
880            let s = serde_saphyr::from_str(service).unwrap();
881            assert!(!is_service_loadbalancer_provisioned().matches_object(Some(&s)))
882        }
883
884        #[test]
885        /// pass if service is not a loadbalancer
886        fn service_lb_provisioned_not_loadbalancer() {
887            use super::{Condition, is_service_loadbalancer_provisioned};
888
889            let service = r"
890                apiVersion: v1
891                kind: Service
892                metadata:
893                  name: test
894                spec:
895                  selector:
896                    app.kubernetes.io/name: test
897                  type: ClusterIP
898                  ports:
899                    - protocol: TCP
900                      port: 80
901                      targetPort: 9376
902                status:
903                  loadBalancer: {}
904            ";
905
906            let s = serde_saphyr::from_str(service).unwrap();
907            assert!(is_service_loadbalancer_provisioned().matches_object(Some(&s)))
908        }
909
910        #[test]
911        /// fail if service does not exist
912        fn service_lb_provisioned_missing() {
913            use super::{Condition, is_service_loadbalancer_provisioned};
914
915            assert!(!is_service_loadbalancer_provisioned().matches_object(None))
916        }
917
918        #[test]
919        /// pass when ingress has received a loadbalancer IP
920        fn ingress_provisioned_ok_ip() {
921            use super::{Condition, is_ingress_provisioned};
922
923            let ingress = r#"
924                apiVersion: networking.k8s.io/v1
925                kind: Ingress
926                metadata:
927                  name: test
928                  namespace: default
929                  resourceVersion: "1401"
930                  uid: d653ee4d-0adb-40d9-b03c-7f84f35d4a67
931                spec:
932                  ingressClassName: nginx
933                  rules:
934                  - host: httpbin.local
935                    http:
936                      paths:
937                      - path: /
938                        backend:
939                          service:
940                            name: httpbin
941                            port:
942                              number: 80
943                status:
944                  loadBalancer:
945                    ingress:
946                      - ip: 10.89.7.3
947            "#;
948
949            let i = serde_saphyr::from_str(ingress).unwrap();
950            assert!(is_ingress_provisioned().matches_object(Some(&i)))
951        }
952
953        #[test]
954        /// pass when ingress has received a loadbalancer hostname
955        fn ingress_provisioned_ok_hostname() {
956            use super::{Condition, is_ingress_provisioned};
957
958            let ingress = r#"
959                apiVersion: networking.k8s.io/v1
960                kind: Ingress
961                metadata:
962                  name: test
963                  namespace: default
964                  resourceVersion: "1401"
965                  uid: d653ee4d-0adb-40d9-b03c-7f84f35d4a67
966                spec:
967                  ingressClassName: nginx
968                  rules:
969                  - host: httpbin.local
970                    http:
971                      paths:
972                      - path: /
973                        backend:
974                          service:
975                            name: httpbin
976                            port:
977                              number: 80
978                status:
979                  loadBalancer:
980                    ingress:
981                      - hostname: example.exposed.service
982            "#;
983
984            let i = serde_saphyr::from_str(ingress).unwrap();
985            assert!(is_ingress_provisioned().matches_object(Some(&i)))
986        }
987
988        #[test]
989        /// fail if ingress is still waiting for a LB
990        fn ingress_provisioned_pending() {
991            use super::{Condition, is_ingress_provisioned};
992
993            let ingress = r#"
994                apiVersion: networking.k8s.io/v1
995                kind: Ingress
996                metadata:
997                  name: test
998                  namespace: default
999                  resourceVersion: "1401"
1000                  uid: d653ee4d-0adb-40d9-b03c-7f84f35d4a67
1001                spec:
1002                  ingressClassName: nginx
1003                  rules:
1004                  - host: httpbin.local
1005                    http:
1006                      paths:
1007                      - path: /
1008                        backend:
1009                          service:
1010                            name: httpbin
1011                            port:
1012                              number: 80
1013                status:
1014                  loadBalancer: {}
1015            "#;
1016
1017            let i = serde_saphyr::from_str(ingress).unwrap();
1018            assert!(!is_ingress_provisioned().matches_object(Some(&i)))
1019        }
1020
1021        #[test]
1022        /// fail if ingress does not exist
1023        fn ingress_provisioned_missing() {
1024            use super::{Condition, is_ingress_provisioned};
1025
1026            assert!(!is_ingress_provisioned().matches_object(None))
1027        }
1028    }
1029}
1030
1031/// Utilities for deleting objects
1032pub mod delete {
1033    use super::{await_condition, conditions};
1034    use kube_client::{Api, Resource, api::DeleteParams};
1035    use serde::de::DeserializeOwned;
1036    use std::fmt::Debug;
1037    use thiserror::Error;
1038
1039    /// Errors from `delete_and_finalize`
1040    #[derive(Debug, Error)]
1041    pub enum Error {
1042        /// No uid found on metadata.uid on deleted object
1043        #[error("deleted object has no UID to wait for")]
1044        NoUid,
1045
1046        /// Apiserver returned an error to the delete call
1047        #[error("failed to delete object: {0}")]
1048        Delete(#[source] kube_client::Error),
1049
1050        /// A watcher failed to probe the object for the `is_deleted` condition
1051        #[error("failed to wait for object to be deleted: {0}")]
1052        Await(#[source] super::Error),
1053    }
1054
1055    /// Delete an object, and wait for it to be removed from the Kubernetes API (including waiting for all finalizers to unregister themselves).
1056    ///
1057    /// # Errors
1058    ///
1059    /// Returns an [`Error`](enum@super::Error) if the object was unable to be deleted, or if the wait was interrupted.
1060    pub async fn delete_and_finalize<K: Clone + Debug + Send + DeserializeOwned + Resource + 'static>(
1061        api: Api<K>,
1062        name: &str,
1063        delete_params: &DeleteParams,
1064    ) -> Result<(), Error> {
1065        let deleted_obj_uid = api
1066            .delete(name, delete_params)
1067            .await
1068            .map_err(Error::Delete)?
1069            .either(
1070                |mut obj| obj.meta_mut().uid.take(),
1071                |status| status.details.map(|details| details.uid),
1072            )
1073            .ok_or(Error::NoUid)?;
1074        await_condition(api, name, conditions::is_deleted(&deleted_obj_uid))
1075            .await
1076            .map_err(Error::Await)?;
1077        Ok(())
1078    }
1079}
kube_runtime/wait.rs

kube_runtime/
wait.rs
