agent_data_plane_config/

shared.rs

//! Cross-cutting values consumed by more than one domain.

use std::collections::HashMap;
use std::path::PathBuf;
use std::str::FromStr;
use std::time::Duration;

use serde::Serialize;

use crate::defaults::{DEFAULT_ENCODER_FLUSH_TIMEOUT, DEFAULT_MAX_METRICS_PER_PAYLOAD, DEFAULT_ZSTD_COMPRESSOR_LEVEL};
use crate::{ConfigValue, Error};

/// Assumed maximum size, in bytes, of a single payload in the retry queue.
///
/// The deprecated `forwarder_retry_queue_max_size` setting bounds the retry queue by payload count,
/// while the queue itself is bounded by bytes. Converting between the two requires assuming a
/// per-payload size, and this value matches the Agent's own assumption so that the same
/// configuration yields the same byte budget in ADP and the Agent.
pub const MAX_PAYLOAD_SIZE_BYTES: u64 = 2 * 1024 * 1024;

/// Cross-cutting configuration shared across domains.
#[derive(Clone, Debug, Default, PartialEq, Serialize)]
pub struct SharedConfiguration {
    /// Primary forwarder endpoints and transport.
    pub endpoints: Endpoints,

    /// Global and host-level tagging.
    pub tags: GlobalTags,

    /// Inputs used to derive deployment-wide static tags.
    pub static_tags: StaticTagSettings,

    /// Tags attached to basic liveness telemetry.
    pub basic_telemetry: BasicTelemetry,

    /// Metrics-encoder settings reused across the metrics-emitting pipelines.
    pub metrics_encoding: MetricsEncoding,

    /// Cluster Agent connection, shared by checks, DogStatsD, and OTLP.
    pub cluster_agent: ClusterAgent,

    /// Autoscaling failover, shared by checks, DogStatsD, and OTLP.
    pub autoscaling_failover: AutoscalingFailover,

    /// Secrets management, read by the Datadog intake forwarders.
    pub secrets: Secrets,

    /// Host and container runtime discovery, read by the environment providers.
    pub environment: Environment,

    /// Verbosity of the internal telemetry emitted about the runtime itself. (not in Datadog Agent
    /// config schema)
    pub metrics_level: String,

    /// Base directory for runtime-state files.
    ///
    /// ADP uses this to derive default locations for forwarder retry data, DogStatsD captures, and
    /// DogStatsD context dumps. Defaults to unset when configuration does not provide a concrete
    /// `run_path`.
    pub run_path: Option<PathBuf>,

    /// Host that network listeners bind to (`bind_host`).
    ///
    /// An unset value leaves each listener at its own default host (`127.0.0.1`), and a blank value is normalized to
    /// unset. Read by the DogStatsD UDP and TCP listeners, and by the APM relay's TCP listener; Unix socket and named
    /// pipe listeners ignore it. `dogstatsd_non_local_traffic` and `apm_config.apm_non_local_traffic` take precedence
    /// for their own listeners and bind to `0.0.0.0` regardless of this value.
    ///
    /// The provenance matters to the APM relay: like the trace-agent, it binds every interface in a container only
    /// when neither this nor `apm_config.apm_non_local_traffic` was set explicitly. The Agent's schema defaults this
    /// to `localhost`, so the value alone can't tell. The value stays an `Option` because a blank value is unset
    /// whether or not it was set explicitly.
    pub bind_host: ConfigValue<Option<String>>,
}

/// Host identity and container runtime discovery inputs.
#[derive(Clone, Debug, Default, PartialEq, Serialize)]
pub struct Environment {
    /// Hostname reported for all emitted data.
    ///
    /// Only read in standalone mode, where it is reported verbatim. In connected mode the hostname comes from the
    /// Datadog Agent and this value is ignored.
    ///
    /// Defaults to empty, and a defaulted or empty value is treated as absent. Operators running standalone must set
    /// it explicitly; startup fails otherwise.
    pub hostname: ConfigValue<String>,

    /// containerd runtime discovery and client timeouts.
    pub containerd: Containerd,

    /// Filesystem roots describing container workloads.
    pub container_roots: ContainerRoots,
}

/// containerd runtime discovery and client timeouts.
#[derive(Clone, Debug, Default, PartialEq, Serialize)]
pub struct Containerd {
    /// containerd gRPC socket. A defaulted empty value enables path probing.
    pub socket_path: ConfigValue<PathBuf>,

    /// Timeout for establishing a containerd gRPC connection. Defaults to 1 second; `0` never connects.
    pub connection_timeout: Duration,

