security_client_rs/fleet/
wait.rs

1//! Waiting for asynchronous Fleet work. Each wait is bounded by
2//! [`PollOptions`]; see [`crate::poll`] for the deadline rules.
3use super::{AgentActionStatus, AgentUpload, Fleet, agents::Agent};
4use crate::{
5    Result,
6    poll::{self, PollOptions, WaitOutcome},
7};
8
9impl Fleet<'_> {
10    /// Waits until an agent action finishes.
11    ///
12    /// Kibana has no lookup by action ID, so each check searches progressively
13    /// larger history windows, up to 10,000 action documents, within the
14    /// deadline. Bulk actions can occupy multiple documents. An action that was
15    /// seen and then can no longer be found, for example because newer actions
16    /// pushed it out of that window, ends the
17    /// wait as [`WaitOutcome::Vanished`] with its last state.
18    ///
19    /// A finished action can contain failed agents. Unknown action statuses also
20    /// end the wait for inspection; check the status, failure count and errors
21    /// on the returned [`AgentActionStatus`].
22    pub async fn wait_for_action(
23        &self,
24        action_id: &str,
25        options: PollOptions,
26    ) -> Result<WaitOutcome<AgentActionStatus>> {
27        poll::wait(
28            options,
29            || self.find_action(action_id),
30            AgentActionStatus::is_finished,
31        )
32        .await
33    }
34
35    async fn find_action(&self, action_id: &str) -> Result<Option<AgentActionStatus>> {
36        for size in [100, 1_000, 10_000] {
37            let actions = self
38                .agent_action_status()
39                .page(0)
40                .per_page(size)
41                .send()
42                .await?
43                .json()
44                .await?;
45            if let Some(action) = actions.items.into_iter().find(|a| a.action_id == action_id) {
46                return Ok(Some(action));
47            }
48        }
49        Ok(None)
50    }
51
52    /// Waits until the upload created by a diagnostics action finishes.
53    /// Download it with [`download_agent_file`](Self::download_agent_file) once
54    /// it is [`Ready`](super::UploadStatus::Ready). An upload that was seen and
55    /// then disappears ends the wait as [`WaitOutcome::Vanished`].
56    pub async fn wait_for_upload(
57        &self,
58        agent_id: &str,
59        action_id: &str,
60        options: PollOptions,
61    ) -> Result<WaitOutcome<AgentUpload>> {
62        poll::wait(
63            options,
64            || async {
65                let uploads = self
66                    .list_agent_uploads(agent_id)
67                    .send()
68                    .await?
69                    .json()
70                    .await?;
71                Ok(uploads.items.into_iter().find(|u| u.action_id == action_id))
72            },
73            AgentUpload::is_finished,
74        )
75        .await
76    }
77
78    /// Waits until the agent is online and reports `policy_id` at `revision` or
79    /// later. An agent that has not reported a revision has not acknowledged it.
80    pub async fn wait_for_agent_policy(
81        &self,
82        agent_id: &str,
83        policy_id: &str,
84        revision: u64,
85        options: PollOptions,
86    ) -> Result<WaitOutcome<Agent>> {
87        poll::wait(
88            options,
89            || async {
90                Ok(Some(
91                    self.get_agent(agent_id).send().await?.json().await?.item,
92                ))
93            },
94            |agent| {
95                agent.policy_id.as_deref() == Some(policy_id)
96                    && agent.status.as_deref() == Some("online")
97                    && agent
98                        .policy_revision
99                        .is_some_and(|current| current >= revision)
100            },
101        )
102        .await
103    }
104}
security_client_rs/fleet/wait.rs

security_client_rs/fleet/
wait.rs
