saluki_components/encoders/datadog/metrics/

mod.rs

use std::{collections::HashMap, collections::VecDeque, ops::Range, time::Duration};

use agent_data_plane_config::shared::{SharedConfiguration, V3SeriesMode};
use async_trait::async_trait;
use ddsketch::DDSketch;
use http::{HeaderValue, Method, Request};
use protobuf::{rt::WireType, CodedOutputStream};
use saluki_common::{
    buf::{ChunkedBytesBuffer, FrozenChunkedBytesBuffer},
    iter::ReusableDeduplicator,
};
use saluki_context::tags::{SharedTagSet, Tag};
use saluki_core::accounting::{MemoryBounds, MemoryBoundsBuilder};
use saluki_core::runtime;
use saluki_core::{
    components::{encoders::*, BuildContext},
    data_model::{
        event::{
            metric::{Metric, MetricOrigin, MetricValues},
            EventType,
        },
        payload::{HttpPayload, Payload, PayloadMetadata, PayloadType},
    },
    observability::ComponentMetricsExt as _,
    topology::{EventsBuffer, PayloadsBuffer},
};
use saluki_error::{generic_error, ErrorContext as _, GenericError};
use saluki_io::compression::{CompressionScheme, Compressor};
use saluki_metrics::MetricsBuilder;
use tokio::{io::AsyncWriteExt as _, select, sync::mpsc, time::sleep};
use tracing::{debug, error, warn};
use url::Url;

use self::v3::{
    V3EncodedMetrics, V3EncodedRequest, V3EncoderStats, V3MetricType, V3PayloadLimits, V3PayloadRequest,
    V3PayloadSplitReason, V3SerializerTelemetry, V3Writer,
};
use crate::{
    common::datadog::{
        clamp_payload_limits,
        config::OpwMetricsConfiguration,
        endpoints::{
            resolve_additional_endpoints, series_v3_config_can_enable_v3, EndpointV3Settings, ResolvedEndpoint,
            V3EndpointConfig,
        },
        io::RB_BUFFER_CHUNK_SIZE,
        protocol::{MetricsPayloadInfo, UseV3ApiConfig, UseV3ApiSeriesConfig, V3ApiConfig},
        request_builder::{RequestBuilder, RequestBuilderError},
        telemetry::ComponentTelemetry,
        DEFAULT_SERIALIZER_COMPRESSED_SIZE_LIMIT, DEFAULT_SERIALIZER_UNCOMPRESSED_SIZE_LIMIT, METRICS_SERIES_V3_PATH,
        METRICS_SKETCHES_V3_PATH,
    },
    encoders::datadog::metrics::v2::MetricsEndpointEncoder,
};

mod endpoint;
use self::endpoint::{EndpointConfiguration, MetricsEndpoint};

mod v1;
mod v2;
mod v3;

const V3_SKETCHES_ENDPOINT_URI: &str = METRICS_SKETCHES_V3_PATH;

/// Encoding mode for a metrics endpoint.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum MetricsEncoderMode {
    /// Send V2 payloads only.
    V2Only,
    /// V3 is enabled for at least one endpoint; generate tagged V2 and V3 payloads so each endpoint
    /// receives the protocol version configured for it.
    V3Enabled,
}

impl MetricsEncoderMode {
    fn from_config(use_v3: bool) -> Self {
        match use_v3 {
            false => Self::V2Only,
            true => Self::V3Enabled,
        }
    }

    fn needs_v3(self) -> bool {
        matches!(self, Self::V3Enabled)
    }

    fn needs_tagging(self) -> bool {
        matches!(self, Self::V3Enabled)
    }
}

fn metrics_encoder_mode_for_config(use_v3: bool, metrics_v3_disabled_by_compressor: bool) -> MetricsEncoderMode {
    let use_v3 = use_v3 && !metrics_v3_disabled_by_compressor;
    MetricsEncoderMode::from_config(use_v3)
}

fn selected_metrics_primary_v3_override(opw_metrics: &OpwMetricsConfiguration) -> Option<bool> {
    selected_metrics_primary_endpoint(opw_metrics).map(|(_, use_v3_series)| use_v3_series)
}

fn selected_metrics_primary_endpoint(opw_metrics: &OpwMetricsConfiguration) -> Option<(&str, bool)> {
    let selected = opw_metrics.selected_endpoint()?;
    let url = selected.url.trim();
    metrics_primary_url_can_resolve(url).then_some((url, selected.use_v3_series))
}

fn metrics_primary_url_can_resolve(url: &str) -> bool {
    if url.is_empty() {
        return false;
    }
    if url.starts_with("http://") || url.starts_with("https://") {
        Url::parse(url).is_ok_and(|url| url.host_str().is_some())
    } else {
        Url::parse(&format!("https://{url}")).is_ok_and(|url| url.host_str().is_some())
    }
}

fn series_v3_can_be_enabled_for_config(
    use_v2_api_series: bool, metrics_primary_v3_override: Option<bool>, has_additional_endpoints: bool,
    series_config: &UseV3ApiSeriesConfig,
) -> bool {
    use_v2_api_series
        && (metrics_primary_v3_override == Some(true)
            || ((metrics_primary_v3_override != Some(false) || has_additional_endpoints)
                && series_v3_config_can_enable_v3(series_config)))
}

/// Datadog Metrics encoder.
///
/// Generates Datadog metrics payloads for the Datadog platform.
#[derive(Clone)]
#[cfg_attr(test, derive(Debug, PartialEq))]
pub struct DatadogMetricsConfiguration {
    /// Maximum number of input metrics to encode into a single request payload.
    ///
    /// This applies both to the series and sketches endpoints.
    max_metrics_per_payload: usize,

    /// Maximum compressed size, in bytes, of generic payloads.
    ///
    /// This applies to V1 JSON series payloads and sketch payloads, matching the Datadog Agent's generic payload
    /// builder. V2 series payloads use the series-specific limit instead. The effective value is clamped to
    /// the Agent's default intake-safe limit of 2,621,440 bytes, so larger configured values do not allow payloads that
    /// intake may reject. If set to `0`, every non-empty compressed payload exceeds the limit and is dropped during
    /// flush.
    max_payload_size: usize,

    /// Maximum uncompressed size, in bytes, of generic payloads.
    ///
    /// This applies to V1 JSON series payloads and sketch payloads, matching the Datadog Agent's generic payload
    /// builder. V2 series payloads use the series-specific limit instead. The effective value
    /// is clamped to the Agent's default intake-safe limit of 4,194,304 bytes, so larger configured values do not allow
    /// payloads that intake may reject. Values smaller than the minimum endpoint framing size prevent the request
    /// builder from starting.
    max_uncompressed_payload_size: usize,

    /// Maximum compressed size, in bytes, of a V2 series payload.
    ///
    /// This applies only when the V2 series API is in use. V1 series and sketches use the generic payload limit
    /// instead. The effective value is clamped to the V2 series API limit of 512,000 bytes, so larger configured values
    /// do not allow payloads that intake would reject. If set to `0`, every non-empty compressed payload exceeds the
    /// limit and is dropped during flush.
    max_series_payload_size: usize,

    /// Maximum uncompressed size, in bytes, of a V2 series payload.
    ///
    /// This applies only when the V2 series API is in use. V1 series and sketches use the generic uncompressed payload
    /// limit instead. The effective value is clamped to the V2 series API limit of 5,242,880 bytes, so larger configured
    /// values do not allow payloads that intake would reject. This limit protects the encoder before compression, so
    /// compressed payload size may still force a separate flush. Values smaller than the minimum endpoint framing size
    /// prevent the request builder from starting.
    max_series_uncompressed_payload_size: usize,

    /// Maximum number of data points, across all series, to encode into a single series request payload.
    ///
    /// This applies only to series metrics (counters, gauges, rates, sets) and not to sketch metrics (histograms,
    /// distributions). A single metric series may contribute multiple data points when it carries more than one
    /// timestamp/value pair. When encoding an input would cause the running data point total to exceed this limit, the
    /// current payload is flushed first and the input is placed in the next payload.
    max_series_points_per_payload: usize,

    /// Flush timeout for pending requests.
    ///
    /// When the destination has written metrics to the in-flight request payload, but it has not yet reached the
    /// payload size limits that would force the payload to be flushed, the destination will wait for a period of time
    /// before flushing the in-flight request payload. This allows for the possibility of other events to be processed
    /// and written into the request payload, thereby maximizing the payload size and reducing the number of requests
    /// generated and sent overall.
    flush_timeout: Duration,

    /// Compression kind to use for the request payloads.
    compressor_kind: String,

    /// Effective zstd compression level for the request payloads.
    zstd_compressor_level: i32,

    /// Whether to use the V2 API for series metrics.
    ///
    /// When `true`, series metrics are sent to the V2 protobuf endpoint (`/api/v2/series`). When `false`, series
    /// metrics are sent to the legacy V1 JSON endpoint (`/api/v1/series`). Sketch metrics always use the V2 endpoint
    /// (`/api/beta/sketches`) regardless of this setting.
    use_v2_series_api: bool,

    /// Whether to log metric payload contents before encoding.
    ///
    /// This logs decoded metric objects, not the encoded JSON/protobuf HTTP body.
    log_payloads: bool,

    /// Additional tags to apply to all forwarded metrics.
    additional_tags: Option<SharedTagSet>,

    /// V3 API configuration for per-endpoint V3 support.
    ///
    /// Configures which endpoints receive V3 payloads.
    v3_api: V3ApiConfig,

    /// Agent-compatible V3 API configuration.
    use_v3_api: UseV3ApiConfig,

    /// Metrics routing to an alternate intake.
    opw_metrics: OpwMetricsConfiguration,

    /// The primary metrics endpoint, as configured and not altered in any way.
    primary_endpoint: String,

    /// Additional endpoints that metrics may be dual-shipped to, keyed by endpoint URL with their API keys.
    additional_endpoints: HashMap<String, Vec<String>>,
}

impl DatadogMetricsConfiguration {
    /// Creates a new `DatadogMetricsConfiguration` from the resolved shared configuration.
    pub fn from_configuration(shared: &SharedConfiguration) -> Self {
        let endpoints = &shared.endpoints;
        let metrics = &shared.metrics_encoding;

        Self {
            max_metrics_per_payload: metrics.max_metrics_per_payload,
            max_payload_size: metrics.max_payload_size,
            max_uncompressed_payload_size: metrics.max_uncompressed_payload_size,
            max_series_payload_size: metrics.max_series_payload_size,
            max_series_uncompressed_payload_size: metrics.max_series_uncompressed_payload_size,
            max_series_points_per_payload: metrics.max_series_points_per_payload,
            flush_timeout: metrics.flush_timeout,
            compressor_kind: endpoints.compression.compressor_kind.clone(),
            zstd_compressor_level: endpoints.compression.effective_zstd_level(),
            use_v2_series_api: metrics.use_v2_series_api,
            log_payloads: metrics.log_payloads,
            additional_tags: None,
            v3_api: (&metrics.v3_api).into(),
            use_v3_api: UseV3ApiConfig { series: metrics.into() },
            opw_metrics: OpwMetricsConfiguration::from_configuration(endpoints),
            primary_endpoint: endpoints.primary_endpoint(),
            additional_endpoints: endpoints.additional_endpoints.clone(),
        }
    }

    /// Sets additional tags to be applied uniformly to all metrics forwarded by this destination.
    pub fn with_additional_tags(mut self, additional_tags: SharedTagSet) -> Self {
        self.additional_tags = Some(additional_tags);
        self
    }

    /// Restricts endpoint-aware protocol selection to a single overridden metrics endpoint.
    ///
    /// This mirrors a forwarder branch that replaces the normal primary endpoint and removes additional and
    /// alternate-intake endpoints, such as Multi-Region Failover.
    pub fn with_metrics_endpoint_override(mut self, dd_url: String) -> Self {
        self.primary_endpoint = dd_url;
        self.additional_endpoints = HashMap::new();
        self.opw_metrics.disable();
        self
    }

    /// Forces series metrics to use V2.
    ///
    /// This is used for local destinations that only accept the V2 series protocol, such as the Cluster Agent.
    pub fn with_v2_series_only(mut self) -> Self {
        self.use_v3_api.series.enabled = V3SeriesMode::Disabled;
        self.use_v3_api.series.endpoints.clear();
        self.opw_metrics.clear_v3_series_overrides();
        self
    }

    fn v3_payload_limits(&self) -> V3PayloadLimits {
        V3PayloadLimits::new(
            self.max_series_payload_size,
            self.max_series_uncompressed_payload_size,
            self.max_metrics_per_payload,
            self.max_series_points_per_payload,
        )
    }

    fn endpoint_v3_settings(
        &self, endpoint: &ResolvedEndpoint, metrics_primary_v3_override: Option<bool>,
    ) -> EndpointV3Settings {
        EndpointV3Settings::from_v3_config(V3EndpointConfig {
            configured_endpoint: endpoint.configured_endpoint(),
            series_config: &self.use_v3_api.series,
            metrics_primary_v3_override,
            serializer_v3_sketches_endpoints: &self.v3_api.sketches.endpoints,
        })
    }

    fn any_series_endpoint_matches(
        &self, mut predicate: impl FnMut(&EndpointV3Settings) -> bool,
    ) -> Result<bool, GenericError> {
        if let Some((metrics_primary_url, metrics_primary_v3_override)) =
            selected_metrics_primary_endpoint(&self.opw_metrics)
        {
            let metrics_primary = ResolvedEndpoint::from_raw_endpoint(metrics_primary_url, "")
                .error_context("Failed parsing/resolving the metrics primary destination endpoint.")?;
            let settings = self.endpoint_v3_settings(&metrics_primary, Some(metrics_primary_v3_override));
            if predicate(&settings) {
                return Ok(true);
            }
        } else {
            let primary = ResolvedEndpoint::from_raw_endpoint(&self.primary_endpoint, "")
                .error_context("Failed parsing/resolving the primary destination endpoint.")?;
            let settings = self.endpoint_v3_settings(&primary, None);
            if predicate(&settings) {
                return Ok(true);
            }
        }

        for endpoint in resolve_additional_endpoints(&self.additional_endpoints)
            .error_context("Failed parsing/resolving the additional destination endpoints.")?
        {
            let settings = self.endpoint_v3_settings(&endpoint, None);
            if predicate(&settings) {
                return Ok(true);
            }
        }

        Ok(false)
    }

    fn requires_v2_series(&self, metrics_v3_disabled_by_compressor: bool) -> Result<bool, GenericError> {
        if !self.use_v2_series_api || metrics_v3_disabled_by_compressor {
            return Ok(true);
        }

        self.any_series_endpoint_matches(|settings| !settings.use_v3_series)
    }

    fn requires_v3_series(&self, metrics_v3_disabled_by_compressor: bool) -> Result<bool, GenericError> {
        if metrics_v3_disabled_by_compressor {
            return Ok(false);
        }

        let metrics_primary_v3_override = selected_metrics_primary_v3_override(&self.opw_metrics);
        if !series_v3_can_be_enabled_for_config(
            self.use_v2_series_api,
            metrics_primary_v3_override,
            !self.additional_endpoints.is_empty(),
            &self.use_v3_api.series,
        ) {
            return Ok(false);
        }

        self.any_series_endpoint_matches(|settings| settings.use_v3_series)
    }
}

#[async_trait]
impl EncoderBuilder for DatadogMetricsConfiguration {
    fn input_event_type(&self) -> EventType {
        EventType::Metric
    }

