Skip to content
Merged
Show file tree
Hide file tree
Changes from 5 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions crates/core/src/api/shared.rs
Original file line number Diff line number Diff line change
Expand Up @@ -220,6 +220,11 @@ pub(crate) fn metadata_with_otel_error(metadata: Option<Json>, error: &FlowError
metadata
.entry("error.type".to_string())
.or_insert_with(|| Json::String(error.otel_error_type().to_string()));
if let Some(exception_type) = error.exception_type() {
metadata
.entry("exception.type".to_string())
.or_insert_with(|| Json::String(exception_type.to_string()));
}
}
metadata
}
Expand Down
19 changes: 18 additions & 1 deletion crates/core/src/error.rs
Original file line number Diff line number Diff line change
Expand Up @@ -129,6 +129,15 @@ pub enum FlowError {
/// An internal runtime error (e.g., lock poisoning).
#[error("internal error: {0}")]
Internal(String),

/// An exception raised by a language-binding callback.
#[error("internal error: {message}")]
CallbackException {
/// Original binding-rendered exception message.
message: String,
/// Original language exception class name.
exception_type: String,
},
}

/// A specialized [`Result`](std::result::Result) type for NeMo Relay operations.
Expand Down Expand Up @@ -158,7 +167,15 @@ impl FlowError {
UpstreamFailureClass::InvalidRequest => "invalid_request",
UpstreamFailureClass::Other => "upstream_error",
},
Self::Internal(_) => "internal_error",
Self::Internal(_) | Self::CallbackException { .. } => "internal_error",
}
}

/// Returns the originating language exception class, when available.
pub(crate) fn exception_type(&self) -> Option<&str> {
match self {
Self::CallbackException { exception_type, .. } => Some(exception_type),
_ => None,
}
}
}
Expand Down
80 changes: 77 additions & 3 deletions crates/core/src/observability/otel.rs
Original file line number Diff line number Diff line change
Expand Up @@ -948,6 +948,8 @@ pub(super) struct ActiveSpan {
span_context: SpanContext,
start_model_name: Option<String>,
projected_attributes: Vec<KeyValue>,
descendant_error_type: Option<String>,
descendant_exception_type: Option<String>,
}

pub(super) struct OtelEventProcessor {
Expand Down Expand Up @@ -1151,12 +1153,15 @@ impl OtelEventProcessor {
span_context,
start_model_name,
projected_attributes,
descendant_error_type: None,
descendant_exception_type: None,
},
);
}

fn process_end(&mut self, event: &Event) {
let Some(mut active_span) = self.active_spans.remove(&event.uuid()) else {
self.propagate_suppressed_error_metadata(event);
return;
};
self.record_completed_span_context(event.uuid(), active_span.span_context.clone());
Expand All @@ -1167,6 +1172,36 @@ impl OtelEventProcessor {
OpenTelemetryType::GenAi => super::otel_genai::end_attributes(event),
OpenTelemetryType::OpenInference => super::openinference::end_attributes(event),
};
let is_error = metadata_string(event, "otel.status_code") == Some("ERROR");
let explicit_error_type = metadata_string(event, "error.type");
let error_type = is_error.then(|| {
explicit_error_type
.map(ToOwned::to_owned)
.or(active_span.descendant_error_type.take())
.unwrap_or_else(|| "_OTHER".to_string())
});
let exception_type = is_error
.then(|| {
metadata_string(event, "exception.type")
.map(ToOwned::to_owned)
.or(active_span.descendant_exception_type.take())
})
.flatten();
if matches!(
self.otel_type,
OpenTelemetryType::Full | OpenTelemetryType::GenAi
) && let Some(error_type) = error_type.as_ref()
{
attributes.retain(|attribute| attribute.key.as_str() != "error.type");
attributes.push(KeyValue::new("error.type", error_type.clone()));
}
if let Some(exception_type) = exception_type.as_ref() {
active_span.span.add_event_with_timestamp(
"exception",
to_system_time(*event.timestamp()),
vec![KeyValue::new("exception.type", exception_type.clone())],
);
}
let end_model_name =
model_name_for_llm_event(event).or_else(|| active_span.start_model_name.take());
if self.otel_type == OpenTelemetryType::Full
Expand All @@ -1187,12 +1222,40 @@ impl OtelEventProcessor {
&self.attribute_mappings,
));
}
if is_error && let Some(parent_span) = self.find_parent_span_mut(event) {
if parent_span.descendant_error_type.is_none() {
parent_span.descendant_error_type = error_type;
}
if parent_span.descendant_exception_type.is_none() {
parent_span.descendant_exception_type = exception_type;
}
}
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Resolved
active_span.span.set_attributes(attributes);
active_span
.span
.end_with_timestamp(to_system_time(*event.timestamp()));
}

