Skip to content
Merged
Show file tree
Hide file tree
Changes from 8 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
7 changes: 7 additions & 0 deletions crates/cli/tests/coverage/plugins_tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -200,6 +200,13 @@ fn typed_editor_model_contains_observability_sections() {
.iter()
.any(|field| field.name == "endpoint")
);
let attribute_mappings = openinference.field("attribute_mappings").unwrap();
assert_eq!(attribute_mappings.kind, EditorFieldKind::List);
let mapping = attribute_mappings.list_item.unwrap();
assert_eq!(mapping.kind, EditorFieldKind::Section);
let mapping_schema = mapping.schema.unwrap()();
assert_eq!(mapping_schema.fields[0].name, "key");
assert_eq!(mapping_schema.fields[1].name, "alias");
}

#[test]
Expand Down
183 changes: 183 additions & 0 deletions crates/core/src/observability/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,29 @@
use crate::api::event::EventNormalizationExt;
use serde::{Deserialize, Serialize};

/// Copies a projected OTLP attribute to a second attribute name.
///
/// `key` names the fully-qualified projected attribute and `alias` names the
/// additional attribute to emit with the same typed value.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "schema", derive(schemars::JsonSchema))]
pub struct OtlpAttributeMapping {
/// Fully-qualified projected attribute to copy.
pub key: String,
/// Additional attribute name receiving the copied value.
pub alias: String,
}

impl OtlpAttributeMapping {
/// Creates an attribute mapping.
pub fn new(key: impl Into<String>, alias: impl Into<String>) -> Self {
Self {
key: key.into(),
alias: alias.into(),
}
}
}

#[cfg(test)]
use std::sync::Mutex;

Expand Down Expand Up @@ -49,6 +72,162 @@ pub(crate) fn default_mark_exclude_names() -> Vec<String> {
vec!["llm.chunk".to_string()]
}

/// Validates OTLP attribute mappings shared by exporter configuration surfaces.
pub fn validate_attribute_mappings(
mappings: &[OtlpAttributeMapping],
) -> std::result::Result<(), String> {
let mut aliases = std::collections::HashSet::new();
for mapping in mappings {
if mapping.key.trim().is_empty() {
return Err("attribute mapping key must not be blank".to_string());
}
if mapping.alias.trim().is_empty() {
return Err("attribute mapping alias must not be blank".to_string());
}
if !aliases.insert(mapping.alias.trim()) {
return Err(format!(
"attribute mapping alias {:?} is duplicated",
mapping.alias
));
}
}
Ok(())
}
Comment thread
willkill07 marked this conversation as resolved.

#[cfg(any(feature = "otel", feature = "openinference"))]
/// Projects only top-level JSON fields as OTLP attributes.
///
/// Nested objects and arrays remain JSON strings so arbitrary payloads do not
/// create ambiguous dotted attribute paths or unbounded attribute sets.
pub(crate) fn push_top_level_json_attributes(
attributes: &mut Vec<opentelemetry::KeyValue>,
prefix: &str,
value: Option<&crate::json::Json>,
) {
let Some(value) = value else {
return;
};
match value {
crate::json::Json::Object(values) => {
for (field, value) in values {
push_top_level_json_value(attributes, &format!("{prefix}.{field}"), value);
}
}
value => push_top_level_json_value(attributes, prefix, value),
}
}

#[cfg(any(feature = "otel", feature = "openinference"))]
/// Serializes a value and projects its top-level JSON fields as OTLP attributes.
pub(crate) fn push_serialized_top_level_attributes<T: Serialize + ?Sized>(
attributes: &mut Vec<opentelemetry::KeyValue>,
prefix: &str,
value: Option<&T>,
) {
let Some(value) = value else {
return;
};
if let Ok(value) = serde_json::to_value(value) {
push_top_level_json_attributes(attributes, prefix, Some(&value));
}
}

#[cfg(any(feature = "otel", feature = "openinference"))]
fn push_top_level_json_value(
attributes: &mut Vec<opentelemetry::KeyValue>,
key: &str,
value: &crate::json::Json,
) {
use opentelemetry::KeyValue;

match value {
crate::json::Json::Null => {}
crate::json::Json::Bool(value) => attributes.push(KeyValue::new(key.to_string(), *value)),
crate::json::Json::String(value) => {
attributes.push(KeyValue::new(key.to_string(), value.clone()))
}
crate::json::Json::Number(value) => {
if let Some(value) = value.as_i64() {
attributes.push(KeyValue::new(key.to_string(), value));
} else if let Some(value) = value.as_u64() {
if let Ok(value) = i64::try_from(value) {
attributes.push(KeyValue::new(key.to_string(), value));
} else {
attributes.push(KeyValue::new(key.to_string(), value.to_string()));
}
} else if let Some(value) = value.as_f64() {
attributes.push(KeyValue::new(key.to_string(), value));
}
}
crate::json::Json::Array(_) | crate::json::Json::Object(_) => {
if let Ok(value) = serde_json::to_string(value) {
attributes.push(KeyValue::new(key.to_string(), value));
}
}
}
}

#[cfg(any(feature = "otel", feature = "openinference"))]
pub(crate) fn apply_attribute_mappings(
attributes: &mut Vec<opentelemetry::KeyValue>,
mappings: &[OtlpAttributeMapping],
) {
attributes.extend(attribute_mapping_aliases(attributes, mappings));
}

/// Keeps the start attributes needed to resolve mappings at the end of a span.
///
/// The final span attributes must still take precedence over mapped aliases, so
/// retain both mapped source keys and aliases that were already present at
/// start. The span itself owns all other start attributes and does not need a
/// second copy in the active-span state.
#[cfg(any(feature = "otel", feature = "openinference"))]
pub(crate) fn attribute_mapping_inputs(
attributes: &[opentelemetry::KeyValue],
mappings: &[OtlpAttributeMapping],
) -> Vec<opentelemetry::KeyValue> {
attributes
.iter()
.filter(|attribute| {
mappings.iter().any(|mapping| {
attribute.key.as_str() == mapping.key || attribute.key.as_str() == mapping.alias
})
})
.cloned()
.collect()
}

/// Resolves typed aliases from a complete set of projected attributes.
///
/// Callers that project a span across multiple lifecycle events must pass every
/// real span attribute so projected fields always take precedence over aliases.
#[cfg(any(feature = "otel", feature = "openinference"))]
pub(crate) fn attribute_mapping_aliases(
projected_attributes: &[opentelemetry::KeyValue],
mappings: &[OtlpAttributeMapping],
) -> Vec<opentelemetry::KeyValue> {
if mappings.is_empty() {
return Vec::new();
}
let existing = projected_attributes
.iter()
.map(|attribute| attribute.key.as_str().to_string())
.collect::<std::collections::HashSet<_>>();
mappings
.iter()
.filter(|mapping| !existing.contains(mapping.alias.as_str()))
.filter_map(|mapping| {
projected_attributes
.iter()
.rev()
.find(|attribute| attribute.key.as_str() == mapping.key)
.map(|attribute| {
opentelemetry::KeyValue::new(mapping.alias.clone(), attribute.value.clone())
})
})
.collect()
}

/// Returns whether a mark matches a configured projection exclusion.
///
/// Agent hook adapters may preserve the canonical event name in metadata while
Expand Down Expand Up @@ -200,3 +379,7 @@ where
};
span.set_status(status);
}

#[cfg(all(test, any(feature = "otel", feature = "openinference")))]
#[path = "../../tests/unit/observability/attribute_projection_tests.rs"]
mod attribute_projection_tests;
Loading