    fn output_payload_type(&self) -> PayloadType {
        PayloadType::Http
    }

    async fn build(&self, context: BuildContext) -> Result<Box<dyn Encoder + Send>, GenericError> {
        let metrics_builder = MetricsBuilder::from_component_context(context.component_context());
        let telemetry = ComponentTelemetry::from_builder(&metrics_builder);
        let v3_serializer_telemetry = V3SerializerTelemetry::from_builder(&metrics_builder);

        let v2_compression_scheme = CompressionScheme::new(&self.compressor_kind, self.zstd_compressor_level);
        let v3_compression_scheme = if self.v3_api.compression_level > 0 {
            CompressionScheme::new(&self.compressor_kind, self.v3_api.compression_level)
        } else {
            v2_compression_scheme
        };
        let series_endpoint_uri = METRICS_SERIES_V3_PATH.to_string();
        let payload_limits = self.v3_payload_limits();

        let v2_endpoint_config = EndpointConfiguration::new(
            v2_compression_scheme,
            self.max_metrics_per_payload,
            self.max_series_points_per_payload,
            self.additional_tags.clone(),
        );
        let endpoint_config = EndpointConfiguration::new(
            v3_compression_scheme,
            self.max_metrics_per_payload,
            // Actually enforced by V3PayloadLimits, required for the
            // constructor shared between V1/V2/V3.
            usize::MAX,
            self.additional_tags.clone(),
        );

        // Derive the encoding mode for each metric type from the configuration.
        let metrics_v3_disabled_by_compressor = matches!(v3_compression_scheme, CompressionScheme::Zlib(_));
        let use_v3_series = self.requires_v3_series(metrics_v3_disabled_by_compressor)?;
        let series_mode = metrics_encoder_mode_for_config(use_v3_series, metrics_v3_disabled_by_compressor);
        let sketches_mode =
            metrics_encoder_mode_for_config(self.v3_api.use_v3_sketches(), metrics_v3_disabled_by_compressor);
        let series_endpoint = if self.use_v2_series_api {
            MetricsEndpoint::SeriesV2
        } else {
            MetricsEndpoint::SeriesV1
        };
        let v3_runtime_config = V3RuntimeConfig {
            endpoint_config,
            payload_limits,
            series_endpoint_uri,
            serializer_telemetry: v3_serializer_telemetry,
        };
        let generic_payload_limits = clamp_payload_limits(
            self.max_uncompressed_payload_size,
            self.max_payload_size,
            DEFAULT_SERIALIZER_UNCOMPRESSED_SIZE_LIMIT,
            DEFAULT_SERIALIZER_COMPRESSED_SIZE_LIMIT,
        );
        let (series_uncompressed_limit, series_compressed_limit) = if series_endpoint == MetricsEndpoint::SeriesV2 {
            clamp_payload_limits(
                self.max_series_uncompressed_payload_size,
                self.max_series_payload_size,
                v2::SERIES_V2_UNCOMPRESSED_SIZE_LIMIT,
                v2::SERIES_V2_COMPRESSED_SIZE_LIMIT,
            )
        } else {
            generic_payload_limits
        };
        let v2_series_builder = if self.requires_v2_series(metrics_v3_disabled_by_compressor)? {
            let mut builder = v2::create_v2_request_builder(series_endpoint, &v2_endpoint_config)
                .await
                .error_context("Failed to create V2 series request builder.")?;
            builder.with_len_limits(series_uncompressed_limit, series_compressed_limit)?;
            Some(builder)
        } else {
            debug!("All metrics series endpoints use authoritative V3; disabling V2 series encoding.");
            None
        };

        let (sketches_uncompressed_limit, sketches_compressed_limit) = generic_payload_limits;
        let mut v2_sketch_builder = v2::create_v2_request_builder(MetricsEndpoint::Sketches, &v2_endpoint_config)
            .await
            .error_context("Failed to create V2 sketches request builder.")?;
        v2_sketch_builder.with_len_limits(sketches_uncompressed_limit, sketches_compressed_limit)?;
        let v2_sketch_builder = Some(v2_sketch_builder);

        let flush_timeout = if self.flush_timeout.is_zero() {
            // We always give ourselves a minimum flush timeout of 10ms to allow for some very minimal amount of
            // batching, while still practically flushing things almost immediately.
            Duration::from_millis(10)
        } else {
            self.flush_timeout
        };

        if series_mode.needs_v3() || sketches_mode.needs_v3() {
            debug!(
                ?series_mode,
                ?sketches_mode,
                v3_sketches_endpoints = ?self.v3_api.sketches.endpoints,
                "V3 encoding support is enabled."
            );
        }

        Ok(Box::new(DatadogMetrics {
            v2_series_builder,
            v2_sketch_builder,
            series_mode,
            sketches_mode,
            v3_runtime_config,
            telemetry,
            flush_timeout,
            log_payloads: self.log_payloads,
        }))
    }
}

impl MemoryBounds for DatadogMetricsConfiguration {
    fn specify_bounds(&self, builder: &mut MemoryBoundsBuilder) {
        // TODO: How do we properly represent the requests we can generate that may be sitting around in-flight?
        //
        // Theoretically, we'll end up being limited by the size of the downstream forwarder's interconnect, and however
        // many payloads it will buffer internally... so realistically the firm limit boils down to the forwarder itself
        // but we'll have a hard time in the forwarder knowing the maximum size of any given payload being sent in, which
        // then makes it hard to calculate a proper firm bound even though we know the rest of the values required to
        // calculate the firm bound.
        builder
            .minimum()
            .with_single_value::<DatadogMetrics>("component struct")
            .with_array::<EventsBuffer>("request builder events channel", 8)
            .with_array::<PayloadsBuffer>("request builder payloads channel", 8);

        builder
            .firm()
            // Capture the size of the "split re-encode" buffers in the request builders, which is where we keep owned
            // versions of metrics that we encode in case we need to actually re-encode them during a split operation.
            .with_array::<Metric>("series metrics split re-encode buffer", self.max_metrics_per_payload)
            .with_array::<Metric>("sketch metrics split re-encode buffer", self.max_metrics_per_payload);
    }
}

pub struct DatadogMetrics {
    v2_series_builder: Option<RequestBuilder<v2::MetricsEndpointEncoder>>,
    v2_sketch_builder: Option<RequestBuilder<v2::MetricsEndpointEncoder>>,
    series_mode: MetricsEncoderMode,
    sketches_mode: MetricsEncoderMode,
    v3_runtime_config: V3RuntimeConfig,
    telemetry: ComponentTelemetry,
    flush_timeout: Duration,
    log_payloads: bool,
}

struct V3RuntimeConfig {
    endpoint_config: EndpointConfiguration,
    payload_limits: V3PayloadLimits,
    series_endpoint_uri: String,
    serializer_telemetry: V3SerializerTelemetry,
}

#[async_trait]
impl Encoder for DatadogMetrics {
    async fn run(mut self: Box<Self>, mut context: EncoderContext) -> Result<(), GenericError> {
        let Self {
            v2_series_builder,
            v2_sketch_builder,
            series_mode,
            sketches_mode,
            v3_runtime_config,
            telemetry,
            flush_timeout,
            log_payloads,
        } = *self;

        let mut health = context.take_health_handle();

        let (events_tx, events_rx) = mpsc::channel(8);
        let (payloads_tx, mut payloads_rx) = mpsc::channel(8);

        // Run our request builder task on the worker pool.
        //
        // The request builder task ignores the shutdown signal on purpose: it drains its incoming event buffer channel
        // until the channel closes, which is what guarantees every buffered metric is encoded and dispatched.
        let request_builder_fut = run_request_builder(
            v2_series_builder,
            v2_sketch_builder,
            series_mode,
            sketches_mode,
            v3_runtime_config,
            telemetry,
            events_rx,
            payloads_tx,
            flush_timeout,
            log_payloads,
        );
        runtime::worker("request_builder", request_builder_fut)
            .on_runtime(context.topology_context().global_thread_pool().clone())
            .spawn();

        health.mark_ready();
        debug!("Datadog Metrics encoder started.");

        loop {
            select! {
                biased;

                _ = health.live() => continue,
                maybe_payload = payloads_rx.recv() => match maybe_payload {
                    Some(payload) => {
                        if let Err(e) = context.dispatcher().dispatch(payload).await {
                            error!("Failed to dispatch payload: {}", e);
                        }
                    }
                    None => break,
                },
                maybe_event_buffer = context.events().next() => match maybe_event_buffer {
                    Some(event_buffer) => {
                        // Both channels between this task and the request builder task are bounded, and each task is
                        // the producer on one and the consumer on the other. Blocking outright on `events_tx` would
                        // stop us draining `payloads_rx`, which can deadlock: the builder task blocks sending a
                        // payload once `payloads_tx` is full, so it never returns to receive from `events_rx`, so
                        // capacity never frees up here. Keep draining payloads while we wait for capacity so that at
                        // least one of the two channels is always making progress.
                        let permit = loop {
                            select! {
                                biased;

                                permit = events_tx.reserve() => break permit
                                    .error_context("Failed to reserve capacity for event buffer.")?,
                                maybe_payload = payloads_rx.recv() => match maybe_payload {
                                    Some(payload) => if let Err(e) = context.dispatcher().dispatch(payload).await {
                                        error!("Failed to dispatch payload: {}", e);
                                    },

                                    // Our payloads channel is gone, which means our request builder task went away unexpectedly.
                                    None => return Err(generic_error!("Request builder task stopped before accepting event buffer.")),
                                }
                            }
                        };

                        permit.send(event_buffer);
                    }
                    None => break,
                },
            }
        }

        // Drop the events sender, which signals the request builder task to stop.
        drop(events_tx);

        // Continue draining the payloads receiver until it is closed.
        while let Some(payload) = payloads_rx.recv().await {
            if let Err(e) = context.dispatcher().dispatch(payload).await {
                error!("Failed to dispatch payload: {}", e);
            }
        }

        // Draining `payloads_rx` to completion already implies the request builder finished: it owns the only sender,
        // so the channel only closes once that child's future has run to completion (or been dropped).
        debug!("Datadog Metrics encoder stopped.");

        Ok(())
    }
}

/// Logs the decoded contents of a metric prior to encoding.
///
/// This logs the metric object itself, not the encoded JSON/protobuf HTTP body.
fn log_metric_payload(metric: &Metric) {
    match metric.values() {
        MetricValues::Counter(..) | MetricValues::Rate(..) | MetricValues::Gauge(..) | MetricValues::Set(..) => {
            debug!(?metric, "Flushing series metric.")
        }
        MetricValues::Histogram(..) | MetricValues::Distribution(..) => {
            debug!(?metric, "Flushing sketch metric.")
        }
    }
}