    /// Per-RPC timeout for containerd API calls. Defaults to 5 seconds; `0` fails every call.
    pub query_timeout: Duration,
}

/// Filesystem roots describing container workloads.
#[derive(Clone, Debug, Default, PartialEq, Serialize)]
pub struct ContainerRoots {
    /// procfs root. Defaults to `/host/proc`, which is only used when set explicitly.
    pub proc_root: ConfigValue<PathBuf>,

    /// cgroupfs root. Defaults to `/host/sys/fs/cgroup/`, which is only used when set explicitly.
    pub cgroup_root: ConfigValue<PathBuf>,
}

/// Inputs used to derive deployment-wide static tags.
#[derive(Clone, Debug, Default, PartialEq, Serialize)]
pub struct StaticTagSettings {
    /// Deployment-provider classification added as `provider_kind:<value>` when non-empty.
    ///
    /// Defaults to empty, which adds no provider-kind tag.
    pub provider_kind: String,

    /// Whether the deployment uses EKS Fargate.
    ///
    /// Defaults to `false`. When enabled, the static-tag resolver adds EKS-specific tags in addition to global tags.
    pub eks_fargate: bool,

    /// Kubernetes node name used for the EKS Fargate node tag.
    ///
    /// Defaults to empty, which omits `eks_fargate_node` and emits a warning when EKS Fargate is enabled.
    pub kubernetes_kubelet_nodename: String,

    /// Kubernetes cluster name used for the EKS Fargate cluster tag.
    ///
    /// Defaults to empty, which omits `kube_cluster_name` unless the configured global tags already provide one.
    pub cluster_name: String,
}

/// Primary outbound endpoints plus the forwarder, proxy, TLS, and compression settings that apply
/// to every pipeline emitting to the intake.
#[derive(Clone, Debug, Default, PartialEq, Serialize)]
pub struct Endpoints {
    /// API key for the primary intake.
    pub api_key: String,

    /// Base site domain for the primary intake (for example, `datadoghq.com`).
    ///
    /// The Datadog schema supplies a default value when nothing sets this key. Provenance is
    /// `Explicit` only when the value was explicitly configured.
    pub site: ConfigValue<String>,

    /// Full primary intake URL, which overrides [`site`](Self::site) when set explicitly.
    ///
    /// The Core Agent supplies this key at its schema default even when not set by the user or
    /// operator. Provenance is preserved so that we know when this was explicitly set and should
    /// override `site`.
    pub dd_url: ConfigValue<String>,

    /// Additional dual-shipping endpoints, keyed by intake URL with their API keys.
    pub additional_endpoints: HashMap<String, Vec<String>>,

    /// Whether metrics may carry arbitrary tags.
    pub allow_arbitrary_tags: bool,

    /// Outbound HTTP proxy settings.
    pub proxy: Proxy,

    /// Outbound TLS client settings.
    pub tls: Tls,

    /// Payload compression settings.
    pub compression: Compression,

    /// Forwarder retry, backoff, worker, and disk-storage settings.
    pub forwarder: Forwarder,

    /// Alternate metrics intake for the Observability Pipelines Worker, used in place of the
    /// default intake when enabled.
    pub opw_intake: AltMetricsIntake,

    /// Alternate metrics intake for Vector, used in place of the default intake when enabled.
    pub vector_intake: AltMetricsIntake,
}

impl Endpoints {
    /// Returns the primary intake endpoint, as configured and without normalization.
    ///
    /// An explicitly configured [`dd_url`](Self::dd_url) overrides [`site`](Self::site), even when
    /// its value equals the schema default: the operator asked for that URL. Otherwise the endpoint
    /// is derived from `site`. An empty `site` cannot produce an endpoint, so the effective `dd_url`
    /// value is used instead; it already carries the source schema's default URL.
    pub fn primary_endpoint(&self) -> String {
        if self.dd_url.is_explicit() || self.site.value.is_empty() {
            self.dd_url.value.clone()
        } else {
            format!("https://app.{}", self.site.value)
        }
    }
}

/// An alternate metrics intake (Observability Pipelines Worker or Vector) that replaces the Datadog
/// intake when enabled.
#[derive(Clone, Debug, Default, PartialEq, Serialize)]
pub struct AltMetricsIntake {
    /// Whether this alternate intake replaces the default one.
    pub enabled: bool,

    /// URL of the alternate metrics intake.
    pub url: String,

    /// Whether metrics ship to this intake over the V3 series protocol
    /// (`observability_pipelines_worker.metrics.use_v3_api.series` / `vector.metrics.use_v3_api.series`).
    pub use_v3_series: bool,
}

