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
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 @@ -210,6 +210,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
Loading