Skip to content

Commit dfa899b

Browse files
ronitanilkumarpront
ronitanilkumar
and
pront
authored
feat(datadog_agent source): accept LLMObs telemetry at /api/v2/llmobs (#25636)
Co-authored-by: Pavlos Rontidis <pavlos.rontidis@gmail.com>
1 parent 63ce10e commit dfa899b

7 files changed

Lines changed: 498 additions & 2 deletions

File tree

‎changelog.d/25441_llmobs_endpoint.feature.md‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,3 @@
1+
The `datadog_agent` source now accepts LLM Observability (LLMObs) telemetry at `/api/v2/llmobs`. When `multiple_outputs` is enabled, LLMObs span events are available as log events on the `llmobs` output port.
2+
3+
authors: ronitanilkumar

‎src/sources/datadog_agent/llmobs.rs‎

Lines changed: 270 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,270 @@
1+
use std::sync::Arc;
2+
3+
use bytes::Bytes;
4+
use chrono::{TimeZone, Utc};
5+
use http::StatusCode;
6+
use serde::Deserialize;
7+
use serde_json::Value;
8+
use vector_lib::{
9+
EstimatedJsonEncodedSizeOf,
10+
config::LegacyKey,
11+
internal_event::{CountByteSize, InternalEventHandle as _},
12+
json_size::JsonSize,
13+
lookup::{event_path, path},
14+
};
15+
use warp::{Filter, filters::BoxedFilter, path as warp_path, path::FullPath, reply::Response};
16+
17+
use super::{ApiKeyQueryParams, DatadogAgentConfig, DatadogAgentSource, RequestHandler};
18+
use crate::{
19+
common::http::ErrorMessage,
20+
event::{Event, LogEvent},
21+
internal_events::DatadogAgentJsonParseError,
22+
sources::util::http::capped_body,
23+
};
24+
25+
pub(super) fn build_warp_filter(
26+
handler: RequestHandler,
27+
source: DatadogAgentSource,
28+
) -> BoxedFilter<(Response,)> {
29+
let direct = warp::post()
30+
.and(warp_path!("api" / "v2" / "llmobs" / ..))
31+
.and(warp::path::full())
32+
.and(warp::header::optional::<String>("content-encoding"))
33+
.and(warp::header::optional::<String>("dd-api-key"))
34+
.and(warp::query::<ApiKeyQueryParams>())
35+
.and(capped_body())
36+
.and_then({
37+
let handler = handler.clone();
38+
let source = source.clone();
39+
move |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_llmobs_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::LLMOBS)
58+
}
59+
});
60+
61+
let evp_proxy = warp::post()
62+
.and(warp_path!(
63+
"evp_proxy" / "v2" / "api" / "v2" / "llmobs" / ..
64+
))
65+
.and(warp::path::full())
66+
.and(warp::header::optional::<String>("content-encoding"))
67+
.and(warp::header::optional::<String>("dd-api-key"))
68+
.and(warp::query::<ApiKeyQueryParams>())
69+
.and(capped_body())
70+
.and_then(
71+
move |path: FullPath,
72+
encoding_header: Option<String>,
73+
api_token: Option<String>,
74+
query_params: ApiKeyQueryParams,
75+
body: Bytes| {
76+
let events = source
77+
.decode(&encoding_header, body, path.as_str())
78+
.and_then(|body| {
79+
decode_llmobs_body(
80+
body,
81+
source.api_key_extractor.extract(
82+
path.as_str(),
83+
api_token,
84+
query_params.dd_api_key,
85+
),
86+
&source,
87+
)
88+
});
89+
handler.clone().handle_request(events, super::LLMOBS)
90+
},
91+
);
92+
93+
direct.or(evp_proxy).unify().boxed()
94+
}
95+
96+
#[derive(Deserialize)]
97+
struct LLMObsEnvelopeItem {
98+
#[serde(rename = "event_type")]
99+
_event_type: Option<String>,
100+
spans: Vec<LLMObsSpan>,
101+
#[serde(rename = "_dd.tracer_version")]
102+
dd_tracer_version: Option<String>,
103+
#[serde(rename = "_dd.scope")]
104+
_dd_scope: Option<String>,
105+
}
106+
107+
/// The Datadog Agent and SDK clients (e.g. dd-trace-py's `LLMObsSpanEncoder`) POST a single
108+
/// JSON object envelope. Some fixtures/tools may send a JSON array of envelopes instead.
109+
/// Accept both shapes.
110+
#[derive(Deserialize)]
111+
#[serde(untagged)]
112+
enum LLMObsEnvelope {
113+
Single(LLMObsEnvelopeItem),
114+
Multiple(Vec<LLMObsEnvelopeItem>),
115+
}
116+
117+
impl LLMObsEnvelope {
118+
fn into_items(self) -> Vec<LLMObsEnvelopeItem> {
119+
match self {
120+
LLMObsEnvelope::Single(item) => vec![item],
121+
LLMObsEnvelope::Multiple(items) => items,
122+
}
123+
}
124+
}
125+
126+
#[derive(Deserialize)]
127+
struct LLMObsSpan {
128+
span_id: String,
129+
trace_id: String,
130+
parent_id: Option<String>,
131+
name: Option<String>,
132+
session_id: Option<String>,
133+
service: Option<String>,
134+
start_ns: Option<i64>,
135+
duration: Option<i64>,
136+
status: Option<String>,
137+
status_message: Option<String>,
138+
meta: Option<Value>,
139+
metrics: Option<Value>,
140+
#[serde(default)]
141+
tags: Vec<String>,
142+
#[serde(rename = "_dd")]
143+
dd: Option<Value>,
144+
span_links: Option<Value>,
145+
#[serde(rename = "config")]
146+
config: Option<Value>,
147+
collection_errors: Option<Value>,
148+
}
149+
150+
pub(crate) fn decode_llmobs_body(
151+
body: Bytes,
152+
api_key: Option<Arc<str>>,
153+
source: &DatadogAgentSource,
154+
) -> Result<Vec<Event>, ErrorMessage> {
155+
let envelope: Vec<LLMObsEnvelopeItem> = serde_json::from_slice::<LLMObsEnvelope>(&body)
156+
.map_err(|error| {
157+
emit!(DatadogAgentJsonParseError { error: &error });
158+
ErrorMessage::new(
159+
StatusCode::BAD_REQUEST,
160+
format!("Error parsing JSON: {error:?}"),
161+
)
162+
})?
163+
.into_items();
164+
165+
let now = Utc::now();
166+
let mut event_bytes_received = JsonSize::zero();
167+
168+
let events: Vec<Event> = envelope
169+
.into_iter()
170+
.flat_map(|item| {
171+
let tracer_version = item.dd_tracer_version.clone();
172+
item.spans.into_iter().map(move |span| {
173+
let mut log = LogEvent::default();
174+
log.insert(event_path!("span_id"), span.span_id);
175+
log.insert(event_path!("trace_id"), span.trace_id);
176+
if let Some(v) = span.parent_id {
177+
log.insert(event_path!("parent_id"), v);
178+
}
179+
if let Some(v) = span.name {
180+
log.insert(event_path!("name"), v);
181+
}
182+
if let Some(v) = span.session_id {
183+
log.insert(event_path!("session_id"), v);
184+
}
185+
if let Some(v) = span.service {
186+
log.insert(event_path!("service"), v);
187+
}
188+
if let Some(ns) = span.start_ns {
189+
log.insert(event_path!("start_ns"), ns);
190+
source.log_namespace.insert_source_metadata(
191+
DatadogAgentConfig::NAME,
192+
&mut log,
193+
Some(LegacyKey::InsertIfEmpty(path!("timestamp"))),
194+
path!("timestamp"),
195+
Utc.timestamp_nanos(ns),
196+
);
197+
}
198+
if let Some(v) = span.duration {
199+
log.insert(event_path!("duration"), v);
200+
}
201+
if let Some(v) = span.status {
202+
log.insert(event_path!("status"), v);
203+
}
204+
if let Some(v) = span.status_message {
205+
log.insert(event_path!("status_message"), v);
206+
}
207+
if let Some(v) = span.meta {
208+
log.insert(event_path!("meta"), v);
209+
}
210+
if let Some(v) = span.metrics {
211+
log.insert(event_path!("metrics"), v);
212+
}
213+
if !span.tags.is_empty() {
214+
log.insert(event_path!("tags"), span.tags.clone());
215+
}
216+
if let Some(v) = span.span_links {
217+
log.insert(event_path!("span_links"), v);
218+
}
219+
if let Some(v) = span.config {
220+
log.insert(event_path!("config"), v);
221+
}
222+
if let Some(v) = span.collection_errors {
223+
log.insert(event_path!("collection_errors"), v);
224+
}
225+
226+
// Extract ml_app: first check span._dd.ml_app, then fall back to tags array.
227+
let ml_app = span
228+
.dd
229+
.as_ref()
230+
.and_then(|dd| dd.get("ml_app"))
231+
.and_then(|v| v.as_str())
232+
.map(str::to_owned)
233+
.or_else(|| {
234+
span.tags
235+
.iter()
236+
.find_map(|tag| tag.strip_prefix("ml_app:").map(str::to_owned))
237+
});
238+
if let Some(app) = ml_app {
239+
log.insert(event_path!("ml_app"), app);
240+
}
241+
242+
if let Some(v) = tracer_version.clone() {
243+
log.insert(event_path!("_dd", "tracer_version"), v);
244+
}
245+
246+
Event::Log(log)
247+
})
248+
})
249+
.map(|mut event| {
250+
if let Event::Log(ref mut log) = event {
251+
event_bytes_received += log.estimated_json_encoded_size_of();
252+
source.log_namespace.insert_standard_vector_source_metadata(
253+
log,
254+
DatadogAgentConfig::NAME,
255+
now,
256+
);
257+
if let Some(k) = &api_key {
258+
log.metadata_mut().set_datadog_api_key(Arc::clone(k));
259+
}
260+
}
261+
event
262+
})
263+
.collect();
264+
265+
source
266+
.events_received
267+
.emit(CountByteSize(events.len(), event_bytes_received));
268+
269+
Ok(events)
270+
}

