prosa_fetcher/
proc.rs

1use std::{
2    convert::Infallible,
3    io,
4    time::{Duration, Instant},
5};
6
7use base64::{DecodeError, Engine as _, engine::general_purpose::URL_SAFE};
8use chrono::{Local, NaiveTime};
9use http::Response;
10use http_body_util::combinators::BoxBody;
11use hyper::{
12    Request,
13    body::{Bytes, Incoming},
14    client::conn::{http1, http2},
15};
16use hyper_util::rt::{TokioExecutor, TokioIo};
17use opentelemetry::KeyValue;
18use prosa::{
19    core::{
20        adaptor::Adaptor,
21        error::ProcError,
22        msg::{InternalMsg, Msg, RequestMsg},
23        proc::{Proc, ProcBusParam as _, proc, proc_settings},
24    },
25    io::stream::TargetSetting,
26};
27use serde::{Deserialize, Serialize};
28use thiserror::Error;
29use tokio::{
30    sync::{mpsc, watch},
31    time,
32};
33use tracing::{debug, error, info, warn};
34
35use crate::adaptor::FetcherAdaptor;
36
37#[derive(Debug, Error)]
38/// ProSA service error when the service can't respond correctly to a request
39pub enum FetcherError<M>
40where
41    M: std::marker::Send,
42{
43    /// IO error
44    #[error("IO error during the fetch `{0}`")]
45    Io(#[from] io::Error),
46    /// Hyper error
47    #[error("Hyper error during the fetch `{0:?}` from `{1}`")]
48    Hyper(hyper::Error, String),
49    /// HTTP error
50    #[error("HTTP error on object parsing `{0}`")]
51    Http(#[from] http::Error),
52    /// Queue error
53    #[error("Fetcher communication error `{0}`")]
54    Queue(#[from] watch::error::SendError<FetchAction<M>>),
55    /// HTTP queue error
56    #[error("No HTTP task available to process the message `{0}`")]
57    HttpQueue(Box<mpsc::error::SendError<http::Request<BoxBody<Bytes, Infallible>>>>),
58    /// Base64 decode error
59    #[error("Can't decode Base64 data `{0}`")]
60    B64Decode(#[from] DecodeError),
61    /// Other error
62    #[error("Fetcher other error `{0}`")]
63    Other(String),
64}
65
66impl<M> From<mpsc::error::SendError<http::Request<BoxBody<Bytes, Infallible>>>> for FetcherError<M>
67where
68    M: std::marker::Send,
69{
70    fn from(error: mpsc::error::SendError<http::Request<BoxBody<Bytes, Infallible>>>) -> Self {
71        FetcherError::<M>::HttpQueue(Box::new(error))
72    }
73}
74
75impl<M> ProcError for FetcherError<M>
76where
77    M: 'static + std::fmt::Debug + std::marker::Send,
78{
79    fn recoverable(&self) -> bool {
80        match self {
81            FetcherError::Io(error) => error.recoverable(),
82            FetcherError::Hyper(_error, _addr) => true,
83            FetcherError::Http(_error) => true,
84            FetcherError::Queue(_send_error) => false,
85            FetcherError::HttpQueue(_send_error) => false,
86            FetcherError::B64Decode(_decode_error) => false,
87            FetcherError::Other(_) => false,
88        }
89    }
90}
91
92#[derive(Debug, Deserialize, Serialize, Copy, Clone)]
93pub struct TimeRange {
94    /// Start period hour
95    pub start: NaiveTime,
96    /// End period hour
97    pub end: NaiveTime,
98}
99
100impl TimeRange {
101    // Méthode pour vérifier si une heure donnée est dans la plage
102    pub fn contains(&self, time: &NaiveTime) -> bool {
103        if self.start <= self.end {
104            time >= &self.start && time <= &self.end
105        } else {
106            time >= &self.start || time <= &self.end
107        }
108    }
109}
110
111/// Settings for Fetcher processor
112#[proc_settings]
113#[derive(Debug, Deserialize, Serialize, Clone)]
114pub struct FetcherSettings {
115    /// Target settings to connect to the remote system
116    target: Option<TargetSetting>,
117    /// Remote service to call in order to fetch information from remote system
118    service_name: Option<String>,
119    /// Authentication with authorization header with provided user password
120    #[serde(default = "FetcherSettings::get_default_authorization")]
121    pub authorization: bool,
122    /// Period where the remote system need to be fetch
123    #[serde(default = "FetcherSettings::get_default_period")]
124    period: Duration,
125    /// Timeout duration for every fetch
126    #[serde(default = "FetcherSettings::get_default_timeout")]
127    timeout: Duration,
128    /// Maximum number of retry to fetch a resource
129    #[serde(default = "FetcherSettings::get_default_max_retry")]
130    max_retry: u8,
131    /// Hour time range when the fetcher execute
132    pub(crate) active_time_range: Option<TimeRange>,
133    #[serde(default)]
134    title_case_headers: bool,
135}
136
137impl FetcherSettings {
138    fn get_default_authorization() -> bool {
139        true
140    }
141
142    fn get_default_period() -> Duration {
143        Duration::from_secs(60)
144    }
145
146    fn get_default_timeout() -> Duration {
147        Duration::from_secs(10)
148    }
149
150    fn get_default_max_retry() -> u8 {
151        2
152    }
153
154    /// Create a new Fetcher settings
155    pub fn new(
156        target: TargetSetting,
157        service_name: String,
158        authorization: bool,
159        period: Duration,
160        timeout: Duration,
161    ) -> FetcherSettings {
162        FetcherSettings {
163            target: Some(target),
164            service_name: Some(service_name),
165            authorization,
166            period,
167            timeout,
168            ..Default::default()
169        }
170    }
171
172    /// Get the username for login
173    pub fn username(&self) -> Option<&str> {
174        self.target.as_ref().map(|t| t.url.username())
175    }
176
177    /// Getter of the URL password (decode from Base64Url)
178    pub fn password(&self) -> Result<Option<Vec<u8>>, DecodeError> {
179        if let Some(password) = self.target.as_ref().and_then(|t| t.url.password()) {
180            Ok(Some(URL_SAFE.decode(password.replace("%3D", "="))?))
181        } else {
182            Ok(None)
183        }
184    }
185
186    /// Method to get a challenged password to authenticate.
187    /// mac is the HMac function to use for your challenge.
188    pub fn challenge_password<H, M>(
189        &self,
190        challenge: &[u8],
191    ) -> Result<Option<bytes::Bytes>, FetcherError<M>>
192    where
193        H: hmac::Mac + hmac::digest::KeyInit,
194        M: Send,
195    {
196        if let Some(password) = self.target.as_ref().and_then(|t| t.url.password()) {
197            let binary_password = URL_SAFE.decode(password.replace("%3D", "="))?;
198            let mut mac =
199                <H as hmac::digest::KeyInit>::new_from_slice(&binary_password).map_err(|e| {
200                    FetcherError::Other(format!("Crypto error on password challenge {e}"))
201                })?;
202            mac.update(challenge);
203            return Ok(Some(bytes::Bytes::copy_from_slice(
204                &mac.finalize().into_bytes(),
205            )));
206        }
207
208        Ok(None)
209    }
210
211    /// Method to know if the fetcher is active depending of the time of the day.
212    /// It only return false if an `active_time_range` is set and the current time is not in range
213    pub fn is_active(&self) -> bool {
214        if let Some(active_time_range) = self.active_time_range {
215            active_time_range.contains(&Local::now().time())
216        } else {
217            true
218        }
219    }
220
221    /// Getter of an HTTP1 context
222    pub fn get_http1_ctx(&self) -> http1::Builder {
223        let mut http1_ctx = http1::Builder::new();
224
225        if self.title_case_headers {
226            // Set HTTP1 context for old HTTP server
227            http1_ctx.title_case_headers(true);
228        }
229
230        http1_ctx
231    }
232}
233
234#[proc_settings]
235impl Default for FetcherSettings {
236    fn default() -> Self {
237        FetcherSettings {
238            target: None,
239            service_name: None,
240            authorization: Self::get_default_authorization(),
241            period: Self::get_default_period(),
242            timeout: Self::get_default_timeout(),
243            max_retry: Self::get_default_max_retry(),
244            active_time_range: None,
245            title_case_headers: false,
246        }
247    }
248}
249
250/// Enum that describe what action should be done everytime
251#[derive(Debug)]
252pub enum FetchAction<M>
253where
254    M: std::marker::Send,
255{
256    /// No further action
257    None,
258    /// Send an HTTP request message
259    Http,
260    /// Send a service request message
261    Srv(String, M),
262}
263
264impl<M> FetchAction<M>
265where
266    M: std::marker::Send,
267{
268    /// Method to know if there is still action to execute
269    pub fn have_action(&self) -> bool {
270        !matches!(self, FetchAction::<M>::None)
271    }
272}
273
274#[proc(settings = FetcherSettings)]
275pub struct FetcherProc {}
276
277#[proc]
278impl FetcherProc {
279    fn spawn_http_fetch(
280        settings: &FetcherSettings,
281        target: TargetSetting,
282        mut req_rx: mpsc::Receiver<Request<BoxBody<Bytes, Infallible>>>,
283        resp_tx: mpsc::Sender<Result<Response<Incoming>, FetcherError<M>>>,
284    ) {
285        let timeout = settings.timeout;
286        let have_time_range = settings.active_time_range.is_some();
287        let http1_ctx = settings.get_http1_ctx();
288        let max_retry = settings.max_retry;
289        tokio::spawn(async move {
290            let mut msg_to_send = None;
291            let mut nb_retry = 0;
292            'conn: loop {
293                // Wait for a message before openning the socket
294                if msg_to_send.is_none() {
295                    msg_to_send = req_rx.recv().await;
296                    if msg_to_send.is_none() {
297                        if let Err(e) = resp_tx.try_send(Err(FetcherError::Other(
298                            "Internal HTTP queue is closed".to_string(),
299                        ))) {
300                            warn!(
301                                addr = target.to_string(),
302                                "Error during message openning: {e}"
303                            );
304                        }
305                        return;
306                    }
307                }
308
309                match target.connect().await {
310                    Ok(stream) => {
311                        let is_http2 = stream.selected_alpn_check(|alpn| alpn == b"h2");
312                        let stream = TokioIo::new(stream);
313
314                        if is_http2 {
315                            match time::timeout(
316                                timeout,
317                                http2::handshake(TokioExecutor::new(), stream),
318                            )
319                            .await
320                            {
321                                Ok(Ok((mut sender, mut connection))) => loop {
322                                    if let Some(msg) = msg_to_send.take() {
323                                        tokio::select! {
324                                            // Closed the socket
325                                            Err(_) = &mut connection => {
326                                                debug!(addr = target.to_string(), "Remote close the HTTP2 socket");
327                                                continue 'conn;
328                                            }
329                                            // Send an HTTP request
330                                            resp = sender.try_send_request(msg) => {
331                                                match resp {
332                                                    Ok(r) => {
333                                                        if let Err(e) = resp_tx.try_send(Ok(r)) {
334                                                            warn!(addr = target.to_string(), "Error during HTTP2 response return: {e}");
335                                                        } else {
336                                                            nb_retry = 0;
337                                                        }
338                                                    }
339                                                    Err(mut e) => {
340                                                        msg_to_send = e.take_message();
341                                                        let hyper_err = e.into_error();
342                                                        if nb_retry < max_retry {
343                                                            // try to send again the message after reconnection
344                                                            nb_retry += 1;
345                                                            continue 'conn;
346                                                        } else {
347                                                            error!(addr = target.to_string(), "Failed to fetch HTTP2 `{msg_to_send:?}`, because of `{hyper_err}`, after {nb_retry}/{max_retry} retries");
348                                                            if let Err(e) = resp_tx.try_send(Err(FetcherError::Hyper(hyper_err, target.to_string()))) {
349                                                                warn!(addr = target.to_string(), "Error during HTTP2 error response return: {e}");
350                                                            }
351                                                            continue 'conn;
352                                                        }
353                                                    }
354                                                }
355                                            }
356                                            // Receive a message to send from the queue (unload also the queue if too many messages arrive)
357                                            Some(mut msg) = req_rx.recv() => {
358                                                *msg.version_mut() = http::Version::HTTP_2;
359                                                msg_to_send = Some(msg);
360                                            }
361                                        }
362                                    } else {
363                                        tokio::select! {
364                                            // Closed the socket
365                                            Err(_) = &mut connection => {
366                                                continue 'conn;
367                                            }
368                                            // Receive a message to send from the queue
369                                            Some(mut msg) = req_rx.recv() => {
370                                                *msg.version_mut() = http::Version::HTTP_2;
371                                                msg_to_send = Some(msg);
372                                            }
373                                        }
374                                    }
375                                },
376                                Ok(Err(handshake_error)) => warn!(
377                                    addr = target.to_string(),
378                                    "HTTP2 handshake error: {handshake_error}"
379                                ),
380                                Err(_) => warn!(
381                                    addr = target.to_string(),
382                                    "HTTP2 handshake timeout after {}ms", target.connect_timeout
383                                ),
384                            }
385                        } else {
386                            match time::timeout(timeout, http1_ctx.handshake(stream)).await {
387                                Ok(Ok((mut sender, mut connection))) => loop {
388                                    if let Some(msg) = msg_to_send.take() {
389                                        tokio::select! {
390                                            // Closed the socket
391                                            Err(_) = &mut connection => {
392                                                debug!(addr = target.to_string(), "Remote close the HTTP1 socket");
393                                                continue 'conn;
394                                            }
395                                            // Send an HTTP request
396                                            resp = sender.try_send_request(msg) => {
397                                                match resp {
398                                                    Ok(r) => {
399                                                        if let Err(e) = resp_tx.try_send(Ok(r)) {
400                                                            warn!(addr = target.to_string(), "Error during HTTP1 response return: {e}");
401                                                        } else {
402                                                            nb_retry = 0;
403                                                        }
404                                                    }
405                                                    Err(mut e) => {
406                                                        msg_to_send = e.take_message();
407                                                        let hyper_err = e.into_error();
408                                                        if nb_retry < max_retry {
409                                                            // try to send again the message after reconnection
410                                                            nb_retry += 1;
411                                                            continue 'conn;
412                                                        } else {
413                                                            error!(addr = target.to_string(), "Failed to fetch HTTP1 `{msg_to_send:?}`, because of `{hyper_err}`, after {nb_retry}/{max_retry} retries");
414                                                            if let Err(e) = resp_tx.try_send(Err(FetcherError::Hyper(hyper_err, target.to_string()))) {
415                                                                warn!(addr = target.to_string(), "Error during HTTP1 error response return: {e}");
416                                                            }
417                                                            continue 'conn;
418                                                        }
419                                                    }
420                                                }
421                                            }
422                                            // Receive a message to send from the queue (unload also the queue if too many messages arrive)
423                                            Some(mut msg) = req_rx.recv() => {
424                                                *msg.version_mut() = http::Version::HTTP_11;
425                                                msg_to_send = Some(msg);
426                                            }
427                                        }
428                                    } else {
429                                        tokio::select! {
430                                            // Closed the socket
431                                            Err(_) = &mut connection => {
432                                                continue 'conn;
433                                            }
434                                            // Receive a message to send from the queue
435                                            Some(mut msg) = req_rx.recv() => {
436                                                *msg.version_mut() = http::Version::HTTP_11;
437                                                msg_to_send = Some(msg);
438                                            }
439                                        }
440                                    }
441                                },
442                                Ok(Err(handshake_error)) => warn!(
443                                    addr = target.to_string(),
444                                    "HTTP1 handshake error: {handshake_error}"
445                                ),
446                                Err(_) => warn!(
447                                    addr = target.to_string(),
448                                    "HTTP1 handshake timeout after {}ms", target.connect_timeout
449                                ),
450                            }
451                        }
452                    }
453                    Err(e) => {
454                        // If the distant have a time range, maybe the distant is not up, so just throw an info log
455                        if have_time_range {
456                            info!(
457                                addr = target.to_string(),
458                                "Can't connect to remote: {:?}", e
459                            );
460                        } else {
461                            warn!(
462                                addr = target.to_string(),
463                                "Can't connect to remote: {:?}", e
464                            );
465                        }
466                    }
467                }
468            }
469        });
470    }
471}
472
473macro_rules! process_action {
474    ($self:ident, $action:ident, $adaptor:ident, $http_req_tx:ident) => {
475        match $action {
476            FetchAction::Http => {
477                let request_builder = if let Some(target) = &$self.settings.target {
478                    let mut authority_url = target.url.clone();
479                    let _ = authority_url.set_username("");
480                    let _ = authority_url.set_password(None);
481                    let mut request_builder = Request::builder().header(hyper::header::HOST, authority_url.authority());
482                    if $self.settings.authorization
483                        && let Some(authorization) = target.get_authentication()
484                    {
485                        request_builder = request_builder.header(hyper::header::AUTHORIZATION, authorization);
486                    }
487
488                    // TOOD add USER agent
489                    request_builder
490                } else {
491                    Request::builder()
492                };
493                let request = $adaptor.create_http_request(request_builder)?;
494                debug!(addr = $self.settings.target.as_ref().map(|t| t.to_string()), "Send: {:?}", request);
495                $http_req_tx.send(request).await.map_err(FetcherError::<M>::from)?;
496            }
497            FetchAction::Srv(service_name, msg) => {
498                debug!("Call Service({}) Fetch action", service_name);
499                if let Some(service) = $self.service.get_proc_service(&service_name) {
500                    let req_msg = RequestMsg::new(service_name.clone(), msg, $self.proc.get_service_queue());
501                    debug!(name: "fetcher_proc", target: "prosa_proc_fetcher::proc", parent: req_msg.get_span(), proc_name = $self.proc.name(), service = service_name, request = format!("{:?}", req_msg.get_data()));
502                    service.proc_queue.send(InternalMsg::Request(req_msg)).await?;
503                }
504            },
505            FetchAction::None => { /* No further action to do */ }
506        }
507    };
508}
509
510// Fetcher processor to fetch information from remote systems
511#[proc]
512impl<A> Proc<A> for FetcherProc
513where
514    A: Adaptor + FetcherAdaptor<M> + std::marker::Send,
515{
516    async fn internal_run(&mut self) -> Result<(), Box<dyn ProcError + Send + Sync>> {
517        // Initiate an adaptor for the fetcher processor
518        let mut adaptor = A::new(self)?;
519
520        // TODO wait for external service to become available if needed.
521
522        // Declare the processor
523        self.proc.add_proc().await?;
524
525        // Interval between each fetch
526        let mut fetch_interval = time::interval(self.settings.period);
527
528        // Spawn HTTP task if needed
529        let (http_req_tx, http_req_rx) = mpsc::channel(1);
530        let (http_resp_tx, mut http_resp_rx) = mpsc::channel(1);
531        if let Some(target) = &self.settings.target {
532            Self::spawn_http_fetch(&self.settings, target.clone(), http_req_rx, http_resp_tx);
533        }
534
535        let meter = self.proc.meter("fetcher");
536        let action_histogram = meter
537            .u64_histogram("prosa_fetcher_duration")
538            .with_description("Fetcher duration histogram")
539            .build();
540
541        let mut is_active = true;
542        let mut sent_time = Instant::now();
543        loop {
544            tokio::select! {
545                _interval = fetch_interval.tick() => if self.settings.is_active() {
546                    is_active = true;
547                    let action = adaptor.fetch()?;
548                    process_action!(self, action, adaptor, http_req_tx);
549                    sent_time = Instant::now();
550                } else if is_active {
551                    is_active = false;
552                    adaptor.end_active_period();
553                },
554                Some(http_resp) = http_resp_rx.recv() => {
555                    let mut histogram_attributes = vec![
556                        KeyValue::new("type", "http"),
557                        KeyValue::new("code", http_resp.as_ref().map(|r| r.status().as_u16()).unwrap_or(502) as i64),
558                    ];
559                    if let Some(target) = &self.settings.target {
560                        histogram_attributes.push(KeyValue::new("target", target.to_string()));
561                    }
562                    action_histogram.record(
563                        sent_time.elapsed().as_millis() as u64,
564                        &histogram_attributes,
565                    );
566
567                    let action = adaptor.process_http_response(http_resp).await?;
568                    process_action!(self, action, adaptor, http_req_tx);
569                    sent_time = Instant::now();
570                }
571                Some(msg) = self.internal_rx_queue.recv() => {
572                    match msg {
573                        InternalMsg::Request(msg) => panic!(
574                            "The fetcher processor {} should not receive a request {:?}",
575                            self.get_proc_id(),
576                            msg
577                        ),
578                        InternalMsg::Response(msg) => {
579                            action_histogram.record(
580                                sent_time.elapsed().as_millis() as u64,
581                                &[
582                                    KeyValue::new("type", "service"),
583                                    KeyValue::new("service", msg.get_service().clone()),
584                                    KeyValue::new("code", 0),
585                                ],
586                            );
587                            let action = adaptor.process_service_response(msg)?;
588                            process_action!(self, action, adaptor, http_req_tx);
589                            sent_time = Instant::now();
590                        },
591                        InternalMsg::Error(err) => {
592                            action_histogram.record(
593                                sent_time.elapsed().as_millis() as u64,
594                                &[
595                                    KeyValue::new("type", "service"),
596                                    KeyValue::new("service", err.get_service().clone()),
597                                    KeyValue::new("code", err.get_err().get_code() as i64),
598                                ],
599                            );
600                            let action = adaptor.process_service_error(err)?;
601                            process_action!(self, action, adaptor, http_req_tx);
602                            sent_time = Instant::now();
603                        },
604                        InternalMsg::Command(_) => todo!(),
605                        InternalMsg::Config => todo!(),
606                        InternalMsg::Service(table) => self.service = table,
607                        InternalMsg::Shutdown => {
608                            // Stop directly the processor
609                            adaptor.terminate();
610                            self.proc.remove_proc(None).await?;
611                            return Ok(());
612                        }
613                    }
614                }
615            }
616        }
617    }
618}
prosa_fetcher/proc.rs

prosa_fetcher/
proc.rs