/// Outbound HTTP proxy settings.
#[derive(Clone, Debug, Default, PartialEq, Serialize)]
pub struct Proxy {
    /// Proxy URL for plain HTTP requests.
    pub http: String,

    /// Proxy URL for HTTPS requests.
    pub https: String,

    /// Hosts that bypass the proxy.
    pub no_proxy: Vec<String>,

    /// Whether no-proxy entries match by suffix rather than exact host.
    pub no_proxy_nonexact_match: bool,

    /// Whether cloud-metadata requests also go through the proxy.
    pub use_proxy_for_cloud_metadata: bool,
}

/// Outbound TLS client settings.
#[derive(Clone, Debug, Default, PartialEq, Serialize)]
pub struct Tls {
    /// Whether server certificate validation is skipped.
    pub skip_ssl_validation: bool,

    /// Minimum TLS version enforced on outbound connections.
    pub min_tls_version: String,

    /// Path to which TLS session keys are logged, for debugging.
    pub sslkeylogfile: String,

    /// Timeout for completing the TLS handshake after a connection is established.
    ///
    /// Defaults to 10 seconds. Bounds only the handshake step, distinct from the overall request timeout. A value
    /// of zero disables the handshake-specific deadline, leaving the overall request timeout as the only bound.
    pub handshake_timeout: Duration,
}

/// Payload compression settings applied before transmission.
#[derive(Clone, Debug, PartialEq, Serialize)]
pub struct Compression {
    /// Which compression algorithm the encoder uses.
    // TODO: enum?
    pub compressor_kind: String,

    /// ADP's own zstd compression level (`data_plane.serializer_zstd_compressor_level`).
    ///
    /// Defaults to `3`, which is higher than the Agent's default of `1` because ADP compresses more
    /// cheaply than the Agent. Its provenance says whether an operator asked for the level, which
    /// separates a configured `3` from the default `3`. Read [`effective_zstd_level`] rather than this
    /// field.
    ///
    /// [`effective_zstd_level`]: Compression::effective_zstd_level
    pub adp_zstd_level: ConfigValue<i32>,

    /// The Core Agent's zstd compression level (`serializer_zstd_compressor_level`).
    ///
    /// The Agent supplies this key at its own default of `1` even when nothing sets it, so only its
    /// provenance says whether an operator asked for the level. Read [`effective_zstd_level`] rather
    /// than this field.
    ///
    /// [`effective_zstd_level`]: Compression::effective_zstd_level
    pub agent_zstd_level: ConfigValue<i32>,
}

impl Compression {
    /// Returns the effective compression level used when the algorithm is zstd.
    ///
    /// Defaults to `3`, which is higher than the Agent's default of `1` because ADP compresses more
    /// cheaply than the Agent. The setting is determined as followed:
    /// - If an operator explicitly sets `data_plane.serializer_zstd_compressor_level`, it wins.
    /// - If an operator explicitly sets `serializer_zstd_compressor_level`, ADP uses it.
    /// - If neither is set, the ADP default of `3` is used.
    pub fn effective_zstd_level(&self) -> i32 {
        // Resolution options that depend on the order in which values are applied to a single field
        // are more brittle and harder to understand. Keeping these fields separate and applying
        // precedence with this helper function makes the override logic easier to understand.

        // If the operator explicitly set ADP zstd level, use it.
        if self.adp_zstd_level.is_explicit() {
            return self.adp_zstd_level.value;
        }

        // If the operator explicitly set Agent zstd level, use it (even if it happens to equal the
        // default).
        if self.agent_zstd_level.is_explicit() {
            return self.agent_zstd_level.value;
        }

        // If nothing was explicitly set, use ADP's default.
        self.adp_zstd_level.value
    }
}

impl Default for Compression {
    fn default() -> Self {
        Self {
            compressor_kind: String::new(),
            adp_zstd_level: ConfigValue::defaulted(DEFAULT_ZSTD_COMPRESSOR_LEVEL),
            agent_zstd_level: ConfigValue::default(),
        }
    }
}

/// HTTP protocol the forwarder negotiates with the intake.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize)]
pub enum ForwarderHttpProtocol {
    #[default]
    Auto,
    Http1,
}

/// Forwarder retry, backoff, worker, and disk-storage settings.
#[derive(Clone, Debug, Default, PartialEq, Serialize)]
pub struct Forwarder {
    /// How often, in seconds, API keys are checked for validity against the intake.
    pub apikey_validation_interval: i64,

    /// Base delay, in seconds, for retry backoff.
    pub backoff_base: f64,

