|
| 1 | +//! `/api/v2/series` handler and validation pipeline. |
| 2 | +//! |
| 3 | +//! `handle_series` fires every payload property's assertion, walks the envelope, |
| 4 | +//! byte-size, and decode checks in order, then returns the first failure status or |
| 5 | +//! `202 Accepted`. |
| 6 | +
|
| 7 | +use std::sync::Arc; |
| 8 | +use std::time::{SystemTime, UNIX_EPOCH}; |
| 9 | + |
| 10 | +use axum::{ |
| 11 | + body::to_bytes, |
| 12 | + extract::{Request, State}, |
| 13 | + http::StatusCode, |
| 14 | + response::{IntoResponse, Response}, |
| 15 | +}; |
| 16 | +use tracing::{debug, error}; |
| 17 | + |
| 18 | +use crate::http::Measurements; |
| 19 | +use crate::properties::payload::{bytes, envelope}; |
| 20 | +use crate::series_observation::SeriesObservation; |
| 21 | + |
| 22 | +/// Memory backstop on the decompressed body buffered in the handler, above the Pyld06 5 MiB spec limit |
| 23 | +const MAX_DECOMPRESSED_BODY_BYTES: usize = 64 * 1024 * 1024; |
| 24 | + |
| 25 | +/// Reasons `handle_series` cannot evaluate a request. |
| 26 | +#[derive(Debug)] |
| 27 | +pub(crate) enum SeriesError { |
| 28 | + /// The measurement middleware did not record `Measurements` on the route. |
| 29 | + MissingMeasurements, |
| 30 | + /// The system clock predates the Unix epoch or overflows i64 seconds. |
| 31 | + Clock, |
| 32 | + /// Reading the request body failed, or the body overran the decompressed cap. |
| 33 | + Body(axum::Error), |
| 34 | +} |
| 35 | + |
| 36 | +impl IntoResponse for SeriesError { |
| 37 | + fn into_response(self) -> Response { |
| 38 | + match self { |
| 39 | + Self::MissingMeasurements => { |
| 40 | + error!("Missing Measurements extension on /api/v2/series, measurement middleware is misconfigured."); |
| 41 | + StatusCode::INTERNAL_SERVER_ERROR |
| 42 | + } |
| 43 | + Self::Clock => { |
| 44 | + error!("System clock is not readable as seconds since the Unix epoch."); |
| 45 | + StatusCode::INTERNAL_SERVER_ERROR |
| 46 | + } |
| 47 | + Self::Body(e) => { |
| 48 | + // `to_bytes` errors on the size cap and on a read failure. Treat both |
| 49 | + // as oversized, matching the wire-side measurement middleware. |
| 50 | + error!(error = %e, cap = MAX_DECOMPRESSED_BODY_BYTES, "Rejected /api/v2/series body at the decompressed cap."); |
| 51 | + StatusCode::PAYLOAD_TOO_LARGE |
| 52 | + } |
| 53 | + } |
| 54 | + .into_response() |
| 55 | + } |
| 56 | +} |
| 57 | + |
| 58 | +/// Handler for `POST /api/v2/series`. |
| 59 | +pub(crate) async fn handle_series( |
| 60 | + State(expected_hostname): State<Arc<str>>, request: Request, |
| 61 | +) -> Result<StatusCode, SeriesError> { |
| 62 | + // Pyld21 bounds points' timestamps against the intake wall clock at request receipt |
| 63 | + let now_secs = now_epoch_secs()?; |
| 64 | + let (parts, body) = request.into_parts(); |
| 65 | + let &Measurements { |
| 66 | + compressed_len, |
| 67 | + decompression_applied, |
| 68 | + declared_content_length, |
| 69 | + } = parts |
| 70 | + .extensions |
| 71 | + .get::<Measurements>() |
| 72 | + .ok_or(SeriesError::MissingMeasurements)?; |
| 73 | + |
| 74 | + let body_bytes = to_bytes(body, MAX_DECOMPRESSED_BODY_BYTES) |
| 75 | + .await |
| 76 | + .map_err(SeriesError::Body)?; |
| 77 | + |
| 78 | + // Datadog Agent sends `{}` to probe connectivity, not a metric payload. The real |
| 79 | + // intake accepts the probe with 202. Match it rather than 200. |
| 80 | + if body_bytes.as_ref() == b"{}" { |
| 81 | + debug!("Received connectivity probe for /api/v2/series, returning 202 Accepted."); |
| 82 | + return Ok(StatusCode::ACCEPTED); |
| 83 | + } |
| 84 | + |
| 85 | + let headers = parts.headers; |
| 86 | + let uncompressed_len = body_bytes.len() as u64; |
| 87 | + |
| 88 | + // Envelope and byte-size properties. |
| 89 | + let api_key_ok = envelope::api_key(&headers); |
| 90 | + let content_type_ok = envelope::content_type(&headers); |
| 91 | + envelope::content_encoding(&headers); |
| 92 | + let compressed_ok = bytes::compressed_size(compressed_len); |
| 93 | + let uncompressed_ok = bytes::uncompressed_size(uncompressed_len, decompression_applied); |
| 94 | + bytes::content_length(declared_content_length, compressed_len); |
| 95 | + |
| 96 | + let (observation, decode_ok) = SeriesObservation::decode(&body_bytes, decompression_applied); |
| 97 | + |
| 98 | + if let Some(observation) = observation.as_ref() { |
| 99 | + observation.assert_payload_properties(now_secs, &expected_hostname); |
| 100 | + debug!( |
| 101 | + bytes = body_bytes.len(), |
| 102 | + series = observation.series_len(), |
| 103 | + "received /api/v2/series" |
| 104 | + ); |
| 105 | + } |
| 106 | + |
| 107 | + // Return the first failure status in pipeline order, or 202 Accepted. |
| 108 | + let failure = first_status_failure(&[ |
| 109 | + (api_key_ok, StatusCode::FORBIDDEN), |
| 110 | + (content_type_ok, StatusCode::BAD_REQUEST), |
| 111 | + (compressed_ok, StatusCode::PAYLOAD_TOO_LARGE), |
| 112 | + (uncompressed_ok, StatusCode::PAYLOAD_TOO_LARGE), |
| 113 | + (decode_ok, StatusCode::BAD_REQUEST), |
| 114 | + ]); |
| 115 | + Ok(failure.unwrap_or(StatusCode::ACCEPTED)) |
| 116 | +} |
| 117 | + |
| 118 | +/// Return the first failed status check, in the given pipeline order, or `None` |
| 119 | +/// when every check holds. |
| 120 | +fn first_status_failure(checks: &[(bool, StatusCode)]) -> Option<StatusCode> { |
| 121 | + checks.iter().find(|(ok, _)| !ok).map(|&(_, status)| status) |
| 122 | +} |
| 123 | + |
| 124 | +/// Return the current time as whole seconds since the Unix epoch. |
| 125 | +/// |
| 126 | +/// Returns `SeriesError::Clock` when the system clock predates the epoch or the second count |
| 127 | +/// overflows `i64`. |
| 128 | +fn now_epoch_secs() -> Result<i64, SeriesError> { |
| 129 | + let secs = SystemTime::now() |
| 130 | + .duration_since(UNIX_EPOCH) |
| 131 | + .map_err(|_| SeriesError::Clock)? |
| 132 | + .as_secs(); |
| 133 | + i64::try_from(secs).map_err(|_| SeriesError::Clock) |
| 134 | +} |
0 commit comments