vector/sources/datadog_agent/
mod.rs

1#[cfg(all(test, feature = "datadog-agent-integration-tests"))]
2mod integration_tests;
3#[cfg(test)]
4mod tests;
5
6pub mod logs;
7pub mod metrics;
8pub mod traces;
9
10#[allow(warnings, clippy::pedantic, clippy::nursery)]
11pub(crate) mod ddmetric_proto {
12    include!(concat!(env!("OUT_DIR"), "/datadog.agentpayload.rs"));
13}
14
15#[allow(warnings)]
16pub(crate) mod ddtrace_proto {
17    include!(concat!(env!("OUT_DIR"), "/dd_trace.rs"));
18}
19
20use std::{convert::Infallible, fmt::Debug, net::SocketAddr, sync::Arc, time::Duration};
21
22use bytes::{Buf, Bytes};
23use chrono::{DateTime, Utc, serde::ts_milliseconds};
24use futures::FutureExt;
25use http::StatusCode;
26use hyper::{Server, service::make_service_fn};
27use regex::Regex;
28use serde::{Deserialize, Serialize};
29use serde_with::serde_as;
30use snafu::Snafu;
31use tokio::net::TcpStream;
32use tower::ServiceBuilder;
33use tracing::Span;
34use vector_lib::{
35    codecs::decoding::{DeserializerConfig, FramingConfig},
36    config::{LegacyKey, LogNamespace},
37    configurable::configurable_component,
38    event::{BatchNotifier, BatchStatus},
39    internal_event::{EventsReceived, Registered},
40    lookup::owned_value_path,
41    schema::meaning,
42    source_sender::SendError,
43    tls::MaybeTlsIncomingStream,
44};
45use vrl::{
46    path::OwnedTargetPath,
47    value::{Kind, kind::Collection},
48};
49use warp::{Filter, Reply, filters::BoxedFilter, reject::Rejection, reply::Response};
50
51use crate::{
52    SourceSender,
53    codecs::{Decoder, DecodingConfig},
54    common::http::ErrorMessage,
55    config::{
56        DataType, GenerateConfig, Resource, SourceAcknowledgementsConfig, SourceConfig,
57        SourceContext, SourceOutput, log_schema,
58    },
59    event::Event,
60    http::{KeepaliveConfig, MaxConnectionAgeLayer, build_http_trace_layer},
61    internal_events::{HttpBytesReceived, StreamClosedError},
62    schema,
63    serde::{bool_or_struct, default_decoding, default_framing_message_based},
64    sources::{
65        self,
66        util::{
67            decompression::{CappedDecoder, max_decompressed_size_bytes},
68            http::emit_decompress_error,
69        },
70    },
71    tls::{MaybeTlsSettings, TlsEnableableConfig},
72};
73
74pub const LOGS: &str = "logs";
75pub const METRICS: &str = "metrics";
76pub const TRACES: &str = "traces";
77
78/// Configuration for the `datadog_agent` source.
79#[configurable_component(source(
80    "datadog_agent",
81    "Receive logs, metrics, and traces collected by a Datadog Agent."
82))]
83#[serde_as]
84#[derive(Clone, Debug)]
85pub struct DatadogAgentConfig {
86    /// The socket address to accept connections on.
87    ///
88    /// It _must_ include a port.
89    #[configurable(metadata(docs::examples = "0.0.0.0:80"))]
90    #[configurable(metadata(docs::examples = "localhost:80"))]
91    address: SocketAddr,
92
93    /// If this is set to `true`, when incoming events contain a Datadog API key, it is
94    /// stored in the event metadata and used if the event is sent to a Datadog sink.
95    #[configurable(metadata(docs::advanced))]
96    #[serde(default = "crate::serde::default_true")]
97    store_api_key: bool,
98
99    /// If this is set to `true`, logs are not accepted by the component.
100    #[configurable(metadata(docs::advanced))]
101    #[serde(default = "crate::serde::default_false")]
102    disable_logs: bool,
103
104    /// If this is set to `true`, metrics (beta) are not accepted by the component.
105    #[configurable(metadata(docs::advanced))]
106    #[serde(default = "crate::serde::default_false")]
107    disable_metrics: bool,
108
109    /// If this is set to `true`, traces (alpha) are not accepted by the component.
110    #[configurable(metadata(docs::advanced))]
111    #[serde(default = "crate::serde::default_false")]
112    disable_traces: bool,
113
114    /// If this is set to `true`, logs, metrics (beta), and traces (alpha) are sent to different outputs.
115    ///
116    ///
117    /// For a source component named `agent`, the received logs, metrics (beta), and traces (alpha) can then be
118    /// configured as input to other components by specifying `agent.logs`, `agent.metrics`, and
119    /// `agent.traces`, respectively.
120    #[configurable(metadata(docs::advanced))]
121    #[serde(default = "crate::serde::default_false")]
122    multiple_outputs: bool,
123
124    /// If this is set to `true`, when log events contain the field `ddtags`, the string value that
125    /// contains a list of key:value pairs set by the Agent is parsed and expanded into an array.
126    #[configurable(metadata(docs::advanced))]
127    #[serde(default = "crate::serde::default_false")]
128    parse_ddtags: bool,
129
130    /// If this is set to `true`, metric names are split at the first '.' into a namespace and name.
131    /// For example, `system.cpu.usage` would be split into namespace `system` and name `cpu.usage`.
132    /// If `false`, the full metric name is used without splitting. This may be useful if you are using a
133    /// default namespace for metrics in sinks connected to this source.
134    #[configurable(metadata(docs::advanced))]
135    #[serde(default = "crate::serde::default_true")]
136    split_metric_namespace: bool,
137
138    /// The namespace to use for logs. This overrides the global setting.
139    #[serde(default)]
140    #[configurable(metadata(docs::hidden))]
141    log_namespace: Option<bool>,
142
143    #[configurable(derived)]
144    tls: Option<TlsEnableableConfig>,
145
146    #[configurable(derived)]
147    #[serde(default = "default_framing_message_based")]
148    framing: FramingConfig,
149
150    #[configurable(derived)]
151    #[serde(default = "default_decoding")]
152    decoding: DeserializerConfig,
153
154    #[configurable(derived)]
155    #[serde(default, deserialize_with = "bool_or_struct")]
156    acknowledgements: SourceAcknowledgementsConfig,
157
158    #[configurable(derived)]
159    #[serde(default)]
160    keepalive: KeepaliveConfig,
161
162    /// The timeout before responding to requests with a HTTP 503 Service Unavailable error.
163    ///
164    /// If not set, responses to completed requests will block indefinitely until connected
165    /// transforms or sinks are ready to receive the events. When this happens, the sending Datadog
166    /// Agent will eventually time out the request and drop the connection, resulting Vector
167    /// generating an "Events dropped." error and incrementing the `component_discarded_events_total`
168    /// internal metric. By setting this option to a value less than the Agent's timeout, Vector
169    /// will instead respond to the Agent with a HTTP 503 Service Unavailable error, emit a warning,
170    /// and increment the `component_timed_out_events_total` internal metric instead.
171    #[serde_as(as = "Option<serde_with::DurationSecondsWithFrac<f64>>")]
172    send_timeout_secs: Option<f64>,
173}
174
175impl GenerateConfig for DatadogAgentConfig {
176    fn generate_config() -> toml::Value {
177        toml::Value::try_from(Self {
178            address: "0.0.0.0:8080".parse().unwrap(),
179            tls: None,
180            store_api_key: true,
181            framing: default_framing_message_based(),
182            decoding: default_decoding(),
183            acknowledgements: SourceAcknowledgementsConfig::default(),
184            disable_logs: false,
185            disable_metrics: false,
186            disable_traces: false,
187            multiple_outputs: false,
188            parse_ddtags: false,
189            split_metric_namespace: true,
190            log_namespace: Some(false),
191            keepalive: KeepaliveConfig::default(),
192            send_timeout_secs: None,
193        })
194        .unwrap()
195    }
196}
197
198#[async_trait::async_trait]
199#[typetag::serde(name = "datadog_agent")]
200impl SourceConfig for DatadogAgentConfig {
201    async fn build(&self, cx: SourceContext) -> crate::Result<sources::Source> {
202        let log_namespace = cx.log_namespace(self.log_namespace);
203
204        let logs_schema_definition = cx
205            .schema_definitions
206            .get(&Some(LOGS.to_owned()))
207            .or_else(|| cx.schema_definitions.get(&None))
208            .cloned();
209
210        let decoder =
211            DecodingConfig::new(self.framing.clone(), self.decoding.clone(), log_namespace)
212                .build()?;
213
214        let tls = MaybeTlsSettings::from_config(self.tls.as_ref(), true)?;
215        let source = DatadogAgentSource::new(
216            self.store_api_key,
217            decoder,
218            tls.http_protocol_name(),
219            logs_schema_definition,
220            log_namespace,
221            self.parse_ddtags,
222            self.split_metric_namespace,
223        );
224        let listener = tls.bind(&self.address).await?;
225        let handler = RequestHandler {
226            acknowledgements: cx.do_acknowledgements(self.acknowledgements),
227            multiple_outputs: self.multiple_outputs,
228            out: cx.out,
229        };
230        let filters = source.build_warp_filters(handler, self)?;
231        let shutdown = cx.shutdown;
232        let keepalive_settings = self.keepalive.clone();
233
234        info!(message = "Building HTTP server.", address = %self.address);
235
236        Ok(Box::pin(async move {
237            let routes = filters.recover(|r: Rejection| async move {
238                if let Some(e_msg) = r.find::<ErrorMessage>() {
239                    let json = warp::reply::json(e_msg);
240                    Ok(warp::reply::with_status(json, e_msg.status_code()))
241                } else {
242                    // other internal error - will return 500 internal server error
243                    Err(r)
244                }
245            });
246
247            let span = Span::current();
248            let make_svc = make_service_fn(move |conn: &MaybeTlsIncomingStream<TcpStream>| {
249                let svc = ServiceBuilder::new()
250                    .layer(build_http_trace_layer(span.clone()))
251                    .option_layer(keepalive_settings.max_connection_age_secs.map(|secs| {
252                        MaxConnectionAgeLayer::new(
253                            Duration::from_secs(secs),
254                            keepalive_settings.max_connection_age_jitter_factor,
255                            conn.peer_addr(),
256                        )
257                    }))
258                    .service(warp::service(routes.clone()));
259                futures_util::future::ok::<_, Infallible>(svc)
260            });
261
262            Server::builder(hyper::server::accept::from_stream(listener.accept_stream()))
263                .serve(make_svc)
264                .with_graceful_shutdown(shutdown.map(|_| ()))
265                .await
266                .map_err(|err| {
267                    error!("An error occurred: {:?}.", err);
268                })?;
269
270            Ok(())
271        }))
272    }
273
274    fn outputs(&self, global_log_namespace: LogNamespace) -> Vec<SourceOutput> {
275        let definition = self
276            .decoding
277            .schema_definition(global_log_namespace.merge(self.log_namespace))
278            // NOTE: "status" is intentionally semantically mapped to "severity",
279            //       since that is what DD designates as the semantic meaning of status
280            // https://docs.datadoghq.com/logs/log_configuration/attributes_naming_convention/?s=severity#reserved-attributes
281            .with_source_metadata(
282                Self::NAME,
283                Some(LegacyKey::InsertIfEmpty(owned_value_path!("status"))),
284                &owned_value_path!("status"),
285                Kind::bytes(),
286                Some(meaning::SEVERITY),
287            )
288            .with_source_metadata(
289                Self::NAME,
290                Some(LegacyKey::InsertIfEmpty(owned_value_path!("timestamp"))),
291                &owned_value_path!("timestamp"),
292                Kind::timestamp(),
293                Some(meaning::TIMESTAMP),
294            )
295            .with_source_metadata(
296                Self::NAME,
297                Some(LegacyKey::InsertIfEmpty(owned_value_path!("hostname"))),
298                &owned_value_path!("hostname"),
299                Kind::bytes(),
300                Some(meaning::HOST),
301            )
302            .with_source_metadata(
303                Self::NAME,
304                Some(LegacyKey::InsertIfEmpty(owned_value_path!("service"))),
305                &owned_value_path!("service"),
306                Kind::bytes(),
307                Some(meaning::SERVICE),
308            )
309            .with_source_metadata(
310                Self::NAME,
311                Some(LegacyKey::InsertIfEmpty(owned_value_path!("ddsource"))),
312                &owned_value_path!("ddsource"),
313                Kind::bytes(),
314                Some(meaning::SOURCE),
315            )
316            .with_source_metadata(
317                Self::NAME,
318                Some(LegacyKey::InsertIfEmpty(owned_value_path!("ddtags"))),
319                &owned_value_path!("ddtags"),
320                if self.parse_ddtags {
321                    Kind::array(Collection::empty().with_unknown(Kind::bytes())).or_undefined()
322                } else {
323                    Kind::bytes()
324                },
325                Some(meaning::TAGS),
326            )
327            .with_standard_vector_source_metadata();
328
329        let mut output = Vec::with_capacity(1);
330
331        if self.multiple_outputs {
332            if !self.disable_logs {
333                output.push(SourceOutput::new_maybe_logs(DataType::Log, definition).with_port(LOGS))
334            }
335            if !self.disable_metrics {
336                output.push(SourceOutput::new_metrics().with_port(METRICS))
337            }
338            if !self.disable_traces {
339                output.push(SourceOutput::new_traces().with_port(TRACES))
340            }
341        } else {
342            output.push(SourceOutput::new_maybe_logs(
343                DataType::all_bits(),
344                definition,
345            ))
346        }
347        output
348    }
349
350    fn resources(&self) -> Vec<Resource> {
351        vec![Resource::tcp(self.address)]
352    }
353
354    fn can_acknowledge(&self) -> bool {
355        true
356    }
357
358    fn send_timeout(&self) -> Option<Duration> {
359        self.send_timeout_secs.map(Duration::from_secs_f64)
360    }
361}
362
363#[derive(Clone, Copy, Debug, Snafu)]
364pub(crate) enum ApiError {
365    ServerShutdown,
366}
367
368impl warp::reject::Reject for ApiError {}
369
370#[derive(Deserialize)]
371pub struct ApiKeyQueryParams {
372    #[serde(rename = "dd-api-key")]
373    pub dd_api_key: Option<String>,
374}
375
376#[derive(Clone)]
377pub(crate) struct DatadogAgentSource {
378    pub(crate) api_key_extractor: ApiKeyExtractor,
379    pub(crate) log_schema_host_key: OwnedTargetPath,
380    pub(crate) log_schema_source_type_key: OwnedTargetPath,
381    pub(crate) log_namespace: LogNamespace,
382    pub(crate) decoder: Decoder,
383    protocol: &'static str,
384    logs_schema_definition: Option<Arc<schema::Definition>>,
385    events_received: Registered<EventsReceived>,
386    parse_ddtags: bool,
387    split_metric_namespace: bool,
388}
389
390#[derive(Clone)]
391pub struct ApiKeyExtractor {
392    matcher: Regex,
393    store_api_key: bool,
394}
395
396impl ApiKeyExtractor {
397    pub fn extract(
398        &self,
399        path: &str,
400        header: Option<String>,
401        query_params: Option<String>,
402    ) -> Option<Arc<str>> {
403        if !self.store_api_key {
404            return None;
405        }
406        // Grab from URL first
407        self.matcher
408            .captures(path)
409            .and_then(|cap| cap.name("api_key").map(|key| key.as_str()).map(Arc::from))
410            // Try from query params
411            .or_else(|| query_params.map(Arc::from))
412            // Try from header next
413            .or_else(|| header.map(Arc::from))
414    }
415}
416
417impl DatadogAgentSource {
418    pub(crate) fn new(
419        store_api_key: bool,
420        decoder: Decoder,
421        protocol: &'static str,
422        logs_schema_definition: Option<schema::Definition>,
423        log_namespace: LogNamespace,
424        parse_ddtags: bool,
425        split_metric_namespace: bool,
426    ) -> Self {
427        Self {
428            api_key_extractor: ApiKeyExtractor {
429                store_api_key,
430                matcher: Regex::new(r"^/v1/input/(?P<api_key>[[:alnum:]]{32})/??")
431                    .expect("static regex always compiles"),
432            },
433            log_schema_host_key: log_schema()
434                .host_key_target_path()
435                .expect("global log_schema.host_key to be valid path")
436                .clone(),
437            log_schema_source_type_key: log_schema()
438                .source_type_key_target_path()
439                .expect("global log_schema.source_type_key to be valid path")
440                .clone(),
441            decoder,
442            protocol,
443            logs_schema_definition: logs_schema_definition.map(Arc::new),
444            log_namespace,
445            events_received: register!(EventsReceived),
446            parse_ddtags,
447            split_metric_namespace,
448        }
449    }
450
451    fn build_warp_filters(
452        &self,
453        handler: RequestHandler,
454        config: &DatadogAgentConfig,
455    ) -> crate::Result<BoxedFilter<(Response,)>> {
456        let mut filters =
457            (!config.disable_logs).then(|| logs::build_warp_filter(handler.clone(), self.clone()));
458
459        if !config.disable_traces {
460            let trace_filter = traces::build_warp_filter(handler.clone(), self.clone());
461            filters = filters
462                .map(|f| f.or(trace_filter.clone()).unify().boxed())
463                .or(Some(trace_filter));
464        }
465
466        if !config.disable_metrics {
467            let metrics_filter = metrics::build_warp_filter(handler, self.clone());
468            filters = filters
469                .map(|f| f.or(metrics_filter.clone()).unify().boxed())
470                .or(Some(metrics_filter));
471        }
472
473        filters.ok_or_else(|| "At least one of the supported data type shall be enabled".into())
474    }
475
476    pub(crate) fn decode(
477        &self,
478        header: &Option<String>,
479        mut body: Bytes,
480        path: &str,
481    ) -> Result<Bytes, ErrorMessage> {
482        if let Some(encodings) = header {
483            for encoding in encodings.rsplit(',').map(str::trim) {
484                body = match encoding {
485                    "identity" => body,
486                    // Cap each decompressed payload so a compression bomb cannot drive
487                    // unbounded allocation on this unauthenticated HTTP listener.
488                    "gzip" | "x-gzip" => CappedDecoder::gzip(body.reader())
489                        .decompress()
490                        .map_err(|error| {
491                            emit_decompress_error(encoding, error, max_decompressed_size_bytes())
492                        })?
493                        .into(),
494                    "zstd" => CappedDecoder::zstd_http(body.reader())
495                        .map_err(|error| {
496                            emit_decompress_error(encoding, error, max_decompressed_size_bytes())
497                        })?
498                        .decompress()
499                        .map_err(|error| {
500                            emit_decompress_error(encoding, error, max_decompressed_size_bytes())
501                        })?
502                        .into(),
503                    "deflate" | "x-deflate" => CappedDecoder::zlib(body.reader())
504                        .decompress()
505                        .map_err(|error| {
506                            emit_decompress_error(encoding, error, max_decompressed_size_bytes())
507                        })?
508                        .into(),
509                    encoding => {
510                        return Err(ErrorMessage::new(
511                            StatusCode::UNSUPPORTED_MEDIA_TYPE,
512                            format!("Unsupported encoding {encoding}"),
513                        ));
514                    }
515                }
516            }
517        }
518        emit!(HttpBytesReceived {
519            byte_size: body.len(),
520            http_path: path,
521            protocol: self.protocol,
522        });
523        Ok(body)
524    }
525}
526
527#[derive(Clone)]
528struct RequestHandler {
529    acknowledgements: bool,
530    multiple_outputs: bool,
531    out: SourceSender,
532}
533
534impl RequestHandler {
535    async fn handle_request(
536        mut self,
537        events: Result<Vec<Event>, ErrorMessage>,
538        output: &'static str,
539    ) -> Result<Response, Rejection> {
540        match events {
541            Ok(events) => self.handle_events(events, output).await,
542            Err(err) => Err(warp::reject::custom(err)),
543        }
544    }
545
546    async fn handle_events(
547        &mut self,
548        mut events: Vec<Event>,
549        output: &'static str,
550    ) -> Result<Response, Rejection> {
551        let receiver = BatchNotifier::maybe_apply_to(self.acknowledgements, &mut events);
552        let count = events.len();
553        let output = self.multiple_outputs.then_some(output);
554
555        let result = if let Some(name) = output {
556            self.out.send_batch_named(name, events).await
557        } else {
558            self.out.send_batch(events).await
559        };
560        match result {
561            Ok(()) => {}
562            Err(SendError::Closed) => {
563                emit!(StreamClosedError { count });
564                return Err(warp::reject::custom(ApiError::ServerShutdown));
565            }
566            Err(SendError::Timeout) => {
567                return Ok(warp::reply::with_status(
568                    "Service unavailable",
569                    StatusCode::SERVICE_UNAVAILABLE,
570                )
571                .into_response());
572            }
573        }
574        match receiver {
575            None => Ok(warp::reply().into_response()),
576            Some(receiver) => match receiver.await {
577                BatchStatus::Delivered => Ok(warp::reply().into_response()),
578                BatchStatus::Errored => Err(warp::reject::custom(ErrorMessage::new(
579                    StatusCode::INTERNAL_SERVER_ERROR,
580                    "Error delivering contents to sink".into(),
581                ))),
582                BatchStatus::Rejected => Err(warp::reject::custom(ErrorMessage::new(
583                    StatusCode::BAD_REQUEST,
584                    "Contents failed to deliver to sink".into(),
585                ))),
586            },
587        }
588    }
589}
590
591// https://github.com/DataDog/datadog-agent/blob/a33248c2bc125920a9577af1e16f12298875a4ad/pkg/logs/processor/json.go#L23-L49
592#[derive(Clone, Debug, Deserialize, Serialize)]
593#[serde(deny_unknown_fields)]
594struct LogMsg {
595    pub message: Bytes,
596    pub status: Bytes,
597    #[serde(
598        deserialize_with = "ts_milliseconds::deserialize",
599        serialize_with = "ts_milliseconds::serialize"
600    )]
601    pub timestamp: DateTime<Utc>,
602    pub hostname: Bytes,
603    pub service: Bytes,
604    pub ddsource: Bytes,
605    pub ddtags: Bytes,
606}
vector/sources/datadog_agent/mod.rs

vector/sources/datadog_agent/
mod.rs