    /// Multiplier applied to the backoff delay after each failed attempt.
    pub backoff_factor: f64,

    /// Maximum retry backoff delay, in seconds.
    pub backoff_max: f64,

    /// How often, in seconds, idle connections are reset.
    pub connection_reset_interval: u64,

    /// Fraction of the in-memory retry queue at which payloads spill to disk.
    pub flush_to_disk_mem_ratio: f64,

    /// Capacity of the high-priority send buffer.
    pub high_prio_buffer_size: usize,

    /// HTTP protocol the forwarder negotiates with the intake.
    pub http_protocol: ForwarderHttpProtocol,

    /// Maximum number of in-flight requests to the intake.
    pub max_concurrent_requests: usize,

    /// Number of forwarder worker tasks.
    pub num_workers: usize,

    /// Age, in days, after which payloads queued on disk are discarded.
    pub outdated_file_in_days: u32,

    /// Number of retry cycles between attempts to recover a failed endpoint.
    pub recovery_interval: u32,

    /// Whether the recovery interval resets after a successful send.
    pub recovery_reset: bool,

    /// Retry-queue capacity expressed as seconds of buffered payloads.
    pub retry_queue_capacity_time_interval_sec: u64,

    /// Maximum number of payloads held in the in-memory retry queue.
    ///
    /// Deprecated in favor of [`retry_queue_payloads_max_size`](Self::retry_queue_payloads_max_size).
    /// This setting counts payloads, not bytes;
    /// [`effective_retry_queue_max_size_bytes`](Self::effective_retry_queue_max_size_bytes) converts
    /// it using [`MAX_PAYLOAD_SIZE_BYTES`]. The Datadog schema supplies `0` when nothing sets this
    /// key. Because `0` is also a value an operator can set, honor this setting only when it is
    /// explicit.
    pub retry_queue_max_size: ConfigValue<u64>,

    /// Maximum total size, in bytes, of payloads held in the retry queue.
    ///
    /// The Datadog schema supplies 15 MiB when nothing sets this key. Takes precedence over
    /// [`retry_queue_max_size`](Self::retry_queue_max_size) when set explicitly.
    pub retry_queue_payloads_max_size: ConfigValue<u64>,

    /// Grace period the forwarder is given to drain before shutdown.
    pub stop_timeout: Duration,

    /// Whether the forwarder waits for in-flight requests to finish when it shuts down.
    ///
    /// When `false`, in-flight requests are aborted as soon as shutdown begins and their transactions are requeued
    /// with the other pending transactions, which are then persisted to disk if disk storage is enabled. When `true`,
    /// in-flight requests are allowed to finish first, bounded by the topology-wide shutdown timeout; if that timeout
    /// elapses first, nothing pending is persisted. The Datadog schema supplies `false` when nothing sets this key.
    pub stop_wait_for_inflight: bool,

    /// Fraction of available disk the on-disk retry store may use.
    pub storage_max_disk_ratio: f64,

    /// Maximum size, in bytes, of the on-disk retry store.
    pub storage_max_size_in_bytes: u64,

    /// Directory where retry payloads are persisted to disk.
    pub storage_path: PathBuf,

    /// Per-request timeout, in seconds, for calls to the intake.
    pub timeout: u64,
}

impl Forwarder {
    /// Returns the effective maximum size, in bytes, of the in-memory retry queue.
    ///
    /// An explicit [`retry_queue_payloads_max_size`](Self::retry_queue_payloads_max_size) wins, then
    /// an explicit [`retry_queue_max_size`](Self::retry_queue_max_size), and otherwise the effective
    /// payload-size value, which carries the source schema's default. Selection cannot look at the
    /// values themselves: `0` is both the deprecated setting's schema default and a value an
    /// operator can mean.
    ///
    /// The deprecated setting counts payloads rather than bytes, so it is scaled by
    /// [`MAX_PAYLOAD_SIZE_BYTES`] to reach a byte budget. A count large enough to overflow saturates
    /// at [`u64::MAX`], which the retry queue treats as effectively unbounded.
    pub fn effective_retry_queue_max_size_bytes(&self) -> u64 {
        if !self.retry_queue_payloads_max_size.is_explicit() && self.retry_queue_max_size.is_explicit() {
            self.retry_queue_max_size.value.saturating_mul(MAX_PAYLOAD_SIZE_BYTES)
        } else {
            self.retry_queue_payloads_max_size.value
        }
    }
}

/// Global / host tagging.
#[derive(Clone, Debug, Default, PartialEq, Serialize)]
pub struct GlobalTags {
    /// Tags configured through `tags` / `DD_TAGS`.
    pub tags: Vec<String>,