#[allow(clippy::too_many_arguments)]
async fn run_request_builder(
    mut v2_series_builder: Option<RequestBuilder<v2::MetricsEndpointEncoder>>,
    mut v2_sketch_builder: Option<RequestBuilder<v2::MetricsEndpointEncoder>>, series_mode: MetricsEncoderMode,
    sketches_mode: MetricsEncoderMode, v3_runtime_config: V3RuntimeConfig, telemetry: ComponentTelemetry,
    mut events_rx: mpsc::Receiver<EventsBuffer>, mut payloads_tx: mpsc::Sender<PayloadsBuffer>,
    flush_timeout: Duration, log_payloads: bool,
) -> Result<(), GenericError> {
    let mut pending_flush = false;
    let pending_flush_timeout = sleep(flush_timeout);
    tokio::pin!(pending_flush_timeout);

    let mut v3_series_metrics = series_mode.needs_v3().then(Vec::<Metric>::new);
    let mut v3_sketch_metrics = sketches_mode.needs_v3().then(Vec::<Metric>::new);
    let mut v3_series_points = 0usize;
    let mut v3_sketch_points = 0usize;
    let mut v3_series_ratio = V3CompressionRatio::default();
    let mut v3_sketch_ratio = V3CompressionRatio::default();

    let tag_series = series_mode.needs_tagging();
    let tag_sketches = sketches_mode.needs_tagging();
    let v3_flush_context = V3FlushContext {
        endpoint_config: &v3_runtime_config.endpoint_config,
        payload_limits: v3_runtime_config.payload_limits,
        series_endpoint_uri: &v3_runtime_config.series_endpoint_uri,
        serializer_telemetry: &v3_runtime_config.serializer_telemetry,
        telemetry: &telemetry,
    };

    loop {
        select! {
            Some(event_buffer) = events_rx.recv() => {
                for event in event_buffer {
                    let metric = match event.try_into_metric() {
                        Some(metric) => metric,
                        None => continue,
                    };

                    if log_payloads {
                        log_metric_payload(&metric);
                    }

                    // A series metric whose points are all non-finite would encode to a series with no points, which
                    // intake rejects as an empty value set. Drop it whole rather than emit an empty series.
                    if !v1::has_emittable_point(&metric) {
                        debug!(metric = %metric.context().name(), "Dropping series metric with no finite points.");
                        telemetry.events_dropped_encoder().increment(1);
                        continue;
                    }

                    // Figure out which endpoint the metric belongs to, and grab the relevant V2 builder/V3 storage.
                    let endpoint = MetricsEndpoint::from_metric(&metric);
                    let (endpoint_mode, maybe_v2_builder, maybe_v3_metrics, v3_points) = match endpoint {
                        MetricsEndpoint::SeriesV1 | MetricsEndpoint::SeriesV2 => (
                            series_mode,
                            &mut v2_series_builder,
                            &mut v3_series_metrics,
                            &mut v3_series_points,
                        ),
                        MetricsEndpoint::Sketches => (
                            sketches_mode,
                            &mut v2_sketch_builder,
                            &mut v3_sketch_metrics,
                            &mut v3_sketch_points,
                        ),
                    };
                    let metric_point_count = metric.values().len();
                    let should_buffer_v3 = endpoint_mode.needs_v3();

                    // Store a copy of the metric in `maybe_v3_metrics` if it's present.
                    //
                    // We have to do this before encoding because `RequestBuilder::encode` consumes the metric. This also means we'll
                    // need to _remove_ the metric if encoding fails.
                    if should_buffer_v3 {
                        if let Some(metrics) = maybe_v3_metrics {
                            metrics.push(metric.clone());
                            *v3_points += metric_point_count;
                        }
                    }

                    // Attempt encoding the metric for V2 if configured.
                    //
                    // If the metric could not be encoded, remove its buffered V3 copy when this endpoint also requires
                    // a V2 encoding.
                    let v2_payload_info = match endpoint {
                        MetricsEndpoint::SeriesV1 | MetricsEndpoint::SeriesV2 => {
                            tag_series.then(MetricsPayloadInfo::v2_series)
                        }
                        MetricsEndpoint::Sketches => tag_sketches.then(MetricsPayloadInfo::v2_sketches),
                    };
                    let _v2_flushed = if let Some(builder) = maybe_v2_builder {
                        let result = encode_v2_metrics(builder, metric, &telemetry, &mut payloads_tx, v2_payload_info).await?;
                        if should_buffer_v3
                            && !result.encoded()
                            && !matches!(endpoint_mode, MetricsEncoderMode::V3Enabled)
                        {
                            if let Some(metrics) = maybe_v3_metrics {
                                let _ = metrics.pop();
                                *v3_points = v3_points.saturating_sub(metric_point_count);
                            }
                        }
                        result.flushed()
                    } else {
                        false
                    };

                    let v3_payload_info = match endpoint {
                        MetricsEndpoint::SeriesV1 | MetricsEndpoint::SeriesV2 => {
                            tag_series.then(MetricsPayloadInfo::v3_series)
                        }
                        MetricsEndpoint::Sketches => tag_sketches.then(MetricsPayloadInfo::v3_sketches),
                    };
                    let mut split_metric = None;
                    let _v3_flushed = if let Some(v3_metrics) = maybe_v3_metrics {
                        let should_flush_v3 = match endpoint_mode {
                            MetricsEncoderMode::V2Only => false,
                            MetricsEncoderMode::V3Enabled => {
                                if v3_flush_context.payload_limits.point_count_fits(metric_point_count)
                                    && v3_flush_context.payload_limits.point_count_exceeds_limit(*v3_points)
                                    && v3_metrics.len() > 1
                                {
                                    v3_flush_context
                                        .serializer_telemetry
                                        .record_split_reason(V3PayloadSplitReason::MaxPoints);
                                    if let Some(metric) = v3_metrics.pop() {
                                        *v3_points = v3_points.saturating_sub(metric.values().len());
                                        split_metric = Some(metric);
                                    }
                                    true
                                } else {
                                    v3_flush_context.payload_limits.should_flush_point_count_limit(*v3_points)
                                        || v3_flush_context
                                            .payload_limits
                                            .should_flush_metric_count_limit(v3_metrics)
                                }
                            }
                        };
                        if should_flush_v3 {
                            encode_and_flush_v3_metrics(
                                endpoint,
                                v3_flush_context,
                                v3_metrics,
                                &mut v3_series_ratio,
                                &mut v3_sketch_ratio,
                                &mut payloads_tx,
                                v3_payload_info,
                            )
                            .await?;
                            *v3_points = 0;
                            true
                        } else {
                            false
                        }
                    } else {
                        false
                    };

                    if matches!(endpoint_mode, MetricsEncoderMode::V3Enabled) {
                        if let Some(m) = split_metric.take() {
                            let point_count = m.values().len();
                            if let Some(metrics) = maybe_v3_metrics {
                                metrics.push(m);
                                *v3_points += point_count;
                            }
                        }
                    }
                }

                debug!("Processed event buffer.");

                // If we're not already pending a flush, we'll start the countdown.
                if !pending_flush {
                    pending_flush_timeout.as_mut().reset(tokio::time::Instant::now() + flush_timeout);
                    pending_flush = true;
                }
            },
            _ = &mut pending_flush_timeout, if pending_flush => {
                debug!("Flushing pending request(s).");

                pending_flush = false;

                // Flush any pending series metrics.
                let v2_series_payload_info = tag_series.then(MetricsPayloadInfo::v2_series);
                let mut v2_series_flush_succeeded = true;
                if let Some(builder) = &mut v2_series_builder {
                    if let Err(e) = flush_v2_metrics(builder, &mut payloads_tx, v2_series_payload_info).await {
                        error!(error = %e, "Failed to flush V2 series metrics: {}", e);
                        v2_series_flush_succeeded = false;
                    }
                }

                let v3_series_payload_info = tag_series.then(MetricsPayloadInfo::v3_series);
                if let Some(metrics) = &mut v3_series_metrics {
                    if v2_series_flush_succeeded || matches!(series_mode, MetricsEncoderMode::V3Enabled) {
                        if let Err(e) = encode_and_flush_v3_series_metrics(
                            v3_flush_context,
                            metrics,
                            &mut v3_series_ratio,
                            &mut payloads_tx,
                            v3_series_payload_info,
                        )
                        .await
                        {
                            error!(error = %e, "Failed to flush V3 series metrics: {}", e);
                        }
                        v3_series_points = 0;
                    } else {
                        warn!("Failed to flush V2 series metrics, skipping V3 series flush.");
                        metrics.clear();
                        v3_series_points = 0;
                    }
                }

                // Flush any pending sketch metrics.
                let v2_sketches_payload_info = tag_sketches.then(MetricsPayloadInfo::v2_sketches);
                let mut v2_sketches_flush_succeeded = true;
                if let Some(builder) = &mut v2_sketch_builder {
                    if let Err(e) = flush_v2_metrics(builder, &mut payloads_tx, v2_sketches_payload_info).await {
                        error!(error = %e, "Failed to flush V2 sketch metrics: {}", e);
                        v2_sketches_flush_succeeded = false;
                    }
                }

                let v3_sketches_payload_info = tag_sketches.then(MetricsPayloadInfo::v3_sketches);
                if let Some(metrics) = &mut v3_sketch_metrics {
                    if v2_sketches_flush_succeeded || matches!(sketches_mode, MetricsEncoderMode::V3Enabled) {
                        if let Err(e) = encode_and_flush_v3_sketch_metrics(
                            v3_flush_context,
                            metrics,
                            &mut v3_sketch_ratio,
                            &mut payloads_tx,
                            v3_sketches_payload_info,
                        )
                        .await
                        {
                            error!(error = %e, "Failed to flush V3 sketch metrics: {}", e);
                        }
                        v3_sketch_points = 0;
                    } else {
                        warn!("Failed to flush V2 sketch metrics, skipping V3 sketch flush.");
                        metrics.clear();
                        v3_sketch_points = 0;
                    }
                }

                debug!("All flushed requests sent to I/O task. Waiting for next event buffer...");
            },

            // Event buffers channel has been closed, and we have no pending flushing, so we're all done.
            else => break,
        }
    }

    Ok(())
}

struct EncodeResult {
    encoded: bool,
    flushed: bool,
}

impl EncodeResult {
    pub const fn new(encoded: bool, flushed: bool) -> Self {
        Self { encoded, flushed }
    }

    pub const fn encoded(&self) -> bool {
        self.encoded
    }

    pub const fn flushed(&self) -> bool {
        self.flushed
    }
}

async fn encode_v2_metrics(
    request_builder: &mut RequestBuilder<v2::MetricsEndpointEncoder>, metric: Metric, telemetry: &ComponentTelemetry,
    payloads_tx: &mut mpsc::Sender<Payload>, payload_info: Option<MetricsPayloadInfo>,
) -> Result<EncodeResult, GenericError> {
    // Encode the metric. If we get it back, that means the current request is full, and we need to
    // flush it before we can try to encode the metric again... so we'll hold on to it in that case
    // before flushing and trying to encode it again.
    let metric_to_retry = match request_builder.encode(metric).await {
        Ok(None) => return Ok(EncodeResult::new(true, false)),
        Ok(Some(metric)) => metric,
        Err(RequestBuilderError::InvalidInput { input }) => {
            debug!(metric_name = %input.context().name(), "Dropping metric with no emittable values.");
            telemetry.events_dropped_encoder().increment(1);
            return Ok(EncodeResult::new(false, false));
        }
        Err(e) => {
            error!(error = %e, "Failed to encode metric.");
            telemetry.events_dropped_encoder().increment(1);
            return Ok(EncodeResult::new(false, false));
        }
    };

    flush_v2_metrics(request_builder, payloads_tx, payload_info).await?;

    // Now try to encode the metric again. If it fails again, we'll just log it because it shouldn't
    // be possible to fail at this point, otherwise we would have already caught that the first
    // time.
    match request_builder.encode(metric_to_retry).await {
        Ok(None) => Ok(EncodeResult::new(true, true)),
        Ok(Some(_)) => unreachable!(
            "failure to encode due to size should never occur after flush for metrics which aren't unencodable"
        ),
        Err(e) => {
            error!(error = %e, "Failed to encode metric.");
            telemetry.events_dropped_encoder().increment(1);
            Ok(EncodeResult::new(false, true))
        }
    }
}

async fn flush_v2_metrics(
    request_builder: &mut RequestBuilder<MetricsEndpointEncoder>, payloads_tx: &mut mpsc::Sender<Payload>,
    payload_info: Option<MetricsPayloadInfo>,
) -> Result<usize, GenericError> {
    let mut requests_flushed = 0;

    let maybe_requests = request_builder.flush().await;
    for maybe_request in maybe_requests {
        match maybe_request {
            Ok((events, data_points, request)) => {
                requests_flushed += 1;

                flush_payload(request, events, data_points, payloads_tx, payload_info).await?;
            }

            // TODO: Increment a counter here that metrics were dropped due to a flush failure.
            Err(e) => {
                if !e.is_recoverable() {
                    return Err(GenericError::from(e).context("Failed to flush request."));
                }
            }
        }
    }

    Ok(requests_flushed)
}

#[derive(Clone, Copy)]
struct V3FlushContext<'a> {
    endpoint_config: &'a EndpointConfiguration,
    payload_limits: V3PayloadLimits,
    series_endpoint_uri: &'a str,
    serializer_telemetry: &'a V3SerializerTelemetry,
    telemetry: &'a ComponentTelemetry,
}

#[allow(clippy::too_many_arguments)]
async fn encode_and_flush_v3_metrics(
    endpoint: MetricsEndpoint, context: V3FlushContext<'_>, metrics: &mut Vec<Metric>,
    series_ratio: &mut V3CompressionRatio, sketches_ratio: &mut V3CompressionRatio,
    payloads_tx: &mut mpsc::Sender<Payload>, payload_info: Option<MetricsPayloadInfo>,
) -> Result<(), GenericError> {
    match endpoint {
        MetricsEndpoint::SeriesV1 | MetricsEndpoint::SeriesV2 => {
            encode_and_flush_v3_series_metrics(context, metrics, series_ratio, payloads_tx, payload_info).await
        }
        MetricsEndpoint::Sketches => {
            encode_and_flush_v3_sketch_metrics(context, metrics, sketches_ratio, payloads_tx, payload_info).await
        }
    }
}

async fn encode_and_flush_v3_series_metrics(
    context: V3FlushContext<'_>, metrics: &mut Vec<Metric>, ratio: &mut V3CompressionRatio,
    payloads_tx: &mut mpsc::Sender<Payload>, payload_info: Option<MetricsPayloadInfo>,
) -> Result<(), GenericError> {
    if metrics.is_empty() {
        return Ok(());
    }
    let metrics_to_flush = std::mem::take(metrics);

    encode_and_flush_v3_payload_requests(
        context.series_endpoint_uri,
        &metrics_to_flush,
        context,
        "series",
        ratio,
        payloads_tx,
        payload_info,
    )
    .await
}

async fn encode_and_flush_v3_sketch_metrics(
    context: V3FlushContext<'_>, metrics: &mut Vec<Metric>, ratio: &mut V3CompressionRatio,
    payloads_tx: &mut mpsc::Sender<Payload>, payload_info: Option<MetricsPayloadInfo>,
) -> Result<(), GenericError> {
    if metrics.is_empty() {
        return Ok(());
    }
    let metrics_to_flush = std::mem::take(metrics);

    encode_and_flush_v3_payload_requests(
        V3_SKETCHES_ENDPOINT_URI,
        &metrics_to_flush,
        context,
        "sketches",
        ratio,
        payloads_tx,
        payload_info,
    )
    .await
}

/// Smallest compression ratio used when translating the compressed size limit into an uncompressed byte target.
///
/// Guards against dividing by a pathologically small ratio and producing an enormous target.
const V3_MIN_COMPRESSION_RATIO: f64 = 0.02;

/// Fraction of the derived byte target that a batch is actually cut at.
///
/// Leaves headroom so that ordinary variation between batches does not push the finalized payload over the real limit
/// and force the expensive re-encode path.
const V3_BATCH_TARGET_MARGIN: f64 = 0.9;

/// Weight given to the newest observation when updating the compression ratio.
const V3_RATIO_SMOOTHING: f64 = 0.25;

/// Tracks the compression ratio observed on emitted V3 payloads, so batches can be sized against the compressed limit.
///
/// V3 payloads are columnar and only compressed once the whole batch is finalized, so unlike the V2 request builder
/// there is no live compressor to consult with a [`CompressionEstimator`] while batching. Instead we remember how well
/// recent payloads compressed and use that to turn the compressed size limit (which is the limit that actually binds
/// for metrics intake) into an uncompressed byte target for the next batch.
#[derive(Default)]
struct V3CompressionRatio {
    ratio: Option<f64>,
}

impl V3CompressionRatio {
    /// Returns the uncompressed byte target that a batch should be cut at.
    fn uncompressed_target(&self, limits: V3PayloadLimits) -> usize {
        // Until a payload has been observed, assume the data will not compress at all. That errs towards batches that
        // are too small, which costs an extra payload, rather than too large, which costs a full re-encode.
        let ratio = self.ratio.unwrap_or(1.0).clamp(V3_MIN_COMPRESSION_RATIO, 1.0);
        let target_from_compressed_limit = (limits.max_compressed_size as f64 / ratio) as usize;
        let target = target_from_compressed_limit.min(limits.max_uncompressed_size);

        // Always leave room for at least one metric, however small the limits are.
        (((target as f64) * V3_BATCH_TARGET_MARGIN) as usize).max(1)
    }

    /// Records the uncompressed and compressed sizes of an emitted payload.
    fn record(&mut self, uncompressed_len: usize, compressed_len: usize) {
        if uncompressed_len == 0 {
            return;
        }

        let observed = compressed_len as f64 / uncompressed_len as f64;
        self.ratio = Some(match self.ratio {
            Some(current) => (current * (1.0 - V3_RATIO_SMOOTHING)) + (observed * V3_RATIO_SMOOTHING),
            None => observed,
        });
    }
}

/// A contiguous run of metrics encoded into a single V3 payload.
struct V3Batch {
    encoded: V3EncodedMetrics,
    event_count: usize,
    data_point_count: usize,
}

/// Encodes metrics starting at `*idx` into a single V3 payload, stopping once a payload limit or
/// `target_uncompressed_len` is reached, and advances `*idx` past everything consumed.
///
/// Sizing the batch while encoding it is what keeps this to a single encode pass. The alternative (encoding a large
/// batch and halving it whenever the result is too big) re-encodes and re-compresses the same metrics once per level of
/// splitting, which is dominated by dictionary rebuilding on high-cardinality data
fn build_v3_batch(
    metrics: &[Metric], idx: &mut usize, context: V3FlushContext<'_>, payload_kind: &'static str,
    target_uncompressed_len: usize,
) -> Option<V3Batch> {
    let mut writer = V3Writer::new();
    let mut tags_deduplicator = ReusableDeduplicator::new();
    let additional_tags = context.endpoint_config.additional_tags();
    let mut data_point_count = 0usize;
    let mut cut_for_size = false;

    while *idx < metrics.len() {
        let metric = &metrics[*idx];

        if !metric_has_emittable_values(metric) {
            debug!(metric_name = %metric.context().name(), "Dropping metric with no emittable values.");
            context.telemetry.events_dropped_encoder().increment(1);
            *idx += 1;
            continue;
        }

        let metric_points = metric.values().len();
        if !context.payload_limits.point_count_fits(metric_points) {
            // This metric exceeds the point limit on its own, so it cannot fit in any payload.
            context.serializer_telemetry.record_item_too_big();
            context
                .serializer_telemetry
                .record_split_reason(V3PayloadSplitReason::ItemTooBig);
            warn!(
                payload_kind,
                data_points = metric_points,
                point_limit = context.payload_limits.max_points_per_payload,
                "Dropping oversized V3 metric that exceeds the point-count limit."
            );
            context.telemetry.events_dropped_encoder().increment(1);
            *idx += 1;
            continue;
        }

        // Limits that have to be honoured before the metric goes in. The first metric of a batch is always admitted,
        // since it has already been checked to fit by itself.
        if writer.metric_count() > 0 {
            if !context
                .payload_limits
                .point_count_fits(data_point_count + metric_points)
            {
                context
                    .serializer_telemetry
                    .record_split_reason(V3PayloadSplitReason::MaxPoints);
                break;
            }

            if context.payload_limits.metric_count_reached(writer.metric_count()) {
                break;
            }
        }

        write_metric_to_v3(&mut writer, metric, additional_tags, &mut tags_deduplicator);
        data_point_count += metric_points;
        *idx += 1;

        // The size contribution of a metric is only knowable once it is written, since dictionary deduplication makes
        // it depend on what came before. Cutting after the fact can overshoot by one metric, which the target's
        // headroom absorbs; the finalized payload is checked against the real limits regardless.
        if writer.estimated_uncompressed_len() >= target_uncompressed_len {
            cut_for_size = true;
            break;
        }
    }

    // Only count this as a payload boundary if metrics actually remain: a batch that ends on the last metric was not
    // split by anything.
    if cut_for_size && *idx < metrics.len() {
        context
            .serializer_telemetry
            .record_split_reason(V3PayloadSplitReason::PayloadFull);
    }

    let event_count = writer.metric_count();
    if event_count == 0 {
        return None;
    }

    match writer.finalize() {
        Ok(encoded) => Some(V3Batch {
            encoded,
            event_count,
            data_point_count,
        }),
        Err(e) => {
            error!(error = %e, payload_kind, events = event_count, "Failed to encode V3 metrics payload request.");
            context.telemetry.events_dropped_encoder().increment(event_count as u64);
            None
        }
    }
}

