vector/sources/datadog_agent/
logs.rs

1use std::sync::Arc;
2
3use bytes::{BufMut, Bytes, BytesMut};
4use chrono::Utc;
5use http::StatusCode;
6use tokio_util::codec::Decoder;
7use vector_lib::{
8    EstimatedJsonEncodedSizeOf,
9    codecs::StreamDecodingError,
10    config::LegacyKey,
11    internal_event::{CountByteSize, InternalEventHandle as _},
12    json_size::JsonSize,
13    lookup::path,
14};
15use vrl::core::Value;
16use warp::{Filter, filters::BoxedFilter, path as warp_path, path::FullPath, reply::Response};
17
18use super::{ApiKeyQueryParams, DatadogAgentConfig, DatadogAgentSource, LogMsg, RequestHandler};
19use crate::{
20    common::{datadog::DDTAGS, http::ErrorMessage},
21    event::Event,
22    internal_events::DatadogAgentJsonParseError,
23    sources::util::http::capped_body,
24};
25
26pub(super) fn build_warp_filter(
27    handler: RequestHandler,
28    source: DatadogAgentSource,
29) -> BoxedFilter<(Response,)> {
30    warp::post()
31        .and(warp_path!("v1" / "input" / ..).or(warp_path!("api" / "v2" / "logs" / ..)))
32        .and(warp::path::full())
33        .and(warp::header::optional::<String>("content-encoding"))
34        .and(warp::header::optional::<String>("dd-api-key"))
35        .and(warp::query::<ApiKeyQueryParams>())
36        .and(capped_body())
37        .and_then(
38            move |_,
39                  path: FullPath,
40                  encoding_header: Option<String>,
41                  api_token: Option<String>,
42                  query_params: ApiKeyQueryParams,
43                  body: Bytes| {
44                let events = source
45                    .decode(&encoding_header, body, path.as_str())
46                    .and_then(|body| {
47                        decode_log_body(
48                            body,
49                            source.api_key_extractor.extract(
50                                path.as_str(),
51                                api_token,
52                                query_params.dd_api_key,
53                            ),
54                            &source,
55                        )
56                    });
57                handler.clone().handle_request(events, super::LOGS)
58            },
59        )
60        .boxed()
61}
62
63pub(crate) fn decode_log_body(
64    body: Bytes,
65    api_key: Option<Arc<str>>,
66    source: &DatadogAgentSource,
67) -> Result<Vec<Event>, ErrorMessage> {
68    if body.is_empty() || body.as_ref() == b"{}" {
69        // The datadog agent may send an empty payload as a keep alive
70        // https://github.com/DataDog/datadog-agent/blob/5a6c5dd75a2233fbf954e38ddcc1484df4c21a35/pkg/logs/client/http/destination.go#L52
71        debug!(message = "Empty payload ignored.");
72        return Ok(Vec::new());
73    }
74
75    let messages: Vec<LogMsg> = serde_json::from_slice(&body).map_err(|error| {
76        emit!(DatadogAgentJsonParseError { error: &error });
77
78        ErrorMessage::new(
79            StatusCode::BAD_REQUEST,
80            format!("Error parsing JSON: {error:?}"),
81        )
82    })?;
83
84    let now = Utc::now();
85    let mut decoded = Vec::new();
86    let mut event_bytes_received = JsonSize::zero();
87
88    for LogMsg {
89        message,
90        status,
91        timestamp,
92        hostname,
93        service,
94        ddsource,
95        ddtags,
96    } in messages
97    {
98        let mut decoder = source.decoder.clone();
99        let mut buffer = BytesMut::new();
100        buffer.put(message);
101
102        loop {
103            match decoder.decode_eof(&mut buffer) {
104                Ok(Some((events, _byte_size))) => {
105                    for mut event in events {
106                        if let Event::Log(ref mut log) = event {
107                            let namespace = &source.log_namespace;
108                            let source_name = "datadog_agent";
109
110                            namespace.insert_source_metadata(
111                                source_name,
112                                log,
113                                Some(LegacyKey::InsertIfEmpty(path!("status"))),
114                                path!("status"),
115                                status.clone(),
116                            );
117                            namespace.insert_source_metadata(
118                                source_name,
119                                log,
120                                Some(LegacyKey::InsertIfEmpty(path!("timestamp"))),
121                                path!("timestamp"),
122                                timestamp,
123                            );
124                            namespace.insert_source_metadata(
125                                source_name,
126                                log,
127                                Some(LegacyKey::InsertIfEmpty(path!("hostname"))),
128                                path!("hostname"),
129                                hostname.clone(),
130                            );
131                            namespace.insert_source_metadata(
132                                source_name,
133                                log,
134                                Some(LegacyKey::InsertIfEmpty(path!("service"))),
135                                path!("service"),
136                                service.clone(),
137                            );
138                            namespace.insert_source_metadata(
139                                source_name,
140                                log,
141                                Some(LegacyKey::InsertIfEmpty(path!("ddsource"))),
142                                path!("ddsource"),
143                                ddsource.clone(),
144                            );
145
146                            let ddtags: Value = if source.parse_ddtags {
147                                parse_ddtags(&ddtags)
148                            } else {
149                                ddtags.clone().into()
150                            };
151
152                            namespace.insert_source_metadata(
153                                source_name,
154                                log,
155                                Some(LegacyKey::InsertIfEmpty(path!(DDTAGS))),
156                                path!(DDTAGS),
157                                ddtags,
158                            );
159
160                            // compute EstimatedJsonSizeOf before enrichment
161                            event_bytes_received += log.estimated_json_encoded_size_of();
162
163                            namespace.insert_standard_vector_source_metadata(
164                                log,
165                                DatadogAgentConfig::NAME,
166                                now,
167                            );
168
169                            if let Some(k) = &api_key {
170                                log.metadata_mut().set_datadog_api_key(Arc::clone(k));
171                            }
172
173                            let logs_schema_definition = source
174                                .logs_schema_definition
175                                .as_ref()
176                                .unwrap_or_else(|| panic!("registered log schema required"));
177
178                            log.metadata_mut()
179                                .set_schema_definition(logs_schema_definition);
180                        }
181
182                        decoded.push(event);
183                    }
184                }
185                Ok(None) => break,
186                Err(error) => {
187                    // Error is logged by `vector_lib::codecs::Decoder`, no further
188                    // handling is needed here.
189                    if !error.can_continue() {
190                        break;
191                    }
192                }
193            }
194        }
195    }
196
197    source
198        .events_received
199        .emit(CountByteSize(decoded.len(), event_bytes_received));
200
201    Ok(decoded)
202}
203
204// ddtags input is a string containing a list of tags which
205// can include both bare tags and key-value pairs.
206// the tag list members are separated by `,` and the
207// tag-value pairs are separated by `:`.
208//
209// The output is an Array regardless of the input string.
210fn parse_ddtags(ddtags_raw: &Bytes) -> Value {
211    if ddtags_raw.is_empty() {
212        return Vec::<Value>::new().into();
213    }
214
215    let ddtags_str = String::from_utf8_lossy(ddtags_raw);
216
217    // There are multiple tags, which could be either bare or pairs
218    let ddtags: Vec<Value> = ddtags_str
219        .split(',')
220        .filter(|kv| !kv.is_empty())
221        .map(|kv| Value::Bytes(Bytes::from(kv.trim().to_string())))
222        .collect();
223
224    if ddtags.is_empty() && !ddtags_str.is_empty() {
225        warn!(
226            message = "`parse_ddtags` set to true and Agent log contains non-empty ddtags string, but no tag-value pairs were parsed."
227        )
228    }
229
230    ddtags.into()
231}
232
233#[cfg(test)]
234mod tests {
235    use similar_asserts::assert_eq;
236    use vrl::value;
237
238    use super::*;
239
240    #[test]
241    fn ddtags_parse_empty() {
242        let raw = Bytes::from(String::from(""));
243        let val = parse_ddtags(&raw);
244
245        assert_eq!(val, value!([]));
246    }
247
248    #[test]
249    fn ddtags_parse_bare() {
250        let raw = Bytes::from(String::from("bare"));
251        let val = parse_ddtags(&raw);
252
253        assert_eq!(val, value!(["bare"]));
254    }
255
256    #[test]
257    fn ddtags_parse_kv_one() {
258        let raw = Bytes::from(String::from("filename:driver.log"));
259        let val = parse_ddtags(&raw);
260
261        assert_eq!(val, value!(["filename:driver.log"]));
262    }
263
264    #[test]
265    fn ddtags_parse_kv_multi() {
266        let raw = Bytes::from(String::from("filename:driver.log,wizard:the_grey"));
267        let val = parse_ddtags(&raw);
268
269        assert_eq!(val, value!(["filename:driver.log", "wizard:the_grey"]));
270    }
271
272    #[test]
273    fn ddtags_parse_kv_bare_combo() {
274        let raw = Bytes::from(String::from("filename:driver.log,debug,wizard:the_grey"));
275        let val = parse_ddtags(&raw);
276
277        assert_eq!(
278            val,
279            value!(["filename:driver.log", "debug", "wizard:the_grey"])
280        );
281    }
282}
vector/sources/datadog_agent/logs.rs

vector/sources/datadog_agent/
logs.rs
