Skip to content
Merged
Show file tree
Hide file tree
Changes from all 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.
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.
}

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