|
| 1 | +use crate::{config::{DataType, SinkConfig, SinkContext}, event::Event, http::HttpClient, sinks::{Healthcheck, UriParseError, VectorSink, datadog::{Region, healthcheck}, util::{Compression, Concurrency, TowerRequestConfig, batch::{BatchConfig, BatchSettings}, retries::RetryLogic}}}; |
| 2 | +use chrono:: Utc; |
| 3 | +use futures::{stream, FutureExt, SinkExt}; |
| 4 | +use http::{uri::InvalidUri, Uri}; |
| 5 | +use serde::{Deserialize, Serialize}; |
| 6 | +use snafu::{ResultExt, Snafu}; |
| 7 | +use tower::ServiceBuilder; |
| 8 | +use std::{ |
| 9 | + future::ready, |
| 10 | + sync::atomic::AtomicI64, |
| 11 | +}; |
| 12 | + |
| 13 | +// TODO: revisit our concurrency and batching defaults |
| 14 | +const DEFAULT_REQUEST_LIMITS: TowerRequestConfig = |
| 15 | + TowerRequestConfig::const_new(Concurrency::None, Concurrency::None).retry_attempts(5); |
| 16 | + |
| 17 | +const DEFAULT_BATCH_SETTINGS: BatchSettings<()> = |
| 18 | + BatchSettings::const_default().events(20).timeout(1); |
| 19 | + |
| 20 | +const MAXIMUM_SERIES_PAYLOAD_COMPRESSED_SIZE: usize = 3_200_000; |
| 21 | +const MAXIMUM_SERIES_PAYLOAD_SIZE: usize = 62_914_560; |
| 22 | + |
| 23 | +#[derive(Debug, Snafu)] |
| 24 | +enum BuildError { |
| 25 | + #[snafu(display("Invalid host {:?}: {:?}", host, source))] |
| 26 | + InvalidHost { host: String, source: InvalidUri }, |
| 27 | +} |
| 28 | + |
| 29 | +/// Various metric type-specific API types. |
| 30 | +/// |
| 31 | +/// Each of these corresponds to a specific request path when making a request to the agent API. |
| 32 | +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] |
| 33 | +enum DatadogMetricsEndpoint { |
| 34 | + Series, |
| 35 | + Distribution, |
| 36 | +} |
| 37 | + |
| 38 | +pub struct DatadogMetricsRetryLogic; |
| 39 | + |
| 40 | +impl RetryLogic for DatadogMetricsRetryLogic { |
| 41 | + type Error = HttpError; |
| 42 | + type Response = DatadogMetricsResponse; |
| 43 | + |
| 44 | + fn is_retriable_error(&self, error: &Self::Error) -> bool { |
| 45 | + todo!() |
| 46 | + } |
| 47 | + |
| 48 | + fn should_retry_response(&self, response: &Self::Response) -> RetryAction { |
| 49 | + todo!() |
| 50 | + } |
| 51 | +} |
| 52 | + |
| 53 | +#[derive(Deserialize, Serialize, Debug, Clone, Default)] |
| 54 | +#[serde(deny_unknown_fields)] |
| 55 | +pub struct DatadogMetricsConfig { |
| 56 | + #[serde(alias = "namespace")] |
| 57 | + pub default_namespace: Option<String>, |
| 58 | + // Deprecated name |
| 59 | + #[serde(alias = "host")] |
| 60 | + pub endpoint: Option<String>, |
| 61 | + // Deprecated, replaced by the site option |
| 62 | + pub region: Option<Region>, |
| 63 | + pub site: Option<String>, |
| 64 | + pub api_key: String, |
| 65 | + #[serde(default)] |
| 66 | + pub batch: BatchConfig, |
| 67 | + #[serde(default)] |
| 68 | + pub request: TowerRequestConfig, |
| 69 | + #[serde(default = "Compression::gzip_default")] |
| 70 | + pub compression: Compression, |
| 71 | +} |
| 72 | + |
| 73 | +impl_generate_config_from_default!(DatadogMetricsConfig); |
| 74 | + |
| 75 | +#[async_trait::async_trait] |
| 76 | +#[typetag::serde(name = "datadog_metrics")] |
| 77 | +impl SinkConfig for DatadogMetricsConfig { |
| 78 | + async fn build(&self, cx: SinkContext) -> crate::Result<(VectorSink, Healthcheck)> { |
| 79 | + let client = HttpClient::new(None, cx.proxy())?; |
| 80 | + |
| 81 | + let client = self.build_client(&cx.proxy)?; |
| 82 | + let healthcheck = self.build_healthcheck(client.clone()); |
| 83 | + let sink = self.build_sink(client, cx)?; |
| 84 | + |
| 85 | + Ok((sink, healthcheck)) |
| 86 | + } |
| 87 | + |
| 88 | + fn input_type(&self) -> DataType { |
| 89 | + DataType::Metric |
| 90 | + } |
| 91 | + |
| 92 | + fn sink_type(&self) -> &'static str { |
| 93 | + "datadog_metrics" |
| 94 | + } |
| 95 | +} |
| 96 | + |
| 97 | +impl DatadogMetricsConfig { |
| 98 | + /// Copies this `DatadogMetricsConfig` with the API key set to the given value. |
| 99 | + pub fn with_api_key<T: Into<String>>(api_key: T) -> Self { |
| 100 | + Self { |
| 101 | + api_key: api_key.into(), |
| 102 | + ..Self::default() |
| 103 | + } |
| 104 | + } |
| 105 | + |
| 106 | + /// Gets the base URI of the Datadog agent API. |
| 107 | + /// |
| 108 | + /// Per the Datadog agent convention, we should include a unique identifier as part of the |
| 109 | + /// domain to indicate that these metrics are being submitted by Vector, including the version, |
| 110 | + /// likely useful for detecting if a specific version of the agent (Vector, in this case) is |
| 111 | + /// doing something wrong, for understanding issues from the API side. |
| 112 | + /// |
| 113 | + /// The `endpoint` configuration field will be used here if it is present. |
| 114 | + fn get_base_agent_endpoint(&self) -> String { |
| 115 | + self.endpoint.clone().unwrap_or_else(|| { |
| 116 | + let version = str::replace(crate::built_info::PKG_VERSION, ".", "-"); |
| 117 | + format!("https://{}-vector.agent.{}", version, self.get_site()) |
| 118 | + }) |
| 119 | + } |
| 120 | + |
| 121 | + /// Generates the full URIs to use for the various type-specific metrics endpoints. |
| 122 | + fn generate_metric_endpoints(&self) -> crate::Result<Vec<(DatadogMetricsEndpoint, Uri)>> { |
| 123 | + let base_uri = self.get_base_metric_endpoint(); |
| 124 | + let series_endpoint = build_uri(&base_uri, "/api/v1/series")?; |
| 125 | + let distribution_endpoint = build_uri(&base_uri, "/api/v1/distribution_points")?; |
| 126 | + |
| 127 | + Ok(vec![ |
| 128 | + (DatadogMetricsEndpoint::Series, series_endpoint), |
| 129 | + (DatadogMetricsEndpoint::Distribution, distribution_endpoint), |
| 130 | + ]) |
| 131 | + } |
| 132 | + |
| 133 | + /// Gets the base URI of the Datadog API. |
| 134 | + /// |
| 135 | + /// The `endpoint` configuration field will be used here if it is present. |
| 136 | + fn get_api_endpoint(&self) -> String { |
| 137 | + self.endpoint |
| 138 | + .clone() |
| 139 | + .unwrap_or_else(|| format!("https://api.{}", self.get_site())) |
| 140 | + } |
| 141 | + |
| 142 | + /// Gets the base domain to use for any calls to Datadog. |
| 143 | + /// |
| 144 | + /// If `site` is not specified, we fallback to `region`, and if that is not specified, we |
| 145 | + /// fallback to the Datadog US domain. |
| 146 | + fn get_site(&self) -> &str { |
| 147 | + self.site.as_deref().unwrap_or_else(|| match self.region { |
| 148 | + Some(Region::Eu) => "datadoghq.eu", |
| 149 | + None | Some(Region::Us) => "datadoghq.com", |
| 150 | + }) |
| 151 | + } |
| 152 | + |
| 153 | + fn build_client(&self, proxy: &ProxyConfig) -> crate::Result<HttpClient> { |
| 154 | + HttpClient::new(None, proxy) |
| 155 | + } |
| 156 | + |
| 157 | + fn build_healthcheck(&self, client: HttpClient) -> Healthcheck { |
| 158 | + healthcheck(self.get_api_endpoint(), self.api_key.clone(), client).boxed() |
| 159 | + } |
| 160 | + |
| 161 | + fn build_sink(&self, client: HttpClient, cx: SinkContext) -> crate::Result<VectorSink> { |
| 162 | + let batch = DEFAULT_BATCH_SETTINGS |
| 163 | + .parse_config(self.batch)?; |
| 164 | + |
| 165 | + let request_limits = self.request.unwrap_with(&DEFAULT_REQUEST_LIMITS); |
| 166 | + let metric_endpoints = self.generate_metric_endpoints()?; |
| 167 | + let service = ServiceBuilder::new() |
| 168 | + .settings(request_limits, DatadogMetricsRetryLogic) |
| 169 | + .service(DatadogMetricsService::new(client)); |
| 170 | + |
| 171 | + let sink = DatadogMetricsSink::new( |
| 172 | + cx, |
| 173 | + service, |
| 174 | + metric_endpoints, |
| 175 | + compression: self.compression, |
| 176 | + ); |
| 177 | + |
| 178 | + Ok(VectorSink::Sink(Box::new(sink))) |
| 179 | + } |
| 180 | +} |
| 181 | + |
| 182 | +fn build_uri(host: &str, endpoint: &str) -> crate::Result<Uri> { |
| 183 | + format!("{}{}", host, endpoint) |
| 184 | + .parse::<Uri>() |
| 185 | + .context(UriParseError) |
| 186 | +} |
| 187 | + |
| 188 | +#[cfg(test)] |
| 189 | +mod tests { |
| 190 | + use super::*; |
| 191 | + use crate::{event::metric::Sample, sinks::util::test::load_sink}; |
| 192 | + |
| 193 | + #[test] |
| 194 | + fn generate_config() { |
| 195 | + crate::test_util::test_generate_config::<DatadogConfig>(); |
| 196 | + } |
| 197 | +} |
0 commit comments