    /// Tags configured through `extra_tags` / `DD_EXTRA_TAGS`.
    pub extra_tags: Vec<String>,

    /// How long, after startup, host tags remain attached to emitted data.
    pub expected_tags_duration: Duration,
}

/// Tagging options for basic liveness telemetry.
#[derive(Clone, Debug, Default, PartialEq, Serialize)]
pub struct BasicTelemetry {
    /// Whether liveness signals include the process container's low-cardinality tags.
    ///
    /// Defaults to `false`. Enable this for containerized deployments that need to associate basic
    /// telemetry with the running container. If the container cannot be resolved, liveness signals
    /// are emitted without these tags.
    pub add_container_tags: bool,
}

/// Metrics-encoder settings reused across the metrics-emitting pipelines (DogStatsD, checks, and
/// OTLP): histogram settings, payload limits, and the encoder flush timeout.
#[derive(Clone, Debug, PartialEq, Serialize)]
pub struct MetricsEncoding {
    /// How long the encoder waits before flushing a partially filled payload. (not in Datadog Agent
    /// config schema)
    ///
    /// Shared by the metrics-emitting pipelines and the traces encoder, all of which read the
    /// `flush_timeout_secs` key. Defaults to 2 seconds.
    pub flush_timeout: Duration,

    /// Maximum number of metrics packed into a single payload. (not in Datadog Agent config schema)
    ///
    /// Defaults to [`DEFAULT_MAX_METRICS_PER_PAYLOAD`].
    pub max_metrics_per_payload: usize,

    /// Maximum compressed payload size, in bytes.
    pub max_payload_size: usize,

    /// Maximum compressed size, in bytes, of a series payload.
    pub max_series_payload_size: usize,

    /// Maximum number of series data points per payload.
    pub max_series_points_per_payload: usize,

    /// Maximum uncompressed size, in bytes, of a series payload.
    pub max_series_uncompressed_payload_size: usize,

    /// Maximum uncompressed payload size, in bytes.
    pub max_uncompressed_payload_size: usize,

    /// Whether series are submitted via the v2 intake API.
    pub use_v2_series_api: bool,

    /// Whether outgoing payloads are logged for debugging.
    pub log_payloads: bool,

    /// Histogram aggregation and encoding settings.
    pub histogram: HistogramEncoding,

    /// Experimental V3 sketches settings (`serializer_experimental_use_v3_api.*`).
    pub v3_api: V3ApiEncoding,

    /// Global V3 series routing mode (`use_v3_api.series.enabled`).
    pub v3_series_mode: V3SeriesMode,

    /// Per-endpoint V3 series routing overrides, keyed by endpoint URL
    /// (`use_v3_api.series.endpoints`).
    pub v3_series_endpoint_modes: HashMap<String, V3SeriesMode>,
}

impl Default for MetricsEncoding {
    fn default() -> Self {
        Self {
            // The `flush_timeout_secs` key is Saluki-only, so its default belongs to the ADP config
            // crate rather than a source schema.
            flush_timeout: DEFAULT_ENCODER_FLUSH_TIMEOUT,
            max_metrics_per_payload: DEFAULT_MAX_METRICS_PER_PAYLOAD,
            max_payload_size: 0,
            max_series_payload_size: 0,
            max_series_points_per_payload: 0,
            max_series_uncompressed_payload_size: 0,
            max_uncompressed_payload_size: 0,
            use_v2_series_api: false,
            log_payloads: false,
            histogram: HistogramEncoding::default(),
            v3_api: V3ApiEncoding::default(),
            v3_series_mode: V3SeriesMode::default(),
            v3_series_endpoint_modes: HashMap::new(),
        }
    }
}

/// Experimental V3 sketches settings (`serializer_experimental_use_v3_api.*`).
#[derive(Clone, Debug, Default, PartialEq, Serialize)]
pub struct V3ApiEncoding {
    /// Endpoints using the V3 sketches intake.
    pub sketches: V3ApiSettings,

    /// zstd compression level for V3 payloads.
    pub compression_level: i32,
}

/// V3 sketches intake settings.
#[derive(Clone, Debug, Default, PartialEq, Serialize)]
pub struct V3ApiSettings {
    /// Endpoints enabled for the V3 intake.
    pub endpoints: Vec<String>,
}

/// Whether series are routed to the V3 metrics intake (`use_v3_api.series.*`).
///
/// Each variant serializes to the spelling [`FromStr`] reads, so a serialized mode round-trips.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize)]
pub enum V3SeriesMode {
    /// Route series to the V3 intake.
    #[serde(rename = "true")]
    Enabled,