/// Turns an encoded batch into a payload request, returning `None` if it does not fit the payload limits.
async fn create_v3_batch_request(
    endpoint_uri: &str, batch: V3Batch, context: V3FlushContext<'_>, payload_kind: &'static str,
    ratio: &mut V3CompressionRatio,
) -> Option<V3PayloadRequest> {
    let V3Batch {
        encoded,
        event_count,
        data_point_count,
    } = batch;

    let mut encoded_request =
        match create_v3_request(endpoint_uri, encoded, context.endpoint_config.compression_scheme()).await {
            Ok(request) => request,
            Err(e) => {
                error!(error = %e, payload_kind, events = event_count, "Failed to create V3 metrics request.");
                context.telemetry.events_dropped_encoder().increment(event_count as u64);
                return None;
            }
        };

    // Feed the observation back even when the request is too big: an oversized batch is exactly the case where the
    // target needs correcting.
    ratio.record(encoded_request.uncompressed_len, encoded_request.compressed_len);

    if !context.payload_limits.request_fits(&encoded_request) {
        return None;
    }

    // Per-column compressed sizes are measured only for requests we actually emit. Measuring them eagerly in
    // `create_v3_request` would compress every column of every oversized attempt that gets discarded, which is pure
    // waste.
    if let Err(e) =
        measure_v3_column_compressed_sizes(&mut encoded_request.stats, context.endpoint_config.compression_scheme())
            .await
    {
        error!(error = %e, payload_kind, "Failed to measure V3 column compressed sizes.");
    } else {
        record_v3_serializer_stats(context.serializer_telemetry, &encoded_request.stats);
    }

    Some(V3PayloadRequest {
        request: encoded_request.request,
        event_count,
        data_point_count,
    })
}

/// Re-encodes an oversized range as progressively smaller ranges until each one fits.
///
/// This is the fallback for a batch that was sized against the byte target but still exceeded the real payload limits.
/// It re-encodes and re-compresses at every level of splitting, so it is deliberately only reached when the target
/// guessed wrong.
async fn split_and_encode_oversized_v3_range(
    endpoint_uri: &str, metrics: &[Metric], range: Range<usize>, context: V3FlushContext<'_>,
    payload_kind: &'static str, ratio: &mut V3CompressionRatio,
) -> Vec<V3PayloadRequest> {
    // `build_v3_batch` advances past metrics it drops. Preserve that decision when the accepted batch has to be
    // re-encoded: retrying the raw source range would otherwise resurrect those metrics.
    let accepted_metrics = metrics[range]
        .iter()
        .filter(|metric| {
            metric_has_emittable_values(metric) && context.payload_limits.point_count_fits(metric.values().len())
        })
        .collect::<Vec<_>>();

    let mut requests = Vec::new();
    let mut pending_ranges = VecDeque::new();
    pending_ranges.push_back(0..accepted_metrics.len());

    while let Some(range) = pending_ranges.pop_front() {
        if range.is_empty() {
            continue;
        }

        let metrics_in_range = &accepted_metrics[range.clone()];
        let event_count = metrics_in_range.len();
        let data_point_count = metrics_in_range.iter().map(|metric| metric.values().len()).sum();

        let encoded = match encode_v3_metrics_batch(
            metrics_in_range.iter().copied(),
            context.endpoint_config.additional_tags(),
        ) {
            Ok(encoded) => encoded,
            Err(e) => {
                error!(error = %e, payload_kind, events = event_count, "Failed to encode V3 metrics payload request.");
                context.telemetry.events_dropped_encoder().increment(event_count as u64);
                continue;
            }
        };
        let batch = V3Batch {
            encoded,
            event_count,
            data_point_count,
        };

        if let Some(request) = create_v3_batch_request(endpoint_uri, batch, context, payload_kind, ratio).await {
            requests.push(request);
            continue;
        }

        if range.len() == 1 {
            // The encoded request is too large and this range cannot be split any further.
            context.serializer_telemetry.record_item_too_big();
            context
                .serializer_telemetry
                .record_split_reason(V3PayloadSplitReason::ItemTooBig);
            warn!(
                payload_kind,
                compressed_limit = context.payload_limits.max_compressed_size,
                uncompressed_limit = context.payload_limits.max_uncompressed_size,
                "Dropping oversized V3 metric that cannot be split further."
            );
            context.telemetry.events_dropped_encoder().increment(1);
            continue;
        }

        // Retry this oversized range as two smaller ranges, preserving the original metric order.
        context
            .serializer_telemetry
            .record_split_reason(V3PayloadSplitReason::PayloadFull);
        let pivot = range.start + range.len() / 2;
        pending_ranges.push_front(pivot..range.end);
        pending_ranges.push_front(range.start..pivot);
    }

    requests
}

#[allow(clippy::too_many_arguments)]
async fn encode_and_flush_v3_payload_requests(
    endpoint_uri: &str, metrics: &[Metric], context: V3FlushContext<'_>, payload_kind: &'static str,
    ratio: &mut V3CompressionRatio, payloads_tx: &mut mpsc::Sender<Payload>, payload_info: Option<MetricsPayloadInfo>,
) -> Result<(), GenericError> {
    let target_uncompressed_len = ratio.uncompressed_target(context.payload_limits);
    let mut idx = 0;

    while idx < metrics.len() {
        let batch_start = idx;
        let Some(batch) = build_v3_batch(metrics, &mut idx, context, payload_kind, target_uncompressed_len) else {
            // `build_v3_batch` always consumes at least one metric when one is available, but guard against spinning
            // on a batch that somehow produced nothing.
            if idx == batch_start {
                error!(
                    payload_kind,
                    "V3 batching made no progress; dropping remaining metrics."
                );
                context
                    .telemetry
                    .events_dropped_encoder()
                    .increment((metrics.len() - idx) as u64);
                break;
            }
            continue;
        };

        let requests = match create_v3_batch_request(endpoint_uri, batch, context, payload_kind, ratio).await {
            Some(request) => vec![request],
            // The batch we sized against the byte target still came out over the real limits, so fall back to
            // re-encoding it as progressively smaller ranges. The target self-corrects from the ratio we record on
            // every emitted payload, so this should be rare.
            None => {
                split_and_encode_oversized_v3_range(
                    endpoint_uri,
                    metrics,
                    batch_start..idx,
                    context,
                    payload_kind,
                    ratio,
                )
                .await
            }
        };

        for payload_request in requests {
            flush_payload(
                payload_request.request,
                payload_request.event_count,
                payload_request.data_point_count,
                payloads_tx,
                payload_info,
            )
            .await?;
            debug!(
                payload_kind,
                events = payload_request.event_count,
                data_points = payload_request.data_point_count,
                "Sent V3 payload."
            );
        }
    }

    Ok(())
}

fn record_v3_serializer_stats(telemetry: &V3SerializerTelemetry, stats: &V3EncoderStats) {
    telemetry.record_values_count(stats.value_encoding_stats);

    for column in &stats.columns {
        let uncompressed_size = column.bytes.len() as u64;
        let compressed_size = column.compressed_len as u64;
        telemetry.record_column_size(column.field_number, uncompressed_size, compressed_size);
    }
}

/// Measures the compressed size of each V3 column, for telemetry purposes.
///
/// This is deliberately separate from building the request: it is only worth paying for columns belonging to a request
/// that will actually be sent, since oversized requests are discarded and re-encoded as smaller ranges.
async fn measure_v3_column_compressed_sizes(
    stats: &mut V3EncoderStats, compression_scheme: CompressionScheme,
) -> Result<(), GenericError> {
    for column in &mut stats.columns {
        column.compressed_len = compressed_v3_len(&column.bytes, compression_scheme)
            .await
            .error_context("Failed to measure V3 column compressed size.")?;
    }

    Ok(())
}

async fn compressed_v3_len(bytes: &[u8], compression_scheme: CompressionScheme) -> Result<usize, GenericError> {
    if matches!(compression_scheme, CompressionScheme::Noop) {
        return Ok(bytes.len());
    }

    let buffer = ChunkedBytesBuffer::new(RB_BUFFER_CHUNK_SIZE);
    let mut compressor = Compressor::from_scheme(compression_scheme, buffer);
    compressor
        .write_all(bytes)
        .await
        .error_context("Failed to compress V3 bytes.")?;
    compressor
        .flush()
        .await
        .error_context("Failed to flush V3 compressor.")?;
    compressor
        .shutdown()
        .await
        .error_context("Failed to shutdown V3 compressor.")?;

    Ok(compressor.into_inner().freeze().len())
}

async fn flush_payload(
    request: Request<FrozenChunkedBytesBuffer>, event_count: usize, data_point_count: usize,
    payloads_tx: &mut mpsc::Sender<Payload>, payload_info: Option<MetricsPayloadInfo>,
) -> Result<(), GenericError> {
    let mut payload_meta = PayloadMetadata::from_event_and_data_point_count(event_count, data_point_count);
    if let Some(info) = payload_info {
        payload_meta = payload_meta.with(info);
    }
    let http_payload = HttpPayload::new(payload_meta, request);
    let payload = Payload::Http(http_payload);

    payloads_tx
        .send(payload)
        .await
        .error_context("Failed to send payload.")?;

    Ok(())
}

// Encodes a batch of metrics to V3 columnar format.
fn encode_v3_metrics_batch<'a>(
    metrics: impl IntoIterator<Item = &'a Metric>, additional_tags: &SharedTagSet,
) -> Result<V3EncodedMetrics, GenericError> {
    let mut writer = V3Writer::new();
    let mut tags_deduplicator = ReusableDeduplicator::new();

    for metric in metrics {
        write_metric_to_v3(&mut writer, metric, additional_tags, &mut tags_deduplicator);
    }

    writer
        .finalize()
        .map_err(|e| generic_error!("Failed to serialize V3 payload: {}", e))
}

/// Writes a single metric to the V3 writer.
pub(super) fn sketch_has_emittable_values(sketch: &DDSketch) -> bool {
    !sketch.is_empty() || !sketch.bins().is_empty()
}

pub(super) fn metric_has_emittable_values(metric: &Metric) -> bool {
    match metric.values() {
        MetricValues::Counter(points) | MetricValues::Rate(points, _) | MetricValues::Gauge(points) => {
            points.into_iter().any(|(_, value)| emittable_scalar_point(value))
        }
        MetricValues::Set(points) => points.into_iter().any(|(_, value)| emittable_scalar_point(value)),
        MetricValues::Distribution(sketches) => sketches
            .into_iter()
            .any(|(_, sketch)| sketch_has_emittable_values(sketch)),
        MetricValues::Histogram(points) => points.into_iter().any(|(_, histogram)| !histogram.samples().is_empty()),
    }
}

fn write_metric_to_v3(
    writer: &mut V3Writer, metric: &Metric, additional_tags: &SharedTagSet,
    tags_deduplicator: &mut ReusableDeduplicator<Tag>,
) {
    if !metric_has_emittable_values(metric) {
        debug!(metric_name = %metric.context().name(), "Dropping metric with no emittable values.");
        return;
    }

    let metric_type = match metric.values() {
        MetricValues::Counter(..) => V3MetricType::Count,
        MetricValues::Rate(..) => V3MetricType::Rate,
        MetricValues::Gauge(..) | MetricValues::Set(..) => V3MetricType::Gauge,
        MetricValues::Histogram(..) | MetricValues::Distribution(..) => V3MetricType::Sketch,
    };
    let is_sketch = metric_type == V3MetricType::Sketch;

    let mut builder = writer.write(metric_type, metric.context().name());

    // Tags - chain instrumented + additional + origin tags
    let chained_tags = metric
        .context()
        .tags()
        .into_iter()
        .chain(additional_tags)
        .chain(metric.context().origin_tags());
    let all_tags = tags_deduplicator
        .deduplicated(chained_tags)
        .filter(|t| is_sketch || !is_v3_series_resource_tag(t) && !is_v3_series_device_tag(t))
        .map(|t| t.as_str());
    builder.set_tags(all_tags);

    // Resources - extract host and, for series, promoted resource tags.
    let mut resources = Vec::new();
    if let Some(host) = metric.context().host().filter(|host| !host.is_empty()) {
        resources.push(("host", host));
    }
    if !is_sketch {
        let mut device_resource = None;
        let chained_tags = metric
            .context()
            .origin_tags()
            .into_iter()
            .chain(metric.context().tags())
            .chain(additional_tags);
        for tag in tags_deduplicator.deduplicated(chained_tags) {
            if is_v3_series_device_tag(tag) {
                device_resource = tag.value().filter(|device| !device.is_empty());
            } else if is_v3_series_resource_tag(tag) {
                if let Some((rtype, rname)) = tag.value().and_then(|value| value.split_once(':')) {
                    if !rtype.is_empty() && !rname.is_empty() {
                        resources.push((rtype, rname));
                    }
                }
            }
        }
        if let Some(device) = device_resource {
            let device_idx = usize::from(metric.context().host().is_some_and(|host| !host.is_empty()));
            resources.insert(device_idx, ("device", device));
        }
    }
    builder.set_resources(&resources);

    // Origin metadata
    if let Some(origin) = metric.metadata().origin() {
        match origin {
            MetricOrigin::SourceType(source_type) => {
                builder.set_source_type(source_type.as_ref());
            }
            MetricOrigin::OriginMetadata {
                product,
                subproduct,
                product_detail,
            } => {
                builder.set_origin(*product, *subproduct, *product_detail, false);
            }
        }
    }

    if metric_type != V3MetricType::Sketch {
        if let Some(unit) = metric.metadata().unit() {
            builder.set_unit(unit);
        }
    }

    // Points based on metric type
    match metric.values() {
        MetricValues::Counter(points) | MetricValues::Gauge(points) => {
            for (ts, val) in points {
                if !emittable_scalar_point(val) {
                    continue;
                }
                let timestamp = ts.map(|t| t.get() as i64).unwrap_or(0);
                builder.add_point(timestamp, val);
            }
        }
        MetricValues::Rate(points, interval) => {
            builder.set_interval(interval.as_secs());
            for (ts, val) in points {
                // Scale by interval as done in V2. A zero interval is an unnormalized rate.
                let scaled = if interval.is_zero() {
                    val
                } else {
                    val / interval.as_secs_f64()
                };
                if !emittable_scalar_point(scaled) {
                    continue;
                }
                let timestamp = ts.map(|t| t.get() as i64).unwrap_or(0);
                builder.add_point(timestamp, scaled);
            }
        }
        MetricValues::Set(points) => {
            // Set values are already converted to count in the iterator
            for (ts, count) in points {
                if !emittable_scalar_point(count) {
                    continue;
                }
                let timestamp = ts.map(|t| t.get() as i64).unwrap_or(0);
                builder.add_point(timestamp, count);
            }
        }
        MetricValues::Distribution(sketches) => {
            for (ts, sketch) in sketches {
                let timestamp = ts.map(|t| t.get() as i64).unwrap_or(0);
                if sketch_has_emittable_values(sketch) {
                    let bin_keys: Vec<i32> = sketch.bins().iter().map(|b| b.key()).collect();
                    let bin_counts: Vec<u32> = sketch.bins().iter().map(|b| b.count()).collect();
                    builder.add_sketch(
                        timestamp,
                        sketch.count() as i64,
                        sketch.stored_sum(),
                        sketch.stored_min(),
                        sketch.stored_max(),
                        &bin_keys,
                        &bin_counts,
                    );
                }
            }
        }
        MetricValues::Histogram(histograms) => {
            for (ts, histogram) in histograms {
                let timestamp = ts.map(|t| t.get() as i64).unwrap_or(0);
                // Convert histogram to DDSketch
                let mut sketch = DDSketch::default();
                for sample in histogram.samples() {
                    sketch.insert_n(sample.value.into_inner(), sample.weight.0 as u64);
                }
                if !sketch.is_empty() {
                    let bin_keys: Vec<i32> = sketch.bins().iter().map(|b| b.key()).collect();
                    let bin_counts: Vec<u32> = sketch.bins().iter().map(|b| b.count()).collect();
                    builder.add_sketch(
                        timestamp,
                        sketch.count() as i64,
                        sketch.sum().unwrap_or(0.0),
                        sketch.min().unwrap_or(0.0),
                        sketch.max().unwrap_or(0.0),
                        &bin_keys,
                        &bin_counts,
                    );
                }
            }
        }
    }

    builder.close();
}

