vector/
http.rs

1#![allow(missing_docs)]
2use std::{
3    collections::HashMap,
4    fmt,
5    net::SocketAddr,
6    task::{Context, Poll},
7    time::Duration,
8};
9
10use futures::future::BoxFuture;
11use headers::{Authorization, HeaderMapExt};
12use http::{
13    HeaderMap, Request, Response, Uri, Version, header::HeaderValue, request::Builder,
14    uri::InvalidUri,
15};
16use hyper::{
17    body::{Body, HttpBody},
18    client,
19    client::{Client, HttpConnector, connect::Connect},
20};
21use hyper_openssl::HttpsConnector;
22use hyper_proxy::ProxyConnector;
23use rand::Rng;
24use serde_with::serde_as;
25use snafu::{ResultExt, Snafu};
26use tokio::time::Instant;
27use tower::{Layer, Service};
28use tower_http::{
29    classify::{ServerErrorsAsFailures, SharedClassifier},
30    trace::TraceLayer,
31};
32use tracing::{Instrument, Span};
33use vector_lib::{configurable::configurable_component, sensitive_string::SensitiveString};
34
35#[cfg(feature = "aws-core")]
36use crate::aws::AwsAuthentication;
37use crate::{
38    config::ProxyConfig,
39    internal_events::{HttpServerRequestReceived, HttpServerResponseSent, http_client},
40    tls::{MaybeTlsSettings, TlsError, tls_connector_builder},
41};
42
43pub mod status {
44    pub const FORBIDDEN: u16 = 403;
45    pub const NOT_FOUND: u16 = 404;
46    pub const TOO_MANY_REQUESTS: u16 = 429;
47}
48
49#[derive(Debug, Snafu)]
50#[snafu(visibility(pub(crate)))]
51pub enum HttpError {
52    #[snafu(display("Failed to build TLS connector: {}", source))]
53    BuildTlsConnector { source: TlsError },
54    #[snafu(display("Failed to build HTTPS connector: {}", source))]
55    MakeHttpsConnector { source: openssl::error::ErrorStack },
56    #[snafu(display("Failed to build Proxy connector: {}", source))]
57    MakeProxyConnector { source: InvalidUri },
58    #[snafu(display("Failed to make HTTP(S) request: {}", source))]
59    CallRequest { source: hyper::Error },
60    #[snafu(display("Failed to build HTTP request: {}", source))]
61    BuildRequest { source: http::Error },
62}
63
64impl HttpError {
65    pub const fn is_retriable(&self) -> bool {
66        match self {
67            HttpError::BuildRequest { .. } | HttpError::MakeProxyConnector { .. } => false,
68            HttpError::CallRequest { .. }
69            | HttpError::BuildTlsConnector { .. }
70            | HttpError::MakeHttpsConnector { .. } => true,
71        }
72    }
73}
74
75pub type HttpClientFuture = <HttpClient as Service<http::Request<Body>>>::Future;
76pub type HttpProxyConnector = ProxyConnector<HttpsConnector<HttpConnector>>;
77
78/// An HTTP client generic over the underlying connector `C`.
79///
80/// `C` defaults to [`HttpProxyConnector`], the proxy-aware TLS connector built from Vector's
81/// [`MaybeTlsSettings`]/[`ProxyConfig`]. Callers that need a custom connector can supply
82/// their own via [`HttpClient::new_with_connector`].
83pub struct HttpClient<B = Body, C = HttpProxyConnector> {
84    client: Client<C, B>,
85    user_agent: HeaderValue,
86    // Proxy-header injection only applies when the client owns a `ProxyConnector`. A custom
87    // connector supplied via `new_with_connector` sets this to `None`.
88    proxy_connector: Option<HttpProxyConnector>,
89}
90
91impl<B> HttpClient<B, HttpProxyConnector>
92where
93    B: fmt::Debug + HttpBody + Send + 'static,
94    B::Data: Send,
95    B::Error: Into<crate::Error>,
96{
97    pub fn new(
98        tls_settings: impl Into<MaybeTlsSettings>,
99        proxy_config: &ProxyConfig,
100    ) -> Result<Self, HttpError> {
101        HttpClient::new_with_custom_client(tls_settings, proxy_config, &mut Client::builder())
102    }
103
104    pub fn new_with_custom_client(
105        tls_settings: impl Into<MaybeTlsSettings>,
106        proxy_config: &ProxyConfig,
107        client_builder: &mut client::Builder,
108    ) -> Result<Self, HttpError> {
109        let proxy_connector = build_proxy_connector(tls_settings.into(), proxy_config)?;
110        let client = client_builder.build(proxy_connector.clone());
111
112        Ok(HttpClient {
113            client,
114            user_agent: default_user_agent(),
115            proxy_connector: Some(proxy_connector),
116        })
117    }
118}
119
120impl<B, C> HttpClient<B, C>
121where
122    B: fmt::Debug + HttpBody + Send + 'static,
123    B::Data: Send,
124    B::Error: Into<crate::Error>,
125    C: Connect + Clone + Send + Sync + 'static,
126{
127    /// Build an `HttpClient` from a pre-constructed connector.
128    ///
129    /// Proxy-header injection is disabled for custom connectors (`proxy_connector` is always `None`).
130    /// It only matters for the case of a plaintext-`http` request sent through an HTTP proxy, where
131    /// `Proxy-Authorization` must ride on the request itself. HTTPS-through-proxy is unaffected:
132    /// proxy auth happens during the CONNECT handshake inside the connector.
133    pub fn new_with_connector(connector: C, client_builder: &mut client::Builder) -> Self {
134        let client = client_builder.build(connector);
135
136        HttpClient {
137            client,
138            user_agent: default_user_agent(),
139            proxy_connector: None,
140        }
141    }
142
143    pub fn send(
144        &self,
145        mut request: Request<B>,
146    ) -> BoxFuture<'static, Result<http::Response<Body>, HttpError>> {
147        let span = tracing::info_span!("http");
148        let _enter = span.enter();
149
150        default_request_headers(&mut request, &self.user_agent);
151        self.maybe_add_proxy_headers(&mut request);
152
153        emit!(http_client::AboutToSendHttpRequest { request: &request });
154
155        let response = self.client.request(request);
156
157        let fut = async move {
158            // Capture the time right before we issue the request.
159            // Request doesn't start the processing until we start polling it.
160            let before = std::time::Instant::now();
161
162            // Send request and wait for the result.
163            let response_result = response.await;
164
165            // Compute the roundtrip time it took to send the request and get
166            // the response or error.
167            let roundtrip = before.elapsed();
168
169            // Handle the errors and extract the response.
170            let response = response_result
171                .inspect_err(|error| {
172                    // Emit the error into the internal events system.
173                    emit!(http_client::GotHttpWarning { error, roundtrip });
174                })
175                .context(CallRequestSnafu)?;
176
177            // Emit the response into the internal events system.
178            emit!(http_client::GotHttpResponse {
179                response: &response,
180                roundtrip
181            });
182            Ok(response)
183        }
184        .instrument(span.clone().or_current());
185
186        Box::pin(fut)
187    }
188
189    fn maybe_add_proxy_headers(&self, request: &mut Request<B>) {
190        let Some(proxy_connector) = &self.proxy_connector else {
191            return;
192        };
193        if let Some(proxy_headers) = proxy_connector.http_headers(request.uri()) {
194            for (k, v) in proxy_headers {
195                let request_headers = request.headers_mut();
196                if !request_headers.contains_key(k) {
197                    request_headers.insert(k, v.into());
198                }
199            }
200        }
201    }
202}
203
204fn default_user_agent() -> HeaderValue {
205    let app_name = crate::get_app_name();
206    let version = crate::get_version();
207    HeaderValue::from_str(&format!("{app_name}/{version}"))
208        .expect("Invalid header value for user-agent!")
209}
210
211pub fn build_proxy_connector(
212    tls_settings: MaybeTlsSettings,
213    proxy_config: &ProxyConfig,
214) -> Result<ProxyConnector<HttpsConnector<HttpConnector>>, HttpError> {
215    // Create dedicated TLS connector for the proxied connection with user TLS settings.
216    let tls = tls_connector_builder(&tls_settings)
217        .context(BuildTlsConnectorSnafu)?
218        .build();
219    let https = build_tls_connector(tls_settings)?;
220    let mut proxy = ProxyConnector::new(https).unwrap();
221    // Make proxy connector aware of user TLS settings by setting the TLS connector:
222    // https://github.com/vectordotdev/vector/issues/13683
223    proxy.set_tls(Some(tls));
224    proxy_config
225        .configure(&mut proxy)
226        .context(MakeProxyConnectorSnafu)?;
227    Ok(proxy)
228}
229
230pub fn build_tls_connector(
231    tls_settings: MaybeTlsSettings,
232) -> Result<HttpsConnector<HttpConnector>, HttpError> {
233    let mut http = HttpConnector::new();
234    http.enforce_http(false);
235
236    let tls = tls_connector_builder(&tls_settings).context(BuildTlsConnectorSnafu)?;
237    let mut https = HttpsConnector::with_connector(http, tls).context(MakeHttpsConnectorSnafu)?;
238
239    let settings = tls_settings.tls().cloned();
240    https.set_callback(move |c, _uri| {
241        if let Some(settings) = &settings {
242            settings.apply_connect_configuration(c)
243        } else {
244            Ok(())
245        }
246    });
247    Ok(https)
248}
249
250fn default_request_headers<B>(request: &mut Request<B>, user_agent: &HeaderValue) {
251    if !request.headers().contains_key("User-Agent") {
252        request
253            .headers_mut()
254            .insert("User-Agent", user_agent.clone());
255    }
256
257    if !request.headers().contains_key("Accept-Encoding") {
258        // hardcoding until we support compressed responses:
259        // https://github.com/vectordotdev/vector/issues/5440
260        request
261            .headers_mut()
262            .insert("Accept-Encoding", HeaderValue::from_static("identity"));
263    }
264}
265
266impl<B, C> Service<Request<B>> for HttpClient<B, C>
267where
268    B: fmt::Debug + HttpBody + Send + 'static,
269    B::Data: Send,
270    B::Error: Into<crate::Error> + Send,
271    C: Connect + Clone + Send + Sync + 'static,
272{
273    type Response = http::Response<Body>;
274    type Error = HttpError;
275    type Future = BoxFuture<'static, Result<Self::Response, Self::Error>>;
276
277    fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
278        Poll::Ready(Ok(()))
279    }
280
281    fn call(&mut self, request: Request<B>) -> Self::Future {
282        self.send(request)
283    }
284}
285
286impl<B, C: Clone> Clone for HttpClient<B, C> {
287    fn clone(&self) -> Self {
288        Self {
289            client: self.client.clone(),
290            user_agent: self.user_agent.clone(),
291            proxy_connector: self.proxy_connector.clone(),
292        }
293    }
294}
295
296impl<B, C> fmt::Debug for HttpClient<B, C> {
297    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
298        f.debug_struct("HttpClient")
299            .field("client", &self.client)
300            .field("user_agent", &self.user_agent)
301            .finish()
302    }
303}
304
305/// Configuration of the authentication strategy for HTTP requests.
306///
307/// HTTP authentication should be used with HTTPS only, as the authentication credentials are passed as an
308/// HTTP header without any additional encryption beyond what is provided by the transport itself.
309#[configurable_component]
310#[derive(Clone, Debug, Eq, PartialEq)]
311#[serde(deny_unknown_fields, rename_all = "snake_case", tag = "strategy")]
312#[configurable(metadata(docs::enum_tag_description = "The authentication strategy to use."))]
313pub enum Auth {
314    /// Basic authentication.
315    ///
316    /// The username and password are concatenated and encoded using [base64][base64].
317    ///
318    /// [base64]: https://en.wikipedia.org/wiki/Base64
319    Basic {
320        /// The basic authentication username.
321        #[configurable(metadata(docs::examples = "${USERNAME}"))]
322        #[configurable(metadata(docs::examples = "username"))]
323        user: String,
324
325        /// The basic authentication password.
326        #[configurable(metadata(docs::examples = "${PASSWORD}"))]
327        #[configurable(metadata(docs::examples = "password"))]
328        password: SensitiveString,
329    },
330
331    /// Bearer authentication.
332    ///
333    /// The bearer token value (OAuth2, JWT, etc.) is passed as-is.
334    Bearer {
335        /// The bearer authentication token.
336        token: SensitiveString,
337    },
338
339    #[cfg(feature = "aws-core")]
340    /// AWS authentication.
341    Aws {
342        /// The AWS authentication configuration.
343        auth: AwsAuthentication,
344
345        /// The AWS service name to use for signing.
346        service: String,
347    },
348
349    /// Custom Authorization Header Value, will be inserted into the headers as `Authorization: < value >`
350    Custom {
351        /// Custom string value of the Authorization header
352        #[configurable(metadata(docs::examples = "${AUTH_HEADER_VALUE}"))]
353        #[configurable(metadata(docs::examples = "CUSTOM_PREFIX ${TOKEN}"))]
354        value: String,
355    },
356}
357
358pub trait MaybeAuth: Sized {
359    fn choose_one(&self, other: &Self) -> crate::Result<Self>;
360}
361
362impl MaybeAuth for Option<Auth> {
363    fn choose_one(&self, other: &Self) -> crate::Result<Self> {
364        if self.is_some() && other.is_some() {
365            Err("Two authorization credentials was provided.".into())
366        } else {
367            Ok(self.clone().or_else(|| other.clone()))
368        }
369    }
370}
371
372impl Auth {
373    pub fn apply<B>(&self, req: &mut Request<B>) {
374        self.apply_headers_map(req.headers_mut())
375    }
376
377    pub fn apply_builder(&self, mut builder: Builder) -> Builder {
378        if let Some(map) = builder.headers_mut() {
379            self.apply_headers_map(map)
380        }
381        builder
382    }
383
384    pub fn apply_headers_map(&self, map: &mut HeaderMap) {
385        match &self {
386            Auth::Basic { user, password } => {
387                let auth = Authorization::basic(user.as_str(), password.inner());
388                map.typed_insert(auth);
389            }
390            Auth::Bearer { token } => match Authorization::bearer(token.inner()) {
391                Ok(auth) => map.typed_insert(auth),
392                Err(error) => error!(message = "Invalid bearer token.", token = %token, %error),
393            },
394            Auth::Custom { value } => {
395                // The value contains just the value for the Authorization header
396                // Expected format: "SSWS token123" or "Bearer token123", etc.
397                match HeaderValue::from_str(value) {
398                    Ok(header_val) => {
399                        map.insert(http::header::AUTHORIZATION, header_val);
400                    }
401                    Err(error) => {
402                        error!(message = "Invalid custom auth header value.", value = %value, %error)
403                    }
404                }
405            }
406            #[cfg(feature = "aws-core")]
407            _ => {}
408        }
409    }
410}
411
412pub fn get_http_scheme_from_uri(uri: &Uri) -> &'static str {
413    // If there's no scheme, we just use "http" since it provides the most semantic relevance without inadvertently
414    // implying things it can't know i.e. returning "https" when we're not actually sure HTTPS was used.
415    uri.scheme_str().map_or("http", |scheme| match scheme {
416        "http" => "http",
417        "https" => "https",
418        // `http::Uri` ensures that we always get "http" or "https" if the URI is created with a well-formed scheme, but
419        // it also supports arbitrary schemes, which is where we bomb out down here, since we can't generate a static
420        // string for an arbitrary input string... and anything other than "http" and "https" makes no sense for an HTTP
421        // client anyways.
422        s => panic!("invalid URI scheme for HTTP client: {s}"),
423    })
424}
425
426/// Builds a [TraceLayer] configured for a HTTP server.
427///
428/// This layer emits HTTP specific telemetry for requests received, responses sent, and handler duration.
429pub fn build_http_trace_layer<T, U>(
430    span: Span,
431) -> TraceLayer<
432    SharedClassifier<ServerErrorsAsFailures>,
433    impl Fn(&Request<T>) -> Span + Clone,
434    impl Fn(&Request<T>, &Span) + Clone,
435    impl Fn(&Response<U>, Duration, &Span) + Clone,
436    (),
437    (),
438    (),
439> {
440    TraceLayer::new_for_http()
441        .make_span_with(move |request: &Request<T>| {
442            // This is an error span so that the labels are always present for metrics.
443            error_span!(
444               parent: &span,
445               "http-request",
446               method = %request.method(),
447               path = %request.uri().path(),
448            )
449        })
450        .on_request(Box::new(|_request: &Request<T>, _span: &Span| {
451            emit!(HttpServerRequestReceived);
452        }))
453        .on_response(|response: &Response<U>, latency: Duration, _span: &Span| {
454            emit!(HttpServerResponseSent { response, latency });
455        })
456        .on_failure(())
457        .on_body_chunk(())
458        .on_eos(())
459}
460
461/// Configuration of HTTP server keepalive parameters.
462#[serde_as]
463#[configurable_component]
464#[derive(Clone, Debug, PartialEq)]
465#[serde(deny_unknown_fields)]
466pub struct KeepaliveConfig {
467    /// The maximum amount of time a connection may exist before it is closed by sending
468    /// a `Connection: close` header on the HTTP response. Set this to a large value like
469    /// `100000000` to "disable" this feature
470    ///
471    ///
472    /// Only applies to HTTP/0.9, HTTP/1.0, and HTTP/1.1 requests.
473    ///
474    /// A random jitter configured by `max_connection_age_jitter_factor` is added
475    /// to the specified duration to spread out connection storms.
476    #[serde(default = "default_max_connection_age")]
477    #[configurable(metadata(docs::examples = 600))]
478    #[configurable(metadata(docs::type_unit = "seconds"))]
479    #[configurable(metadata(docs::human_name = "Maximum Connection Age"))]
480    pub max_connection_age_secs: Option<u64>,
481
482    /// The factor by which to jitter the `max_connection_age_secs` value.
483    ///
484    /// A value of 0.1 means that the actual duration will be between 90% and 110% of the
485    /// specified maximum duration.
486    #[serde(default = "default_max_connection_age_jitter_factor")]
487    #[configurable(validation(range(min = 0.0, max = 1.0)))]
488    pub max_connection_age_jitter_factor: f64,
489}
490
491const fn default_max_connection_age() -> Option<u64> {
492    Some(300) // 5 minutes
493}
494
495const fn default_max_connection_age_jitter_factor() -> f64 {
496    0.1
497}
498
499impl Default for KeepaliveConfig {
500    fn default() -> Self {
501        Self {
502            max_connection_age_secs: default_max_connection_age(),
503            max_connection_age_jitter_factor: default_max_connection_age_jitter_factor(),
504        }
505    }
506}
507
508/// A layer that limits the maximum duration of a client connection. It does so by adding a
509/// `Connection: close` header to the response if `max_connection_duration` time has elapsed
510/// since `start_reference`.
511///
512/// **Notes:**
513/// - This is intended to be used in a Hyper server (or similar) that will automatically close
514///   the connection after a response with a `Connection: close` header is sent.
515/// - This layer assumes that it is instantiated once per connection, which is true within the
516///   Hyper framework.
517pub struct MaxConnectionAgeLayer {
518    start_reference: Instant,
519    max_connection_age: Duration,
520    peer_addr: SocketAddr,
521}
522
523impl MaxConnectionAgeLayer {
524    pub fn new(max_connection_age: Duration, jitter_factor: f64, peer_addr: SocketAddr) -> Self {
525        Self {
526            start_reference: Instant::now(),
527            max_connection_age: Self::jittered_duration(max_connection_age, jitter_factor),
528            peer_addr,
529        }
530    }
531
532    fn jittered_duration(duration: Duration, jitter_factor: f64) -> Duration {
533        // Ensure the jitter_factor is between 0.0 and 1.0
534        let jitter_factor = jitter_factor.clamp(0.0, 1.0);
535        // Generate a random jitter factor between `1 - jitter_factor`` and `1 + jitter_factor`.
536        let mut rng = rand::rng();
537        let random_jitter_factor = rng.random_range(-jitter_factor..=jitter_factor) + 1.;
538        duration.mul_f64(random_jitter_factor)
539    }
540}
541
542impl<S> Layer<S> for MaxConnectionAgeLayer
543where
544    S: Service<Request<Body>, Response = Response<Body>> + Clone + Send + 'static,
545    S::Future: Send + 'static,
546{
547    type Service = MaxConnectionAgeService<S>;
548
549    fn layer(&self, service: S) -> Self::Service {
550        MaxConnectionAgeService {
551            service,
552            start_reference: self.start_reference,
553            max_connection_age: self.max_connection_age,
554            peer_addr: self.peer_addr,
555        }
556    }
557}
558
559/// A service that limits the maximum age of a client connection. It does so by adding a
560/// `Connection: close` header to the response if `max_connection_age` time has elapsed
561/// since `start_reference`.
562///
563/// **Notes:**
564/// - This is intended to be used in a Hyper server (or similar) that will automatically close
565///   the connection after a response with a `Connection: close` header is sent.
566/// - This service assumes that it is instantiated once per connection, which is true within the
567///   Hyper framework.
568#[derive(Clone)]
569pub struct MaxConnectionAgeService<S> {
570    service: S,
571    start_reference: Instant,
572    max_connection_age: Duration,
573    peer_addr: SocketAddr,
574}
575
576impl<S, E> Service<Request<Body>> for MaxConnectionAgeService<S>
577where
578    S: Service<Request<Body>, Response = Response<Body>, Error = E> + Clone + Send + 'static,
579    S::Future: Send + 'static,
580{
581    type Response = S::Response;
582    type Error = E;
583    type Future = BoxFuture<'static, Result<Self::Response, E>>;
584
585    fn poll_ready(
586        &mut self,
587        cx: &mut std::task::Context<'_>,
588    ) -> std::task::Poll<Result<(), Self::Error>> {
589        self.service.poll_ready(cx)
590    }
591
592    fn call(&mut self, req: Request<Body>) -> Self::Future {
593        let start_reference = self.start_reference;
594        let max_connection_age = self.max_connection_age;
595        let peer_addr = self.peer_addr;
596        let version = req.version();
597        let future = self.service.call(req);
598        Box::pin(async move {
599            let mut response = future.await?;
600            match version {
601                Version::HTTP_09 | Version::HTTP_10 | Version::HTTP_11
602                    if start_reference.elapsed() >= max_connection_age =>
603                {
604                    debug!(
605                        message = "Closing connection due to max connection age.",
606                        ?max_connection_age,
607                        connection_age = ?start_reference.elapsed(),
608                        ?peer_addr,
609                    );
610                    // Tell the client to close this connection.
611                    // Hyper will automatically close the connection after the response is sent.
612                    response.headers_mut().insert(
613                        hyper::header::CONNECTION,
614                        hyper::header::HeaderValue::from_static("close"),
615                    );
616                }
617                Version::HTTP_09 | Version::HTTP_10 | Version::HTTP_11 => (),
618                // TODO need to send GOAWAY frame
619                Version::HTTP_2 => (),
620                // TODO need to send GOAWAY frame
621                Version::HTTP_3 => (),
622                _ => (),
623            }
624            Ok(response)
625        })
626    }
627}
628
629/// The type of a query parameter's value, determines if it's treated as a plain string or a VRL expression.
630#[configurable_component]
631#[derive(Clone, Debug, Default, Eq, PartialEq)]
632#[serde(rename_all = "snake_case")]
633pub enum ParamType {
634    /// The parameter value is a plain string.
635    #[default]
636    String,
637    /// The parameter value is a VRL expression that is evaluated before each request.
638    Vrl,
639}
640
641impl ParamType {
642    fn is_default(&self) -> bool {
643        *self == Self::default()
644    }
645}
646
647/// Represents a query parameter value, which can be a simple string or a typed object
648/// indicating whether the value is a string or a VRL expression.
649#[configurable_component]
650#[derive(Clone, Debug, Eq, PartialEq)]
651#[serde(untagged)]
652pub enum ParameterValue {
653    /// A simple string value. For backwards compatibility.
654    String(String),
655    /// A value with an explicit type.
656    Typed {
657        /// The raw value of the parameter.
658        value: String,
659        /// The parameter type, indicating how the `value` should be treated.
660        #[serde(
661            default,
662            skip_serializing_if = "ParamType::is_default",
663            rename = "type"
664        )]
665        r#type: ParamType,
666    },
667}
668
669impl ParameterValue {
670    /// Returns true if the parameter is a VRL expression.
671    pub const fn is_vrl(&self) -> bool {
672        match self {
673            ParameterValue::String(_) => false,
674            ParameterValue::Typed { r#type, .. } => matches!(r#type, ParamType::Vrl),
675        }
676    }
677
678    /// Returns the raw string value of the parameter.
679    #[allow(clippy::missing_const_for_fn)]
680    pub fn value(&self) -> &str {
681        match self {
682            ParameterValue::String(s) => s,
683            ParameterValue::Typed { value, .. } => value,
684        }
685    }
686
687    /// Consumes the `ParameterValue` and returns the owned raw string value.
688    pub fn into_value(self) -> String {
689        match self {
690            ParameterValue::String(s) => s,
691            ParameterValue::Typed { value, .. } => value,
692        }
693    }
694}
695
696/// Configuration of the query parameter value for HTTP requests.
697#[configurable_component]
698#[derive(Clone, Debug, Eq, PartialEq)]
699#[serde(untagged)]
700#[configurable(metadata(docs::enum_tag_description = "Query parameter value"))]
701pub enum QueryParameterValue {
702    /// Query parameter with single value
703    SingleParam(ParameterValue),
704    /// Query parameter with multiple values
705    MultiParams(Vec<ParameterValue>),
706}
707
708impl QueryParameterValue {
709    /// Returns an iterator over the contained `ParameterValue`s.
710    pub fn iter(&self) -> impl Iterator<Item = &ParameterValue> {
711        match self {
712            QueryParameterValue::SingleParam(param) => std::slice::from_ref(param).iter(),
713            QueryParameterValue::MultiParams(params) => params.iter(),
714        }
715    }
716
717    /// Convert to `Vec<ParameterValue>` for owned iteration.
718    fn into_vec(self) -> Vec<ParameterValue> {
719        match self {
720            QueryParameterValue::SingleParam(param) => vec![param],
721            QueryParameterValue::MultiParams(params) => params,
722        }
723    }
724}
725
726// Implement IntoIterator for owned QueryParameterValue
727impl IntoIterator for QueryParameterValue {
728    type Item = ParameterValue;
729    type IntoIter = std::vec::IntoIter<ParameterValue>;
730
731    fn into_iter(self) -> Self::IntoIter {
732        self.into_vec().into_iter()
733    }
734}
735
736pub type QueryParameters = HashMap<String, QueryParameterValue>;
737
738#[cfg(test)]
739mod tests {
740    use std::convert::Infallible;
741
742    use hyper::{Server, server::conn::AddrStream, service::make_service_fn};
743    use proptest::prelude::*;
744    use tower::ServiceBuilder;
745
746    use super::*;
747    use crate::test_util::addr::next_addr;
748
749    #[test]
750    fn test_default_request_headers_defaults() {
751        let user_agent = HeaderValue::from_static("vector");
752        let mut request = Request::post("http://example.com").body(()).unwrap();
753        default_request_headers(&mut request, &user_agent);
754        assert_eq!(
755            request.headers().get("Accept-Encoding"),
756            Some(&HeaderValue::from_static("identity")),
757        );
758        assert_eq!(request.headers().get("User-Agent"), Some(&user_agent));
759    }
760
761    #[test]
762    fn test_default_request_headers_does_not_overwrite() {
763        let mut request = Request::post("http://example.com")
764            .header("Accept-Encoding", "gzip")
765            .header("User-Agent", "foo")
766            .body(())
767            .unwrap();
768        default_request_headers(&mut request, &HeaderValue::from_static("vector"));
769        assert_eq!(
770            request.headers().get("Accept-Encoding"),
771            Some(&HeaderValue::from_static("gzip")),
772        );
773        assert_eq!(
774            request.headers().get("User-Agent"),
775            Some(&HeaderValue::from_static("foo"))
776        );
777    }
778
779    proptest! {
780        #[test]
781        fn test_jittered_duration(duration_in_secs in 0u64..120, jitter_factor in 0.0..1.0) {
782            let duration = Duration::from_secs(duration_in_secs);
783            let jittered_duration = MaxConnectionAgeLayer::jittered_duration(duration, jitter_factor);
784
785            // Check properties based on the range of inputs
786            if jitter_factor == 0.0 {
787                // When jitter_factor is 0, jittered_duration should be equal to the original duration
788                prop_assert_eq!(
789                    jittered_duration,
790                    duration,
791                    "jittered_duration {:?} should be equal to duration {:?}",
792                    jittered_duration,
793                    duration,
794                );
795            } else if duration_in_secs > 0 {
796                // Check the bounds when duration is non-zero and jitter_factor is non-zero
797                let lower_bound = duration.mul_f64(1.0 - jitter_factor);
798                let upper_bound = duration.mul_f64(1.0 + jitter_factor);
799                prop_assert!(
800                    jittered_duration >= lower_bound && jittered_duration <= upper_bound,
801                    "jittered_duration {:?} should be between {:?} and {:?}",
802                    jittered_duration,
803                    lower_bound,
804                    upper_bound,
805                );
806            } else {
807                // When duration is zero, jittered_duration should also be zero
808                prop_assert_eq!(
809                    jittered_duration,
810                    Duration::from_secs(0),
811                    "jittered_duration {:?} should be equal to zero",
812                    jittered_duration,
813                );
814            }
815        }
816    }
817
818    #[tokio::test]
819    async fn test_max_connection_age_service() {
820        tokio::time::pause();
821
822        let start_reference = Instant::now();
823        let max_connection_age = Duration::from_secs(1);
824        let mut service = MaxConnectionAgeService {
825            service: tower::service_fn(|_req: Request<Body>| async {
826                Ok::<Response<Body>, hyper::Error>(Response::new(Body::empty()))
827            }),
828            start_reference,
829            max_connection_age,
830            peer_addr: "1.2.3.4:1234".parse().unwrap(),
831        };
832
833        let req = Request::get("http://example.com")
834            .body(Body::empty())
835            .unwrap();
836        let response = service.call(req).await.unwrap();
837        assert_eq!(response.headers().get("Connection"), None);
838
839        tokio::time::advance(Duration::from_millis(500)).await;
840        let req = Request::get("http://example.com")
841            .body(Body::empty())
842            .unwrap();
843        let response = service.call(req).await.unwrap();
844        assert_eq!(response.headers().get("Connection"), None);
845
846        tokio::time::advance(Duration::from_millis(500)).await;
847        let req = Request::get("http://example.com")
848            .body(Body::empty())
849            .unwrap();
850        let response = service.call(req).await.unwrap();
851        assert_eq!(
852            response.headers().get("Connection"),
853            Some(&HeaderValue::from_static("close"))
854        );
855    }
856
857    #[tokio::test]
858    async fn test_max_connection_age_service_http2() {
859        tokio::time::pause();
860
861        let start_reference = Instant::now();
862        let max_connection_age = Duration::from_secs(0);
863        let mut service = MaxConnectionAgeService {
864            service: tower::service_fn(|_req: Request<Body>| async {
865                Ok::<Response<Body>, hyper::Error>(Response::new(Body::empty()))
866            }),
867            start_reference,
868            max_connection_age,
869            peer_addr: "1.2.3.4:1234".parse().unwrap(),
870        };
871
872        let mut req = Request::get("http://example.com")
873            .body(Body::empty())
874            .unwrap();
875        *req.version_mut() = Version::HTTP_2;
876        let response = service.call(req).await.unwrap();
877        assert_eq!(response.headers().get("Connection"), None);
878    }
879
880    #[tokio::test]
881    async fn test_max_connection_age_service_http3() {
882        tokio::time::pause();
883
884        let start_reference = Instant::now();
885        let max_connection_age = Duration::from_secs(0);
886        let mut service = MaxConnectionAgeService {
887            service: tower::service_fn(|_req: Request<Body>| async {
888                Ok::<Response<Body>, hyper::Error>(Response::new(Body::empty()))
889            }),
890            start_reference,
891            max_connection_age,
892            peer_addr: "1.2.3.4:1234".parse().unwrap(),
893        };
894
895        let mut req = Request::get("http://example.com")
896            .body(Body::empty())
897            .unwrap();
898        *req.version_mut() = Version::HTTP_3;
899        let response = service.call(req).await.unwrap();
900        assert_eq!(response.headers().get("Connection"), None);
901    }
902
903    #[tokio::test]
904    async fn test_max_connection_age_service_zero_duration() {
905        tokio::time::pause();
906
907        let start_reference = Instant::now();
908        let max_connection_age = Duration::from_millis(0);
909        let mut service = MaxConnectionAgeService {
910            service: tower::service_fn(|_req: Request<Body>| async {
911                Ok::<Response<Body>, hyper::Error>(Response::new(Body::empty()))
912            }),
913            start_reference,
914            max_connection_age,
915            peer_addr: "1.2.3.4:1234".parse().unwrap(),
916        };
917
918        let req = Request::get("http://example.com")
919            .body(Body::empty())
920            .unwrap();
921        let response = service.call(req).await.unwrap();
922        assert_eq!(
923            response.headers().get("Connection"),
924            Some(&HeaderValue::from_static("close"))
925        );
926    }
927
928    // Note that we unfortunately cannot mock the time in this test because the client calls
929    // sleep internally, which advances the clock.  However, this test shouldn't be flakey given
930    // the time bounds provided.
931    #[tokio::test]
932    async fn test_max_connection_age_service_with_hyper_server() {
933        // Create a hyper server with the max connection age layer.
934        let max_connection_age = Duration::from_secs(1);
935        let (_guard, addr) = next_addr();
936        let make_svc = make_service_fn(move |conn: &AddrStream| {
937            let svc = ServiceBuilder::new()
938                .layer(MaxConnectionAgeLayer::new(
939                    max_connection_age,
940                    0.,
941                    conn.remote_addr(),
942                ))
943                .service(tower::service_fn(|_req: Request<Body>| async {
944                    Ok::<Response<Body>, hyper::Error>(Response::new(Body::empty()))
945                }));
946            futures_util::future::ok::<_, Infallible>(svc)
947        });
948
949        tokio::spawn(async move {
950            Server::bind(&addr).serve(make_svc).await.unwrap();
951        });
952
953        // Wait for the server to start.
954        tokio::time::sleep(Duration::from_millis(10)).await;
955
956        // Create a client, which has its own connection pool.
957        let client = HttpClient::new(None, &ProxyConfig::default()).unwrap();
958
959        // Responses generated before the client's max connection age has elapsed do not
960        // include a `Connection: close` header in the response.
961        let req = Request::get(format!("http://{addr}/"))
962            .body(Body::empty())
963            .unwrap();
964        let response = client.send(req).await.unwrap();
965        assert_eq!(response.headers().get("Connection"), None);
966
967        let req = Request::get(format!("http://{addr}/"))
968            .body(Body::empty())
969            .unwrap();
970        let response = client.send(req).await.unwrap();
971        assert_eq!(response.headers().get("Connection"), None);
972
973        // The first response generated after the client's max connection age has elapsed should
974        // include the `Connection: close` header.
975        tokio::time::sleep(Duration::from_secs(1)).await;
976        let req = Request::get(format!("http://{addr}/"))
977            .body(Body::empty())
978            .unwrap();
979        let response = client.send(req).await.unwrap();
980        assert_eq!(
981            response.headers().get("Connection"),
982            Some(&HeaderValue::from_static("close")),
983        );
984
985        // The next request should establish a new connection.
986        // Importantly, this also confirms that each connection has its own independent
987        // connection age timer.
988        let req = Request::get(format!("http://{addr}/"))
989            .body(Body::empty())
990            .unwrap();
991        let response = client.send(req).await.unwrap();
992        assert_eq!(response.headers().get("Connection"), None);
993    }
994}
vector/http.rs

vector/
http.rs
