From 4f75a71998014f74438cfd1b8e52aaa41ede0b30 Mon Sep 17 00:00:00 2001 From: ktf Date: Thu, 27 May 2021 15:32:49 +0200 Subject: [PATCH 01/16] Initial Signed-off-by: ktf --- src/internal_events/datadog_events.rs | 35 ++ src/internal_events/mod.rs | 4 + src/sinks/datadog/events.rs | 475 ++++++++++++++++++++++++++ src/sinks/datadog/metrics.rs | 26 +- src/sinks/datadog/mod.rs | 25 ++ 5 files changed, 546 insertions(+), 19 deletions(-) create mode 100644 src/internal_events/datadog_events.rs create mode 100644 src/sinks/datadog/events.rs diff --git a/src/internal_events/datadog_events.rs b/src/internal_events/datadog_events.rs new file mode 100644 index 0000000000000..59572941eb931 --- /dev/null +++ b/src/internal_events/datadog_events.rs @@ -0,0 +1,35 @@ +use super::InternalEvent; +use metrics::counter; + +#[derive(Debug)] +pub struct DatadogEventsProcessed { + pub byte_size: usize, +} + +impl InternalEvent for DatadogEventsProcessed { + fn emit_metrics(&self) { + counter!("processed_bytes_total", self.byte_size as u64); + } +} + +#[derive(Debug)] +pub struct DatadogEventsFieldInvalid<'a> { + pub field: &'a str, +} + +impl<'a> InternalEvent for DatadogEventsFieldInvalid<'a> { + fn emit_logs(&self) { + debug!( + message = "Required field is missing.", + field = %self.field, + internal_log_rate_secs = 10 + ); + } + + fn emit_metrics(&self) { + counter!( + "processing_errors_total", 1, + "error_type" => "field_missing", + "field" => self.field.to_owned()); + } +} diff --git a/src/internal_events/mod.rs b/src/internal_events/mod.rs index 4bdc1990066d3..0263be7f10631 100644 --- a/src/internal_events/mod.rs +++ b/src/internal_events/mod.rs @@ -30,6 +30,8 @@ mod concat; #[cfg(feature = "sinks-console")] mod console; #[cfg(feature = "sinks-datadog")] +mod datadog_events; +#[cfg(feature = "sinks-datadog")] mod datadog_logs; #[cfg(feature = "transforms-dedupe")] mod dedupe; @@ -151,6 +153,8 @@ pub use self::concat::*; #[cfg(feature = "sinks-console")] pub use self::console::*; #[cfg(feature = "sinks-datadog")] +pub use self::datadog_events::*; +#[cfg(feature = "sinks-datadog")] pub use self::datadog_logs::*; #[cfg(feature = "transforms-dedupe")] pub(crate) use self::dedupe::*; diff --git a/src/sinks/datadog/events.rs b/src/sinks/datadog/events.rs new file mode 100644 index 0000000000000..fe95c0eaff900 --- /dev/null +++ b/src/sinks/datadog/events.rs @@ -0,0 +1,475 @@ +use super::healthcheck; +use crate::{ + config::{log_schema, DataType, GenerateConfig, SinkConfig, SinkContext, SinkDescription}, + event::{Event, Value}, + http::HttpClient, + internal_events::{DatadogEventsFieldInvalid, DatadogEventsProcessed}, + sinks::{ + util::{ + batch::{Batch, BatchError}, + encode_event, + encoding::{EncodingConfig, EncodingConfiguration, TimestampFormat}, + http::{HttpSink, PartitionHttpSink}, + BatchConfig, BatchSettings, BoxedRawValue, Compression, EncodedEvent, Encoding, + JsonArrayBuffer, PartitionBuffer, PartitionInnerBuffer, TowerRequestConfig, VecBuffer, + }, + Healthcheck, VectorSink, + }, + tls::{MaybeTlsSettings, TlsConfig}, +}; +use bytes::Bytes; +use flate2::write::GzEncoder; +use futures::{FutureExt, SinkExt}; +use http::{Request, StatusCode}; +use hyper::body::Body; +use indoc::indoc; +use serde::{Deserialize, Serialize}; +use serde_json::json; +use std::{io::Write, sync::Arc, time::Duration}; + +#[derive(Deserialize, Serialize, Debug, Clone)] +#[serde(deny_unknown_fields)] +pub struct DatadogEventsConfig { + endpoint: Option, + + #[serde(default = "default_site")] + site: String, + default_api_key: String, + encoding: EncodingConfig<()>, + + tls: Option, + + #[serde(default)] + compression: Option, + + #[serde(default)] + batch: BatchConfig, + + #[serde(default)] + request: TowerRequestConfig, +} + +type ApiKey = Arc; + +fn default_site() -> String { + "datadoghq.com".to_owned() +} + +inventory::submit! { + SinkDescription::new::("datadog_events") +} + +impl GenerateConfig for DatadogEventsConfig { + fn generate_config() -> toml::Value { + toml::from_str(indoc! {r#" + default_api_key = "${DATADOG_API_KEY_ENV_VAR}" + "#}) + .unwrap() + } +} + +impl DatadogEventsConfig { + fn get_uri(&self) -> String { + format!("{}/api/v1/events", self.get_api_endpoint()) + } + + // The API endpoint is used for healtcheck/API key validation, it is derived from the `site` or `region` option + // the same way the offical Datadog Agent does but the `endpoint` option still override both the main and the + // API endpoint. + fn get_api_endpoint(&self) -> String { + self.endpoint + .clone() + .unwrap_or_else(|| format!("https://api.{}", &self.site)) + } + + fn batch_settings(&self) -> Result, BatchError> { + BatchSettings::default() + .bytes(bytesize::kib(100u64)) + .events(20) + .timeout(1) + .parse_config(self.batch) + } + + /// Builds the required BatchedHttpSink. + /// Since the DataDog sink can create one of two different sinks, this + /// extracts most of the shared functionality required to create either sink. + fn build_sink( + &self, + cx: SinkContext, + service: T, + batch: B, + timeout: Duration, + ) -> crate::Result<(VectorSink, Healthcheck)> + where + O: 'static, + B: Batch> + std::marker::Send + 'static, + B::Output: std::marker::Send + Clone, + B::Input: std::marker::Send, + T: HttpSink< + Input = PartitionInnerBuffer, + Output = PartitionInnerBuffer, + > + Clone, + { + let request_settings = self.request.unwrap_with(&TowerRequestConfig::default()); + + let tls_settings = MaybeTlsSettings::from_config( + &Some(self.tls.clone().unwrap_or_else(TlsConfig::enabled)), + false, + )?; + + let client = HttpClient::new(tls_settings)?; + let healthcheck = healthcheck( + self.get_api_endpoint().clone(), + self.default_api_key.clone(), + client.clone(), + ) + .boxed(); + let sink = PartitionHttpSink::new( + service, + PartitionBuffer::new(batch), + request_settings, + timeout, + client, + cx.acker(), + ) + .sink_map_err(|error| error!(message = "Fatal datadog_metrics sink error.", %error)); + + Ok((VectorSink::Sink(Box::new(sink)), healthcheck)) + } + + /// Build the request, GZipping the contents if the config specifies. + fn build_request( + &self, + uri: &str, + api_key: &str, + content_type: &str, + body: Vec, + ) -> crate::Result>> { + let request = Request::post(uri) + .header("Content-Type", content_type) + .header("DD-API-KEY", api_key); + + let compression = self.compression.unwrap_or(Compression::Gzip(None)); + + let (request, body) = match compression { + Compression::None => (request, body), + Compression::Gzip(level) => { + // Default the compression level to 6, which is similar to datadog agent. + // https://docs.datadoghq.com/agent/logs/log_transport/?tab=https#log-compression + let level = level.unwrap_or(6); + let mut encoder = + GzEncoder::new(Vec::new(), flate2::Compression::new(level as u32)); + + encoder.write_all(&body)?; + ( + request.header("Content-Encoding", "gzip"), + encoder.finish()?, + ) + } + }; + + request + .header("Content-Length", body.len()) + .body(body) + .map_err(Into::into) + } +} + +#[async_trait::async_trait] +#[typetag::serde(name = "datadog_events")] +impl SinkConfig for DatadogEventsConfig { + async fn build(&self, cx: SinkContext) -> crate::Result<(VectorSink, Healthcheck)> { + let mut config = self.clone(); + + // We modify encoding to fit DataDog requirements + config.encoding.timestamp_format = config + .encoding + .timestamp_format + .or(Some(TimestampFormat::Unix)); + + let batch_settings = self.batch_settings()?; + self.build_sink( + cx, + DatadogEventsService { + config: self.clone(), + uri: self.get_uri(), + default_api_key: Arc::from(self.default_api_key.clone()), + }, + JsonArrayBuffer::new(batch_settings.size), + batch_settings.timeout, + ) + } + + fn input_type(&self) -> DataType { + DataType::Log + } + + fn sink_type(&self) -> &'static str { + "datadog_events" + } +} + +#[derive(Clone)] +struct DatadogEventsService { + config: DatadogEventsConfig, + // Used to store the complete URI and avoid calling `get_uri` for each request + uri: String, + default_api_key: ApiKey, +} + +#[async_trait::async_trait] +impl HttpSink for DatadogEventsService { + type Input = PartitionInnerBuffer; + type Output = PartitionInnerBuffer, ApiKey>; + + fn encode_event(&self, mut event: Event) -> Option> { + let log = event.as_mut_log(); + + if !log.contains("title") { + emit!(DatadogEventsFieldInvalid { field: "title" }); + } + + if let Some(message) = log.remove(log_schema().message_key()) { + log.insert("text", message); + } else if !log.contains("text") { + emit!(DatadogEventsFieldInvalid { + field: log_schema().message_key() + }); + } + + if let Some(host) = log.remove(log_schema().host_key()) { + log.insert("host", host); + } + + self.config.encoding.apply_rules(&mut event); + + if let Some(timestamp) = event.as_mut_log().remove(log_schema().timestamp_key()) { + event.as_mut_log().insert("date_happened", timestamp); + } + + let (fields, metadata) = event.into_log().into_parts(); + let json_event = json!(fields); + let api_key = metadata + .datadog_api_key() + .as_ref() + .unwrap_or(&self.default_api_key); + + Some(EncodedEvent { + item: PartitionInnerBuffer::new(json_event, Arc::clone(api_key)), + metadata: Some(metadata), + }) + } + + async fn build_request(&self, events: Self::Output) -> crate::Result>> { + let (events, api_key) = events.into_parts(); + + let body = serde_json::to_vec(&events)?; + // check the number of events to ignore health-check requests + if !events.is_empty() { + emit!(DatadogEventsProcessed { + byte_size: body.len(), + }); + } + self.config + .build_request(self.uri.as_str(), &api_key[..], "application/json", body) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::{ + config::SinkConfig, + sinks::util::test::{build_test_server_status, load_sink}, + test_util::{next_addr, random_lines_with_stream}, + }; + use futures::{ + channel::mpsc::{Receiver, TryRecvError}, + stream, StreamExt, + }; + use hyper::StatusCode; + use indoc::indoc; + use pretty_assertions::assert_eq; + use vector_core::event::{BatchNotifier, BatchStatus}; + + #[test] + fn generate_config() { + crate::test_util::test_generate_config::(); + } + + fn event_with_api_key(msg: &str, key: &str) -> Event { + let mut e = Event::from(msg); + e.as_mut_log() + .metadata_mut() + .set_datadog_api_key(Some(Arc::from(key))); + e + } + #[tokio::test] + async fn smoke() { + let (expected, rx) = start_test(StatusCode::OK, BatchStatus::Delivered).await; + + let output = rx.take(expected.len()).collect::>().await; + + for (i, val) in output.iter().enumerate() { + assert_eq!( + val.0.headers.get("Content-Type").unwrap(), + "application/json" + ); + + let mut json = serde_json::Deserializer::from_slice(&val.1[..]) + .into_iter::() + .map(|v| v.expect("decoding json")); + + let json = json.next().unwrap(); + + // The json we send to Datadog is an array of events. + // As we have set batch.max_events to 1, each entry will be + // an array containing a single record. + let message = json + .get(0) + .unwrap() + .get("message") + .unwrap() + .as_str() + .unwrap(); + assert_eq!(message, expected[i]); + } + } + + #[tokio::test] + async fn handles_failure() { + let (_expected, mut rx) = start_test(StatusCode::FORBIDDEN, BatchStatus::Failed).await; + + assert!(matches!(rx.try_next(), Err(TryRecvError { .. }))); + } + + async fn start_test( + http_status: StatusCode, + batch_status: BatchStatus, + ) -> (Vec, Receiver<(http::request::Parts, Bytes)>) { + let config = indoc! {r#" + default_api_key = "atoken" + compression = "none" + batch.max_events = 1 + "#}; + let (mut config, cx) = load_sink::(&config).unwrap(); + + let addr = next_addr(); + // Swap out the endpoint so we can force send it + // to our local server + let endpoint = format!("http://{}", addr); + config.endpoint = Some(endpoint.clone()); + + let (sink, _) = config.build(cx).await.unwrap(); + + let (rx, _trigger, server) = build_test_server_status(addr, http_status); + tokio::spawn(server); + + let (batch, mut receiver) = BatchNotifier::new_with_receiver(); + let (expected, events) = random_lines_with_stream(100, 10, Some(batch)); + + let _ = sink.run(events).await.unwrap(); + + assert_eq!(receiver.try_recv(), Ok(batch_status)); + + (expected, rx) + } + + #[tokio::test] + async fn api_key_in_metadata() { + let (mut config, cx) = load_sink::(indoc! {r#" + default_api_key = "atoken" + compression = "none" + batch.max_events = 1 + "#}) + .unwrap(); + + let addr = next_addr(); + // Swap out the endpoint so we can force send it + // to our local server + let endpoint = format!("http://{}", addr); + config.endpoint = Some(endpoint.clone()); + + let (sink, _) = config.build(cx).await.unwrap(); + + let (rx, _trigger, server) = build_test_server_status(addr, StatusCode::OK); + tokio::spawn(server); + + let (expected, events) = random_lines_with_stream(100, 10, None); + + let mut events = events.map(|mut e| { + e.as_mut_log() + .metadata_mut() + .set_datadog_api_key(Some(Arc::from("from_metadata"))); + Ok(e) + }); + + let _ = sink.into_sink().send_all(&mut events).await.unwrap(); + let output = rx.take(expected.len()).collect::>().await; + + for (i, val) in output.iter().enumerate() { + assert_eq!(val.0.headers.get("DD-API-KEY").unwrap(), "from_metadata"); + + assert_eq!( + val.0.headers.get("Content-Type").unwrap(), + "application/json" + ); + + let mut json = serde_json::Deserializer::from_slice(&val.1[..]) + .into_iter::() + .map(|v| v.expect("decoding json")); + + let json = json.next().unwrap(); + + // The json we send to Datadog is an array of events. + // As we have set batch.max_events to 1, each entry will be + // an array containing a single record. + let message = json + .get(0) + .unwrap() + .get("message") + .unwrap() + .as_str() + .unwrap(); + assert_eq!(message, expected[i]); + } + } + + #[tokio::test] + async fn multiple_api_keys() { + let (mut config, cx) = load_sink::(indoc! {r#" + default_api_key = "atoken" + compression = "none" + batch.max_events = 1 + "#}) + .unwrap(); + + let addr = next_addr(); + // Swap out the endpoint so we can force send it + // to our local server + let endpoint = format!("http://{}", addr); + config.endpoint = Some(endpoint.clone()); + + let (sink, _) = config.build(cx).await.unwrap(); + + let (rx, _trigger, server) = build_test_server_status(addr, StatusCode::OK); + tokio::spawn(server); + + let events = vec![ + event_with_api_key("mow", "pkc"), + event_with_api_key("pnh", "vvo"), + Event::from("no API key in metadata"), + ]; + + let _ = sink.run(stream::iter(events)).await.unwrap(); + + let mut keys = rx + .take(3) + .map(|r| r.0.headers.get("DD-API-KEY").unwrap().clone()) + .collect::>() + .await; + + keys.sort(); + assert_eq!(keys, vec!["atoken", "pkc", "vvo"]) + } +} diff --git a/src/sinks/datadog/metrics.rs b/src/sinks/datadog/metrics.rs index 1d6168433f499..836e4919656ce 100644 --- a/src/sinks/datadog/metrics.rs +++ b/src/sinks/datadog/metrics.rs @@ -1,3 +1,4 @@ +use super::healthcheck; use crate::{ config::{DataType, SinkConfig, SinkContext, SinkDescription}, event::metric::{Metric, MetricKind, MetricValue, Sample, StatisticKind}, @@ -182,7 +183,12 @@ impl_generate_config_from_default!(DatadogConfig); impl SinkConfig for DatadogConfig { async fn build(&self, cx: SinkContext) -> crate::Result<(VectorSink, Healthcheck)> { let client = HttpClient::new(None)?; - let healthcheck = healthcheck(self.clone(), client.clone()).boxed(); + let healthcheck = healthcheck( + self.get_api_endpoint().clone(), + self.api_key.clone(), + client.clone(), + ) + .boxed(); let batch = BatchSettings::default() .events(20) @@ -279,24 +285,6 @@ fn build_uri(host: &str, endpoint: &'static str) -> crate::Result { Ok(uri) } -async fn healthcheck(config: DatadogConfig, client: HttpClient) -> crate::Result<()> { - let uri = format!("{}/api/v1/validate", config.get_api_endpoint()) - .parse::() - .context(UriParseError)?; - - let request = Request::get(uri) - .header("DD-API-KEY", config.api_key) - .body(hyper::Body::empty()) - .unwrap(); - - let response = client.send(request).await?; - - match response.status() { - StatusCode::OK => Ok(()), - other => Err(HealthcheckError::UnexpectedStatus { status: other }.into()), - } -} - fn encode_tags(tags: &BTreeMap) -> Vec { let mut pairs: Vec<_> = tags .iter() diff --git a/src/sinks/datadog/mod.rs b/src/sinks/datadog/mod.rs index 1224c93cd9c14..ec3571224e71c 100644 --- a/src/sinks/datadog/mod.rs +++ b/src/sinks/datadog/mod.rs @@ -1,5 +1,12 @@ +use crate::{ + http::HttpClient, + sinks::{Healthcheck, HealthcheckError, UriParseError, VectorSink}, +}; +use http::{uri::InvalidUri, Request, StatusCode, Uri}; use serde::{Deserialize, Serialize}; +use snafu::{ResultExt, Snafu}; +pub mod events; pub mod logs; pub mod metrics; @@ -9,3 +16,21 @@ pub enum Region { Us, Eu, } + +async fn healthcheck(endpoint: String, api_key: String, client: HttpClient) -> crate::Result<()> { + let uri = format!("{}/api/v1/validate", endpoint) + .parse::() + .context(UriParseError)?; + + let request = Request::get(uri) + .header("DD-API-KEY", api_key) + .body(hyper::Body::empty()) + .unwrap(); + + let response = client.send(request).await?; + + match response.status() { + StatusCode::OK => Ok(()), + other => Err(HealthcheckError::UnexpectedStatus { status: other }.into()), + } +} From 737016364897ca881944736c033056727895be06 Mon Sep 17 00:00:00 2001 From: ktf Date: Fri, 28 May 2021 18:42:46 +0200 Subject: [PATCH 02/16] Fixes Signed-off-by: ktf --- src/internal_events/datadog_events.rs | 8 ++-- src/sinks/datadog/events.rs | 65 ++++++++++++++------------- 2 files changed, 39 insertions(+), 34 deletions(-) diff --git a/src/internal_events/datadog_events.rs b/src/internal_events/datadog_events.rs index 59572941eb931..f6a403e3657cd 100644 --- a/src/internal_events/datadog_events.rs +++ b/src/internal_events/datadog_events.rs @@ -13,11 +13,11 @@ impl InternalEvent for DatadogEventsProcessed { } #[derive(Debug)] -pub struct DatadogEventsFieldInvalid<'a> { - pub field: &'a str, +pub struct DatadogEventsFieldInvalid { + pub field: &'static str, } -impl<'a> InternalEvent for DatadogEventsFieldInvalid<'a> { +impl InternalEvent for DatadogEventsFieldInvalid { fn emit_logs(&self) { debug!( message = "Required field is missing.", @@ -30,6 +30,6 @@ impl<'a> InternalEvent for DatadogEventsFieldInvalid<'a> { counter!( "processing_errors_total", 1, "error_type" => "field_missing", - "field" => self.field.to_owned()); + "field" => self.field); } } diff --git a/src/sinks/datadog/events.rs b/src/sinks/datadog/events.rs index fe95c0eaff900..996bb8fa4b285 100644 --- a/src/sinks/datadog/events.rs +++ b/src/sinks/datadog/events.rs @@ -8,10 +8,12 @@ use crate::{ util::{ batch::{Batch, BatchError}, encode_event, - encoding::{EncodingConfig, EncodingConfiguration, TimestampFormat}, + encoding::{ + EncodingConfig, EncodingConfigWithDefault, EncodingConfiguration, TimestampFormat, + }, http::{HttpSink, PartitionHttpSink}, - BatchConfig, BatchSettings, BoxedRawValue, Compression, EncodedEvent, Encoding, - JsonArrayBuffer, PartitionBuffer, PartitionInnerBuffer, TowerRequestConfig, VecBuffer, + BatchConfig, BatchSettings, BoxedRawValue, Compression, EncodedEvent, JsonArrayBuffer, + PartitionBuffer, PartitionInnerBuffer, TowerRequestConfig, VecBuffer, }, Healthcheck, VectorSink, }, @@ -27,6 +29,18 @@ use serde::{Deserialize, Serialize}; use serde_json::json; use std::{io::Write, sync::Arc, time::Duration}; +#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum Encoding { + Json, +} + +impl Default for Encoding { + fn default() -> Self { + Self::Json + } +} + #[derive(Deserialize, Serialize, Debug, Clone)] #[serde(deny_unknown_fields)] pub struct DatadogEventsConfig { @@ -35,7 +49,11 @@ pub struct DatadogEventsConfig { #[serde(default = "default_site")] site: String, default_api_key: String, - encoding: EncodingConfig<()>, + #[serde( + skip_serializing_if = "crate::serde::skip_serializing_if_default", + default + )] + encoding: EncodingConfigWithDefault, tls: Option, @@ -179,14 +197,6 @@ impl DatadogEventsConfig { #[typetag::serde(name = "datadog_events")] impl SinkConfig for DatadogEventsConfig { async fn build(&self, cx: SinkContext) -> crate::Result<(VectorSink, Healthcheck)> { - let mut config = self.clone(); - - // We modify encoding to fit DataDog requirements - config.encoding.timestamp_format = config - .encoding - .timestamp_format - .or(Some(TimestampFormat::Unix)); - let batch_settings = self.batch_settings()?; self.build_sink( cx, @@ -241,11 +251,18 @@ impl HttpSink for DatadogEventsService { log.insert("host", host); } - self.config.encoding.apply_rules(&mut event); - - if let Some(timestamp) = event.as_mut_log().remove(log_schema().timestamp_key()) { - event.as_mut_log().insert("date_happened", timestamp); + match log.remove(log_schema().timestamp_key()) { + Some(Value::Integer(timestamp)) => Some(timestamp), + Some(Value::Timestamp(timestamp)) => Some(timestamp.timestamp()), + Some(value) => { + log.insert(log_schema().timestamp_key(), value); + None + } + _ => None, } + .map(|timestamp| log.insert("date_happened", timestamp)); + + self.config.encoding.apply_rules(&mut event); let (fields, metadata) = event.into_log().into_parts(); let json_event = json!(fields); @@ -325,13 +342,7 @@ mod tests { // The json we send to Datadog is an array of events. // As we have set batch.max_events to 1, each entry will be // an array containing a single record. - let message = json - .get(0) - .unwrap() - .get("message") - .unwrap() - .as_str() - .unwrap(); + let message = json.get(0).unwrap().get("text").unwrap().as_str().unwrap(); assert_eq!(message, expected[i]); } } @@ -424,13 +435,7 @@ mod tests { // The json we send to Datadog is an array of events. // As we have set batch.max_events to 1, each entry will be // an array containing a single record. - let message = json - .get(0) - .unwrap() - .get("message") - .unwrap() - .as_str() - .unwrap(); + let message = json.get(0).unwrap().get("text").unwrap().as_str().unwrap(); assert_eq!(message, expected[i]); } } From e1d3e88ae890cb77d555a53694697a7042c918d3 Mon Sep 17 00:00:00 2001 From: ktf Date: Fri, 28 May 2021 22:55:23 +0200 Subject: [PATCH 03/16] Fixes Signed-off-by: ktf --- .../src/event/util/log/path_iter.rs | 6 + src/sinks/datadog/events.rs | 196 ++++++++---------- src/sinks/datadog/metrics.rs | 4 +- src/sinks/datadog/mod.rs | 6 +- 4 files changed, 98 insertions(+), 114 deletions(-) diff --git a/lib/vector-core/src/event/util/log/path_iter.rs b/lib/vector-core/src/event/util/log/path_iter.rs index fc53c2e640642..c0c6aa96f1f72 100644 --- a/lib/vector-core/src/event/util/log/path_iter.rs +++ b/lib/vector-core/src/event/util/log/path_iter.rs @@ -17,6 +17,12 @@ pub enum PathComponent { Invalid, } +impl<'a> From<&'a str> for PathComponent { + fn from(key: &'a str) -> Self { + Self::Key(key.to_string()) + } +} + /// Iterator over components of paths specified in form `a.b[0].c[2]`. pub struct PathIter<'a> { path: &'a str, diff --git a/src/sinks/datadog/events.rs b/src/sinks/datadog/events.rs index 996bb8fa4b285..ac16b4a6c8714 100644 --- a/src/sinks/datadog/events.rs +++ b/src/sinks/datadog/events.rs @@ -1,33 +1,27 @@ use super::healthcheck; use crate::{ config::{log_schema, DataType, GenerateConfig, SinkConfig, SinkContext, SinkDescription}, - event::{Event, Value}, + event::Event, http::HttpClient, internal_events::{DatadogEventsFieldInvalid, DatadogEventsProcessed}, sinks::{ util::{ - batch::{Batch, BatchError}, - encode_event, - encoding::{ - EncodingConfig, EncodingConfigWithDefault, EncodingConfiguration, TimestampFormat, - }, + batch::Batch, + encoding::{EncodingConfigWithDefault, EncodingConfiguration, TimestampFormat}, http::{HttpSink, PartitionHttpSink}, - BatchConfig, BatchSettings, BoxedRawValue, Compression, EncodedEvent, JsonArrayBuffer, - PartitionBuffer, PartitionInnerBuffer, TowerRequestConfig, VecBuffer, + BatchConfig, BatchSettings, BoxedRawValue, EncodedEvent, JsonArrayBuffer, + PartitionBuffer, PartitionInnerBuffer, TowerRequestConfig, }, Healthcheck, VectorSink, }, tls::{MaybeTlsSettings, TlsConfig}, }; -use bytes::Bytes; -use flate2::write::GzEncoder; use futures::{FutureExt, SinkExt}; -use http::{Request, StatusCode}; -use hyper::body::Body; +use http::Request; use indoc::indoc; use serde::{Deserialize, Serialize}; use serde_json::json; -use std::{io::Write, sync::Arc, time::Duration}; +use std::{sync::Arc, time::Duration}; #[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)] #[serde(rename_all = "snake_case")] @@ -49,20 +43,9 @@ pub struct DatadogEventsConfig { #[serde(default = "default_site")] site: String, default_api_key: String, - #[serde( - skip_serializing_if = "crate::serde::skip_serializing_if_default", - default - )] - encoding: EncodingConfigWithDefault, tls: Option, - #[serde(default)] - compression: Option, - - #[serde(default)] - batch: BatchConfig, - #[serde(default)] request: TowerRequestConfig, } @@ -91,26 +74,12 @@ impl DatadogEventsConfig { format!("{}/api/v1/events", self.get_api_endpoint()) } - // The API endpoint is used for healtcheck/API key validation, it is derived from the `site` or `region` option - // the same way the offical Datadog Agent does but the `endpoint` option still override both the main and the - // API endpoint. fn get_api_endpoint(&self) -> String { self.endpoint .clone() .unwrap_or_else(|| format!("https://api.{}", &self.site)) } - fn batch_settings(&self) -> Result, BatchError> { - BatchSettings::default() - .bytes(bytesize::kib(100u64)) - .events(20) - .timeout(1) - .parse_config(self.batch) - } - - /// Builds the required BatchedHttpSink. - /// Since the DataDog sink can create one of two different sinks, this - /// extracts most of the shared functionality required to create either sink. fn build_sink( &self, cx: SinkContext, @@ -154,57 +123,21 @@ impl DatadogEventsConfig { Ok((VectorSink::Sink(Box::new(sink)), healthcheck)) } - - /// Build the request, GZipping the contents if the config specifies. - fn build_request( - &self, - uri: &str, - api_key: &str, - content_type: &str, - body: Vec, - ) -> crate::Result>> { - let request = Request::post(uri) - .header("Content-Type", content_type) - .header("DD-API-KEY", api_key); - - let compression = self.compression.unwrap_or(Compression::Gzip(None)); - - let (request, body) = match compression { - Compression::None => (request, body), - Compression::Gzip(level) => { - // Default the compression level to 6, which is similar to datadog agent. - // https://docs.datadoghq.com/agent/logs/log_transport/?tab=https#log-compression - let level = level.unwrap_or(6); - let mut encoder = - GzEncoder::new(Vec::new(), flate2::Compression::new(level as u32)); - - encoder.write_all(&body)?; - ( - request.header("Content-Encoding", "gzip"), - encoder.finish()?, - ) - } - }; - - request - .header("Content-Length", body.len()) - .body(body) - .map_err(Into::into) - } } #[async_trait::async_trait] #[typetag::serde(name = "datadog_events")] impl SinkConfig for DatadogEventsConfig { async fn build(&self, cx: SinkContext) -> crate::Result<(VectorSink, Healthcheck)> { - let batch_settings = self.batch_settings()?; + let batch_settings = BatchSettings::default() + .bytes(bytesize::kib(100u64)) + .events(1) + .timeout(0) + .parse_config(BatchConfig::default())?; + self.build_sink( cx, - DatadogEventsService { - config: self.clone(), - uri: self.get_uri(), - default_api_key: Arc::from(self.default_api_key.clone()), - }, + DatadogEventsService::new(self), JsonArrayBuffer::new(batch_settings.size), batch_settings.timeout, ) @@ -225,6 +158,41 @@ struct DatadogEventsService { // Used to store the complete URI and avoid calling `get_uri` for each request uri: String, default_api_key: ApiKey, + encoding: EncodingConfigWithDefault<()>, +} + +impl DatadogEventsService { + fn new(config: &DatadogEventsConfig) -> Self { + let mut encoding = EncodingConfigWithDefault::default(); + + // DataDog Event API allows only some fields, and refuses + // to accept event if it contains any other field. + encoding.only_fields = Some(vec![ + vec!["aggregation_key".into()], + vec!["alert_type".into()], + vec!["date_happened".into()], + vec!["device_name".into()], + vec!["host".into()], + vec!["priority".into()], + vec!["related_event_id".into()], + vec!["source_type_name".into()], + vec!["tags".into()], + vec!["text".into()], + vec!["title".into()], + ]); + + // DataDog Event API requires unix timestamp. + encoding.timestamp_format = Some(TimestampFormat::Unix); + + Self { + default_api_key: Arc::from(config.default_api_key.clone()), + + uri: config.get_uri(), + + encoding, + config: config.clone(), + } + } } #[async_trait::async_trait] @@ -237,32 +205,39 @@ impl HttpSink for DatadogEventsService { if !log.contains("title") { emit!(DatadogEventsFieldInvalid { field: "title" }); + return None; } - if let Some(message) = log.remove(log_schema().message_key()) { - log.insert("text", message); - } else if !log.contains("text") { - emit!(DatadogEventsFieldInvalid { - field: log_schema().message_key() - }); + if !log.contains("text") { + if let Some(message) = log.remove(log_schema().message_key()) { + log.insert("text", message); + } else { + emit!(DatadogEventsFieldInvalid { + field: log_schema().message_key() + }); + return None; + } } - if let Some(host) = log.remove(log_schema().host_key()) { - log.insert("host", host); + if !log.contains("host") { + if let Some(host) = log.remove(log_schema().host_key()) { + log.insert("host", host); + } + } + + if !log.contains("date_happened") { + if let Some(timestamp) = log.remove(log_schema().timestamp_key()) { + log.insert("date_happened", timestamp); + } } - match log.remove(log_schema().timestamp_key()) { - Some(Value::Integer(timestamp)) => Some(timestamp), - Some(Value::Timestamp(timestamp)) => Some(timestamp.timestamp()), - Some(value) => { - log.insert(log_schema().timestamp_key(), value); - None + if !log.contains("source_type_name") { + if let Some(name) = log.remove(log_schema().source_type_key()) { + log.insert("source_type_name", name); } - _ => None, } - .map(|timestamp| log.insert("date_happened", timestamp)); - self.config.encoding.apply_rules(&mut event); + self.encoding.apply_rules(&mut event); let (fields, metadata) = event.into_log().into_parts(); let json_event = json!(fields); @@ -278,17 +253,20 @@ impl HttpSink for DatadogEventsService { } async fn build_request(&self, events: Self::Output) -> crate::Result>> { - let (events, api_key) = events.into_parts(); - - let body = serde_json::to_vec(&events)?; - // check the number of events to ignore health-check requests - if !events.is_empty() { - emit!(DatadogEventsProcessed { - byte_size: body.len(), - }); - } - self.config - .build_request(self.uri.as_str(), &api_key[..], "application/json", body) + let (mut events, api_key) = events.into_parts(); + + let body = serde_json::to_vec(&events.pop().unwrap())?; + + emit!(DatadogEventsProcessed { + byte_size: body.len(), + }); + + Request::post(self.uri.as_str()) + .header("Content-Type", "application/json") + .header("DD-API-KEY", &api_key[..]) + .header("Content-Length", body.len()) + .body(body) + .map_err(Into::into) } } diff --git a/src/sinks/datadog/metrics.rs b/src/sinks/datadog/metrics.rs index 836e4919656ce..064ad98774235 100644 --- a/src/sinks/datadog/metrics.rs +++ b/src/sinks/datadog/metrics.rs @@ -13,12 +13,12 @@ use crate::{ EncodedEvent, PartitionBatchSink, PartitionBuffer, PartitionInnerBuffer, TowerRequestConfig, }, - Healthcheck, HealthcheckError, UriParseError, VectorSink, + Healthcheck, UriParseError, VectorSink, }, }; use chrono::{DateTime, Utc}; use futures::{stream, FutureExt, SinkExt}; -use http::{uri::InvalidUri, Request, StatusCode, Uri}; +use http::{uri::InvalidUri, Request, Uri}; use lazy_static::lazy_static; use serde::{Deserialize, Serialize}; use snafu::{ResultExt, Snafu}; diff --git a/src/sinks/datadog/mod.rs b/src/sinks/datadog/mod.rs index ec3571224e71c..be6c773fe8cae 100644 --- a/src/sinks/datadog/mod.rs +++ b/src/sinks/datadog/mod.rs @@ -1,10 +1,10 @@ use crate::{ http::HttpClient, - sinks::{Healthcheck, HealthcheckError, UriParseError, VectorSink}, + sinks::{HealthcheckError, UriParseError}, }; -use http::{uri::InvalidUri, Request, StatusCode, Uri}; +use http::{Request, StatusCode, Uri}; use serde::{Deserialize, Serialize}; -use snafu::{ResultExt, Snafu}; +use snafu::ResultExt; pub mod events; pub mod logs; From 907c2e7302471068b443db21243160fef9651507 Mon Sep 17 00:00:00 2001 From: ktf Date: Sat, 29 May 2021 10:06:28 +0200 Subject: [PATCH 04/16] Tests Signed-off-by: ktf --- src/sinks/datadog/events.rs | 174 ++++++++++++++++-------------------- 1 file changed, 77 insertions(+), 97 deletions(-) diff --git a/src/sinks/datadog/events.rs b/src/sinks/datadog/events.rs index ac16b4a6c8714..c03e947ceece7 100644 --- a/src/sinks/datadog/events.rs +++ b/src/sinks/datadog/events.rs @@ -9,7 +9,7 @@ use crate::{ batch::Batch, encoding::{EncodingConfigWithDefault, EncodingConfiguration, TimestampFormat}, http::{HttpSink, PartitionHttpSink}, - BatchConfig, BatchSettings, BoxedRawValue, EncodedEvent, JsonArrayBuffer, + BatchConfig, BatchSettings, BoxedRawValue, Concurrency, EncodedEvent, JsonArrayBuffer, PartitionBuffer, PartitionInnerBuffer, TowerRequestConfig, }, Healthcheck, VectorSink, @@ -23,18 +23,6 @@ use serde::{Deserialize, Serialize}; use serde_json::json; use std::{sync::Arc, time::Duration}; -#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)] -#[serde(rename_all = "snake_case")] -pub enum Encoding { - Json, -} - -impl Default for Encoding { - fn default() -> Self { - Self::Json - } -} - #[derive(Deserialize, Serialize, Debug, Clone)] #[serde(deny_unknown_fields)] pub struct DatadogEventsConfig { @@ -47,7 +35,7 @@ pub struct DatadogEventsConfig { tls: Option, #[serde(default)] - request: TowerRequestConfig, + request: TowerRequestConfig, } type ApiKey = Arc; @@ -97,7 +85,11 @@ impl DatadogEventsConfig { Output = PartitionInnerBuffer, > + Clone, { - let request_settings = self.request.unwrap_with(&TowerRequestConfig::default()); + let mut request_opts = self.request.clone(); + // Since we are sending only one event per request we should try to send as much + // requests in parallel as possible. + request_opts.concurrency = request_opts.concurrency.if_none(Concurrency::Adaptive); + let request_settings = request_opts.unwrap_with(&TowerRequestConfig::default()); let tls_settings = MaybeTlsSettings::from_config( &Some(self.tls.clone().unwrap_or_else(TlsConfig::enabled)), @@ -155,7 +147,6 @@ impl SinkConfig for DatadogEventsConfig { #[derive(Clone)] struct DatadogEventsService { config: DatadogEventsConfig, - // Used to store the complete URI and avoid calling `get_uri` for each request uri: String, default_api_key: ApiKey, encoding: EncodingConfigWithDefault<()>, @@ -186,9 +177,7 @@ impl DatadogEventsService { Self { default_api_key: Arc::from(config.default_api_key.clone()), - uri: config.get_uri(), - encoding, config: config.clone(), } @@ -237,7 +226,7 @@ impl HttpSink for DatadogEventsService { } } - self.encoding.apply_rules(&mut event); + self.encoding.apply_rules(&mut event); let (fields, metadata) = event.into_log().into_parts(); let json_event = json!(fields); @@ -255,7 +244,8 @@ impl HttpSink for DatadogEventsService { async fn build_request(&self, events: Self::Output) -> crate::Result>> { let (mut events, api_key) = events.into_parts(); - let body = serde_json::to_vec(&events.pop().unwrap())?; + assert_eq!(events.len(), 1); + let body = serde_json::to_vec(&events.pop().expect("One event"))?; emit!(DatadogEventsProcessed { byte_size: body.len(), @@ -278,9 +268,11 @@ mod tests { sinks::util::test::{build_test_server_status, load_sink}, test_util::{next_addr, random_lines_with_stream}, }; + use bytes::Bytes; use futures::{ channel::mpsc::{Receiver, TryRecvError}, - stream, StreamExt, + stream::Stream, + StreamExt, }; use hyper::StatusCode; use indoc::indoc; @@ -292,13 +284,52 @@ mod tests { crate::test_util::test_generate_config::(); } - fn event_with_api_key(msg: &str, key: &str) -> Event { - let mut e = Event::from(msg); - e.as_mut_log() - .metadata_mut() - .set_datadog_api_key(Some(Arc::from(key))); - e + fn random_events_with_stream( + len: usize, + count: usize, + batch: Option>, + ) -> (Vec, impl Stream) { + let (lines, stream) = random_lines_with_stream(len, count, batch); + ( + lines, + stream.map(|mut event| { + event.as_mut_log().insert("title", "All!"); + event.as_mut_log().insert("invalid", "Tik"); + event + }), + ) } + + async fn start_test( + http_status: StatusCode, + batch_status: BatchStatus, + ) -> (Vec, Receiver<(http::request::Parts, Bytes)>) { + let config = indoc! {r#" + default_api_key = "atoken" + "#}; + let (mut config, cx) = load_sink::(&config).unwrap(); + + let addr = next_addr(); + // Swap out the endpoint so we can force send it + // to our local server + let endpoint = format!("http://{}", addr); + config.endpoint = Some(endpoint.clone()); + + let (sink, _) = config.build(cx).await.unwrap(); + + let (rx, _trigger, server) = build_test_server_status(addr, http_status); + tokio::spawn(server); + + let (batch, mut receiver) = BatchNotifier::new_with_receiver(); + let (expected, events) = random_events_with_stream(100, 10, Some(batch)); + + let _ = sink.run(events).await.unwrap(); + + assert_eq!(receiver.try_recv(), Ok(batch_status)); + + (expected, rx) + } + #[tokio::test] async fn smoke() { let (expected, rx) = start_test(StatusCode::OK, BatchStatus::Delivered).await; @@ -320,7 +351,7 @@ mod tests { // The json we send to Datadog is an array of events. // As we have set batch.max_events to 1, each entry will be // an array containing a single record. - let message = json.get(0).unwrap().get("text").unwrap().as_str().unwrap(); + let message = json.get("text").unwrap().as_str().unwrap(); assert_eq!(message, expected[i]); } } @@ -332,44 +363,10 @@ mod tests { assert!(matches!(rx.try_next(), Err(TryRecvError { .. }))); } - async fn start_test( - http_status: StatusCode, - batch_status: BatchStatus, - ) -> (Vec, Receiver<(http::request::Parts, Bytes)>) { - let config = indoc! {r#" - default_api_key = "atoken" - compression = "none" - batch.max_events = 1 - "#}; - let (mut config, cx) = load_sink::(&config).unwrap(); - - let addr = next_addr(); - // Swap out the endpoint so we can force send it - // to our local server - let endpoint = format!("http://{}", addr); - config.endpoint = Some(endpoint.clone()); - - let (sink, _) = config.build(cx).await.unwrap(); - - let (rx, _trigger, server) = build_test_server_status(addr, http_status); - tokio::spawn(server); - - let (batch, mut receiver) = BatchNotifier::new_with_receiver(); - let (expected, events) = random_lines_with_stream(100, 10, Some(batch)); - - let _ = sink.run(events).await.unwrap(); - - assert_eq!(receiver.try_recv(), Ok(batch_status)); - - (expected, rx) - } - #[tokio::test] async fn api_key_in_metadata() { let (mut config, cx) = load_sink::(indoc! {r#" default_api_key = "atoken" - compression = "none" - batch.max_events = 1 "#}) .unwrap(); @@ -384,7 +381,7 @@ mod tests { let (rx, _trigger, server) = build_test_server_status(addr, StatusCode::OK); tokio::spawn(server); - let (expected, events) = random_lines_with_stream(100, 10, None); + let (expected, events) = random_events_with_stream(100, 10, None); let mut events = events.map(|mut e| { e.as_mut_log() @@ -410,49 +407,32 @@ mod tests { let json = json.next().unwrap(); - // The json we send to Datadog is an array of events. - // As we have set batch.max_events to 1, each entry will be - // an array containing a single record. - let message = json.get(0).unwrap().get("text").unwrap().as_str().unwrap(); + let message = json.get("text").unwrap().as_str().unwrap(); assert_eq!(message, expected[i]); } } #[tokio::test] - async fn multiple_api_keys() { - let (mut config, cx) = load_sink::(indoc! {r#" - default_api_key = "atoken" - compression = "none" - batch.max_events = 1 - "#}) - .unwrap(); - - let addr = next_addr(); - // Swap out the endpoint so we can force send it - // to our local server - let endpoint = format!("http://{}", addr); - config.endpoint = Some(endpoint.clone()); - - let (sink, _) = config.build(cx).await.unwrap(); + async fn filter_out_fields() { + let (expected, rx) = start_test(StatusCode::OK, BatchStatus::Delivered).await; - let (rx, _trigger, server) = build_test_server_status(addr, StatusCode::OK); - tokio::spawn(server); + let output = rx.take(expected.len()).collect::>().await; - let events = vec![ - event_with_api_key("mow", "pkc"), - event_with_api_key("pnh", "vvo"), - Event::from("no API key in metadata"), - ]; + for (i, val) in output.iter().enumerate() { + assert_eq!( + val.0.headers.get("Content-Type").unwrap(), + "application/json" + ); - let _ = sink.run(stream::iter(events)).await.unwrap(); + let mut json = serde_json::Deserializer::from_slice(&val.1[..]) + .into_iter::() + .map(|v| v.expect("decoding json")); - let mut keys = rx - .take(3) - .map(|r| r.0.headers.get("DD-API-KEY").unwrap().clone()) - .collect::>() - .await; + let json = json.next().unwrap(); - keys.sort(); - assert_eq!(keys, vec!["atoken", "pkc", "vvo"]) + let message = json.get("text").unwrap().as_str().unwrap(); + assert_eq!(message, expected[i]); + assert!(json.get("invalid").is_none()); + } } } From 2c3de72c3fad9735cd6af92464e18c80c20774e6 Mon Sep 17 00:00:00 2001 From: ktf Date: Sat, 29 May 2021 10:31:34 +0200 Subject: [PATCH 05/16] Add docs Signed-off-by: ktf --- .../components/sinks/datadog_events.cue | 61 +++++++++++++++++++ docs/reference/services/datadog_events.cue | 10 +++ docs/reference/urls.cue | 2 + src/sinks/datadog/events.rs | 1 + 4 files changed, 74 insertions(+) create mode 100644 docs/reference/components/sinks/datadog_events.cue create mode 100644 docs/reference/services/datadog_events.cue diff --git a/docs/reference/components/sinks/datadog_events.cue b/docs/reference/components/sinks/datadog_events.cue new file mode 100644 index 0000000000000..9280c1d161df2 --- /dev/null +++ b/docs/reference/components/sinks/datadog_events.cue @@ -0,0 +1,61 @@ +package metadata + +components: sinks: datadog_events: { + title: "Datadog Events" + + classes: sinks._datadog.classes + + features: { + buffer: enabled: true + healthcheck: enabled: true + send: { + batch: enabled: false + compression: enabled: false + encoding: enabled: false + request: enabled: true + tls: { + enabled: true + can_enable: true + can_verify_certificate: true + can_verify_hostname: true + enabled_default: true + } + to: { + service: services.datadog_events + + interface: { + socket: { + api: { + title: "Datadog events API" + url: urls.datadog_events_endpoints + } + direction: "outgoing" + protocols: ["http"] + ssl: "required" + } + } + } + } + } + + support: sinks._datadog.support + + configuration: { + default_api_key: { + description: "Default Datadog [API key](https://docs.datadoghq.com/api/?lang=bash#authentication), if an event has a key set in its metadata it will prevail over the one set here." + required: true + warnings: [] + type: string: { + examples: ["${DATADOG_API_KEY_ENV_VAR}", "ef8d5de700e7989468166c40fc8a0ccd"] + syntax: "literal" + } + } + endpoint: sinks._datadog.configuration.endpoint + site: sinks._datadog.configuration.site + } + + input: { + logs: true + metrics: null + } +} diff --git a/docs/reference/services/datadog_events.cue b/docs/reference/services/datadog_events.cue new file mode 100644 index 0000000000000..ae120bc8fdecb --- /dev/null +++ b/docs/reference/services/datadog_events.cue @@ -0,0 +1,10 @@ +package metadata + +services: datadog_events: { + name: "Datadog events" + thing: "a \(name) index" + url: urls.datadog_events + versions: null + + description: services._datadog.description +} diff --git a/docs/reference/urls.cue b/docs/reference/urls.cue index d33451ea5e318..9bd21a8f14308 100644 --- a/docs/reference/urls.cue +++ b/docs/reference/urls.cue @@ -109,6 +109,8 @@ urls: { datadog_docs: "https://docs.datadoghq.com" datadog_logs: "\(datadog_docs)/logs/" datadog_logs_endpoints: "\(datadog_docs)/logs/log_collection/?tab=http#datadog-logs-endpoints" + datadog_events: "\(datadog_docs)/events/" + datadog_events_endpoints: "\(datadog_docs)/api/latest/events/#post-an-event datadog_metrics: "\(datadog_docs)/metrics/" datadog_metrics_endpoints: "\(datadog_docs)/api/v1/metrics/" date: "https://man7.org/linux/man-pages/man1/date.1.html" diff --git a/src/sinks/datadog/events.rs b/src/sinks/datadog/events.rs index c03e947ceece7..bfbeb4e79b1b5 100644 --- a/src/sinks/datadog/events.rs +++ b/src/sinks/datadog/events.rs @@ -121,6 +121,7 @@ impl DatadogEventsConfig { #[typetag::serde(name = "datadog_events")] impl SinkConfig for DatadogEventsConfig { async fn build(&self, cx: SinkContext) -> crate::Result<(VectorSink, Healthcheck)> { + // Datadog Event API doesn't support batching. let batch_settings = BatchSettings::default() .bytes(bytesize::kib(100u64)) .events(1) From e55a75155b47434e39e0e0183fbf60e045eab5ba Mon Sep 17 00:00:00 2001 From: ktf Date: Sat, 29 May 2021 10:39:28 +0200 Subject: [PATCH 06/16] Add semantic Signed-off-by: ktf --- .github/semantic.yml | 1 + 1 file changed, 1 insertion(+) diff --git a/.github/semantic.yml b/.github/semantic.yml index 91546279e281e..bf69f8288a5ee 100644 --- a/.github/semantic.yml +++ b/.github/semantic.yml @@ -180,6 +180,7 @@ scopes: - blackhole sink # Anything `blackhole` sink related - clickhouse sink # Anything `clickhouse` sink related - console sink # Anything `console` sink related + - datadog_events sink # Anything `datadog_events` sink related - datadog_logs sink # Anything `datadog_logs` sink related - datadog_metrics sink # Anything `datadog_metrics` sink related - elasticsearch sink # Anything `elasticsearch` sink related From 99a74ca068a3f738e278c57d225c4d60b0c01e80 Mon Sep 17 00:00:00 2001 From: ktf Date: Sat, 29 May 2021 10:42:43 +0200 Subject: [PATCH 07/16] Move url Signed-off-by: ktf --- docs/reference/urls.cue | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/docs/reference/urls.cue b/docs/reference/urls.cue index 9bd21a8f14308..d0f114ebd049f 100644 --- a/docs/reference/urls.cue +++ b/docs/reference/urls.cue @@ -107,10 +107,10 @@ urls: { datadog_agent: "https://github.com/DataDog/datadog-agent" datadog_distribution: "\(datadog_docs)/developers/metrics/types/?tab=distribution#definition" datadog_docs: "https://docs.datadoghq.com" - datadog_logs: "\(datadog_docs)/logs/" - datadog_logs_endpoints: "\(datadog_docs)/logs/log_collection/?tab=http#datadog-logs-endpoints" datadog_events: "\(datadog_docs)/events/" datadog_events_endpoints: "\(datadog_docs)/api/latest/events/#post-an-event + datadog_logs: "\(datadog_docs)/logs/" + datadog_logs_endpoints: "\(datadog_docs)/logs/log_collection/?tab=http#datadog-logs-endpoints" datadog_metrics: "\(datadog_docs)/metrics/" datadog_metrics_endpoints: "\(datadog_docs)/api/v1/metrics/" date: "https://man7.org/linux/man-pages/man1/date.1.html" From f4163a2baa4696acfc0a7f522606cf8cf6d6de46 Mon Sep 17 00:00:00 2001 From: ktf Date: Sat, 29 May 2021 10:47:20 +0200 Subject: [PATCH 08/16] Fix url Signed-off-by: ktf --- docs/reference/urls.cue | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/reference/urls.cue b/docs/reference/urls.cue index d0f114ebd049f..34fc75dd69a50 100644 --- a/docs/reference/urls.cue +++ b/docs/reference/urls.cue @@ -108,7 +108,7 @@ urls: { datadog_distribution: "\(datadog_docs)/developers/metrics/types/?tab=distribution#definition" datadog_docs: "https://docs.datadoghq.com" datadog_events: "\(datadog_docs)/events/" - datadog_events_endpoints: "\(datadog_docs)/api/latest/events/#post-an-event + datadog_events_endpoints: "\(datadog_docs)/api/latest/events/#post-an-event" datadog_logs: "\(datadog_docs)/logs/" datadog_logs_endpoints: "\(datadog_docs)/logs/log_collection/?tab=http#datadog-logs-endpoints" datadog_metrics: "\(datadog_docs)/metrics/" From 3497017a11f7fec49e806295ede562d3cca98e7e Mon Sep 17 00:00:00 2001 From: ktf Date: Sat, 29 May 2021 10:56:51 +0200 Subject: [PATCH 09/16] Add request docs Signed-off-by: ktf --- docs/reference/components/sinks/datadog_events.cue | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) diff --git a/docs/reference/components/sinks/datadog_events.cue b/docs/reference/components/sinks/datadog_events.cue index 9280c1d161df2..5b414ea9fb29a 100644 --- a/docs/reference/components/sinks/datadog_events.cue +++ b/docs/reference/components/sinks/datadog_events.cue @@ -12,7 +12,17 @@ components: sinks: datadog_events: { batch: enabled: false compression: enabled: false encoding: enabled: false - request: enabled: true + request: { + enabled: true + adaptive_concurrency: true + concurrency: 5 + rate_limit_duration_secs: 1 + rate_limit_num: 5 + retry_initial_backoff_secs: 1 + retry_max_duration_secs: 10 + timeout_secs: 30 + headers: false + } tls: { enabled: true can_enable: true From 26be43ee19ae2a9eb507944b88d53ce1bca13c02 Mon Sep 17 00:00:00 2001 From: ktf Date: Sat, 29 May 2021 11:25:21 +0200 Subject: [PATCH 10/16] Add batch docs Signed-off-by: ktf --- docs/reference/components/sinks/datadog_events.cue | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/docs/reference/components/sinks/datadog_events.cue b/docs/reference/components/sinks/datadog_events.cue index 5b414ea9fb29a..ef30d85bddf6c 100644 --- a/docs/reference/components/sinks/datadog_events.cue +++ b/docs/reference/components/sinks/datadog_events.cue @@ -9,7 +9,11 @@ components: sinks: datadog_events: { buffer: enabled: true healthcheck: enabled: true send: { - batch: enabled: false + batch: { + enabled: false + common: false + timeout_secs: 0 + } compression: enabled: false encoding: enabled: false request: { From 928fb6b5c5c41903afd7d54414475118687cd18c Mon Sep 17 00:00:00 2001 From: ktf Date: Sat, 29 May 2021 11:29:32 +0200 Subject: [PATCH 11/16] Bump Signed-off-by: ktf From c380977f52bf4bdf2d924d1bdc58944937f65c2e Mon Sep 17 00:00:00 2001 From: ktf Date: Sat, 29 May 2021 12:05:03 +0200 Subject: [PATCH 12/16] Clippy Signed-off-by: ktf --- src/sinks/datadog/events.rs | 44 ++++++++++++++++++------------------ src/sinks/datadog/metrics.rs | 2 +- 2 files changed, 23 insertions(+), 23 deletions(-) diff --git a/src/sinks/datadog/events.rs b/src/sinks/datadog/events.rs index bfbeb4e79b1b5..029d0622fe36d 100644 --- a/src/sinks/datadog/events.rs +++ b/src/sinks/datadog/events.rs @@ -85,7 +85,7 @@ impl DatadogEventsConfig { Output = PartitionInnerBuffer, > + Clone, { - let mut request_opts = self.request.clone(); + let mut request_opts = self.request; // Since we are sending only one event per request we should try to send as much // requests in parallel as possible. request_opts.concurrency = request_opts.concurrency.if_none(Concurrency::Adaptive); @@ -98,7 +98,7 @@ impl DatadogEventsConfig { let client = HttpClient::new(tls_settings)?; let healthcheck = healthcheck( - self.get_api_endpoint().clone(), + self.get_api_endpoint(), self.default_api_key.clone(), client.clone(), ) @@ -155,26 +155,26 @@ struct DatadogEventsService { impl DatadogEventsService { fn new(config: &DatadogEventsConfig) -> Self { - let mut encoding = EncodingConfigWithDefault::default(); - - // DataDog Event API allows only some fields, and refuses - // to accept event if it contains any other field. - encoding.only_fields = Some(vec![ - vec!["aggregation_key".into()], - vec!["alert_type".into()], - vec!["date_happened".into()], - vec!["device_name".into()], - vec!["host".into()], - vec!["priority".into()], - vec!["related_event_id".into()], - vec!["source_type_name".into()], - vec!["tags".into()], - vec!["text".into()], - vec!["title".into()], - ]); - - // DataDog Event API requires unix timestamp. - encoding.timestamp_format = Some(TimestampFormat::Unix); + let encoding = EncodingConfigWithDefault { + // DataDog Event API allows only some fields, and refuses + // to accept event if it contains any other field. + only_fields: Some(vec![ + vec!["aggregation_key".into()], + vec!["alert_type".into()], + vec!["date_happened".into()], + vec!["device_name".into()], + vec!["host".into()], + vec!["priority".into()], + vec!["related_event_id".into()], + vec!["source_type_name".into()], + vec!["tags".into()], + vec!["text".into()], + vec!["title".into()], + ]), + // DataDog Event API requires unix timestamp. + timestamp_format: Some(TimestampFormat::Unix), + ..EncodingConfigWithDefault::default() + }; Self { default_api_key: Arc::from(config.default_api_key.clone()), diff --git a/src/sinks/datadog/metrics.rs b/src/sinks/datadog/metrics.rs index 064ad98774235..85005f93057b3 100644 --- a/src/sinks/datadog/metrics.rs +++ b/src/sinks/datadog/metrics.rs @@ -184,7 +184,7 @@ impl SinkConfig for DatadogConfig { async fn build(&self, cx: SinkContext) -> crate::Result<(VectorSink, Healthcheck)> { let client = HttpClient::new(None)?; let healthcheck = healthcheck( - self.get_api_endpoint().clone(), + self.get_api_endpoint(), self.api_key.clone(), client.clone(), ) From 54094d96d9721dbbf940a0974782f2169d0adb34 Mon Sep 17 00:00:00 2001 From: ktf Date: Tue, 1 Jun 2021 12:51:36 +0200 Subject: [PATCH 13/16] Apply feedback Signed-off-by: ktf --- .../src/event/util/log/path_iter.rs | 6 ---- src/internal_events/datadog_events.rs | 2 +- src/sinks/datadog/events.rs | 33 +++++++++++-------- 3 files changed, 20 insertions(+), 21 deletions(-) diff --git a/lib/vector-core/src/event/util/log/path_iter.rs b/lib/vector-core/src/event/util/log/path_iter.rs index c0c6aa96f1f72..fc53c2e640642 100644 --- a/lib/vector-core/src/event/util/log/path_iter.rs +++ b/lib/vector-core/src/event/util/log/path_iter.rs @@ -17,12 +17,6 @@ pub enum PathComponent { Invalid, } -impl<'a> From<&'a str> for PathComponent { - fn from(key: &'a str) -> Self { - Self::Key(key.to_string()) - } -} - /// Iterator over components of paths specified in form `a.b[0].c[2]`. pub struct PathIter<'a> { path: &'a str, diff --git a/src/internal_events/datadog_events.rs b/src/internal_events/datadog_events.rs index f6a403e3657cd..73c21243c71ca 100644 --- a/src/internal_events/datadog_events.rs +++ b/src/internal_events/datadog_events.rs @@ -19,7 +19,7 @@ pub struct DatadogEventsFieldInvalid { impl InternalEvent for DatadogEventsFieldInvalid { fn emit_logs(&self) { - debug!( + error!( message = "Required field is missing.", field = %self.field, internal_log_rate_secs = 10 diff --git a/src/sinks/datadog/events.rs b/src/sinks/datadog/events.rs index 029d0622fe36d..fb77d1e3d7363 100644 --- a/src/sinks/datadog/events.rs +++ b/src/sinks/datadog/events.rs @@ -1,7 +1,7 @@ use super::healthcheck; use crate::{ config::{log_schema, DataType, GenerateConfig, SinkConfig, SinkContext, SinkDescription}, - event::Event, + event::{Event, PathComponent}, http::HttpClient, internal_events::{DatadogEventsFieldInvalid, DatadogEventsProcessed}, sinks::{ @@ -158,19 +158,24 @@ impl DatadogEventsService { let encoding = EncodingConfigWithDefault { // DataDog Event API allows only some fields, and refuses // to accept event if it contains any other field. - only_fields: Some(vec![ - vec!["aggregation_key".into()], - vec!["alert_type".into()], - vec!["date_happened".into()], - vec!["device_name".into()], - vec!["host".into()], - vec!["priority".into()], - vec!["related_event_id".into()], - vec!["source_type_name".into()], - vec!["tags".into()], - vec!["text".into()], - vec!["title".into()], - ]), + only_fields: Some( + [ + "aggregation_key", + "alert_type", + "date_happened", + "device_name", + "host", + "priority", + "related_event_id", + "source_type_name", + "tags", + "text", + "title", + ] + .iter() + .map(|field| vec![PathComponent::Key(field.to_string())]) + .collect(), + ), // DataDog Event API requires unix timestamp. timestamp_format: Some(TimestampFormat::Unix), ..EncodingConfigWithDefault::default() From 5dfddcd5e0c6064dcfe2841c59b8e39057d6e999 Mon Sep 17 00:00:00 2001 From: ktf Date: Sat, 5 Jun 2021 16:47:17 +0200 Subject: [PATCH 14/16] Apply feedback Signed-off-by: ktf --- src/sinks/datadog/events.rs | 6 ++---- src/sinks/datadog/logs.rs | 3 +-- src/sinks/datadog/mod.rs | 2 ++ 3 files changed, 5 insertions(+), 6 deletions(-) diff --git a/src/sinks/datadog/events.rs b/src/sinks/datadog/events.rs index fb77d1e3d7363..cd74cb6eb5a8a 100644 --- a/src/sinks/datadog/events.rs +++ b/src/sinks/datadog/events.rs @@ -1,4 +1,4 @@ -use super::healthcheck; +use super::{healthcheck, ApiKey}; use crate::{ config::{log_schema, DataType, GenerateConfig, SinkConfig, SinkContext, SinkDescription}, event::{Event, PathComponent}, @@ -38,8 +38,6 @@ pub struct DatadogEventsConfig { request: TowerRequestConfig, } -type ApiKey = Arc; - fn default_site() -> String { "datadoghq.com".to_owned() } @@ -111,7 +109,7 @@ impl DatadogEventsConfig { client, cx.acker(), ) - .sink_map_err(|error| error!(message = "Fatal datadog_metrics sink error.", %error)); + .sink_map_err(|error| error!(message = "Fatal datadog_events sink error.", %error)); Ok((VectorSink::Sink(Box::new(sink)), healthcheck)) } diff --git a/src/sinks/datadog/logs.rs b/src/sinks/datadog/logs.rs index 17dc9ee6bbd20..4334db53ddecc 100644 --- a/src/sinks/datadog/logs.rs +++ b/src/sinks/datadog/logs.rs @@ -1,3 +1,4 @@ +use super::ApiKey; use crate::{ config::{log_schema, DataType, GenerateConfig, SinkConfig, SinkContext, SinkDescription}, event::Event, @@ -49,8 +50,6 @@ pub struct DatadogLogsConfig { request: TowerRequestConfig, } -type ApiKey = Arc; - #[derive(Clone)] struct DatadogLogsJsonService { config: DatadogLogsConfig, diff --git a/src/sinks/datadog/mod.rs b/src/sinks/datadog/mod.rs index be6c773fe8cae..895fc71a2a3a4 100644 --- a/src/sinks/datadog/mod.rs +++ b/src/sinks/datadog/mod.rs @@ -10,6 +10,8 @@ pub mod events; pub mod logs; pub mod metrics; +type ApiKey = Arc; + #[derive(Clone, Debug, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub enum Region { From b8ef075198cd7430ab237f0fcb61fcfd1df40feb Mon Sep 17 00:00:00 2001 From: ktf Date: Sun, 6 Jun 2021 14:48:10 +0200 Subject: [PATCH 15/16] Add use Signed-off-by: ktf --- src/sinks/datadog/mod.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/src/sinks/datadog/mod.rs b/src/sinks/datadog/mod.rs index 895fc71a2a3a4..065bdefa51286 100644 --- a/src/sinks/datadog/mod.rs +++ b/src/sinks/datadog/mod.rs @@ -5,6 +5,7 @@ use crate::{ use http::{Request, StatusCode, Uri}; use serde::{Deserialize, Serialize}; use snafu::ResultExt; +use std::sync::Arc; pub mod events; pub mod logs; From ad31123495dd9f5b9b0b1bae79542ce6c7947601 Mon Sep 17 00:00:00 2001 From: ktf Date: Tue, 8 Jun 2021 14:24:46 +0200 Subject: [PATCH 16/16] Bump Signed-off-by: ktf