    /// Route series to the older intake.
    #[serde(rename = "false")]
    Disabled,

    /// Route series to the V3 intake only for endpoints that are Datadog intake URLs.
    #[default]
    #[serde(rename = "datadog_only")]
    DatadogOnly,
}

impl FromStr for V3SeriesMode {
    type Err = Error;

    // The Agent reads this setting as a string and then interprets it, accepting more spellings than
    // `strconv.ParseBool` does, so the accepted set is wider than that of a `boolean` leaf.
    fn from_str(value: &str) -> Result<Self, Self::Err> {
        match value.trim().to_ascii_lowercase().as_str() {
            "true" | "1" | "t" | "yes" | "on" => Ok(Self::Enabled),
            "false" | "0" | "f" | "no" | "off" | "" => Ok(Self::Disabled),
            "datadog_only" => Ok(Self::DatadogOnly),
            other => Err(Error::new_without_source(format!(
                "unknown V3 series mode `{other}`; expected a boolean or `datadog_only`"
            ))),
        }
    }
}

/// Histogram aggregation/encoding settings, shared by the DogStatsD and checks metrics pipelines.
#[derive(Clone, Debug, Default, PartialEq, Serialize)]
pub struct HistogramEncoding {
    /// Which histogram aggregations (for example, `max` or `median`) are computed.
    pub aggregates: Vec<String>,

    /// Whether histograms are also emitted as distributions.
    pub copy_to_distribution: bool,

    /// Metric-name prefix applied to the distribution copies.
    pub copy_to_distribution_prefix: String,

    /// Which percentile aggregations are computed for histograms.
    pub percentiles: Vec<String>,
}

/// Cluster Agent connection, shared by checks, DogStatsD, and OTLP.
///
/// The defaults named on each field are the Datadog schema defaults, which translation writes whenever the key is
/// absent. They are not the values `Default` produces: that is the zero value of each field, which for
/// `kubernetes_service_name` is the empty string and therefore not the schema default.
#[derive(Clone, Debug, Default, PartialEq, Serialize)]
pub struct ClusterAgent {
    /// Whether the Cluster Agent connection is used.
    ///
    /// Defaults to `false`. Turn it on in a deployment that runs a Cluster Agent; while it is off, nothing talks to it.
    pub enabled: bool,

    /// URL of the Cluster Agent.
    ///
    /// Defaults to unset, which leaves the endpoint to Kubernetes service discovery through
    /// `kubernetes_service_name`. A blank value is normalized to unset. Set this in a deployment where the Cluster
    /// Agent is not reachable through an injected Kubernetes service, and give an `https` endpoint: consumers use only
    /// `https`.
    pub url: Option<String>,

    /// Token used to authenticate to the Cluster Agent.
    ///
    /// Defaults to unset, which leaves the Cluster Agent unreachable: there is no anonymous access. Set it wherever the
    /// Cluster Agent is enabled, to that Agent's own token; a blank value is normalized to unset.
    pub auth_token: Option<String>,

    /// Kubernetes service name used to discover the Cluster Agent.
    ///
    /// Defaults to `datadog-cluster-agent`. The name is turned into the `<NAME>_SERVICE_HOST` and
    /// `<NAME>_SERVICE_PORT` environment variables that Kubernetes injects into the pod. Set this when the Cluster
    /// Agent runs under a different service name, or set it to the empty string to turn the lookup off, which leaves
    /// `url` as the only way to reach the Cluster Agent.
    pub kubernetes_service_name: String,
}

/// Secrets management, as configured for the Core Agent.
///
/// ADP resolves no secrets itself; the Core Agent does. These settings mirror the Agent's own configuration, and ADP
/// reads them for one purpose: to decide whether a rejected API key might be replaced. See
/// [`in_use`](Self::in_use).
#[derive(Clone, Debug, Default, PartialEq, Serialize)]
pub struct Secrets {
    /// Path to the executable the Core Agent runs to fetch secrets.
    ///
    /// Defaults to unset, and a blank value is normalized to unset. ADP does not run it. A configured command makes
    /// [`in_use`](Self::in_use) true, which makes an intake's `403 Forbidden` response retriable. Set this only to match
    /// the Core Agent's own configuration.
    pub backend_command: Option<String>,

    /// Minutes between the secret refreshes the Core Agent triggers after an API key is rejected.
    ///
    /// Defaults to `0`, which turns those refreshes off. A negative value from the source means the same thing and is
    /// clamped to `0`. A positive value makes [`in_use`](Self::in_use) true on its own, and `0` does not make it false
    /// when [`backend_command`](Self::backend_command) is set. Set this only to match the Core Agent's own
    /// configuration.
    pub refresh_on_api_key_failure_interval: u64,
}

