saluki_io/net/client/http/
telemetry.rs

1use std::{
2    collections::HashMap,
3    future::Future,
4    pin::Pin,
5    sync::{Arc, Mutex, OnceLock},
6    task::{ready, Context, Poll},
7};
8
9use http::{uri::Authority, Request, Response, StatusCode, Uri, Version};
10use http_body::Body;
11use metrics::Counter;
12use pin_project_lite::pin_project;
13use saluki_metrics::MetricsBuilder;
14use stringtheory::MetaString;
15use tower::{Layer, Service};
16
17pub type EndpointNameFn = dyn Fn(&Uri) -> Option<MetaString> + Send + Sync;
18
19const ERROR_TYPE_CLIENT: &str = "client_error";
20pub(super) const ERROR_TYPE_CONNECTION: &str = "connection_error";
21pub(super) const ERROR_TYPE_DNS: &str = "dns_error";
22pub(super) const ERROR_TYPE_TLS: &str = "tls_error";
23pub(super) const ERROR_TYPE_WROTE_REQUEST: &str = "wrote_request_error";
24const ERROR_TYPE_CANT_SEND: &str = "cant_send";
25const ERROR_TYPE_GT_400: &str = "gt_400";
26const ERROR_SCOPE_PHASE: &str = "phase";
27const ERROR_SCOPE_TRANSACTION: &str = "transaction";
28
29/// Emits lifecycle and transaction error telemetry for HTTP requests.
30#[derive(Clone)]
31pub(crate) struct HttpTransactionErrorTelemetry {
32    dns_errors: Counter,
33    connection_errors: Counter,
34    tls_errors: Counter,
35    wrote_request_errors: Counter,
36    send_errors: Counter,
37    http_errors: Counter,
38}
39
40impl HttpTransactionErrorTelemetry {
41    /// Creates a new `HttpTransactionErrorTelemetry` from a metrics builder.
42    pub(crate) fn from_builder(builder: &MetricsBuilder) -> Self {
43        // Mirror Core Agent forwarder buckets by counting lifecycle failures at their source. See
44        // datadog-agent/comp/forwarder/defaultforwarder/transaction/transaction.go::GetClientTrace.
45        Self {
46            dns_errors: register_scoped_error(builder, ERROR_TYPE_DNS, ERROR_SCOPE_PHASE),
47            connection_errors: register_scoped_error(builder, ERROR_TYPE_CONNECTION, ERROR_SCOPE_PHASE),
48            tls_errors: register_scoped_error(builder, ERROR_TYPE_TLS, ERROR_SCOPE_PHASE),
49            wrote_request_errors: register_scoped_error(builder, ERROR_TYPE_WROTE_REQUEST, ERROR_SCOPE_PHASE),
50            send_errors: register_scoped_error(builder, ERROR_TYPE_CANT_SEND, ERROR_SCOPE_TRANSACTION),
51            http_errors: register_scoped_error(builder, ERROR_TYPE_GT_400, ERROR_SCOPE_TRANSACTION),
52        }
53    }
54
55    pub(crate) fn dns_errors(&self) -> Counter {
56        self.dns_errors.clone()
57    }
58
59    pub(crate) fn increment_connection_error(&self) {
60        self.connection_errors.increment(1);
61    }
62
63    pub(crate) fn increment_tls_error(&self) {
64        self.tls_errors.increment(1);
65    }
66
67    pub(crate) fn increment_wrote_request_error(&self) {
68        self.wrote_request_errors.increment(1);
69    }
70
71    fn increment_send_error(&self) {
72        self.send_errors.increment(1);
73    }
74
75    fn increment_http_error(&self) {
76        self.http_errors.increment(1);
77    }
78}
79
80fn register_scoped_error(builder: &MetricsBuilder, error_type: &'static str, error_scope: &'static str) -> Counter {
81    builder.register_counter_with_tags(
82        "network_http_requests_errors_total",
83        [("error_type", error_type), ("error_scope", error_scope)],
84    )
85}
86
87/// Emit telemetry about the status of HTTP transactions.
88///
89/// This layer can be used with services that deal with `http::Request` and `http::Response`, and wraps them to provide
90/// telemetry about the status of an HTTP "transaction": a full round-trip of request and response.
91///
92/// ## Metrics
93///
94/// The following metrics are emitted:
95///
96/// - `network_http_requests_failed_total`: The total number of HTTP requests that failed with a status code of 400,
97///   403, or 413.
98/// - `network_http_requests_success_total`: The total number of successful HTTP requests. (any response with a
99///   non-4xx/5xx status code) This metric is additionally tagged with the response protocol as `proto_version`.
100/// - `network_http_requests_success_sent_bytes_total`: The total number of body bytes sent in successful HTTP requests.
101///   (see note below on how this is calculated)
102/// - `network_http_requests_errors_total`: The total number of HTTP requests that had an error, either during the
103///   sending of the request or in the response. This is further broken down by the `error_type` label.
104///   - For all responses with a status code greater than 400, `error_type` will be `client_error` and `code` will be
105///     the string version of the status code.
106///   - When there is an error during the sending of the request, `error_type` classifies the request failure.
107///
108/// All metrics are emitted with two base tags:
109///
110/// - `domain`: The full domain of the request, including scheme and port, but excluding any credentials.
111/// - `endpoint`: The endpoint name, which is derived from the URI path by default but can be customized. (See
112///   [`EndpointTelemetryLayer::with_endpoint_name_fn`] for information on customization and how the endpoint name,
113///   overall, is sanitized.)
114///
115/// Successful request counts also include `proto_version`, formatted as the response's HTTP version (for example,
116/// `HTTP/1.1` or `HTTP/2.0`). Versions without dedicated support are grouped under `unknown`. Successful request bytes
117/// remain grouped only by the base tags.
118///
119/// ### Success bytes calculation
120///
121/// We calculate the number of bytes sent by examining the body length itself, which is done via [`Body::size_hint`].
122/// This requires that an exact body size is known, which isn't always the case. If the body size isn't known, this
123/// metric won't be emitted on a successful response.
124///
125/// For common body types, like [`FrozenChunkedBytesBuffer`][saluki_common::buf::FrozenChunkedBytesBuffer], the size
126/// hint is always exact and so this functionality should work as intended.
127#[derive(Clone, Default)]
128pub struct EndpointTelemetryLayer {
129    builder: MetricsBuilder,
130    endpoint_name_fn: Option<Arc<EndpointNameFn>>,
131    error_telemetry: Option<HttpTransactionErrorTelemetry>,
132}
133
134impl EndpointTelemetryLayer {
135    /// Create a new `EndpointTelemetryLayer` with the given `ComponentContext`.
136    ///
137    /// The component context is used when creating metrics, which ensures they're tagged in a consistent way that
138    /// attributes the metrics to the component issuing the HTTP requests.
139    pub fn with_metrics_builder(mut self, builder: MetricsBuilder) -> Self {
140        self.builder = builder;
141        self
142    }
143
144    pub(super) fn with_error_telemetry(mut self, error_telemetry: HttpTransactionErrorTelemetry) -> Self {
145        self.error_telemetry = Some(error_telemetry);
146        self
147    }
148
149    /// Sets the function used to extract the "endpoint name" from a URI.
150    ///
151    /// The value returned by this function will be sanitized to ensure it can be used as a tag value, and is limited
152    /// to: ASCII alphanumerics, hyphens, underscores, slashes, and periods. Any non-conforming character will be
153    /// replaced with an underscore. Characters will be converted to lowercase.
154    ///
155    /// The value returned by this function is also cached for the given URI, and so the function shouldn't rely on
156    /// non-deterministic behavior, or state, that could change the generated endpoint name for subsequent calls with
157    /// the same input URI.
158    pub fn with_endpoint_name_fn<F>(mut self, endpoint_name_fn: F) -> Self
159    where
160        F: Fn(&Uri) -> Option<MetaString> + Send + Sync + 'static,
161    {
162        self.endpoint_name_fn = Some(Arc::new(endpoint_name_fn));
163        self
164    }
165}
166
167impl<S> Layer<S> for EndpointTelemetryLayer {
168    type Service = EndpointTelemetry<S>;
169
170    fn layer(&self, service: S) -> Self::Service {
171        EndpointTelemetry {
172            service,
173            builder: self.builder.clone(),
174            endpoint_name_fn: self.endpoint_name_fn.clone(),
175            error_telemetry: self.error_telemetry.clone(),
176            domains: HashMap::new(),
177            endpoint_name_cache: HashMap::new(),
178        }
179    }
180}
181
182#[derive(Default)]
183struct SuccessCounters {
184    http_09: OnceLock<Counter>,
185    http_10: OnceLock<Counter>,
186    http_11: OnceLock<Counter>,
187    http_2: OnceLock<Counter>,
188    http_3: OnceLock<Counter>,
189    fallback: OnceLock<Counter>,
190}
191
192impl SuccessCounters {
193    fn slot_for_version(&self, version: Version) -> (&OnceLock<Counter>, &'static str) {
194        match version {
195            Version::HTTP_09 => (&self.http_09, "HTTP/0.9"),
196            Version::HTTP_10 => (&self.http_10, "HTTP/1.0"),
197            Version::HTTP_11 => (&self.http_11, "HTTP/1.1"),
198            Version::HTTP_2 => (&self.http_2, "HTTP/2.0"),
199            Version::HTTP_3 => (&self.http_3, "HTTP/3.0"),
200            _ => (&self.fallback, "unknown"),
201        }
202    }
203}
204
205struct PerEndpointTelemetry {
206    builder: MetricsBuilder,
207    dropped: Counter,
208    success: SuccessCounters,
209    success_bytes: Counter,
210    http_errors_map: Mutex<HashMap<StatusCode, Counter>>,
211}
212
213impl PerEndpointTelemetry {
214    fn new(builder: MetricsBuilder, uri: &Uri, endpoint_name: &str) -> Self {
215        // Reconstruct the full domain from the URI, including scheme and port, but leaving out any credentials.
216        let mut domain = format!("{}://{}", uri.scheme_str().unwrap(), uri.host().unwrap());
217        if let Some(port) = uri.port() {
218            domain.push(':');
219            domain.push_str(port.as_str());
220        }
221
222        let builder = builder
223            .add_default_tag(("domain", domain))
224            .add_default_tag(("endpoint", endpoint_name.to_string()));
225
226        let dropped = builder.register_counter("network_http_requests_failed_total");
227        let success = SuccessCounters::default();
228        let success_bytes = builder.register_counter("network_http_requests_success_sent_bytes_total");
229        let http_errors_map = Mutex::new(HashMap::new());
230
231        Self {
232            builder,
233            dropped,
234            success,
235            success_bytes,
236            http_errors_map,
237        }
238    }
239
240    fn increment_dropped(&self) {
241        self.dropped.increment(1);
242    }
243
244    fn increment_success(&self, version: Version) {
245        let (slot, proto_version) = self.success.slot_for_version(version);
246        slot.get_or_init(|| {
247            self.builder.register_counter_with_tags(
248                "network_http_requests_success_total",
249                [("proto_version", proto_version)],
250            )
251        })
252        .increment(1);
253    }
254
255    fn increment_success_bytes(&self, len: u64) {
256        self.success_bytes.increment(len);
257    }
258
259    fn increment_http_error(&self, status: StatusCode) {
260        let mut http_errors_map = self.http_errors_map.lock().unwrap();
261        let counter = http_errors_map.entry(status).or_insert_with(move || {
262            self.builder.register_counter_with_tags(
263                "network_http_requests_errors_total",
264                [
265                    ("error_type", ERROR_TYPE_CLIENT.to_string()),
266                    ("code", status.as_str().to_string()),
267                ],
268            )
269        });
270        counter.increment(1);
271    }
272}
273
274/// Emit telemetry about the status of HTTP transactions.
275#[derive(Clone)]
276pub struct EndpointTelemetry<S> {
277    service: S,
278    builder: MetricsBuilder,
279    endpoint_name_fn: Option<Arc<EndpointNameFn>>,
280    error_telemetry: Option<HttpTransactionErrorTelemetry>,
281    domains: HashMap<Authority, HashMap<MetaString, Arc<PerEndpointTelemetry>>>,
282    endpoint_name_cache: HashMap<Uri, MetaString>,
283}
284
285impl<S> EndpointTelemetry<S> {
286    fn get_telemetry_handle<B>(&mut self, req: &Request<B>) -> Option<Arc<PerEndpointTelemetry>>
287    where
288        B: Body,
289    {
290        // We require a scheme and a host in the URI to emit telemetry.
291        if req.uri().scheme().is_none() || req.uri().host().is_none() {
292            return None;
293        }
294
295        // `Authority` is underpinned by `Bytes` so cloning is cheap.
296        let authority = req.uri().authority()?.clone();
297        let domain = self.domains.entry(authority).or_default();
298
299        // Look up the per-endpoint telemetry handle, or create a new one if it doesn't exist.
300        //
301        // We do some caching of the endpoint name to avoid repeatedly calling the endpoint name function, which could
302        // be expensive due to the sanitization we perform on it.
303        let endpoint_telemetry = match self.endpoint_name_cache.get(req.uri()) {
304            Some(endpoint_name) => domain
305                .get(endpoint_name)
306                .expect("per-endpoint telemetry must exist if name is cached"),
307            None => {
308                // Generate our endpoint name, and then cache it.
309                let endpoint_name = self
310                    .endpoint_name_fn
311                    .as_ref()
312                    .and_then(|f| f(req.uri()))
313                    .map(sanitize_endpoint_name)
314                    .unwrap_or_else(|| sanitize_endpoint_name(req.uri().path().into()));
315
316                self.endpoint_name_cache
317                    .insert(req.uri().clone(), endpoint_name.clone());
318
319                // Now we'll create the per-endpoint telemetry.
320                domain.entry(endpoint_name).or_insert_with_key(|endpoint_name| {
321                    Arc::new(PerEndpointTelemetry::new(
322                        self.builder.clone(),
323                        req.uri(),
324                        endpoint_name,
325                    ))
326                })
327            }
328        };
329
330        Some(Arc::clone(endpoint_telemetry))
331    }
332}
333
334impl<B, B2, S> Service<Request<B>> for EndpointTelemetry<S>
335where
336    S: Service<Request<B>, Response = http::Response<B2>>,
337    B: Body,
338    B2: Body,
339{
340    type Response = S::Response;
341    type Error = S::Error;
342    type Future = EndpointTelemetryFuture<S::Future>;
343
344    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
345        self.service.poll_ready(cx)
346    }
347
348    fn call(&mut self, req: Request<B>) -> Self::Future {
349        let maybe_body_len = req.body().size_hint().exact();
350        let per_endpoint = self.get_telemetry_handle(&req);
351        let fut = self.service.call(req);
352
353        EndpointTelemetryFuture {
354            per_endpoint,
355            error_telemetry: self.error_telemetry.clone(),
356            maybe_body_len,
357            fut,
358        }
359    }
360}
361
362pin_project! {
363    /// Response future from [`EndpointTelemetry`] services.
364    pub struct EndpointTelemetryFuture<F> {
365        per_endpoint: Option<Arc<PerEndpointTelemetry>>,
366        error_telemetry: Option<HttpTransactionErrorTelemetry>,
367        maybe_body_len: Option<u64>,
368
369        #[pin]
370        fut: F,
371    }
372}
373
374impl<F, B, E> Future for EndpointTelemetryFuture<F>
375where
376    F: Future<Output = Result<Response<B>, E>>,
377    B: Body,
378{
379    type Output = Result<Response<B>, E>;
380
381    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
382        let this = self.project();
383        match ready!(this.fut.poll(cx)) {
384            Ok(response) => {
385                if let Some(per_endpoint) = this.per_endpoint.as_ref() {
386                    let status = response.status();
387                    if status.is_client_error() || status.is_server_error() {
388                        // Always increment the HTTP error total by grouped over the actual status code.
389                        per_endpoint.increment_http_error(status);
390
391                        let status_code = status.as_u16();
392                        if status_code == 400 || status_code == 403 || status_code == 413 {
393                            // There's some specific errors where we're not going to retry them, so we can reasonable
394                            // classify these requests as being dropped: they won't be retried, etc.
395                            per_endpoint.increment_dropped()
396                        } else if let Some(error_telemetry) = this.error_telemetry.as_ref() {
397                            error_telemetry.increment_http_error();
398                        }
399                    } else {
400                        per_endpoint.increment_success(response.version());
401                        if let Some(body_len) = this.maybe_body_len {
402                            per_endpoint.increment_success_bytes(*body_len);
403                        }
404                    }
405                }
406
407                Poll::Ready(Ok(response))
408            }
409            Err(e) => {
410                if let Some(error_telemetry) = this.error_telemetry.as_ref() {
411                    error_telemetry.increment_send_error();
412                }
413
414                Poll::Ready(Err(e))
415            }
416        }
417    }
418}
419
420fn is_sanitized_endpoint_name(s: &str) -> bool {
421    s.chars().all(|c| {
422        c.is_ascii_lowercase() || c.is_ascii_digit() || c == '_' || c == '-' || c == '/' || c == '\\' || c == '.'
423    })
424}
425
426fn sanitize_endpoint_name(endpoint_name: MetaString) -> MetaString {
427    // Check if the endpoint name is already sanitized, and if so, just return it as-is.
428    if is_sanitized_endpoint_name(&endpoint_name) {
429        return endpoint_name;
430    }
431
432    endpoint_name
433        .chars()
434        .map(|c| {
435            if c.is_ascii_alphanumeric() || c == '_' || c == '-' || c == '/' || c == '\\' || c == '.' {
436                c.to_ascii_lowercase()
437            } else {
438                '_'
439            }
440        })
441        .collect::<String>()
442        .into()
443}
444
445#[cfg(test)]
446mod tests {
447    use std::convert::Infallible;
448
449    use bytes::Bytes;
450    use http_body_util::{Empty, Full};
451    use metrics::{Key, Label};
452    use metrics_util::{
453        debugging::{DebugValue, DebuggingRecorder},
454        CompositeKey, MetricKind,
455    };
456    use proptest::prelude::*;
457
458    use super::*;
459
460    #[test]
461    fn success_counter_slots_cover_supported_http_versions() {
462        let counters = SuccessCounters::default();
463        let cases = [
464            (Version::HTTP_09, &counters.http_09, "HTTP/0.9"),
465            (Version::HTTP_10, &counters.http_10, "HTTP/1.0"),
466            (Version::HTTP_11, &counters.http_11, "HTTP/1.1"),
467            (Version::HTTP_2, &counters.http_2, "HTTP/2.0"),
468            (Version::HTTP_3, &counters.http_3, "HTTP/3.0"),
469        ];
470
471        for (version, expected_slot, expected_label) in cases {
472            let (slot, label) = counters.slot_for_version(version);
473            assert!(std::ptr::eq(slot, expected_slot));
474            assert_eq!(label, expected_label);
475        }
476    }
477
478    #[test]
479    fn successful_requests_are_counted_by_response_protocol_without_splitting_bytes() {
480        let recorder = DebuggingRecorder::new();
481        let snapshotter = recorder.snapshotter();
482        let mut response_versions = [http::Version::HTTP_11, http::Version::HTTP_2].into_iter();
483        let inner = tower::service_fn(move |_request: Request<Full<Bytes>>| {
484            let version = response_versions.next().expect("response version should be available");
485            async move {
486                Ok::<_, Infallible>(
487                    Response::builder()
488                        .version(version)
489                        .body(Empty::<Bytes>::new())
490                        .expect("response should be valid"),
491                )
492            }
493        });
494        let mut service = EndpointTelemetryLayer::default().layer(inner);
495
496        let first_request = Request::builder()
497            .uri("https://example.com/api/v1/series")
498            .version(http::Version::HTTP_10)
499            .body(Full::new(Bytes::from_static(b"hello")))
500            .expect("request should be valid");
501        metrics::with_local_recorder(&recorder, || {
502            tokio_test::block_on(service.call(first_request)).expect("request should succeed")
503        });
504
505        let second_request = Request::builder()
506            .uri("https://example.com/api/v1/series")
507            .version(http::Version::HTTP_10)
508            .body(Full::new(Bytes::from_static(b"goodbye")))
509            .expect("request should be valid");
510        metrics::with_local_recorder(&recorder, || {
511            tokio_test::block_on(service.call(second_request)).expect("request should succeed")
512        });
513
514        let snapshot = snapshotter.snapshot().into_hashmap();
515        let success_keys = snapshot
516            .keys()
517            .filter(|key| key.key().name() == "network_http_requests_success_total")
518            .count();
519        assert_eq!(success_keys, 2);
520        for proto_version in ["HTTP/1.1", "HTTP/2.0"] {
521            let key = CompositeKey::new(
522                MetricKind::Counter,
523                Key::from_parts(
524                    "network_http_requests_success_total",
525                    vec![
526                        Label::new("domain", "https://example.com"),
527                        Label::new("endpoint", "/api/v1/series"),
528                        Label::new("proto_version", proto_version),
529                    ],
530                ),
531            );
532            let (_, _, value) = snapshot
533                .get(&key)
534                .expect("protocol-specific success counter should exist");
535            assert_eq!(value, &DebugValue::Counter(1));
536        }
537
538        let success_bytes_keys = snapshot
539            .keys()
540            .filter(|key| key.key().name() == "network_http_requests_success_sent_bytes_total")
541            .count();
542        assert_eq!(success_bytes_keys, 1);
543        let success_bytes_key = CompositeKey::new(
544            MetricKind::Counter,
545            Key::from_parts(
546                "network_http_requests_success_sent_bytes_total",
547                vec![
548                    Label::new("domain", "https://example.com"),
549                    Label::new("endpoint", "/api/v1/series"),
550                ],
551            ),
552        );
553        let (_, _, value) = snapshot
554            .get(&success_bytes_key)
555            .expect("protocol-agnostic success bytes counter should exist");
556        assert_eq!(value, &DebugValue::Counter(12));
557    }
558
559    proptest! {
560        #[test]
561        fn property_test_sanitize_endpoint_name(input in ".*") {
562            let sanitized = sanitize_endpoint_name(input.into());
563            prop_assert!(is_sanitized_endpoint_name(&sanitized));
564        }
565    }
566}
saluki_io/net/client/http/telemetry.rs

saluki_io/net/client/http/
telemetry.rs