#[inline]
fn emittable_scalar_point(point: f64) -> bool {
    point.is_finite()
}

fn is_v3_series_device_tag(tag: &Tag) -> bool {
    tag.name() == "device" && tag.value().is_some()
}

fn is_v3_series_resource_tag(tag: &Tag) -> bool {
    tag.name() == "dd.internal.resource" && tag.value().is_some()
}

/// Creates a V3 HTTP request from encoded payload data.
async fn create_v3_request(
    endpoint_uri: &str, encoded: V3EncodedMetrics, compression_scheme: CompressionScheme,
) -> Result<V3EncodedRequest, GenericError> {
    // Keep the wire payload as one continuous compressed stream. Per-column compressed sizes are measured
    // independently for telemetry.
    let mut header_buf = [0; 16];
    let header_len = {
        let mut header_writer = CodedOutputStream::bytes(&mut header_buf);
        header_writer.write_tag(3, WireType::LengthDelimited)?;
        header_writer.write_uint64_no_tag(encoded.payload.len() as u64)?;
        header_writer.flush()?;
        header_writer.total_bytes_written() as usize
    };

    let uncompressed_len = header_len + encoded.payload.len();

    let buffer = ChunkedBytesBuffer::new(RB_BUFFER_CHUNK_SIZE);
    let mut compressor = Compressor::from_scheme(compression_scheme, buffer);
    compressor
        .write_all(&header_buf[..header_len])
        .await
        .error_context("Failed to compress V3 payload.")?;
    compressor
        .write_all(&encoded.payload)
        .await
        .error_context("Failed to compress V3 payload.")?;
    compressor
        .flush()
        .await
        .error_context("Failed to flush V3 compressor.")?;
    compressor
        .shutdown()
        .await
        .error_context("Failed to shutdown V3 compressor.")?;

    let compressed_buf = compressor.into_inner().freeze();
    let compressed_len = compressed_buf.len();

    let mut builder = Request::builder()
        .method(Method::POST)
        .uri(endpoint_uri)
        .header(http::header::CONTENT_TYPE, "application/x-protobuf");

    if let Some(encoding) = content_encoding_for_scheme(compression_scheme) {
        builder = builder.header(http::header::CONTENT_ENCODING, encoding);
    }

    let request = builder
        .body(compressed_buf)
        .map_err(|e| generic_error!("Failed to build V3 request: {}", e))?;

    Ok(V3EncodedRequest {
        request,
        compressed_len,
        uncompressed_len,
        stats: encoded.stats,
    })
}

fn content_encoding_for_scheme(compression_scheme: CompressionScheme) -> Option<HeaderValue> {
    match compression_scheme {
        CompressionScheme::Noop => None,
        CompressionScheme::Gzip(_) => Some(HeaderValue::from_static("gzip")),
        CompressionScheme::Zlib(_) => Some(HeaderValue::from_static("deflate")),
        CompressionScheme::Zstd(_) => Some(HeaderValue::from_static("zstd")),
    }
}

#[cfg(test)]
mod tests {
    use std::{collections::HashMap, io::Cursor};

    use agent_data_plane_config::{
        shared::{
            AltMetricsIntake, MetricsEncoding as TypedMetricsEncoding, V3ApiSettings as TypedV3ApiSettings,
            V3SeriesMode,
        },
        ConfigValue,
    };
    use bytes::Bytes;
    use datadog_protos::metrics::v3::MetricData as V3MetricData;
    use protobuf::Message as _;
    use saluki_context::{
        tags::{Tag, TagSet},
        Context,
    };
    use saluki_core::data_model::{
        event::{metric::MetricMetadata, Event},
        payload::Payload,
    };
    use saluki_metrics::test::TestRecorder;
    use stringtheory::MetaString;
    use tokio::time::timeout;

    use super::*;
    use crate::common::datadog::test_util::shared_configuration;

    /// Returns an encoder built from shared configuration shaped like a translated Agent configuration.
    fn metrics_config_from(shared: &SharedConfiguration) -> DatadogMetricsConfiguration {
        DatadogMetricsConfiguration::from_configuration(shared)
    }

    /// Returns the Agent-compatible V3 series settings a default configuration resolves to.
    fn agent_series_config() -> UseV3ApiSeriesConfig {
        (&TypedMetricsEncoding::default()).into()
    }

    #[test]
    fn v3_api_settings_come_from_resolved_configuration() {
        let mut shared = shared_configuration();
        shared.metrics_encoding.v3_api.compression_level = 7;
        shared.metrics_encoding.v3_api.sketches = TypedV3ApiSettings {
            endpoints: vec!["https://app.datadoghq.eu".to_string()],
        };

        let config = metrics_config_from(&shared);

        assert_eq!(7, config.v3_api.compression_level);
        assert_eq!(
            Some("https://app.datadoghq.eu"),
            config.v3_api.sketches.endpoints.first().map(String::as_str)
        );
    }

    #[test]
    fn metrics_routing_comes_from_resolved_configuration() {
        let mut shared = shared_configuration();
        shared.endpoints.compression.compressor_kind = "zstd".to_string();
        shared.endpoints.opw_intake = AltMetricsIntake {
            enabled: true,
            url: "https://opw.example.com".to_string(),
            use_v3_series: true,
        };
        shared.metrics_encoding.use_v2_series_api = false;
        shared.metrics_encoding.v3_api.compression_level = 7;
        shared.metrics_encoding.v3_series_mode = V3SeriesMode::Disabled;
        shared.metrics_encoding.v3_series_endpoint_modes =
            HashMap::from([("https://app.datadoghq.com".to_string(), V3SeriesMode::Enabled)]);

        let config = metrics_config_from(&shared);

        assert_eq!("zstd", config.compressor_kind);
        assert!(!config.use_v2_series_api);
        assert_eq!(7, config.v3_api.compression_level);
        assert_eq!(V3SeriesMode::Disabled, config.use_v3_api.series.enabled);
        assert_eq!(
            Some(&V3SeriesMode::Enabled),
            config.use_v3_api.series.endpoints.get("https://app.datadoghq.com")
        );
        // A selected endpoint means alternate-intake routing is enabled; it carries the URL and V3 override.
        let opw = config
            .opw_metrics
            .selected_endpoint()
            .expect("alternate intake routing should be selected");
        assert_eq!("https://opw.example.com", opw.url);
        assert!(opw.use_v3_series);
    }

    #[test]
    fn metrics_v3_disabled_by_compressor_uses_v2_only() {
        assert_eq!(MetricsEncoderMode::V2Only, metrics_encoder_mode_for_config(true, true));
        assert_eq!(
            MetricsEncoderMode::V3Enabled,
            metrics_encoder_mode_for_config(true, false)
        );
    }

    #[test]
    fn mixed_v2_and_v3_endpoints_require_both_series_encoders() {
        // The primary Datadog endpoint is V3-authoritative under `datadog_only`, while the additional
        // endpoint is not a Datadog URL and stays on V2.
        let mut shared = shared_configuration();
        shared.metrics_encoding.v3_series_mode = V3SeriesMode::DatadogOnly;
        shared.endpoints.additional_endpoints = HashMap::from([(
            "https://custom.example.com".to_string(),
            vec!["additional-api-key".to_string()],
        )]);
        let config = metrics_config_from(&shared);

        assert!(config.requires_v2_series(false).expect("endpoints should resolve"));
        assert!(config.requires_v3_series(false).expect("endpoints should resolve"));
    }

    #[test]
    fn all_v2_endpoints_do_not_require_v3_series() {
        let mut shared = shared_configuration();
        shared.metrics_encoding.v3_series_mode = V3SeriesMode::DatadogOnly;
        shared.endpoints.dd_url = ConfigValue::explicit("http://127.0.0.1:9091".to_string());
        let config = metrics_config_from(&shared);

        assert!(config.requires_v2_series(false).expect("endpoints should resolve"));
        assert!(!config.requires_v3_series(false).expect("endpoints should resolve"));
    }

    #[test]
    fn enabled_v3_series_mode_requires_only_v3_series() {
        let mut shared = shared_configuration();
        shared.metrics_encoding.v3_series_mode = V3SeriesMode::Enabled;
        let config = metrics_config_from(&shared);

        assert!(!config.requires_v2_series(false).expect("endpoints should resolve"));
        assert!(config.requires_v3_series(false).expect("endpoints should resolve"));
    }

    #[test]
    fn v1_series_configuration_takes_precedence_over_v3() {
        let mut shared = shared_configuration();
        shared.metrics_encoding.use_v2_series_api = false;
        shared.metrics_encoding.v3_series_mode = V3SeriesMode::Enabled;
        let config = metrics_config_from(&shared);

        let series_v3_can_be_enabled =
            series_v3_can_be_enabled_for_config(config.use_v2_series_api, None, false, &config.use_v3_api.series);

        assert!(!series_v3_can_be_enabled);

        assert_eq!(
            MetricsEncoderMode::V2Only,
            metrics_encoder_mode_for_config(series_v3_can_be_enabled, false)
        );

        assert!(
            config.requires_v2_series(false).expect("endpoint should resolve"),
            "V1 configuration must retain the legacy series builder even when V3 is enabled"
        );
    }

    #[test]
    fn endpoint_override_uses_the_overridden_endpoint_protocol() {
        let mut shared = shared_configuration();
        shared.metrics_encoding.v3_series_mode = V3SeriesMode::Disabled;
        shared.endpoints.dd_url = ConfigValue::explicit("https://primary.example.com".to_string());
        shared.metrics_encoding.v3_series_endpoint_modes =
            HashMap::from([("https://v3-mrf.example.com".to_string(), V3SeriesMode::Enabled)]);
        let config = metrics_config_from(&shared);

        let v2_mrf_config = config
            .clone()
            .with_metrics_endpoint_override("https://v2-mrf.example.com".to_string());
        let v3_mrf_config = config.with_metrics_endpoint_override("https://v3-mrf.example.com".to_string());

        assert!(v2_mrf_config
            .requires_v2_series(false)
            .expect("V2 MRF endpoint should resolve"));
        assert!(!v2_mrf_config
            .requires_v3_series(false)
            .expect("V2 MRF endpoint should resolve"));
        assert!(!v3_mrf_config
            .requires_v2_series(false)
            .expect("V3 MRF endpoint should resolve"));
        assert!(v3_mrf_config
            .requires_v3_series(false)
            .expect("V3 MRF endpoint should resolve"));
    }

    #[test]
    fn an_endpoint_override_drops_the_configured_additional_and_alternate_endpoints() {
        let mut shared = shared_configuration();
        shared.endpoints.dd_url = ConfigValue::explicit("https://primary.example.com".to_string());
        shared.endpoints.additional_endpoints = HashMap::from([(
            "https://additional.example.com".to_string(),
            vec!["additional-api-key".to_string()],
        )]);
        shared.endpoints.opw_intake = AltMetricsIntake {
            enabled: true,
            url: "https://opw.example.com".to_string(),
            use_v3_series: true,
        };

        let config = metrics_config_from(&shared).with_metrics_endpoint_override("https://mrf.example.com".to_string());

        assert_eq!("https://mrf.example.com", config.primary_endpoint);
        assert!(config.additional_endpoints.is_empty());
        assert_eq!(None, config.opw_metrics.selected_endpoint().map(|opw| opw.url));
    }

    #[test]
    fn v2_series_only_override_requires_only_v2_series() {
        let mut shared = shared_configuration();
        shared.metrics_encoding.v3_series_mode = V3SeriesMode::Enabled;
        let config = metrics_config_from(&shared).with_v2_series_only();

        assert_eq!(V3SeriesMode::Disabled, config.use_v3_api.series.enabled);
        assert!(config.use_v3_api.series.endpoints.is_empty());
        assert!(config.requires_v2_series(false).expect("endpoint should resolve"));
        assert!(!config.requires_v3_series(false).expect("endpoint should resolve"));
    }

    #[test]
    fn agent_default_v3_does_not_enable_opw_only_encoder_mode() {
        let series_config = agent_series_config();
        // An alternate intake whose URL cannot resolve does not fall through to the Vector intake.
        let mut shared = shared_configuration();
        shared.endpoints.opw_intake = AltMetricsIntake {
            enabled: true,
            url: "http://[::1".to_string(),
            use_v3_series: false,
        };
        shared.endpoints.vector_intake = AltMetricsIntake {
            enabled: true,
            url: "http://vector.example.com".to_string(),
            use_v3_series: false,
        };
        let opw_metrics = OpwMetricsConfiguration::from_configuration(&shared.endpoints);
        let invalid_metrics_primary_override = selected_metrics_primary_v3_override(&opw_metrics);

        assert!(!series_v3_can_be_enabled_for_config(
            true,
            Some(false),
            false,
            &series_config
        ));
        assert!(series_v3_can_be_enabled_for_config(
            true,
            Some(true),
            false,
            &series_config
        ));
        assert!(series_v3_can_be_enabled_for_config(
            true,
            Some(false),
            true,
            &series_config
        ));
        assert_eq!(None, invalid_metrics_primary_override);
        assert!(series_v3_can_be_enabled_for_config(
            true,
            invalid_metrics_primary_override,
            false,
            &series_config
        ));
    }