impl Secrets {
    /// Returns whether secret resolution might replace a rejected API key.
    ///
    /// This is true when a [`backend_command`](Self::backend_command) is configured or
    /// [`refresh_on_api_key_failure_interval`](Self::refresh_on_api_key_failure_interval) is positive. Either says the
    /// key an intake just rejected may be a secret that gets re-resolved, so the same request is worth retrying. When
    /// neither is configured, nothing is going to replace the key, and retrying the request only wastes it.
    pub const fn in_use(&self) -> bool {
        self.refresh_on_api_key_failure_interval > 0 || self.backend_command.is_some()
    }
}

/// Autoscaling failover, shared by checks, DogStatsD, and OTLP.
#[derive(Clone, Debug, Default, PartialEq, Serialize)]
pub struct AutoscalingFailover {
    /// Whether metrics designated for autoscaling failover are forwarded to the Cluster Agent.
    ///
    /// Defaults to `false`. Also needs `cluster_agent.enabled`, `cluster_agent.auth_token`, a resolvable Cluster Agent
    /// endpoint, and a non-empty `metrics`; otherwise the branch is not built and primary forwarding continues.
    pub enabled: bool,

    /// Names of the metrics designated for autoscaling failover.
    ///
    /// Defaults to `container.memory.usage` and `container.cpu.usage`. An empty list turns the failover branch off even
    /// when `enabled` is set, because there is nothing left to forward. Set this when autoscaling reads metrics other
    /// than the two defaults.
    pub metrics: Vec<String>,
}

#[cfg(test)]
mod tests {
    use super::{Compression, Endpoints, Forwarder, V3SeriesMode};
    use crate::defaults::DEFAULT_ZSTD_COMPRESSOR_LEVEL;
    use crate::ConfigValue;

    #[test]
    fn v3_series_mode_parses_every_form_the_agent_interprets() {
        for (value, expected) in [
            ("true", V3SeriesMode::Enabled),
            ("TRUE", V3SeriesMode::Enabled),
            ("1", V3SeriesMode::Enabled),
            ("t", V3SeriesMode::Enabled),
            ("yes", V3SeriesMode::Enabled),
            ("on", V3SeriesMode::Enabled),
            ("false", V3SeriesMode::Disabled),
            ("0", V3SeriesMode::Disabled),
            ("f", V3SeriesMode::Disabled),
            ("no", V3SeriesMode::Disabled),
            ("off", V3SeriesMode::Disabled),
            ("", V3SeriesMode::Disabled),
            (" datadog_only ", V3SeriesMode::DatadogOnly),
        ] {
            assert_eq!(
                value.parse::<V3SeriesMode>().expect("mode should parse"),
                expected,
                "{value}"
            );
        }
    }

    #[test]
    fn v3_series_mode_rejects_an_uninterpretable_value() {
        let error = "sometimes"
            .parse::<V3SeriesMode>()
            .expect_err("an uninterpretable mode should be rejected");

        assert_eq!(
            error.to_string(),
            "unknown V3 series mode `sometimes`; expected a boolean or `datadog_only`"
        );
    }

    #[test]
    fn v3_series_mode_defaults_to_datadog_only() {
        assert_eq!(V3SeriesMode::default(), V3SeriesMode::DatadogOnly);
    }

    #[test]
    fn explicit_dd_url_overrides_site_even_at_the_schema_default() {
        // The source supplies `dd_url` at its schema default even when nothing set it, so an
        // explicit URL equal to that default still expresses an override.
        let endpoints = Endpoints {
            site: ConfigValue::explicit("datadoghq.eu".to_string()),
            dd_url: ConfigValue::explicit("https://app.datadoghq.com".to_string()),
            ..Default::default()
        };

        assert_eq!("https://app.datadoghq.com", endpoints.primary_endpoint());
    }

    #[test]
    fn defaulted_dd_url_leaves_the_endpoint_to_site() {
        let endpoints = Endpoints {
            site: ConfigValue::explicit("datadoghq.eu".to_string()),
            dd_url: ConfigValue::defaulted("https://app.datadoghq.com".to_string()),
            ..Default::default()
        };

        assert_eq!("https://app.datadoghq.eu", endpoints.primary_endpoint());
    }