fn propagate_suppressed_error_metadata(&mut self, event: &Event) {
if self.otel_type != OpenTelemetryType::GenAi
|| !self.suppressed_parent_contexts.contains_key(&event.uuid())
|| metadata_string(event, "otel.status_code") != Some("ERROR")
{
return;
}
let error_type = metadata_string(event, "error.type").map(ToOwned::to_owned);
let exception_type = metadata_string(event, "exception.type").map(ToOwned::to_owned);
let Some(parent_span) = self.find_parent_span_mut(event) else {
return;
};
if parent_span.descendant_error_type.is_none() {
parent_span.descendant_error_type = error_type;
}
if parent_span.descendant_exception_type.is_none() {
parent_span.descendant_exception_type = exception_type;
}
}

fn process_mark(&mut self, event: &Event) {
if self.otel_type == OpenTelemetryType::GenAi {
return;
Expand Down Expand Up @@ -1310,9 +1373,16 @@ impl OtelEventProcessor {
}

fn parent_span_uuid(&self, event: &Event) -> Option<Uuid> {
event
.parent_uuid()
.filter(|uuid| self.active_spans.contains_key(uuid))
let parent_uuid = event.parent_uuid()?;
if self.active_spans.contains_key(&parent_uuid) {
return Some(parent_uuid);
}
let suppressed_parent = self.suppressed_parent_contexts.get(&parent_uuid)?;
self.active_spans.iter().find_map(|(uuid, active_span)| {
(active_span.span_context.trace_id() == suppressed_parent.trace_id()
&& active_span.span_context.span_id() == suppressed_parent.span_id())
.then_some(*uuid)
})
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Resolved
}

fn find_parent_span(&self, event: &Event) -> Option<&ActiveSpan> {
Expand Down Expand Up @@ -1368,6 +1438,10 @@ impl OtelEventProcessor {
}
}

fn metadata_string<'a>(event: &'a Event, key: &str) -> Option<&'a str> {
event.metadata()?.get(key)?.as_str()
}

fn span_kind(event: &Event) -> SpanKind {
match semantic_scope_type(event) {
Some(ScopeType::Llm) => SpanKind::Client,
Expand Down
4 changes: 3 additions & 1 deletion crates/core/src/plugin/dynamic/native.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1293,7 +1293,9 @@ fn status_from_flow_error(err: FlowError) -> NemoRelayStatus {
FlowError::InvalidArgument(_) => NemoRelayStatus::InvalidArg,
FlowError::ScopeStackEmpty => NemoRelayStatus::ScopeStackEmpty,
FlowError::GuardrailRejected(_) => NemoRelayStatus::GuardrailRejected,
FlowError::Upstream(_) | FlowError::Internal(_) => NemoRelayStatus::Internal,
FlowError::Upstream(_) | FlowError::Internal(_) | FlowError::CallbackException { .. } => {
NemoRelayStatus::Internal
}
}
}

Expand Down
6 changes: 6 additions & 0 deletions crates/core/tests/coverage/error_tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -104,6 +104,12 @@ fn otel_error_type_maps_internal_failures_to_generic_code() {
FlowError::Internal("application callback failed".into()).otel_error_type(),
"internal_error"
);
let external = FlowError::CallbackException {
message: "ValueError: boom".into(),
exception_type: "ValueError".into(),
};
assert_eq!(external.otel_error_type(), "internal_error");
assert_eq!(external.exception_type(), Some("ValueError"));
}

#[test]
Expand Down
Loading
Loading