‎src/sources/datadog_agent/mod.rs‎

Lines changed: 66 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@ mod integration_tests;
33
#[cfg(test)]
44
mod tests;
55

6+
pub mod llmobs;
67
pub mod logs;
78
pub mod metrics;
89
pub mod traces;
@@ -74,6 +75,7 @@ use crate::{
7475
pub const LOGS: &str = "logs";
7576
pub const METRICS: &str = "metrics";
7677
pub const TRACES: &str = "traces";
78+
pub const LLMOBS: &str = "llmobs";
7779

7880
/// Configuration for the `datadog_agent` source.
7981
#[configurable_component(source(
@@ -107,6 +109,11 @@ pub struct DatadogAgentConfig {
107109
#[serde(default = "crate::serde::default_false")]
108110
disable_traces: bool,
109111

112+
/// If this is set to `true`, LLM Observability events are not accepted by the component.
113+
#[configurable(metadata(docs::advanced))]
114+
#[serde(default = "crate::serde::default_false")]
115+
disable_llmobs: bool,
116+
110117
/// If this is set to `true`, logs, metrics (beta), and traces (alpha) are sent to different outputs.
111118
///
112119
///
@@ -177,6 +184,7 @@ impl GenerateConfig for DatadogAgentConfig {
177184
disable_logs: false,
178185
disable_metrics: false,
179186
disable_traces: false,
187+
disable_llmobs: false,
180188
multiple_outputs: false,
181189
parse_ddtags: false,
182190
split_metric_namespace: true,
@@ -319,6 +327,50 @@ impl SourceConfig for DatadogAgentConfig {
319327
)
320328
.with_standard_vector_source_metadata();
321329

330+
let log_namespace = global_log_namespace.merge(self.log_namespace);
331+
let llmobs_definition = schema::Definition::new_with_default_metadata(
332+
Kind::object(
333+
Collection::empty()
334+
.with_known("span_id", Kind::bytes())
335+
.with_known("trace_id", Kind::bytes())
336+
.with_known("parent_id", Kind::bytes().or_undefined())
337+
.with_known("name", Kind::bytes().or_undefined())
338+
.with_known("session_id", Kind::bytes().or_undefined())
339+
.with_known("service", Kind::bytes().or_undefined())
340+
.with_known("start_ns", Kind::integer().or_undefined())
341+
.with_known("duration", Kind::integer().or_undefined())
342+
.with_known("status", Kind::bytes().or_undefined())
343+
.with_known("status_message", Kind::bytes().or_undefined())
344+
.with_known("ml_app", Kind::bytes().or_undefined())
345+
.with_known("meta", Kind::object(Collection::any()).or_undefined())
346+
.with_known("metrics", Kind::object(Collection::any()).or_undefined())
347+
.with_known(
348+
"tags",
349+
Kind::array(Collection::empty().with_unknown(Kind::bytes())).or_undefined(),
350+
)
351+
.with_known("span_links", Kind::any().or_undefined())
352+
.with_known("config", Kind::any().or_undefined())
353+
.with_known("collection_errors", Kind::any().or_undefined())
354+
.with_known(
355+
"_dd",
356+
Kind::object(
357+
Collection::empty()
358+
.with_known("tracer_version", Kind::bytes().or_undefined()),
359+
)
360+
.or_undefined(),
361+
),
362+
),
363+
[log_namespace],
364+
)
365+
.with_source_metadata(
366+
Self::NAME,
367+
Some(LegacyKey::InsertIfEmpty(owned_value_path!("timestamp"))),
368+
&owned_value_path!("timestamp"),
369+
Kind::timestamp(),
370+
Some(meaning::TIMESTAMP),
371+
)
372+
.with_standard_vector_source_metadata();
373+
322374
let mut output = Vec::with_capacity(1);
323375

324376
if self.multiple_outputs {
@@ -331,6 +383,12 @@ impl SourceConfig for DatadogAgentConfig {
331383
if !self.disable_traces {
332384
output.push(SourceOutput::new_traces().with_port(TRACES))
333385
}
386+
if !self.disable_llmobs {
387+
output.push(
388+
SourceOutput::new_maybe_logs(DataType::Log, llmobs_definition)
389+
.with_port(LLMOBS),
390+
)
391+
}
334392
} else {
335393
output.push(SourceOutput::new_maybe_logs(
336394
DataType::all_bits(),
@@ -457,12 +515,19 @@ impl DatadogAgentSource {
457515
}
458516

459517
if !config.disable_metrics {
460-
let metrics_filter = metrics::build_warp_filter(handler, self.clone());
518+
let metrics_filter = metrics::build_warp_filter(handler.clone(), self.clone());
461519
filters = filters
462520
.map(|f| f.or(metrics_filter.clone()).unify().boxed())
463521
.or(Some(metrics_filter));
464522
}
465523

524+
if !config.disable_llmobs {
525+
let llmobs_filter = llmobs::build_warp_filter(handler.clone(), self.clone());
526+
filters = filters
527+
.map(|f| f.or(llmobs_filter.clone()).unify().boxed())
528+
.or(Some(llmobs_filter));
529+
}
530+
466531
filters.ok_or_else(|| "At least one of the supported data type shall be enabled".into())
467532
}
468533

0 commit comments

Comments
 (0)