    #[test]
    fn v3_encodes_zero_count_distribution_with_bins() {
        let mut sketch = DDSketch::default();
        sketch.insert_n(42.0, 10);
        sketch.set_count(0);
        sketch.set_min(40.0);
        sketch.set_max(45.0);
        sketch.set_sum(123.0);
        sketch.set_avg(f64::NAN);

        let metric = Metric::from_parts(
            Context::from_static_parts("zero.count.distribution", &[]),
            MetricValues::distribution((123_u64, sketch)),
            MetricMetadata::default(),
        );

        let encoded = encode_v3_metrics_batch(&[metric], &SharedTagSet::default())
            .expect("zero-count distribution should encode to V3");
        let metric_data = V3MetricData::parse_from_bytes(&encoded.payload).expect("V3 metric data should decode");

        assert_eq!(metric_data.numPoints, vec![1]);
    }

    #[test]
    fn v3_skips_empty_distribution() {
        let metric = Metric::from_parts(
            Context::from_static_parts("empty.distribution", &[]),
            MetricValues::distribution((123_u64, DDSketch::default())),
            MetricMetadata::default(),
        );

        let encoded =
            encode_v3_metrics_batch(&[metric], &SharedTagSet::default()).expect("empty distribution should encode");
        let metric_data = V3MetricData::parse_from_bytes(&encoded.payload).expect("V3 metric data should decode");

        assert!(metric_data.numPoints.is_empty());
    }

    #[tokio::test]
    async fn create_v3_request_uses_configured_endpoint_uri() {
        let encoded = V3Writer::new().finalize().expect("empty V3 payload should encode");
        let request = create_v3_request("/api/intake/metrics/custom/series", encoded, CompressionScheme::noop())
            .await
            .expect("request should be created");

        assert_eq!("/api/intake/metrics/custom/series", request.request.uri());
    }

    #[tokio::test]
    async fn create_v3_request_uses_single_stream_body_and_column_telemetry() {
        let metrics = vec![Metric::counter("v3.single.stream", 42.0)];
        let encoded = encode_v3_metrics_batch(&metrics, &SharedTagSet::default()).expect("metrics should encode to V3");
        let expected_payload = encoded.payload.clone();

        let mut request = create_v3_request(METRICS_SERIES_V3_PATH, encoded, CompressionScheme::noop())
            .await
            .expect("request should be created");
        measure_v3_column_compressed_sizes(&mut request.stats, CompressionScheme::noop())
            .await
            .expect("column compressed sizes should be measured");

        for column in &request.stats.columns {
            assert_eq!(column.compressed_len, column.bytes.len());
        }

        let mut expected_body = Vec::new();
        {
            let mut os = CodedOutputStream::vec(&mut expected_body);
            os.write_tag(3, WireType::LengthDelimited).unwrap();
            os.write_uint64_no_tag(expected_payload.len() as u64).unwrap();
            os.flush().unwrap();
        }
        expected_body.extend_from_slice(&expected_payload);

        assert_eq!(request.request.into_body().into_bytes(), Bytes::from(expected_body));
    }

    #[tokio::test]
    async fn create_v3_request_zstd_body_decodes_to_metric_payload() {
        let metrics = vec![Metric::counter("v3.single.stream.zstd", 42.0)];
        let encoded = encode_v3_metrics_batch(&metrics, &SharedTagSet::default()).expect("metrics should encode to V3");
        let expected_payload = encoded.payload.clone();

        let mut request = create_v3_request(METRICS_SERIES_V3_PATH, encoded, CompressionScheme::zstd_default())
            .await
            .expect("request should be created");
        measure_v3_column_compressed_sizes(&mut request.stats, CompressionScheme::zstd_default())
            .await
            .expect("column compressed sizes should be measured");

        for column in &request.stats.columns {
            let expected_compressed_len = compressed_v3_len(&column.bytes, CompressionScheme::zstd_default())
                .await
                .expect("column compressed size should be measured");
            assert_eq!(column.compressed_len, expected_compressed_len);
        }

        let mut expected_body = Vec::new();
        {
            let mut os = CodedOutputStream::vec(&mut expected_body);
            os.write_tag(3, WireType::LengthDelimited).unwrap();
            os.write_uint64_no_tag(expected_payload.len() as u64).unwrap();
            os.flush().unwrap();
        }
        expected_body.extend_from_slice(&expected_payload);

        let decoded_body = zstd::stream::decode_all(Cursor::new(request.request.into_body().into_bytes()))
            .expect("compressed V3 body should decode");
        assert_eq!(decoded_body, expected_body);
    }

    #[test]
    fn v3_drops_non_finite_scalar_points() {
        const NUM_POINTS_FIELD_NUMBER: u32 = 15;
        const VALUE_SINT64_FIELD_NUMBER: u32 = 17;

        let metrics = vec![
            Metric::from_parts(
                Context::from_static_parts("v3.finite.counter", &[]),
                MetricValues::counter([(1, 1.0_f64), (2, f64::NAN), (3, f64::INFINITY), (4, 2.0)]),
                MetricMetadata::default(),
            ),
            Metric::from_parts(
                Context::from_static_parts("v3.finite.gauge", &[]),
                MetricValues::gauge([(1, 3.0_f64), (2, f64::NAN), (3, 4.0)]),
                MetricMetadata::default(),
            ),
            Metric::from_parts(
                Context::from_static_parts("v3.finite.rate", &[]),
                MetricValues::rate(
                    [(1, 30.0_f64), (2, f64::INFINITY), (3, f64::NAN)],
                    Duration::from_secs(10),
                ),
                MetricMetadata::default(),
            ),
            Metric::set("v3.finite.set", "alpha"),
        ];

        let encoded = encode_v3_metrics_batch(&metrics, &SharedTagSet::default()).expect("metrics should encode to V3");
        let column = |field_number| {
            encoded
                .stats
                .columns
                .iter()
                .find(|column| column.field_number == field_number)
                .map(|column| column.bytes.as_slice())
        };

        assert_eq!(column(NUM_POINTS_FIELD_NUMBER), Some(&[2, 2, 1, 1][..]));
        // Values are 1, 2, 3, 4, 3 (rate scaled by 10s), and 1 (set cardinality), encoded as sint64.
        assert_eq!(column(VALUE_SINT64_FIELD_NUMBER), Some(&[2, 4, 6, 8, 6, 2][..]));
        assert_eq!(encoded.stats.value_encoding_stats.sint64, 6);
        assert_eq!(encoded.stats.value_encoding_stats.float64, 0);
    }

    #[test]
    fn v3_encodes_zero_interval_rate_without_scaling() {
        const NUM_POINTS_FIELD_NUMBER: u32 = 15;
        const VALUE_SINT64_FIELD_NUMBER: u32 = 17;

        let metrics = vec![Metric::rate("v3.unnormalized.rate", 42.0, Duration::ZERO)];
        let encoded = encode_v3_metrics_batch(&metrics, &SharedTagSet::default()).expect("rate should encode to V3");
        let column = |field_number| {
            encoded
                .stats
                .columns
                .iter()
                .find(|column| column.field_number == field_number)
                .map(|column| column.bytes.as_slice())
        };

        assert_eq!(column(NUM_POINTS_FIELD_NUMBER), Some(&[1][..]));
        assert_eq!(column(VALUE_SINT64_FIELD_NUMBER), Some(&[84][..]));
    }

    #[tokio::test]
    async fn v3_serializer_stats_record_agent_style_value_and_column_counters() {
        const VALUE_SINT64_FIELD_NUMBER: u32 = 17;

        let recorder = TestRecorder::default();
        let _local = metrics::set_default_local_recorder(&recorder);
        let serializer_telemetry = V3SerializerTelemetry::from_builder(&MetricsBuilder::default());
        let metrics = vec![
            Metric::gauge("v3.telemetry.zero", [(123, 0.0), (124, 0.0)]),
            Metric::counter("v3.telemetry.sint64", [(123, 100.0), (124, 200.0)]),
            Metric::gauge("v3.telemetry.float32", [(123, 1.5), (124, 2.25)]),
            Metric::gauge("v3.telemetry.float64", [(123, (1i64 << 30) as f64), (124, 1.5)]),
        ];
        let encoded = encode_v3_metrics_batch(&metrics, &SharedTagSet::default()).expect("metrics should encode to V3");
        let mut request = create_v3_request(METRICS_SERIES_V3_PATH, encoded, CompressionScheme::noop())
            .await
            .expect("request should be created");
        measure_v3_column_compressed_sizes(&mut request.stats, CompressionScheme::noop())
            .await
            .expect("column compressed sizes should be measured");

        let value_sint64_column = request
            .stats
            .columns
            .iter()
            .find(|column| column.field_number == VALUE_SINT64_FIELD_NUMBER)
            .expect("sint64 value column should be present");
        let value_sint64_len = value_sint64_column.bytes.len() as u64;
        assert_eq!(value_sint64_column.compressed_len as u64, value_sint64_len);

        record_v3_serializer_stats(&serializer_telemetry, &request.stats);

        assert_eq!(
            recorder.counter(("serializer.v3_values_count", &[("type", "zero")])),
            Some(2)
        );
        assert_eq!(
            recorder.counter(("serializer.v3_values_count", &[("type", "sint64")])),
            Some(2)
        );
        assert_eq!(
            recorder.counter(("serializer.v3_values_count", &[("type", "float32")])),
            Some(2)
        );
        assert_eq!(
            recorder.counter(("serializer.v3_values_count", &[("type", "float64")])),
            Some(2)
        );
        assert_eq!(
            recorder.counter((
                "serializer.v3_column_size",
                &[("column", "ValueSint64"), ("compressed", "uncompressed")]
            )),
            Some(value_sint64_len)
        );
        assert_eq!(
            recorder.counter((
                "serializer.v3_column_size",
                &[("column", "ValueSint64"), ("compressed", "compressed")]
            )),
            Some(value_sint64_len)
        );
    }

    async fn create_v3_test_request(metrics: &[Metric]) -> V3EncodedRequest {
        let encoded = encode_v3_metrics_batch(metrics, &SharedTagSet::default()).expect("metrics should encode to V3");
        create_v3_request(METRICS_SERIES_V3_PATH, encoded, CompressionScheme::noop())
            .await
            .expect("request should be created")
    }

