icinga2_api/api/
async_client.rs

1//! Main API object (async version)
2
3use std::{path::Path, str::from_utf8};
4
5use futures::Stream;
6use futures::StreamExt as _;
7use futures::stream::TryStreamExt as _;
8use serde::{Serialize, de::DeserializeOwned};
9use tokio::io::AsyncBufReadExt as _;
10use tokio_stream::wrappers::LinesStream;
11use tokio_util::io::StreamReader;
12
13use crate::config::Icinga2Instance;
14use crate::types::{
15    enums::event_stream_type::IcingaEventStreamType,
16    event_stream::IcingaEvent,
17    filter::IcingaFilter,
18    rest::{RestApiEndpoint, RestApiResponse},
19};
20
21/// the runtime object for an Icinga2 instance (blocking variant)
22#[derive(Debug, Clone)]
23pub struct Icinga2Async {
24    /// the HTTP client to use
25    client: reqwest::Client,
26    /// the base URL for the Icinga API
27    pub url: url::Url,
28    /// username
29    pub username: String,
30    /// password
31    password: String,
32}
33
34impl Icinga2Async {
35    /// create a new Icinga2 instance from a config that was
36    /// either manually created or previously loaded via [Icinga2Instance::from_config_file]
37    ///
38    /// # Errors
39    /// this fails if the CA certificate file mentioned in the configuration
40    /// can not be found or parsed
41    pub fn from_instance_config(config: &Icinga2Instance) -> Result<Self, crate::error::Error> {
42        let client_builder = reqwest::ClientBuilder::new();
43        let client_builder = client_builder.user_agent(concat!(
44            env!("CARGO_PKG_NAME"),
45            "/",
46            env!("CARGO_PKG_VERSION")
47        ));
48        let mut headers = reqwest::header::HeaderMap::new();
49        headers.insert(
50            "Content-Type",
51            reqwest::header::HeaderValue::from_static("application/json"),
52        );
53        headers.insert(
54            "Accept",
55            reqwest::header::HeaderValue::from_static("application/json"),
56        );
57        let client_builder = client_builder.default_headers(headers);
58        let client_builder = if let Some(ca_certificate) = &config.ca_certificate {
59            let ca_cert_content = fs_err::read(ca_certificate)
60                .map_err(crate::error::Error::CouldNotReadCACertFile)?;
61            let ca_cert = reqwest::Certificate::from_pem(&ca_cert_content)
62                .map_err(crate::error::Error::CouldNotParsePEMCACertificate)?;
63            client_builder.tls_certs_only([ca_cert])
64        } else {
65            client_builder
66        };
67        let client = client_builder
68            .build()
69            .map_err(crate::error::Error::CouldNotBuildReqwestClientFromSuppliedInformation)?;
70        let url =
71            url::Url::parse(&config.url).map_err(crate::error::Error::CouldNotParseUrlInConfig)?;
72        let username = config.username.clone();
73        let password = config.password.clone();
74        Ok(Self {
75            client,
76            url,
77            username,
78            password,
79        })
80    }
81
82    /// create a new Icinga2 instance from a TOML config file
83    ///
84    /// # Errors
85    /// this fails if the configuration file can not be found or parsed
86    /// or the CA certificate file mentioned in the configuration file
87    /// can not be found or parsed
88    pub fn from_config_file(path: &Path) -> Result<Self, crate::error::Error> {
89        let icinga_instance = Icinga2Instance::from_config_file(path)?;
90        Self::from_instance_config(&icinga_instance)
91    }
92
93    /// common code for the REST API calls
94    ///
95    /// # Errors
96    ///
97    /// this returns an error if encoding, the actual request, or decoding of the response fail
98    #[expect(
99        clippy::future_not_send,
100        reason = "neither ApiEndpoint nor its RequestBody is required to be Send; callers that need a Send future can wrap with their own bounds"
101    )]
102    pub async fn rest<ApiEndpoint, Res>(
103        &self,
104        api_endpoint: ApiEndpoint,
105    ) -> Result<Res, crate::error::Error>
106    where
107        ApiEndpoint: RestApiEndpoint,
108        <ApiEndpoint as RestApiEndpoint>::RequestBody: Clone + Serialize + std::fmt::Debug,
109        Res: DeserializeOwned + std::fmt::Debug + RestApiResponse<ApiEndpoint>,
110    {
111        let method = api_endpoint.method()?;
112        let url = api_endpoint.url(&self.url)?;
113        let request_body: Option<std::borrow::Cow<<ApiEndpoint as RestApiEndpoint>::RequestBody>> =
114            api_endpoint.request_body()?;
115        let actual_method = if method == reqwest::Method::GET && request_body.is_some() {
116            reqwest::Method::POST
117        } else {
118            method.to_owned()
119        };
120        let mut req = self.client.request(actual_method, url.to_owned());
121        if method == reqwest::Method::GET && request_body.is_some() {
122            tracing::trace!("Sending GET request with body as POST via X-HTTP-Method-Override");
123            req = req.header(
124                "X-HTTP-Method-Override",
125                reqwest::header::HeaderValue::from_static("GET"),
126            );
127        }
128        req = req.basic_auth(&self.username, Some(&self.password));
129        if let Some(request_body) = request_body {
130            tracing::trace!("Request body:\n{:#?}", request_body);
131            req = req.json(&request_body);
132        }
133        let result = req.send().await;
134        if let Err(ref e) = result {
135            tracing::error!(%url, %method, "Icinga2 send error: {:?}", e);
136        }
137        let result = result?;
138        let status = result.status();
139        let response_body = result.bytes().await?;
140        match from_utf8(&response_body) {
141            Ok(response_body) => {
142                tracing::trace!("Response body:\n{}", &response_body);
143            }
144            Err(e) => {
145                tracing::trace!(
146                    "Response body that could not be parsed as utf8 because of {}:\n{:?}",
147                    &e,
148                    &response_body
149                );
150            }
151        }
152        if status.is_client_error() {
153            tracing::error!(%url, %method, "Icinga2 status error (client error): {:?}", status);
154        } else if status.is_server_error() {
155            tracing::error!(%url, %method, "Icinga2 status error (server error): {:?}", status);
156        }
157        if response_body.is_empty() {
158            Err(crate::error::Error::EmptyResponseBody(status))
159        } else {
160            let jd = &mut serde_json::Deserializer::from_slice(&response_body);
161            match serde_path_to_error::deserialize(jd) {
162                Ok(response_body) => {
163                    tracing::trace!("Parsed response body:\n{:#?}", response_body);
164                    Ok(response_body)
165                }
166                Err(e) => {
167                    let path = e.path();
168                    tracing::error!("Parsing failed at path {}: {}", path.to_string(), e.inner());
169                    if let Ok(response_body) = serde_json::from_slice(&response_body) {
170                        let mut response_body: serde_json::Value = response_body;
171                        for segment in path {
172                            let next = match (&response_body, segment) {
173                                (
174                                    serde_json::Value::Array(vs),
175                                    serde_path_to_error::Segment::Seq { index },
176                                ) => {
177                                    if let Some(v) = vs.get(*index) {
178                                        v.clone()
179                                    } else {
180                                        // if we can not find the element serde_path_to_error references fall back to just returning the error
181                                        return Err(e.into());
182                                    }
183                                }
184                                (
185                                    serde_json::Value::Object(m),
186                                    serde_path_to_error::Segment::Map { key },
187                                ) => {
188                                    if let Some(v) = m.get(key) {
189                                        v.clone()
190                                    } else {
191                                        // if we can not find the element serde_path_to_error references fall back to just returning the error
192                                        return Err(e.into());
193                                    }
194                                }
195                                _ => {
196                                    // if we can not find the element serde_path_to_error references fall back to just returning the error
197                                    return Err(e.into());
198                                }
199                            };
200                            response_body = next;
201                        }
202                        tracing::error!("Value in location path references is: {}", response_body);
203                    }
204                    Err(e.into())
205                }
206            }
207        }
208    }
209
210    /// Long-polling on an event stream
211    ///
212    /// # Errors
213    ///
214    /// this returns an error if encoding or the actual request fail
215    pub async fn event_stream(
216        &self,
217        types: &[IcingaEventStreamType],
218        queue: &str,
219        filter: Option<IcingaFilter>,
220    ) -> Result<impl Stream<Item = Result<IcingaEvent, std::io::Error>>, crate::error::Error> {
221        let method = reqwest::Method::POST;
222        let mut url = self
223            .url
224            .join("v1/events")
225            .map_err(crate::error::Error::CouldNotParseUrlFragment)?;
226        for t in types {
227            url.query_pairs_mut().append_pair("types", &t.to_string());
228        }
229        url.query_pairs_mut().append_pair("queue", queue);
230        let request_body = filter;
231        let mut req = self.client.request(method.to_owned(), url.to_owned());
232        req = req.basic_auth(&self.username, Some(&self.password));
233        if let Some(request_body) = request_body {
234            tracing::trace!("Request body:\n{:#?}", request_body);
235            req = req.json(&request_body);
236        }
237        let result = req.send().await;
238        if let Err(ref e) = result {
239            tracing::error!(%url, %method, "Icinga2 send error: {:?}", e);
240        }
241        let result = result?;
242        let status = result.status();
243        if status.is_client_error() {
244            tracing::error!(%url, %method, "Icinga2 status error (client error): {:?}", status);
245        } else if status.is_server_error() {
246            tracing::error!(%url, %method, "Icinga2 status error (server error): {:?}", status);
247        }
248        let byte_chunk_stream = result.bytes_stream().map_err(std::io::Error::other);
249        let stream_reader = StreamReader::new(byte_chunk_stream);
250        let line_reader = LinesStream::new(stream_reader.lines());
251        let event_reader = line_reader.map(|l| match l {
252            Ok(l) => {
253                tracing::trace!("Icinga2 received raw event:\n{}", &l);
254                let jd = &mut serde_json::Deserializer::from_str(&l);
255                match serde_path_to_error::deserialize(jd) {
256                    Ok(event) => {
257                        tracing::trace!("Icinga2 received event:\n{:#?}", &event);
258                        Ok(event)
259                    }
260                    Err(e) => Err(std::io::Error::other(e)),
261                }
262            }
263            Err(e) => Err(e),
264        });
265        Ok(event_reader)
266    }
267}
268
269#[cfg(test)]
270mod test {
271    use super::*;
272    use std::error::Error;
273    use tracing_test::traced_test;
274
275    #[traced_test]
276    #[tokio::test]
277    async fn test_event_stream_async() -> Result<(), Box<dyn Error>> {
278        dotenvy::dotenv()?;
279        let icinga2 = Icinga2Async::from_config_file(std::path::Path::new(&std::env::var(
280            "ICINGA_TEST_INSTANCE_CONFIG",
281        )?))?;
282        let mut stream = icinga2
283            .event_stream(
284                &[
285                    IcingaEventStreamType::CheckResult,
286                    IcingaEventStreamType::StateChange,
287                    IcingaEventStreamType::Notification,
288                    IcingaEventStreamType::AcknowledgementSet,
289                    IcingaEventStreamType::AcknowledgementCleared,
290                    IcingaEventStreamType::CommentAdded,
291                    IcingaEventStreamType::CommentRemove,
292                    IcingaEventStreamType::DowntimeAdded,
293                    IcingaEventStreamType::DowntimeRemoved,
294                    IcingaEventStreamType::DowntimeStarted,
295                    IcingaEventStreamType::DowntimeTriggered,
296                    IcingaEventStreamType::ObjectCreated,
297                    IcingaEventStreamType::ObjectDeleted,
298                    IcingaEventStreamType::ObjectModified,
299                    IcingaEventStreamType::Flapping,
300                ],
301                "test",
302                None,
303            )
304            .await?;
305        for _ in 0..100 {
306            let event = stream.next().await;
307            tracing::trace!("Got event:\n{:#?}", event);
308            if let Some(event) = event {
309                event?;
310            }
311        }
312        Ok(())
313    }
314}
icinga2_api/api/async_client.rs

icinga2_api/api/
async_client.rs
