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 diff --git a/docs/reference/components/sinks/datadog_events.cue b/docs/reference/components/sinks/datadog_events.cue new file mode 100644 index 0000000000000..ef30d85bddf6c --- /dev/null +++ b/docs/reference/components/sinks/datadog_events.cue @@ -0,0 +1,75 @@ +package metadata + +components: sinks: datadog_events: { + title: "Datadog Events" + + classes: sinks._datadog.classes + + features: { + buffer: enabled: true + healthcheck: enabled: true + send: { + batch: { + enabled: false + common: false + timeout_secs: 0 + } + compression: enabled: false + encoding: enabled: false + 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 + 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..34fc75dd69a50 100644 --- a/docs/reference/urls.cue +++ b/docs/reference/urls.cue @@ -107,6 +107,8 @@ 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_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/" diff --git a/src/internal_events/datadog_events.rs b/src/internal_events/datadog_events.rs new file mode 100644 index 0000000000000..73c21243c71ca --- /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 { + pub field: &'static str, +} + +impl InternalEvent for DatadogEventsFieldInvalid { + fn emit_logs(&self) { + error!( + 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); + } +} 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..cd74cb6eb5a8a --- /dev/null +++ b/src/sinks/datadog/events.rs @@ -0,0 +1,442 @@ +use super::{healthcheck, ApiKey}; +use crate::{ + config::{log_schema, DataType, GenerateConfig, SinkConfig, SinkContext, SinkDescription}, + event::{Event, PathComponent}, + http::HttpClient, + internal_events::{DatadogEventsFieldInvalid, DatadogEventsProcessed}, + sinks::{ + util::{ + batch::Batch, + encoding::{EncodingConfigWithDefault, EncodingConfiguration, TimestampFormat}, + http::{HttpSink, PartitionHttpSink}, + BatchConfig, BatchSettings, BoxedRawValue, Concurrency, EncodedEvent, JsonArrayBuffer, + PartitionBuffer, PartitionInnerBuffer, TowerRequestConfig, + }, + Healthcheck, VectorSink, + }, + tls::{MaybeTlsSettings, TlsConfig}, +}; +use futures::{FutureExt, SinkExt}; +use http::Request; +use indoc::indoc; +use serde::{Deserialize, Serialize}; +use serde_json::json; +use std::{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, + + tls: Option, + + #[serde(default)] + request: TowerRequestConfig, +} + +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()) + } + + fn get_api_endpoint(&self) -> String { + self.endpoint + .clone() + .unwrap_or_else(|| format!("https://api.{}", &self.site)) + } + + 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 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); + let request_settings = request_opts.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(), + 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_events sink error.", %error)); + + Ok((VectorSink::Sink(Box::new(sink)), healthcheck)) + } +} + +#[async_trait::async_trait] +#[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) + .timeout(0) + .parse_config(BatchConfig::default())?; + + self.build_sink( + cx, + DatadogEventsService::new(self), + 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, + uri: String, + default_api_key: ApiKey, + encoding: EncodingConfigWithDefault<()>, +} + +impl DatadogEventsService { + fn new(config: &DatadogEventsConfig) -> Self { + let encoding = EncodingConfigWithDefault { + // DataDog Event API allows only some fields, and refuses + // to accept event if it contains any other field. + 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() + }; + + Self { + default_api_key: Arc::from(config.default_api_key.clone()), + uri: config.get_uri(), + encoding, + config: config.clone(), + } + } +} + +#[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" }); + return None; + } + + 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 !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); + } + } + + if !log.contains("source_type_name") { + if let Some(name) = log.remove(log_schema().source_type_key()) { + log.insert("source_type_name", name); + } + } + + self.encoding.apply_rules(&mut event); + + 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 (mut events, api_key) = events.into_parts(); + + assert_eq!(events.len(), 1); + let body = serde_json::to_vec(&events.pop().expect("One event"))?; + + 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) + } +} + +#[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 bytes::Bytes; + use futures::{ + channel::mpsc::{Receiver, TryRecvError}, + stream::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 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; + + 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("text").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 { .. }))); + } + + #[tokio::test] + async fn api_key_in_metadata() { + let (mut config, cx) = load_sink::(indoc! {r#" + default_api_key = "atoken" + "#}) + .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_events_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(); + + let message = json.get("text").unwrap().as_str().unwrap(); + assert_eq!(message, expected[i]); + } + } + + #[tokio::test] + async fn filter_out_fields() { + 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(); + + let message = json.get("text").unwrap().as_str().unwrap(); + assert_eq!(message, expected[i]); + assert!(json.get("invalid").is_none()); + } + } +} 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/metrics.rs b/src/sinks/datadog/metrics.rs index 1d6168433f499..85005f93057b3 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}, @@ -12,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}; @@ -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(), + 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..065bdefa51286 100644 --- a/src/sinks/datadog/mod.rs +++ b/src/sinks/datadog/mod.rs @@ -1,11 +1,39 @@ +use crate::{ + http::HttpClient, + sinks::{HealthcheckError, UriParseError}, +}; +use http::{Request, StatusCode, Uri}; use serde::{Deserialize, Serialize}; +use snafu::ResultExt; +use std::sync::Arc; +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 { 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()), + } +}