    fn test_v3_flush_context<'a>(
        ep_config: &'a EndpointConfiguration, payload_limits: V3PayloadLimits,
        serializer_telemetry: &'a V3SerializerTelemetry, telemetry: &'a ComponentTelemetry,
    ) -> V3FlushContext<'a> {
        V3FlushContext {
            endpoint_config: ep_config,
            payload_limits,
            series_endpoint_uri: METRICS_SERIES_V3_PATH,
            serializer_telemetry,
            telemetry,
        }
    }

    /// Collected form of a payload emitted by `encode_and_flush_v3_payload_requests`.
    struct CollectedV3Payload {
        event_count: usize,
        data_point_count: usize,
        request: Request<FrozenChunkedBytesBuffer>,
    }

    /// Drives `encode_and_flush_v3_payload_requests` and collects the payloads it emits.
    ///
    /// The channel is sized generously because nothing drains it concurrently here; production drains it from the
    /// encoder's main task.
    async fn collect_v3_payload_requests(metrics: &[Metric], context: V3FlushContext<'_>) -> Vec<CollectedV3Payload> {
        let (mut payloads_tx, mut payloads_rx) = tokio::sync::mpsc::channel(64);
        let mut ratio = V3CompressionRatio::default();
        encode_and_flush_v3_payload_requests(
            METRICS_SERIES_V3_PATH,
            metrics,
            context,
            "series",
            &mut ratio,
            &mut payloads_tx,
            None,
        )
        .await
        .expect("payload requests should encode and flush");
        drop(payloads_tx);

        let mut collected = Vec::new();
        while let Some(payload) = payloads_rx.recv().await {
            let Payload::Http(http_payload) = payload else {
                panic!("expected HTTP payload");
            };
            let (metadata, request) = http_payload.into_parts();
            collected.push(CollectedV3Payload {
                event_count: metadata.event_count(),
                data_point_count: metadata.data_point_count(),
                request,
            });
        }

        collected
    }

    #[tokio::test]
    async fn v3_payload_requests_split_by_compressed_size_limit() {
        let metrics = vec![
            Metric::counter("v3.compressed.split.one", 1.0),
            Metric::counter("v3.compressed.split.two", 2.0),
        ];
        let single_request = create_v3_test_request(&metrics[..1]).await;
        let combined_request = create_v3_test_request(&metrics).await;
        assert!(combined_request.compressed_len > single_request.compressed_len);

        let limits = V3PayloadLimits::new(single_request.compressed_len, usize::MAX, 10_000, 10_000);
        let ep_config = EndpointConfiguration::new(CompressionScheme::noop(), 10_000, usize::MAX, None);
        let telemetry = ComponentTelemetry::from_builder(&MetricsBuilder::default());
        let serializer_telemetry = V3SerializerTelemetry::from_builder(&MetricsBuilder::default());

        let context = test_v3_flush_context(&ep_config, limits, &serializer_telemetry, &telemetry);
        let requests = collect_v3_payload_requests(&metrics, context).await;

        assert_eq!(2, requests.len());
        assert_eq!(
            vec![1, 1],
            requests.iter().map(|request| request.event_count).collect::<Vec<_>>()
        );
        assert!(requests
            .iter()
            .all(|request| request.request.body().len() <= limits.max_compressed_size));
    }

    #[tokio::test]
    async fn v3_byte_fallback_does_not_resurrect_metric_over_point_limit() {
        let point_oversized_metric = Metric::counter("v3.fallback.over.point.limit", [(123, 1.0), (124, 2.0)]);

        let byte_oversized_metric = Metric::counter(
            concat!(
                "v3.fallback.over.byte.limit.",
                "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
                "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb",
                "cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc",
                "dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd",
            ),
            3.0,
        );

        let point_oversized_request = create_v3_test_request(std::slice::from_ref(&point_oversized_metric)).await;
        let byte_oversized_request = create_v3_test_request(std::slice::from_ref(&byte_oversized_metric)).await;

        assert!(byte_oversized_request.compressed_len > point_oversized_request.compressed_len);

        let limits = V3PayloadLimits::new(point_oversized_request.compressed_len, usize::MAX, 10_000, 1);

        let ep_config = EndpointConfiguration::new(CompressionScheme::noop(), 10_000, usize::MAX, None);
        let telemetry = ComponentTelemetry::from_builder(&MetricsBuilder::default());
        let serializer_telemetry = V3SerializerTelemetry::from_builder(&MetricsBuilder::default());
        let context = test_v3_flush_context(&ep_config, limits, &serializer_telemetry, &telemetry);

        let requests = collect_v3_payload_requests(&[point_oversized_metric, byte_oversized_metric], context).await;

        let emitted_point_counts = requests
            .iter()
            .map(|request| request.data_point_count)
            .collect::<Vec<_>>();

        assert!(
            emitted_point_counts
                .iter()
                .all(|point_count| *point_count <= limits.max_points_per_payload),
            "fallback emitted request point counts {emitted_point_counts:?} above limit {}",
            limits.max_points_per_payload
        );
    }

    #[tokio::test]
    async fn v3_serializer_stats_record_payload_full_split_reason() {
        let metrics = vec![
            Metric::counter("v3.telemetry.payload_full.one", 1.0),
            Metric::counter("v3.telemetry.payload_full.two", 2.0),
        ];
        let single_request = create_v3_test_request(&metrics[..1]).await;
        let limits = V3PayloadLimits::new(single_request.compressed_len, usize::MAX, 10_000, 10_000);
        let ep_config = EndpointConfiguration::new(CompressionScheme::noop(), 10_000, usize::MAX, None);
        let recorder = TestRecorder::default();
        let _local = metrics::set_default_local_recorder(&recorder);
        let telemetry = ComponentTelemetry::from_builder(&MetricsBuilder::default());
        let serializer_telemetry = V3SerializerTelemetry::from_builder(&MetricsBuilder::default());

        let context = test_v3_flush_context(&ep_config, limits, &serializer_telemetry, &telemetry);
        let requests = collect_v3_payload_requests(&metrics, context).await;

        assert_eq!(2, requests.len());
        assert_eq!(
            recorder.counter(("serializer.v3_payload_split_reason", &[("reason", "payload_full")])),
            Some(1)
        );
    }

    #[tokio::test]
    async fn v3_serializer_stats_record_item_too_big_split_reason() {
        let metrics = vec![Metric::counter("v3.telemetry.item_too_big", 1.0)];
        let request = create_v3_test_request(&metrics).await;
        let limits = V3PayloadLimits::new(request.compressed_len - 1, usize::MAX, 10_000, 10_000);
        let ep_config = EndpointConfiguration::new(CompressionScheme::noop(), 10_000, usize::MAX, None);
        let recorder = TestRecorder::default();
        let _local = metrics::set_default_local_recorder(&recorder);
        let telemetry = ComponentTelemetry::from_builder(&MetricsBuilder::default());
        let serializer_telemetry = V3SerializerTelemetry::from_builder(&MetricsBuilder::default());

        let context = test_v3_flush_context(&ep_config, limits, &serializer_telemetry, &telemetry);
        let requests = collect_v3_payload_requests(&metrics, context).await;

        assert!(requests.is_empty());
        assert_eq!(recorder.counter("serializer.v3_item_too_big"), Some(1));
        assert_eq!(
            recorder.counter(("serializer.v3_payload_split_reason", &[("reason", "item_too_big")])),
            Some(1)
        );
    }

    #[tokio::test]
    async fn v3_payload_requests_split_by_uncompressed_size_limit() {
        let metrics = vec![
            Metric::counter("v3.uncompressed.split.one", 1.0),
            Metric::counter("v3.uncompressed.split.two", 2.0),
        ];
        let single_request = create_v3_test_request(&metrics[..1]).await;
        let combined_request = create_v3_test_request(&metrics).await;
        assert!(combined_request.uncompressed_len > single_request.uncompressed_len);

        let limits = V3PayloadLimits::new(usize::MAX, single_request.uncompressed_len, 10_000, 10_000);
        let ep_config = EndpointConfiguration::new(CompressionScheme::noop(), 10_000, usize::MAX, None);
        let telemetry = ComponentTelemetry::from_builder(&MetricsBuilder::default());
        let serializer_telemetry = V3SerializerTelemetry::from_builder(&MetricsBuilder::default());

        let context = test_v3_flush_context(&ep_config, limits, &serializer_telemetry, &telemetry);
        let requests = collect_v3_payload_requests(&metrics, context).await;

        assert_eq!(2, requests.len());
        assert_eq!(
            vec![1, 1],
            requests.iter().map(|request| request.event_count).collect::<Vec<_>>()
        );
    }

    /// Drives `build_v3_batch` to exhaustion, returning the metric range and event count of each batch it cut.
    fn collect_v3_batches(
        metrics: &[Metric], context: V3FlushContext<'_>, target_uncompressed_len: usize,
    ) -> Vec<(Range<usize>, usize)> {
        let mut batches = Vec::new();
        let mut idx = 0;
        while idx < metrics.len() {
            let start = idx;
            match build_v3_batch(metrics, &mut idx, context, "series", target_uncompressed_len) {
                Some(batch) => batches.push((start..idx, batch.event_count)),
                None if idx == start => break,
                None => {}
            }
        }

        batches
    }

    #[test]
    fn v3_batches_are_cut_at_the_point_limit() {
        let metrics = vec![
            Metric::counter("v3.points.split.one", [(123, 1.0), (124, 2.0)]),
            Metric::counter("v3.points.split.two", [(123, 3.0), (124, 4.0)]),
            Metric::counter("v3.points.split.three", 5.0),
        ];
        let limits = V3PayloadLimits::new(usize::MAX, usize::MAX, 10_000, 3);
        let ep_config = EndpointConfiguration::new(CompressionScheme::noop(), 10_000, usize::MAX, None);
        let telemetry = ComponentTelemetry::from_builder(&MetricsBuilder::default());
        let serializer_telemetry = V3SerializerTelemetry::from_builder(&MetricsBuilder::default());
        let context = test_v3_flush_context(&ep_config, limits, &serializer_telemetry, &telemetry);

        let batches = collect_v3_batches(&metrics, context, usize::MAX);

        assert_eq!(vec![(0..1, 1), (1..3, 2)], batches);
    }

    #[test]
    fn v3_batches_are_cut_at_the_metric_limit() {
        let metrics = vec![
            Metric::counter("v3.metric.limit.a", 1.0),
            Metric::counter("v3.metric.limit.b", 2.0),
            Metric::counter("v3.metric.limit.c", 3.0),
            Metric::counter("v3.metric.limit.d", 4.0),
            Metric::counter("v3.metric.limit.e", 5.0),
        ];
        let limits = V3PayloadLimits::new(usize::MAX, usize::MAX, 2, 10_000);
        let ep_config = EndpointConfiguration::new(CompressionScheme::noop(), 2, usize::MAX, None);
        let telemetry = ComponentTelemetry::from_builder(&MetricsBuilder::default());
        let serializer_telemetry = V3SerializerTelemetry::from_builder(&MetricsBuilder::default());
        let context = test_v3_flush_context(&ep_config, limits, &serializer_telemetry, &telemetry);

        let batches = collect_v3_batches(&metrics, context, usize::MAX);

        assert_eq!(vec![(0..2, 2), (2..4, 2), (4..5, 1)], batches);
    }

    #[test]
    fn v3_batches_are_cut_at_the_uncompressed_byte_target() {
        let metrics = vec![
            Metric::counter("v3.byte.target.a", 1.0),
            Metric::counter("v3.byte.target.b", 2.0),
            Metric::counter("v3.byte.target.c", 3.0),
            Metric::counter("v3.byte.target.d", 4.0),
        ];
        let limits = V3PayloadLimits::new(usize::MAX, usize::MAX, 10_000, 10_000);
        let ep_config = EndpointConfiguration::new(CompressionScheme::noop(), 10_000, usize::MAX, None);
        let telemetry = ComponentTelemetry::from_builder(&MetricsBuilder::default());
        let serializer_telemetry = V3SerializerTelemetry::from_builder(&MetricsBuilder::default());
        let context = test_v3_flush_context(&ep_config, limits, &serializer_telemetry, &telemetry);

        // A target of one byte forces a cut after every metric, since the estimate is non-zero once anything is
        // written. This is the mechanism that keeps high-cardinality batches from being encoded and then re-split.
        let batches = collect_v3_batches(&metrics, context, 1);

        assert_eq!(4, batches.len());
        assert!(batches.iter().all(|(_, event_count)| *event_count == 1));
    }

    #[test]
    fn v3_serializer_stats_record_max_points_split_reason() {
        let metrics = vec![
            Metric::counter("v3.telemetry.max_points.one", [(123, 1.0), (124, 2.0)]),
            Metric::counter("v3.telemetry.max_points.two", [(123, 3.0), (124, 4.0)]),
        ];
        let limits = V3PayloadLimits::new(usize::MAX, usize::MAX, 10_000, 2);
        let ep_config = EndpointConfiguration::new(CompressionScheme::noop(), 10_000, usize::MAX, None);
        let recorder = TestRecorder::default();
        let _local = metrics::set_default_local_recorder(&recorder);
        let telemetry = ComponentTelemetry::from_builder(&MetricsBuilder::default());
        let serializer_telemetry = V3SerializerTelemetry::from_builder(&MetricsBuilder::default());
        let context = test_v3_flush_context(&ep_config, limits, &serializer_telemetry, &telemetry);

        let batches = collect_v3_batches(&metrics, context, usize::MAX);

        assert_eq!(vec![(0..1, 1), (1..2, 1)], batches);
        assert_eq!(
            recorder.counter(("serializer.v3_payload_split_reason", &[("reason", "max_points")])),
            Some(1)
        );
    }

    #[test]
    fn v3_batches_drop_oversized_metric_and_keep_batching() {
        let metrics = vec![
            Metric::counter("v3.points.oversized.before", [(123, 1.0), (124, 2.0)]),
            Metric::counter(
                "v3.points.oversized.too_big",
                [(123, 3.0), (124, 4.0), (125, 5.0), (126, 6.0)],
            ),
            Metric::counter("v3.points.oversized.after", 7.0),
        ];
        let limits = V3PayloadLimits::new(usize::MAX, usize::MAX, 10_000, 3);
        let ep_config = EndpointConfiguration::new(CompressionScheme::noop(), 10_000, usize::MAX, None);
        let recorder = TestRecorder::default();
        let _local = metrics::set_default_local_recorder(&recorder);
        let telemetry = ComponentTelemetry::from_builder(&MetricsBuilder::default());
        let serializer_telemetry = V3SerializerTelemetry::from_builder(&MetricsBuilder::default());
        let context = test_v3_flush_context(&ep_config, limits, &serializer_telemetry, &telemetry);

        let batches = collect_v3_batches(&metrics, context, usize::MAX);

        // Dropping the oversized metric no longer forces a payload boundary: the metrics either side of it share a
        // batch, since together they still fit the point limit.
        assert_eq!(vec![(0..3, 2)], batches);
        assert_eq!(recorder.counter("serializer.v3_item_too_big"), Some(1));
    }

    #[test]
    fn v3_batches_skip_zero_point_metrics_without_splitting() {
        let metrics = vec![
            Metric::counter("v3.points.zero.before", 1.0),
            Metric::counter("v3.points.zero.empty", &[] as &[f64]),
            Metric::counter("v3.points.zero.after", 2.0),
        ];
        let limits = V3PayloadLimits::new(usize::MAX, usize::MAX, 10_000, 10_000);
        let ep_config = EndpointConfiguration::new(CompressionScheme::noop(), 10_000, usize::MAX, None);
        let telemetry = ComponentTelemetry::from_builder(&MetricsBuilder::default());
        let serializer_telemetry = V3SerializerTelemetry::from_builder(&MetricsBuilder::default());
        let context = test_v3_flush_context(&ep_config, limits, &serializer_telemetry, &telemetry);

        let batches = collect_v3_batches(&metrics, context, usize::MAX);

        assert_eq!(vec![(0..3, 2)], batches);
    }

    #[test]
    fn v3_compression_ratio_targets_the_compressed_limit() {
        let limits = V3PayloadLimits::new(500_000, 5_000_000, 10_000, 10_000);
        let mut ratio = V3CompressionRatio::default();

        // With nothing observed yet, assume no compression: the target tracks the compressed limit directly so the
        // first batch cannot wildly overshoot.
        let initial_target = ratio.uncompressed_target(limits);
        assert_eq!((500_000.0 * V3_BATCH_TARGET_MARGIN) as usize, initial_target);

        // After observing 4:1 compression, the target grows towards the uncompressed budget those bytes imply.
        ratio.record(4_000_000, 1_000_000);
        let observed_target = ratio.uncompressed_target(limits);
        assert!(
            observed_target > initial_target,
            "target should grow once compression is observed: {observed_target} vs {initial_target}"
        );

        // It is still capped by the uncompressed limit, however well the data compresses.
        ratio.record(4_000_000, 40_000);
        ratio.record(4_000_000, 40_000);
        ratio.record(4_000_000, 40_000);
        ratio.record(4_000_000, 40_000);
        ratio.record(4_000_000, 40_000);
        assert!(ratio.uncompressed_target(limits) <= limits.max_uncompressed_size);
    }

    #[test]
    fn v3_series_metric_unit_refs_are_encoded_sparsely() {
        let context = Context::from_static_parts("my.timer.avg", &[]);
        let metadata = MetricMetadata::default().with_unit(MetaString::from_static("millisecond"));
        let gauge = Metric::from_parts(context, MetricValues::gauge([1.0_f64]), metadata);
        let context = Context::from_static_parts("my.counter", &[]);
        let no_unit = Metric::from_parts(context, MetricValues::gauge([2.0_f64]), MetricMetadata::default());
        let context = Context::from_static_parts("my.timer.max", &[]);
        let metadata = MetricMetadata::default().with_unit(MetaString::from_static("millisecond"));
        let same_unit = Metric::from_parts(context, MetricValues::gauge([3.0_f64]), metadata);

        let payload = encode_v3_metrics_batch(&[gauge, no_unit, same_unit], &SharedTagSet::default())
            .expect("V3 metric should encode successfully")
            .payload;

        let expected_unit_dict = [
            0xca, 0x01, // field 25, length-delimited.
            0x0c, // field payload length: varint string length + string bytes.
            0x0b, b'm', b'i', b'l', b'l', b'i', b's', b'e', b'c', b'o', b'n', b'd',
        ];
        assert!(
            payload
                .windows(expected_unit_dict.len())
                .any(|window| window == expected_unit_dict),
            "V3 payload should contain DictUnitStr field for 'millisecond', got bytes: {:?}",
            payload
        );

        let expected_unit_ref = [
            0xd2, 0x01, // field 26, length-delimited.
            0x02, // packed field payload length.
            0x02, 0x00, // sparse unit refs for metrics 1 and 3 only: refs [1, 1] -> deltas [1, 0].
        ];
        assert!(
            payload
                .windows(expected_unit_ref.len())
                .any(|window| window == expected_unit_ref),
            "V3 payload should contain UnitRef field for 'millisecond', got bytes: {:?}",
            payload
        );
    }

    #[test]
    fn v3_sketch_metric_unit_not_encoded() {
        let context = Context::from_static_parts("my.histogram", &[]);
        let metadata = MetricMetadata::default().with_unit(MetaString::from_static("millisecond"));
        let histogram = Metric::from_parts(context, MetricValues::histogram([1.0_f64]), metadata);

        let payload = encode_v3_metrics_batch(&[histogram], &SharedTagSet::default())
            .expect("V3 sketch metric should encode successfully")
            .payload;

        assert!(
            !payload
                .windows(b"millisecond".len())
                .any(|window| window == b"millisecond"),
            "V3 sketch payload should not contain unit bytes, matching the Agent V3 sketch builder: {:?}",
            payload
        );
    }

    #[test]
    fn v3_series_promotes_device_and_internal_resource_tags_to_resources() {
        let context = Context::from_static_parts(
            "series.resources",
            &[
                "env:prod",
                "device:switch1",
                "dd.internal.resource:pod:pod-a",
                "dd.internal.resource:malformed",
            ],
        );
        let context = context.with_host(Some(MetaString::from_static("host-a")));
        let metadata = MetricMetadata::default();
        let metric = Metric::from_parts(context, MetricValues::gauge([1.0_f64]), metadata);

        let payload = encode_v3_metrics_batch(&[metric], &SharedTagSet::default())
            .expect("V3 series should encode successfully")
            .payload;

        assert_contains_bytes(&payload, b"env:prod");
        assert!(!contains_bytes(&payload, b"device:switch1"));
        assert!(!contains_bytes(&payload, b"dd.internal.resource:pod:pod-a"));
        assert!(!contains_bytes(&payload, b"dd.internal.resource:malformed"));

        let expected_resource_dict = [
            0x22, // field 4, length-delimited.
            0x25, // field payload length.
            0x04, b'h', b'o', b's', b't', 0x06, b'h', b'o', b's', b't', b'-', b'a', 0x06, b'd', b'e', b'v', b'i', b'c',
            b'e', 0x07, b's', b'w', b'i', b't', b'c', b'h', b'1', 0x03, b'p', b'o', b'd', 0x05, b'p', b'o', b'd', b'-',
            b'a',
        ];
        assert_contains_bytes(&payload, &expected_resource_dict);
    }

    #[test]
    fn v3_series_promotes_additional_and_origin_resource_tags_without_empty_host() {
        let context = Context::from_static_parts("series.additional_origin_resources", &["env:prod"])
            .with_origin_tags(tag_set(["dd.internal.resource:pod:pod-origin"]));
        let additional_tags = SharedTagSet::from(tag_set([
            "team:core",
            "device:switch1",
            "dd.internal.resource:container:container-a",
        ]));
        let context = context.with_host(Some(MetaString::empty()));
        let metadata = MetricMetadata::default();
        let metric = Metric::from_parts(context, MetricValues::gauge([1.0_f64]), metadata);

        let payload = encode_v3_metrics_batch(&[metric], &additional_tags)
            .expect("V3 series should encode successfully")
            .payload;

        assert_contains_bytes(&payload, b"env:prod");
        assert_contains_bytes(&payload, b"team:core");
        assert!(!contains_bytes(&payload, b"device:switch1"));
        assert!(!contains_bytes(&payload, b"dd.internal.resource:container:container-a"));
        assert!(!contains_bytes(&payload, b"dd.internal.resource:pod:pod-origin"));

        let expected_resource_dict = [
            0x22, // field 4, length-delimited.
            0x34, // field payload length.
            0x06, b'd', b'e', b'v', b'i', b'c', b'e', 0x07, b's', b'w', b'i', b't', b'c', b'h', b'1', 0x03, b'p', b'o',
            b'd', 0x0a, b'p', b'o', b'd', b'-', b'o', b'r', b'i', b'g', b'i', b'n', 0x09, b'c', b'o', b'n', b't', b'a',
            b'i', b'n', b'e', b'r', 0x0b, b'c', b'o', b'n', b't', b'a', b'i', b'n', b'e', b'r', b'-', b'a',
        ];
        assert_contains_bytes(&payload, &expected_resource_dict);
        assert!(!contains_bytes(&payload, b"host"));
    }

    #[test]
    fn v3_sketch_keeps_device_and_internal_resource_tags_as_tags() {
        let context = Context::from_static_parts(
            "sketch.resources",
            &["env:prod", "device:switch1", "dd.internal.resource:pod:pod-a"],
        );
        let context = context.with_host(Some(MetaString::from_static("host-a")));
        let metadata = MetricMetadata::default();
        let metric = Metric::from_parts(context, MetricValues::histogram([1.0_f64]), metadata);

        let payload = encode_v3_metrics_batch(&[metric], &SharedTagSet::default())
            .expect("V3 sketch should encode successfully")
            .payload;

        assert_contains_bytes(&payload, b"env:prod");
        assert_contains_bytes(&payload, b"device:switch1");
        assert_contains_bytes(&payload, b"dd.internal.resource:pod:pod-a");

        let expected_resource_dict = [
            0x22, // field 4, length-delimited.
            0x0c, // field payload length.
            0x04, b'h', b'o', b's', b't', 0x06, b'h', b'o', b's', b't', b'-', b'a',
        ];
        assert_contains_bytes(&payload, &expected_resource_dict);
    }

    /// Distinct inputs for a [`run_request_builder`] integration test.
    struct RequestBuilderScenario {
        series_mode: MetricsEncoderMode,
        sketches_mode: MetricsEncoderMode,
        payload_limits: V3PayloadLimits,
        flush_timeout: Duration,
    }

    impl RequestBuilderScenario {
        /// Creates a scenario with unbounded V3 size/point limits and a 10 ms flush timeout.
        fn new(series_mode: MetricsEncoderMode, sketches_mode: MetricsEncoderMode) -> Self {
            Self {
                series_mode,
                sketches_mode,
                payload_limits: V3PayloadLimits::new(usize::MAX, usize::MAX, 10_000, 10_000),
                flush_timeout: Duration::from_millis(10),
            }
        }

        /// Sets the V3 per-payload point limit that drives point-count split flushes.
        fn with_max_points_per_payload(mut self, max_points_per_payload: usize) -> Self {
            self.payload_limits = V3PayloadLimits::new(usize::MAX, usize::MAX, 10_000, max_points_per_payload);
            self
        }

        /// Overrides the flush timeout used to bound pending flushes.
        fn with_flush_timeout(mut self, flush_timeout: Duration) -> Self {
            self.flush_timeout = flush_timeout;
            self
        }

        /// Spawns [`run_request_builder`] for this scenario, returning a harness for pushing metrics and draining
        /// flushed payloads.
        async fn spawn(self) -> RequestBuilderHarness {
            let v3_endpoint_config = EndpointConfiguration::new(CompressionScheme::noop(), 10_000, usize::MAX, None);
            let metrics_builder = MetricsBuilder::default();
            let telemetry = ComponentTelemetry::from_builder(&metrics_builder);
            let serializer_telemetry = V3SerializerTelemetry::from_builder(&metrics_builder);
            let (events_tx, events_rx) = tokio::sync::mpsc::channel(1);
            let (payloads_tx, payloads_rx) = tokio::sync::mpsc::channel(8);

            let handle = tokio::spawn(run_request_builder(
                None,
                None,
                self.series_mode,
                self.sketches_mode,
                V3RuntimeConfig {
                    endpoint_config: v3_endpoint_config,
                    payload_limits: self.payload_limits,
                    series_endpoint_uri: METRICS_SERIES_V3_PATH.to_string(),
                    serializer_telemetry,
                },
                telemetry,
                events_rx,
                payloads_tx,
                self.flush_timeout,
                false,
            ));

            RequestBuilderHarness {
                events_tx,
                payloads_rx,
                handle,
            }
        }
    }

    /// A spawned [`run_request_builder`] task plus its input/output channels.
    struct RequestBuilderHarness {
        events_tx: tokio::sync::mpsc::Sender<EventsBuffer>,
        payloads_rx: tokio::sync::mpsc::Receiver<PayloadsBuffer>,
        handle: tokio::task::JoinHandle<Result<(), GenericError>>,
    }

    impl RequestBuilderHarness {
        /// Sends a single event buffer carrying `metrics` to the request builder.
        async fn push_metrics(&self, metrics: impl IntoIterator<Item = Metric>) {
            let mut events = EventsBuffer::default();
            for metric in metrics {
                assert!(
                    events.try_push(Event::Metric(metric)).is_none(),
                    "event buffer should accept metric"
                );
            }
            self.events_tx
                .send(events)
                .await
                .expect("events should be sent to request builder");
        }

        /// Receives the next flushed payload, unwrapping it into its HTTP metadata and request.
        async fn next_http_request(&mut self) -> (PayloadMetadata, Request<FrozenChunkedBytesBuffer>) {
            let payload = timeout(Duration::from_secs(1), self.payloads_rx.recv())
                .await
                .expect("payload should arrive before timeout")
                .expect("payload channel should remain open");
            match payload {
                Payload::Http(http_payload) => http_payload.into_parts(),
                _ => panic!("expected HTTP payload"),
            }
        }

        /// Asserts that no payload is flushed within `window`.
        async fn assert_no_payload_within(&mut self, window: Duration) {
            assert!(
                timeout(window, self.payloads_rx.recv()).await.is_err(),
                "no payload should be flushed within {window:?}"
            );
        }

        /// Drops the events sender and waits for the request builder task to stop cleanly.
        async fn shutdown(self) {
            drop(self.events_tx);
            self.handle
                .await
                .expect("request builder task should complete")
                .expect("request builder should stop cleanly");
        }
    }

    #[tokio::test]
    async fn authoritative_v3_flushes_previous_point_limit_batch() {
        // Authoritative V3 batches independently of V2. With a 3-point limit, the first two-point metric fits but
        // the second would exceed it, so the first flushes as a point-limit split and the second flushes on the
        // pending-flush timeout.
        let recorder = TestRecorder::default();
        let _local = metrics::set_default_local_recorder(&recorder);

        let mut harness = RequestBuilderScenario::new(MetricsEncoderMode::V3Enabled, MetricsEncoderMode::V2Only)
            .with_max_points_per_payload(3)
            .with_flush_timeout(Duration::from_millis(250))
            .spawn()
            .await;

        harness
            .push_metrics([
                Metric::counter("authoritative.v3.points.one", [(123, 1.0), (124, 2.0)]),
                Metric::counter("authoritative.v3.points.two", [(123, 3.0), (124, 4.0)]),
            ])
            .await;

        // Point-count split flushes the first metric before the second exceeds the limit.
        let (_, request) = harness.next_http_request().await;
        assert_eq!(METRICS_SERIES_V3_PATH, request.uri());
        harness.assert_no_payload_within(Duration::from_millis(50)).await;

        // The carried-over metric flushes when the pending-flush timeout fires.
        let (_, request) = harness.next_http_request().await;
        assert_eq!(METRICS_SERIES_V3_PATH, request.uri());
        assert_eq!(
            recorder.counter(("serializer.v3_payload_split_reason", &[("reason", "max_points")])),
            Some(1)
        );

        harness.shutdown().await;
    }

    #[tokio::test]
    async fn authoritative_v3_sketches_flush_previous_point_limit_batch() {
        // The same authoritative point-limit split applies to sketches, which always target the V3 sketches
        // endpoint: the first distribution flushes as a point-limit split, the second on the pending-flush timeout.
        let recorder = TestRecorder::default();
        let _local = metrics::set_default_local_recorder(&recorder);

        let mut harness = RequestBuilderScenario::new(MetricsEncoderMode::V2Only, MetricsEncoderMode::V3Enabled)
            .with_max_points_per_payload(3)
            .with_flush_timeout(Duration::from_millis(250))
            .spawn()
            .await;

        harness
            .push_metrics([
                Metric::distribution("authoritative.v3.sketch.points.one", [(123, 1.0), (124, 2.0)]),
                Metric::distribution("authoritative.v3.sketch.points.two", [(123, 3.0), (124, 4.0)]),
            ])
            .await;

        for stage in ["point-limit split", "timeout"] {
            let (_, request) = harness.next_http_request().await;
            assert_eq!(
                V3_SKETCHES_ENDPOINT_URI,
                request.uri(),
                "{stage} sketches payload should target the V3 sketches endpoint"
            );
        }
        assert_eq!(
            recorder.counter(("serializer.v3_payload_split_reason", &[("reason", "max_points")])),
            Some(1)
        );

        harness.shutdown().await;
    }

    fn contains_bytes(haystack: &[u8], needle: &[u8]) -> bool {
        haystack.windows(needle.len()).any(|window| window == needle)
    }

    fn assert_contains_bytes(haystack: &[u8], needle: &[u8]) {
        assert!(
            contains_bytes(haystack, needle),
            "expected payload to contain bytes {:?}, got {:?}",
            needle,
            haystack
        );
    }

    fn tag_set<const N: usize>(tags: [&'static str; N]) -> TagSet {
        tags.into_iter().map(Tag::from_static).collect()
    }

    // Regression test to ensure the V2 series request builder enforces `max_series_points_per_payload`.
    //
    // The test encodes more total points than the configured limit and asserts the builder splits them across
    // multiple payloads without any payload exceeding the limit.
    #[tokio::test]
    async fn v2_series_builder_enforces_max_series_points_per_payload() {
        let v2_endpoint_config = EndpointConfiguration::new(CompressionScheme::noop(), 10_000, 10_000, None);
        let mut builder = v2::create_v2_request_builder(MetricsEndpoint::SeriesV2, &v2_endpoint_config)
            .await
            .expect("V2 request builder should be created");
        builder
            .with_len_limits(usize::MAX, usize::MAX)
            .expect("byte limits should be accepted");

        let mut total_points = 0;
        for i in 0..6_000u32 {
            let metric = Metric::gauge(
                Context::from_parts(MetaString::from(format!("g{i}")), TagSet::default()),
                [(1u64, 1.0), (2u64, 2.0)],
            );
            let mut pending = Some(metric);
            while let Some(metric) = pending.take() {
                if let Some(returned) = builder.encode(metric).await.expect("encode should not error") {
                    for request in builder.flush().await {
                        let (_, points, _) = request.expect("request should build");
                        assert!(
                            points <= 10_000,
                            "payload carried {} points, over the 10000 limit",
                            points
                        );
                        total_points += points;
                    }
                    pending = Some(returned);
                }
            }
        }
        for request in builder.flush().await {
            let (_, points, _) = request.expect("request should build");
            assert!(
                points <= 10_000,
                "final payload carried {} points, over the 10000 limit",
                points
            );
            total_points += points;
        }

        assert_eq!(
            total_points, 12_000,
            "all points should be emitted across the split payloads"
        );
    }
}