    #[test]
    fn an_override_url_is_used_verbatim() {
        let endpoints = Endpoints {
            site: ConfigValue::defaulted("datadoghq.com".to_string()),
            dd_url: ConfigValue::explicit("https://proxy.internal.example.com:3128".to_string()),
            ..Default::default()
        };

        assert_eq!("https://proxy.internal.example.com:3128", endpoints.primary_endpoint());
    }

    #[test]
    fn an_empty_site_falls_back_to_the_effective_dd_url() {
        // `https://app.` is not an endpoint, and the model does not restate the source schema's
        // default site. The effective `dd_url` already carries the schema default URL.
        let endpoints = Endpoints {
            site: ConfigValue::explicit(String::new()),
            dd_url: ConfigValue::defaulted("https://app.datadoghq.com".to_string()),
            ..Default::default()
        };

        assert_eq!("https://app.datadoghq.com", endpoints.primary_endpoint());
    }

    #[test]
    fn retry_queue_size_prefers_the_explicit_payload_size() {
        let forwarder = Forwarder {
            retry_queue_payloads_max_size: ConfigValue::explicit(2048),
            retry_queue_max_size: ConfigValue::explicit(1024),
            ..Default::default()
        };

        assert_eq!(2048, forwarder.effective_retry_queue_max_size_bytes());
    }

    #[test]
    fn retry_queue_size_falls_back_to_the_explicit_deprecated_size() {
        // The deprecated setting is a payload count, so 1024 payloads is a 2 GiB byte budget. Using
        // the count as a byte budget would leave a queue too small to hold a single payload.
        let forwarder = Forwarder {
            retry_queue_payloads_max_size: ConfigValue::defaulted(15 * 1024 * 1024),
            retry_queue_max_size: ConfigValue::explicit(1024),
            ..Default::default()
        };

        assert_eq!(1024 * 2 * 1024 * 1024, forwarder.effective_retry_queue_max_size_bytes());
    }

    #[test]
    fn a_deprecated_retry_queue_size_that_would_overflow_saturates() {
        // A count this large cannot be scaled into a u64. Wrapping would turn an enormous queue into
        // a tiny one, so the conversion saturates into an effectively unbounded queue instead.
        let forwarder = Forwarder {
            retry_queue_payloads_max_size: ConfigValue::defaulted(15 * 1024 * 1024),
            retry_queue_max_size: ConfigValue::explicit(u64::MAX),
            ..Default::default()
        };

        assert_eq!(u64::MAX, forwarder.effective_retry_queue_max_size_bytes());
    }

    #[test]
    fn the_effective_zstd_level_prefers_adp_then_an_explicit_agent_level() {
        let defaulted = Compression::default();
        assert_eq!(DEFAULT_ZSTD_COMPRESSOR_LEVEL, defaulted.effective_zstd_level());

        // The Agent supplies its own default of 1 on every load, so only an explicit value counts.
        let agent_defaulted = Compression {
            agent_zstd_level: ConfigValue::defaulted(1),
            ..Default::default()
        };
        assert_eq!(DEFAULT_ZSTD_COMPRESSOR_LEVEL, agent_defaulted.effective_zstd_level());

        let agent_explicit = Compression {
            agent_zstd_level: ConfigValue::explicit(1),
            ..Default::default()
        };
        assert_eq!(1, agent_explicit.effective_zstd_level());

        // ADP's own key wins over an explicit Agent level, including at ADP's default value.
        let both_explicit = Compression {
            adp_zstd_level: ConfigValue::explicit(DEFAULT_ZSTD_COMPRESSOR_LEVEL),
            agent_zstd_level: ConfigValue::explicit(5),
            ..Default::default()
        };
        assert_eq!(DEFAULT_ZSTD_COMPRESSOR_LEVEL, both_explicit.effective_zstd_level());
    }

    #[test]
    fn an_explicit_zero_retry_queue_size_is_honored() {
        // Zero is the deprecated setting's schema default and also a value an operator can mean, so
        // only provenance can tell the two apart.
        let explicitly_zero = Forwarder {
            retry_queue_payloads_max_size: ConfigValue::defaulted(15 * 1024 * 1024),
            retry_queue_max_size: ConfigValue::explicit(0),
            ..Default::default()
        };
        // Zero payloads scales to zero bytes either way.
        assert_eq!(0, explicitly_zero.effective_retry_queue_max_size_bytes());

        let defaulted_zero = Forwarder {
            retry_queue_payloads_max_size: ConfigValue::defaulted(15 * 1024 * 1024),
            retry_queue_max_size: ConfigValue::defaulted(0),
            ..Default::default()
        };
        assert_eq!(15 * 1024 * 1024, defaulted_zero.effective_retry_queue_max_size_bytes());
    }
}