#[cfg(test)]
mod payload_limits {
    use super::{v2, DatadogMetricsConfiguration};
    use crate::common::datadog::{clamp_payload_limits, test_util::shared_configuration};

    #[test]
    fn payload_limits_come_from_resolved_configuration() {
        let mut shared = shared_configuration();
        shared.metrics_encoding.max_payload_size = 4321;
        shared.metrics_encoding.max_uncompressed_payload_size = 8765;
        shared.metrics_encoding.max_series_payload_size = 1234;
        shared.metrics_encoding.max_series_uncompressed_payload_size = 5678;
        shared.metrics_encoding.max_series_points_per_payload = 500;
        shared.metrics_encoding.max_metrics_per_payload = 42;

        let config = DatadogMetricsConfiguration::from_configuration(&shared);

        assert_eq!(4321, config.max_payload_size);
        assert_eq!(8765, config.max_uncompressed_payload_size);
        assert_eq!(1234, config.max_series_payload_size);
        assert_eq!(5678, config.max_series_uncompressed_payload_size);
        assert_eq!(500, config.max_series_points_per_payload);
        assert_eq!(500, config.v3_payload_limits().max_points_per_payload);
        assert_eq!(42, config.max_metrics_per_payload);
    }

    #[test]
    fn clamps_series_payload_limit_keys_to_api_limits() {
        let (uncompressed_limit, compressed_limit) = clamp_payload_limits(
            v2::SERIES_V2_UNCOMPRESSED_SIZE_LIMIT + 1,
            v2::SERIES_V2_COMPRESSED_SIZE_LIMIT + 1,
            v2::SERIES_V2_UNCOMPRESSED_SIZE_LIMIT,
            v2::SERIES_V2_COMPRESSED_SIZE_LIMIT,
        );
        assert_eq!(uncompressed_limit, v2::SERIES_V2_UNCOMPRESSED_SIZE_LIMIT);
        assert_eq!(compressed_limit, v2::SERIES_V2_COMPRESSED_SIZE_LIMIT);

        let (uncompressed_limit, compressed_limit) = clamp_payload_limits(
            5678,
            1234,
            v2::SERIES_V2_UNCOMPRESSED_SIZE_LIMIT,
            v2::SERIES_V2_COMPRESSED_SIZE_LIMIT,
        );
        assert_eq!(uncompressed_limit, 5678);
        assert_eq!(compressed_limit, 1234);
    }
}