diff --git a/splunk-ao-a2a/src/splunk_ao_a2a/_constants.py b/splunk-ao-a2a/src/splunk_ao_a2a/_constants.py index 6dade0cb..eed3a454 100644 --- a/splunk-ao-a2a/src/splunk_ao_a2a/_constants.py +++ b/splunk-ao-a2a/src/splunk_ao_a2a/_constants.py @@ -10,7 +10,7 @@ SPLUNK_AO_OBSERVE_KEY = "splunk_ao_observe" AGNTCY_OBSERVE_KEY = "observe" # compatibility with AGNTCY Observe SDK -# A2A span attribute keys (match API-side A2A extension expectations) +# A2A protocol span attributes expected by the ingest API A2A_TASK_ID = "a2a.task.id" A2A_CONTEXT_ID = "a2a.context_id" A2A_RPC_METHOD = "a2a.rpc.method" @@ -19,7 +19,7 @@ # OTel GenAI semantic convention attributes — span type determination GENAI_OPERATION_NAME = "gen_ai.operation.name" GENAI_AGENT_NAME = "gen_ai.agent.name" -GENAI_SYSTEM = "gen_ai.system" +GENAI_CONVERSATION_ID = "gen_ai.conversation.id" GENAI_TOOL_NAME = "gen_ai.tool.name" # OTel GenAI semantic convention attributes — input/output content @@ -36,9 +36,6 @@ # Finish reasons FINISH_REASON_STOP = "stop" -# Session correlation -SESSION_ID = "session.id" - # Span link attributes for cross-agent correlation LINK_TYPE = "link.type" LINK_TYPE_AGENT_HANDOFF = "agent_handoff" diff --git a/splunk-ao-a2a/src/splunk_ao_a2a/_context.py b/splunk-ao-a2a/src/splunk_ao_a2a/_context.py index 12b9a1de..e938aa47 100644 --- a/splunk-ao-a2a/src/splunk_ao_a2a/_context.py +++ b/splunk-ao-a2a/src/splunk_ao_a2a/_context.py @@ -12,10 +12,10 @@ from splunk_ao_a2a._constants import ( AGNTCY_OBSERVE_KEY, - SPLUNK_AO_OBSERVE_KEY, LINK_FROM_AGENT, LINK_TYPE, LINK_TYPE_AGENT_HANDOFF, + SPLUNK_AO_OBSERVE_KEY, ) _logger = logging.getLogger(__name__) diff --git a/splunk-ao-a2a/src/splunk_ao_a2a/_spans.py b/splunk-ao-a2a/src/splunk_ao_a2a/_spans.py index 2e37f88c..ead1ac40 100644 --- a/splunk-ao-a2a/src/splunk_ao_a2a/_spans.py +++ b/splunk-ao-a2a/src/splunk_ao_a2a/_spans.py @@ -17,15 +17,14 @@ ERROR_STATES, FINISH_REASON_STOP, GENAI_AGENT_NAME, + GENAI_CONVERSATION_ID, GENAI_INPUT_MESSAGES, GENAI_OPERATION_NAME, GENAI_OUTPUT_MESSAGES, GENAI_RESPONSE_FINISH_REASONS, - GENAI_SYSTEM, GENAI_TOOL_NAME, ROLE_ASSISTANT, ROLE_USER, - SESSION_ID, ) @@ -68,7 +67,6 @@ def set_client_attributes( ) -> None: """Set standard A2A and GenAI attributes on a client span.""" span.set_attribute(GENAI_OPERATION_NAME, "invoke_agent") - span.set_attribute(GENAI_SYSTEM, "a2a") span.set_attribute(A2A_RPC_METHOD, rpc_method) if agent_name: @@ -76,7 +74,7 @@ def set_client_attributes( if hasattr(request, "context_id") and request.context_id: span.set_attribute(A2A_CONTEXT_ID, str(request.context_id)) - span.set_attribute(SESSION_ID, str(request.context_id)) + span.set_attribute(GENAI_CONVERSATION_ID, str(request.context_id)) if hasattr(request, "task_id") and request.task_id: span.set_attribute(A2A_TASK_ID, str(request.task_id)) @@ -90,7 +88,6 @@ def set_server_attributes( ) -> None: """Set standard A2A and GenAI attributes on a server span.""" span.set_attribute(GENAI_OPERATION_NAME, "invoke_agent") - span.set_attribute(GENAI_SYSTEM, "a2a") span.set_attribute(A2A_RPC_METHOD, rpc_method) if agent_name: @@ -101,7 +98,7 @@ def set_server_attributes( context_id = getattr(message, "context_id", None) if context_id: span.set_attribute(A2A_CONTEXT_ID, str(context_id)) - span.set_attribute(SESSION_ID, str(context_id)) + span.set_attribute(GENAI_CONVERSATION_ID, str(context_id)) task_id = getattr(message, "task_id", None) if task_id: @@ -111,7 +108,6 @@ def set_server_attributes( def set_tool_attributes(span: trace.Span, rpc_method: str) -> None: """Set attributes for tool-like A2A operations (get_task, cancel_task, get_card).""" span.set_attribute(GENAI_OPERATION_NAME, "execute_tool") - span.set_attribute(GENAI_SYSTEM, "a2a") span.set_attribute(A2A_RPC_METHOD, rpc_method) span.set_attribute(GENAI_TOOL_NAME, rpc_method) @@ -180,7 +176,7 @@ def set_simple_input(span: trace.Span, args: tuple, rpc_method: str) -> None: def track_task_state(span: trace.Span, obj: Any) -> None: """Record A2A task state and ID on *span*. - Sets ``a2a.task.state``, ``a2a.task.id``, and ``gen_ai.response.finish_reasons``. + Sets the A2A task state and ID plus ``gen_ai.response.finish_reasons``. Marks the span as an error when the task enters a terminal error state. """ if obj is None: @@ -195,9 +191,9 @@ def track_task_state(span: trace.Span, obj: Any) -> None: if state_value in ERROR_STATES: span.set_status(StatusCode.ERROR, f"A2A task {state_value}") - span.set_attribute(GENAI_RESPONSE_FINISH_REASONS, json.dumps([state_value])) + span.set_attribute(GENAI_RESPONSE_FINISH_REASONS, (state_value,)) elif state_value == "completed": - span.set_attribute(GENAI_RESPONSE_FINISH_REASONS, json.dumps([FINISH_REASON_STOP])) + span.set_attribute(GENAI_RESPONSE_FINISH_REASONS, (FINISH_REASON_STOP,)) task_id = getattr(obj, "id", None) if task_id: diff --git a/splunk-ao-a2a/tests/test_spans.py b/splunk-ao-a2a/tests/test_spans.py index 39e4f196..7b410d80 100644 --- a/splunk-ao-a2a/tests/test_spans.py +++ b/splunk-ao-a2a/tests/test_spans.py @@ -14,13 +14,12 @@ A2A_TASK_ID, A2A_TASK_STATE, GENAI_AGENT_NAME, + GENAI_CONVERSATION_ID, GENAI_INPUT_MESSAGES, GENAI_OPERATION_NAME, GENAI_OUTPUT_MESSAGES, GENAI_RESPONSE_FINISH_REASONS, - GENAI_SYSTEM, GENAI_TOOL_NAME, - SESSION_ID, ) @@ -55,12 +54,12 @@ def test_sets_standard_attributes(self, span): # Then: all expected attributes are set span.set_attribute.assert_any_call(GENAI_OPERATION_NAME, "invoke_agent") - span.set_attribute.assert_any_call(GENAI_SYSTEM, "a2a") span.set_attribute.assert_any_call(A2A_RPC_METHOD, "SendMessage") span.set_attribute.assert_any_call(GENAI_AGENT_NAME, "my-agent") span.set_attribute.assert_any_call(A2A_CONTEXT_ID, "ctx-1") - span.set_attribute.assert_any_call(SESSION_ID, "ctx-1") + span.set_attribute.assert_any_call(GENAI_CONVERSATION_ID, "ctx-1") span.set_attribute.assert_any_call(A2A_TASK_ID, "task-1") + assert "gen_ai.system" not in [call.args[0] for call in span.set_attribute.call_args_list] def test_skips_agent_name_when_none(self, span): # Given: no agent name @@ -84,7 +83,7 @@ def test_sets_attributes_from_message(self, span): # Then: attributes extracted from message span.set_attribute.assert_any_call(A2A_CONTEXT_ID, "ctx-2") - span.set_attribute.assert_any_call(SESSION_ID, "ctx-2") + span.set_attribute.assert_any_call(GENAI_CONVERSATION_ID, "ctx-2") span.set_attribute.assert_any_call(A2A_TASK_ID, "task-2") @@ -183,7 +182,7 @@ def test_sets_completed_state(self, span): # Then: state and finish reason set span.set_attribute.assert_any_call(A2A_TASK_STATE, "completed") - span.set_attribute.assert_any_call(GENAI_RESPONSE_FINISH_REASONS, json.dumps(["stop"])) + span.set_attribute.assert_any_call(GENAI_RESPONSE_FINISH_REASONS, ("stop",)) span.set_attribute.assert_any_call(A2A_TASK_ID, "t-1") @pytest.mark.parametrize("state", ["failed", "rejected", "canceled"]) diff --git a/splunk-ao-a2a/tests/test_splunk_ao_compatibility.py b/splunk-ao-a2a/tests/test_splunk_ao_compatibility.py index 34398893..421405f8 100644 --- a/splunk-ao-a2a/tests/test_splunk_ao_compatibility.py +++ b/splunk-ao-a2a/tests/test_splunk_ao_compatibility.py @@ -73,8 +73,14 @@ def exporter_factory(**kwargs: object) -> RecordingExporter: } exported = delegate.spans[0] assert exported.instrumentation_scope.name == INSTRUMENTOR_NAME - assert exported.attributes["gen_ai.system"] == "a2a" + assert "gen_ai.system" not in exported.attributes + assert exported.attributes["splunk_ao.system"] == "splunk_ao_python" assert exported.attributes["a2a.rpc.method"] == "SendMessage" + assert "splunk_ao.a2a.rpc.method" not in exported.attributes + assert exported.attributes["gen_ai.conversation.id"] == "context-id" + assert exported.attributes["splunk_ao.session.id"] == "context-id" + assert exported.attributes["gen_ai.operation.name"] == "invoke_agent" + assert exported.attributes["splunk_ao.operation.name"] == "invoke_agent" assert exported.resource.attributes["splunk_ao.project.name"] == "a2a-project" assert exported.resource.attributes["splunk_ao.logstream.name"] == "a2a-agent-stream" assert "splunk_ao.project.name" not in exported.attributes diff --git a/src/splunk_ao/converter/__init__.py b/src/splunk_ao/converter/__init__.py new file mode 100644 index 00000000..654ae5dd --- /dev/null +++ b/src/splunk_ao/converter/__init__.py @@ -0,0 +1,15 @@ +"""Conversion helpers shared by Splunk AO telemetry paths.""" + +from splunk_ao.converter.attribute_mapping import ( + CONTENT_ALIAS_BY_GEN_AI, + SPLUNK_ALIAS_BY_GEN_AI, + build_span_attributes, + normalize_attributes_for_export, +) + +__all__ = [ + "CONTENT_ALIAS_BY_GEN_AI", + "SPLUNK_ALIAS_BY_GEN_AI", + "build_span_attributes", + "normalize_attributes_for_export", +] diff --git a/src/splunk_ao/converter/attribute_mapping.py b/src/splunk_ao/converter/attribute_mapping.py new file mode 100644 index 00000000..af87a887 --- /dev/null +++ b/src/splunk_ao/converter/attribute_mapping.py @@ -0,0 +1,511 @@ +"""Canonical Galileo-field and OTLP wire-attribute mapping.""" + +from __future__ import annotations + +import json +from collections.abc import Mapping, MutableMapping, Sequence +from enum import Enum +from typing import Any + +from opentelemetry.util.types import AttributeValue +from pydantic import BaseModel + +from galileo_core.schemas.logging.span import AgentSpan, LlmSpan, RetrieverSpan, ToolSpan, WorkflowSpan +from galileo_core.schemas.logging.step import BaseStep, StepType + +SPLUNK_AO_SYSTEM = "splunk_ao.system" +SPLUNK_AO_SYSTEM_VALUE = "splunk_ao_python" + +CONTENT_ALIAS_BY_GEN_AI: Mapping[str, str] = { + "gen_ai.input.messages": "splunk_ao.input.messages", + "gen_ai.system_instructions": "splunk_ao.system_instructions", + "gen_ai.output.messages": "splunk_ao.output.messages", + "gen_ai.tool.call.arguments": "splunk_ao.tool.call.arguments", + "gen_ai.tool.call.result": "splunk_ao.tool.call.result", + "gen_ai.retrieval.documents": "splunk_ao.retrieval.documents", + "gen_ai.tool.definitions": "splunk_ao.tool.definitions", +} + +SPLUNK_ALIAS_BY_GEN_AI: Mapping[str, str] = { + "gen_ai.operation.name": "splunk_ao.operation.name", + "gen_ai.conversation.id": "splunk_ao.session.id", + "gen_ai.workflow.name": "splunk_ao.workflow.name", + "gen_ai.agent.name": "splunk_ao.agent.name", + "gen_ai.agent.id": "splunk_ao.agent.id", + "gen_ai.agent.description": "splunk_ao.agent.description", + "gen_ai.agent.version": "splunk_ao.agent.version", + "gen_ai.provider.name": "splunk_ao.provider.name", + "gen_ai.request.model": "splunk_ao.request.model", + "gen_ai.request.temperature": "splunk_ao.request.temperature", + "gen_ai.request.top_p": "splunk_ao.request.top_p", + "gen_ai.request.top_k": "splunk_ao.request.top_k", + "gen_ai.request.max_tokens": "splunk_ao.request.max_tokens", + "gen_ai.request.stop_sequences": "splunk_ao.request.stop_sequences", + "gen_ai.request.frequency_penalty": "splunk_ao.request.frequency_penalty", + "gen_ai.request.presence_penalty": "splunk_ao.request.presence_penalty", + "gen_ai.request.seed": "splunk_ao.request.seed", + "gen_ai.response.finish_reasons": "splunk_ao.response.finish_reasons", + "gen_ai.response.model": "splunk_ao.response.model", + "gen_ai.response.id": "splunk_ao.response.id", + "gen_ai.output.type": "splunk_ao.output.type", + "gen_ai.tool.call.id": "splunk_ao.tool.call.id", + "gen_ai.tool.name": "splunk_ao.tool.name", + "gen_ai.tool.description": "splunk_ao.tool.description", + "gen_ai.tool.type": "splunk_ao.tool.type", + "gen_ai.retrieval.top_k": "splunk_ao.retrieval.top_k", + "gen_ai.retrieval.query.text": "splunk_ao.retrieval.query.text", + "gen_ai.usage.input_tokens": "splunk_ao.llm.usage.input_tokens", + "gen_ai.usage.output_tokens": "splunk_ao.llm.usage.output_tokens", + "gen_ai.usage.cache_creation.input_tokens": "splunk_ao.llm.usage.cache_creation.input_tokens", + "gen_ai.usage.cache_read.input_tokens": "splunk_ao.llm.usage.cache_read.input_tokens", + "gen_ai.usage.reasoning.output_tokens": "splunk_ao.llm.usage.reasoning.output_tokens", + "gen_ai.response.time_to_first_chunk": "splunk_ao.llm.time_to_first_token_ns", + **CONTENT_ALIAS_BY_GEN_AI, +} + +_OPERATION_BY_STEP_TYPE = { + StepType.llm: "chat", + StepType.retriever: "retrieval", + StepType.tool: "execute_tool", + StepType.workflow: "invoke_workflow", + StepType.agent: "invoke_agent", +} + + +def _json_compatible(value: Any) -> Any: + if isinstance(value, BaseModel): + return value.model_dump(mode="json", exclude_none=True) + if isinstance(value, Enum): + return value.value + if isinstance(value, Mapping): + return {str(key): _json_compatible(item) for key, item in value.items()} + if isinstance(value, Sequence) and not isinstance(value, str | bytes | bytearray): + return [_json_compatible(item) for item in value] + return value + + +def _json_string(value: Any) -> str: + return json.dumps(_json_compatible(value), separators=(",", ":"), sort_keys=True, default=str) + + +def _content_value(value: Any) -> str: + return value if isinstance(value, str) else _json_string(value) + + +def _mapping_value(value: Any) -> dict[str, Any] | None: + if isinstance(value, BaseModel): + return value.model_dump(mode="json", exclude_none=True) + if isinstance(value, Mapping): + return {str(key): _json_compatible(item) for key, item in value.items()} + return None + + +def _mapping_view(value: Any) -> Mapping[str, Any] | None: + if isinstance(value, BaseModel): + return value.model_dump(mode="json", exclude_none=True) + return value if isinstance(value, Mapping) else None + + +def _parse_json_string(value: Any) -> Any: + if not isinstance(value, str): + return value + try: + return json.loads(value) + except (TypeError, ValueError): + return value + + +def _parse_json_value(value: Any) -> Any: + parsed = _parse_json_string(value) + return parsed if isinstance(value, str) else _json_compatible(parsed) + + +def _text_part(value: Any) -> dict[str, Any]: + content = value if isinstance(value, str) else _json_string(value) + return {"type": "text", "content": content} + + +def _content_part(value: Any) -> dict[str, Any] | None: + part = _mapping_value(value) + if part is None or "type" not in part: + return None + + part_type = _json_compatible(part["type"]) + part["type"] = str(part_type) + if part_type == "text" and "content" not in part and "text" in part: + part["content"] = part.pop("text") + return part + + +def _content_parts(value: Any) -> list[dict[str, Any]]: + part = _content_part(value) + if part is not None: + return [part] + + if isinstance(value, Sequence) and not isinstance(value, str | bytes | bytearray): + parts = [_content_part(item) for item in value] + if all(item is not None for item in parts): + return [item for item in parts if item is not None] + + return [_text_part(value)] + + +def _tool_call_part(value: Any) -> dict[str, Any]: + call = _mapping_value(value) or {} + function = _mapping_value(call.pop("function", None)) or {} + result = {**call, "type": "tool_call"} + if "name" in function: + result["name"] = function.pop("name") + if "arguments" in function: + result["arguments"] = _parse_json_value(function.pop("arguments")) + result.update(function) + return result + + +def _mapped_message(source: dict[str, Any], default_role: str) -> dict[str, Any]: + role = _json_compatible(source.pop("role")) + content = source.pop("content", None) + source_parts = source.pop("parts", None) + tool_call_id = source.pop("tool_call_id", None) + tool_calls = source.pop("tool_calls", None) + + if source_parts is not None: + parts = _content_parts(source_parts) + elif role == "tool": + response = {"type": "tool_call_response", "response": _parse_json_value(content)} + if tool_call_id is not None: + response["id"] = str(tool_call_id) + parts = [response] + else: + parts = [] if content in (None, "") and tool_calls else _content_parts("" if content is None else content) + + if tool_calls: + parts.extend(_tool_call_part(tool_call) for tool_call in tool_calls) + + return {**source, "role": str(role), "parts": parts} + + +def _message(value: Any, default_role: str) -> dict[str, Any]: + source = _mapping_value(value) + if source is None or "role" not in source: + return {"role": default_role, "parts": _content_parts(value)} + return _mapped_message(source, default_role) + + +def _message_sequence(value: Any, default_role: str) -> list[dict[str, Any]]: + if not isinstance(value, Sequence) or isinstance(value, str | bytes | bytearray): + return [_message(value, default_role)] + if not value: + return [] + + messages = [_mapping_value(item) for item in value] + if all(message is not None and "role" in message for message in messages): + return [_mapped_message(message, default_role) for message in messages if message is not None] + return [_message(value, default_role)] + + +def _message_container(value: Any) -> tuple[Any | None, bool]: + parsed = _parse_json_string(value) + + source = _mapping_view(parsed) + if source is not None: + update = _mapping_view(_parse_json_string(source.get("update"))) + for container in (update, source): + if container is None or "messages" not in container: + continue + messages = _parse_json_string(container["messages"]) + if _is_message_sequence(messages): + return messages, container is source + + if "role" in source or "type" in source: + return parsed, False + return None, False + + if isinstance(parsed, Sequence) and not isinstance(parsed, str | bytes | bytearray): + if _is_message_sequence(parsed) or _is_content_part_sequence(parsed): + return parsed, False + return (parsed, False) if not isinstance(value, str | bytes | bytearray) else (None, False) + + return parsed, False + + +def _is_message_sequence(value: Any) -> bool: + if not isinstance(value, Sequence) or isinstance(value, str | bytes | bytearray): + message = _mapping_view(value) + return message is not None and "role" in message + return not value or all((message := _mapping_view(item)) is not None and "role" in message for item in value) + + +def _is_content_part_sequence(value: Any) -> bool: + return bool(value) and all((part := _mapping_view(item)) is not None and "type" in part for item in value) + + +def _orchestration_messages(value: Any, default_role: str) -> tuple[list[dict[str, Any]] | None, bool]: + container, full_history = _message_container(value) + messages = None if container is None else _message_sequence(container, default_role) + return messages, full_history + + +def _with_finish_reasons(messages: list[dict[str, Any]], finish_reason: str | None = None) -> list[dict[str, Any]]: + for message in messages: + source_finish_reason = message.get("finish_reason") + message["finish_reason"] = finish_reason or source_finish_reason or "unknown" + return messages + + +def _input_messages(value: Any) -> str: + return _json_string(_message_sequence(value, "user")) + + +def _output_messages(value: Any, finish_reason: str | None = None) -> str: + return _json_string(_with_finish_reasons(_message_sequence(value, "assistant"), finish_reason)) + + +def _tool_definitions(value: Sequence[Any]) -> str: + definitions: list[Any] = [] + for tool in value: + definition = _mapping_value(tool) + if definition is None: + definitions.append(_json_compatible(tool)) + continue + + function = _mapping_value(definition.pop("function", None)) + if function is not None: + definition = {**function, **definition} + definition.setdefault("type", "function") + definitions.append(definition) + return _json_string(definitions) + + +def _object_content(value: Any) -> str: + parsed = _parse_json_value(value) + content = parsed if isinstance(parsed, Mapping) else {"value": parsed} + return _json_string(content) + + +def _set_if_present(attrs: MutableMapping[str, AttributeValue], key: str, value: AttributeValue | None) -> None: + if value is not None: + attrs[key] = value + + +def _field(source: Any, name: str) -> Any: + value = getattr(source, name, None) + if value is not None: + return value + extra = getattr(source, "model_extra", None) + return extra.get(name) if extra else None + + +def set_common_attributes( + attrs: MutableMapping[str, AttributeValue], span: BaseStep, session_id: str | None = None +) -> None: + """Map fields shared by every proprietary span type.""" + if span.redacted_input is not None: + attrs["splunk_ao.redacted_input"] = _content_value(span.redacted_input) + if span.redacted_output is not None: + attrs["splunk_ao.redacted_output"] = _content_value(span.redacted_output) + if span.user_metadata: + attrs["splunk_ao.metadata"] = _json_string(span.user_metadata) + if span.tags: + attrs["splunk_ao.tags"] = tuple(span.tags) + if span.status_code is not None: + attrs["splunk_ao.status_code"] = span.status_code + if span.status_code >= 400: + attrs["error.type"] = str(span.status_code) + + step_number = getattr(span, "step_number", None) + if step_number is not None: + attrs["splunk_ao.step_number"] = step_number + + resolved_session_id = session_id or getattr(span, "session_id", None) + if resolved_session_id is not None: + attrs["gen_ai.conversation.id"] = str(resolved_session_id) + + +def set_dataset_attributes(attrs: MutableMapping[str, AttributeValue], span: BaseStep) -> None: + """Map optional dataset context fields.""" + _set_if_present(attrs, "splunk_ao.dataset.input", span.dataset_input) + _set_if_present(attrs, "splunk_ao.dataset.output", span.dataset_output) + if span.dataset_metadata: + attrs["splunk_ao.dataset.metadata"] = _json_string(span.dataset_metadata) + + +def _set_operation(attrs: MutableMapping[str, AttributeValue], span_type: StepType) -> None: + operation = _OPERATION_BY_STEP_TYPE.get(span_type) + if operation is not None: + attrs["gen_ai.operation.name"] = operation + + +def set_llm_attributes(attrs: MutableMapping[str, AttributeValue], span: LlmSpan) -> None: + """Map LLM request, response, content, and metric fields.""" + _set_operation(attrs, StepType.llm) + attrs["gen_ai.input.messages"] = _input_messages(span.input) + attrs["gen_ai.output.messages"] = _output_messages(span.output, span.finish_reason) + _set_if_present(attrs, "gen_ai.request.model", span.model) + _set_if_present(attrs, "gen_ai.request.temperature", span.temperature) + + for field_name, attribute_name in ( + ("provider", "gen_ai.provider.name"), + ("top_p", "gen_ai.request.top_p"), + ("top_k", "gen_ai.request.top_k"), + ("max_tokens", "gen_ai.request.max_tokens"), + ("stop_sequences", "gen_ai.request.stop_sequences"), + ("frequency_penalty", "gen_ai.request.frequency_penalty"), + ("presence_penalty", "gen_ai.request.presence_penalty"), + ("seed", "gen_ai.request.seed"), + ("response_model", "gen_ai.response.model"), + ("response_id", "gen_ai.response.id"), + ): + _set_if_present(attrs, attribute_name, _field(span, field_name)) + + if span.finish_reason is not None: + attrs["gen_ai.response.finish_reasons"] = (span.finish_reason,) + if span.tools is not None: + attrs["gen_ai.tool.definitions"] = _tool_definitions(span.tools) + if span.events is not None: + attrs["splunk_ao.llm.events"] = _json_string(span.events) + + metrics = span.metrics + _set_if_present(attrs, "gen_ai.usage.input_tokens", metrics.num_input_tokens) + _set_if_present(attrs, "gen_ai.usage.output_tokens", metrics.num_output_tokens) + _set_if_present(attrs, "splunk_ao.llm.usage.total_tokens", metrics.num_total_tokens) + if metrics.time_to_first_token_ns is not None: + attrs["gen_ai.response.time_to_first_chunk"] = metrics.time_to_first_token_ns / 1_000_000_000 + attrs["splunk_ao.llm.time_to_first_token_ns"] = metrics.time_to_first_token_ns + + for field_name, attribute_name in ( + ("input_cost", "splunk_ao.llm.cost.input_usd"), + ("output_cost", "splunk_ao.llm.cost.output_usd"), + ("total_cost", "splunk_ao.llm.cost.total_usd"), + ("cache_creation_input_tokens", "gen_ai.usage.cache_creation.input_tokens"), + ("cache_read_input_tokens", "gen_ai.usage.cache_read.input_tokens"), + ("reasoning_output_tokens", "gen_ai.usage.reasoning.output_tokens"), + ): + _set_if_present(attrs, attribute_name, _field(metrics, field_name)) + + for field_name, attribute_name in ( + ("log_probs", "splunk_ao.llm.log_probs"), + ("top_logprobs", "splunk_ao.llm.top_logprobs"), + ("response_format", "splunk_ao.llm.response_format"), + ("tool_use_allowed", "splunk_ao.llm.tool_use_allowed"), + ("structured_output_name", "splunk_ao.llm.structured_output.name"), + ("structured_output_input", "splunk_ao.llm.structured_output.input"), + ): + _set_if_present(attrs, attribute_name, _field(span, field_name)) + + +def set_tool_attributes(attrs: MutableMapping[str, AttributeValue], span: ToolSpan) -> None: + """Map tool execution fields.""" + _set_operation(attrs, StepType.tool) + attrs["gen_ai.tool.name"] = span.name + attrs["gen_ai.tool.call.arguments"] = _object_content(span.input) + if span.output is not None: + attrs["gen_ai.tool.call.result"] = _object_content(span.output) + _set_if_present(attrs, "gen_ai.tool.call.id", span.tool_call_id) + + +def set_retriever_attributes(attrs: MutableMapping[str, AttributeValue], span: RetrieverSpan) -> None: + """Map retrieval query and document fields.""" + _set_operation(attrs, StepType.retriever) + attrs["gen_ai.retrieval.query.text"] = span.input + attrs["gen_ai.retrieval.documents"] = _content_value(span.output) + attrs["splunk_ao.retrieval.documents.count"] = len(span.output) + attrs["db.operation"] = "search" + requested_top_k = _field(span, "num_documents") + _set_if_present(attrs, "gen_ai.retrieval.top_k", requested_top_k) + + +def _set_orchestration_content(attrs: MutableMapping[str, AttributeValue], span: WorkflowSpan | AgentSpan) -> None: + input_messages, _ = _orchestration_messages(span.input, "user") + if input_messages is not None: + attrs["gen_ai.input.messages"] = _json_string(input_messages) + + if span.output is None: + return + + output_messages, full_history = _orchestration_messages(span.output, "assistant") + if output_messages is None: + return + if full_history and input_messages is not None and output_messages[: len(input_messages)] == input_messages: + output_messages = output_messages[len(input_messages) :] + if output_messages: + attrs["gen_ai.output.messages"] = _json_string(_with_finish_reasons(output_messages)) + + +def set_workflow_attributes(attrs: MutableMapping[str, AttributeValue], span: WorkflowSpan) -> None: + """Map workflow fields.""" + _set_operation(attrs, StepType.workflow) + attrs["gen_ai.workflow.name"] = span.name + _set_orchestration_content(attrs, span) + + +def set_agent_attributes(attrs: MutableMapping[str, AttributeValue], span: AgentSpan) -> None: + """Map agent fields.""" + _set_operation(attrs, StepType.agent) + attrs["gen_ai.agent.name"] = span.name + attrs["splunk_ao.agent.type"] = span.agent_type.value + _set_orchestration_content(attrs, span) + + +def _set_generic_content(attrs: MutableMapping[str, AttributeValue], span: BaseStep) -> None: + if span.input is not None: + attrs["gen_ai.input.messages"] = _input_messages(span.input) + if span.output is not None: + attrs["gen_ai.output.messages"] = _output_messages(span.output) + + +def build_span_attributes(span: BaseStep, session_id: str | None = None) -> dict[str, AttributeValue]: + """Build preliminary attributes for one proprietary Galileo span.""" + attrs: dict[str, AttributeValue] = {} + set_common_attributes(attrs, span, session_id) + set_dataset_attributes(attrs, span) + + if isinstance(span, LlmSpan): + set_llm_attributes(attrs, span) + elif isinstance(span, ToolSpan): + set_tool_attributes(attrs, span) + elif isinstance(span, RetrieverSpan): + set_retriever_attributes(attrs, span) + elif isinstance(span, AgentSpan): + set_agent_attributes(attrs, span) + elif isinstance(span, WorkflowSpan): + set_workflow_attributes(attrs, span) + elif getattr(span.type, "value", span.type) == "control": + attrs["splunk_ao.operation.name"] = "control" + _set_generic_content(attrs, span) + elif span.type == StepType.trace: + _set_generic_content(attrs, span) + else: + raise TypeError(f"Unsupported span type: {type(span).__name__}") + + return attrs + + +def _alias_value(source_key: str, value: AttributeValue) -> AttributeValue: + if source_key == "gen_ai.response.time_to_first_chunk": + try: + return int(float(value) * 1_000_000_000) + except (TypeError, ValueError): + return value + return value + + +def normalize_attributes_for_export( + attrs: Mapping[str, AttributeValue], *, enabled: bool = True +) -> dict[str, AttributeValue]: + """Return final wire attributes without mutating the source mapping.""" + result = dict(attrs) + if not enabled: + return result + + for source_key, destination_key in SPLUNK_ALIAS_BY_GEN_AI.items(): + if source_key in attrs: + if destination_key == "splunk_ao.llm.time_to_first_token_ns" and destination_key in attrs: + continue + result[destination_key] = _alias_value(source_key, attrs[source_key]) + + for source_key in CONTENT_ALIAS_BY_GEN_AI: + result.pop(source_key, None) + + result[SPLUNK_AO_SYSTEM] = SPLUNK_AO_SYSTEM_VALUE + return result diff --git a/src/splunk_ao/exporter/config.py b/src/splunk_ao/exporter/config.py index 176165da..2c282bc7 100644 --- a/src/splunk_ao/exporter/config.py +++ b/src/splunk_ao/exporter/config.py @@ -2,8 +2,13 @@ from collections.abc import Callable from dataclasses import dataclass +from typing import Any from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter +from opentelemetry.sdk.resources import Resource +from opentelemetry.sdk.trace.export import SpanExporter + +from splunk_ao.exporter.span_transform import NormalizingSpanExporter @dataclass @@ -25,7 +30,7 @@ class RoutingAttrs: experiment_id: str | None = None -ExporterFactory = Callable[..., OTLPSpanExporter] +ExporterFactory = Callable[..., SpanExporter] def resolve_exporter_config(endpoint: str, auth_header: tuple[str, str], routing: RoutingAttrs) -> ExporterConfig: @@ -69,7 +74,9 @@ def build_exporter( auth_header: tuple[str, str], routing: RoutingAttrs, _exporter_factory: ExporterFactory = OTLPSpanExporter, -) -> OTLPSpanExporter: + **exporter_kwargs: Any, +) -> SpanExporter: """Build an OTLP HTTP exporter from shared resolved configuration.""" config = resolve_exporter_config(endpoint, auth_header, routing) - return _exporter_factory(endpoint=config.endpoint, headers=config.headers) + delegate = _exporter_factory(endpoint=config.endpoint, headers=config.headers, **exporter_kwargs) + return NormalizingSpanExporter(delegate, Resource(routing_resource_attributes(routing))) diff --git a/src/splunk_ao/exporter/o11y.py b/src/splunk_ao/exporter/o11y.py index d6584770..69b00ec5 100644 --- a/src/splunk_ao/exporter/o11y.py +++ b/src/splunk_ao/exporter/o11y.py @@ -1,6 +1,9 @@ """Splunk Observability Cloud OTLP exporter construction.""" +from typing import Any + from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter +from opentelemetry.sdk.trace.export import SpanExporter from splunk_ao.deployment import O11yConfig from splunk_ao.exporter.config import ( @@ -22,7 +25,12 @@ def resolve_o11y_exporter_config(config: O11yConfig, routing: RoutingAttrs) -> E def build_o11y_exporter( - config: O11yConfig, routing: RoutingAttrs, _exporter_factory: ExporterFactory = OTLPSpanExporter -) -> OTLPSpanExporter: + config: O11yConfig, + routing: RoutingAttrs, + _exporter_factory: ExporterFactory = OTLPSpanExporter, + **exporter_kwargs: Any, +) -> SpanExporter: """Build an OTLP exporter authenticated for Splunk Observability Cloud.""" - return build_exporter(config.otlp_endpoint, _o11y_auth_header(config), routing, _exporter_factory) + return build_exporter( + config.otlp_endpoint, _o11y_auth_header(config), routing, _exporter_factory, **exporter_kwargs + ) diff --git a/src/splunk_ao/exporter/span_transform.py b/src/splunk_ao/exporter/span_transform.py new file mode 100644 index 00000000..91ea14d0 --- /dev/null +++ b/src/splunk_ao/exporter/span_transform.py @@ -0,0 +1,99 @@ +"""Immutable OTLP span normalization before serialization.""" + +from __future__ import annotations + +import os +from collections.abc import Sequence + +from opentelemetry.sdk.resources import Resource +from opentelemetry.sdk.trace import ReadableSpan +from opentelemetry.sdk.trace.export import SpanExporter, SpanExportResult + +from splunk_ao.converter.attribute_mapping import normalize_attributes_for_export + +ROUTING_ATTRIBUTE_KEYS = frozenset( + { + "splunk_ao.project.name", + "splunk_ao.project.id", + "splunk_ao.logstream.name", + "splunk_ao.logstream.id", + "splunk_ao.experiment.id", + } +) + +_NORMALIZATION_ENV = "SPLUNK_AO_DEV_ENABLE_ATTRIBUTE_NORMALIZATION" +_FALSE_VALUES = frozenset({"0", "false", "no", "off"}) + + +def _normalization_enabled() -> bool: + value = os.environ.get(_NORMALIZATION_ENV) + return value is None or value.strip().lower() not in _FALSE_VALUES + + +def copy_span_for_export( + span: ReadableSpan, routing_resource: Resource | None = None, *, normalize_attributes: bool = True +) -> ReadableSpan: + """Return an immutable span copy with final attributes and routing.""" + source_attributes = { + key: value for key, value in (span.attributes or {}).items() if key not in ROUTING_ATTRIBUTE_KEYS + } + attributes = normalize_attributes_for_export(source_attributes, enabled=normalize_attributes) + + source_resource = span.resource or Resource({}) + base_resource = Resource( + {key: value for key, value in source_resource.attributes.items() if key not in ROUTING_ATTRIBUTE_KEYS}, + schema_url=source_resource.schema_url, + ) + resource = base_resource.merge(routing_resource) if routing_resource is not None else base_resource + + return ReadableSpan( + name=span.name, + context=span.context, + parent=span.parent, + resource=resource, + attributes=attributes, + events=span.events, + links=span.links, + kind=span.kind, + status=span.status, + start_time=span.start_time, + end_time=span.end_time, + instrumentation_scope=span.instrumentation_scope, + ) + + +class NormalizingSpanExporter(SpanExporter): + """Apply canonical attributes and routing once before delegating export.""" + + def __init__( + self, + delegate: SpanExporter, + routing_resource: Resource | None = None, + *, + normalize_attributes: bool | None = None, + ) -> None: + self._delegate = delegate + self._routing_resource = routing_resource + self._normalize_attributes = _normalization_enabled() if normalize_attributes is None else normalize_attributes + + @property + def delegate(self) -> SpanExporter: + """Return the wrapped exporter for SDK composition and diagnostics.""" + return self._delegate + + def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult: + """Export normalized immutable copies of the supplied spans.""" + return self._delegate.export( + tuple( + copy_span_for_export(span, self._routing_resource, normalize_attributes=self._normalize_attributes) + for span in spans + ) + ) + + def force_flush(self, timeout_millis: int = 30000) -> bool: + """Delegate flushing exactly once.""" + return self._delegate.force_flush(timeout_millis) + + def shutdown(self) -> None: + """Delegate shutdown exactly once.""" + self._delegate.shutdown() diff --git a/src/splunk_ao/exporter/standalone.py b/src/splunk_ao/exporter/standalone.py index a1034aed..0aab409c 100644 --- a/src/splunk_ao/exporter/standalone.py +++ b/src/splunk_ao/exporter/standalone.py @@ -1,6 +1,9 @@ """Standalone Splunk AO OTLP exporter construction.""" +from typing import Any + from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter +from opentelemetry.sdk.trace.export import SpanExporter from splunk_ao.deployment import StandaloneConfig from splunk_ao.exporter.config import ( @@ -22,7 +25,12 @@ def resolve_standalone_exporter_config(config: StandaloneConfig, routing: Routin def build_standalone_exporter( - config: StandaloneConfig, routing: RoutingAttrs, _exporter_factory: ExporterFactory = OTLPSpanExporter -) -> OTLPSpanExporter: + config: StandaloneConfig, + routing: RoutingAttrs, + _exporter_factory: ExporterFactory = OTLPSpanExporter, + **exporter_kwargs: Any, +) -> SpanExporter: """Build an OTLP exporter authenticated for standalone Splunk AO.""" - return build_exporter(config.otlp_endpoint, _standalone_auth_header(config), routing, _exporter_factory) + return build_exporter( + config.otlp_endpoint, _standalone_auth_header(config), routing, _exporter_factory, **exporter_kwargs + ) diff --git a/src/splunk_ao/otel.py b/src/splunk_ao/otel.py index 780066c5..91c6c108 100644 --- a/src/splunk_ao/otel.py +++ b/src/splunk_ao/otel.py @@ -9,14 +9,14 @@ from opentelemetry import context, trace from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter -from opentelemetry.sdk.resources import Resource from opentelemetry.sdk.trace import ReadableSpan, Span, SpanProcessor from opentelemetry.sdk.trace.export import BatchSpanProcessor, SpanExporter, SpanExportResult from opentelemetry.trace import Tracer -from galileo_core.schemas.logging.span import AgentSpan, RetrieverSpan, ToolSpan, WorkflowSpan +from galileo_core.schemas.logging.span import AgentSpan, WorkflowSpan from galileo_core.schemas.logging.span import Span as GalileoSpan from splunk_ao.config import SplunkAOConfig +from splunk_ao.converter import build_span_attributes from splunk_ao.decorator import ( _dataset_input_context, _dataset_metadata_context, @@ -27,12 +27,7 @@ _session_id_context, ) from splunk_ao.deployment import DeploymentMode, O11yConfig, StandaloneConfig -from splunk_ao.exporter import ( - RoutingAttrs, - resolve_o11y_exporter_config, - resolve_standalone_exporter_config, - routing_resource_attributes, -) +from splunk_ao.exporter import RoutingAttrs, build_o11y_exporter, build_standalone_exporter from splunk_ao.utils.env_helpers import ( _get_log_stream_from_env, _get_log_stream_id_from_env, @@ -41,7 +36,6 @@ _get_project_id_from_env, _get_project_or_default, ) -from splunk_ao.utils.retrievers import document_adapter logger = logging.getLogger(__name__) @@ -62,17 +56,6 @@ def get_tracer( _TRACE_PROVIDER_CONTEXT_VAR: ContextVar[TracerProvider | None] = ContextVar("galileo_trace_provider", default=None) - -ROUTING_ATTRIBUTE_KEYS = frozenset( - { - "splunk_ao.project.name", - "splunk_ao.project.id", - "splunk_ao.logstream.name", - "splunk_ao.logstream.id", - "splunk_ao.experiment.id", - } -) - _LEGACY_ROUTING_OPTIONS = {"logstream": "agentstream", "log_stream_id": "agent_stream_id"} @@ -139,30 +122,6 @@ def _resolve_routing( ) -def _with_routing_resource(span: ReadableSpan, routing_resource: Resource) -> ReadableSpan: - """Return an immutable copy with authoritative routing in its Resource.""" - attributes = {key: value for key, value in (span.attributes or {}).items() if key not in ROUTING_ATTRIBUTE_KEYS} - source_resource = span.resource or Resource({}) - base_resource = Resource( - {key: value for key, value in source_resource.attributes.items() if key not in ROUTING_ATTRIBUTE_KEYS}, - schema_url=source_resource.schema_url, - ) - return ReadableSpan( - name=span.name, - context=span.context, - parent=span.parent, - resource=base_resource.merge(routing_resource), - attributes=attributes, - events=span.events, - links=span.links, - kind=span.kind, - status=span.status, - start_time=span.start_time, - end_time=span.end_time, - instrumentation_scope=span.instrumentation_scope, - ) - - class SplunkAOOTLPExporter(SpanExporter): """ OpenTelemetry OTLP span exporter preconfigured for Splunk AO. @@ -211,21 +170,21 @@ def __init__( deployment = config.resolve_deployment() self._routing = _resolve_routing(deployment, project, project_id, agentstream, agent_stream_id, experiment_id) if deployment == DeploymentMode.O11Y: - exporter_config = resolve_o11y_exporter_config(O11yConfig.from_env(), self._routing) + self._delegate = build_o11y_exporter(O11yConfig.from_env(), self._routing, _exporter_factory, **kwargs) else: - exporter_config = resolve_standalone_exporter_config(StandaloneConfig.from_env(), self._routing) + self._delegate = build_standalone_exporter( + StandaloneConfig.from_env(), self._routing, _exporter_factory, **kwargs + ) self.project = self._routing.project_name self.project_id = self._routing.project_id self.agentstream = self._routing.log_stream_name self.agent_stream_id = self._routing.log_stream_id self.experiment_id = self._routing.experiment_id - self._routing_resource = Resource(routing_resource_attributes(self._routing)) - self._delegate = _exporter_factory(endpoint=exporter_config.endpoint, headers=exporter_config.headers, **kwargs) def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult: - """Export immutable copies with authoritative routing Resources.""" - return self._delegate.export(tuple(_with_routing_resource(span, self._routing_resource) for span in spans)) + """Export through the shared immutable normalization pipeline.""" + return self._delegate.export(spans) def force_flush(self, timeout_millis: int = 30000) -> bool: """Flush the delegate exporter.""" @@ -318,7 +277,7 @@ def on_start(self, span: Span, parent_context: context.Context | None = None) -> session_id = _session_id_context.get(None) if session_id: - span.set_attribute("splunk_ao.session.id", session_id) + span.set_attribute("gen_ai.conversation.id", session_id) # Set dataset attributes for ground truth/reference output support _apply_dataset_attributes( @@ -369,34 +328,6 @@ def add_splunk_ao_span_processor( return resolved_processor -def _set_retriever_span_attributes(span: trace.Span, galileo_span: RetrieverSpan) -> None: - span.set_attribute("db.operation", "search") - span.set_attribute("gen_ai.input.messages", json.dumps([{"role": "user", "content": galileo_span.input}])) - span.set_attribute( - "gen_ai.output.messages", - json.dumps( - [ - { - "role": "assistant", - "content": {"documents": document_adapter.dump_python(galileo_span.output, mode="json")}, - } - ] - ), - ) - - -def _set_tool_span_attributes(span: trace.Span, galileo_span: ToolSpan) -> None: - span.set_attribute("gen_ai.operation.name", "execute_tool") - span.set_attribute("gen_ai.tool.name", galileo_span.name) - span.set_attribute("gen_ai.tool.call.arguments", galileo_span.input) - span.set_attribute("gen_ai.input.messages", json.dumps([{"role": "tool", "content": galileo_span.input}])) - if galileo_span.output is not None: - span.set_attribute("gen_ai.tool.call.result", galileo_span.output) - span.set_attribute("gen_ai.output.messages", json.dumps([{"role": "tool", "content": galileo_span.output}])) - if galileo_span.tool_call_id is not None: - span.set_attribute("gen_ai.tool.call.id", galileo_span.tool_call_id) - - def _apply_dataset_attributes( span: trace.Span, dataset_input: str | None, dataset_output: str | None, dataset_metadata: dict[str, Any] | None ) -> None: @@ -409,48 +340,6 @@ def _apply_dataset_attributes( span.set_attribute("splunk_ao.dataset.metadata", json.dumps(dataset_metadata)) -def _set_workflow_span_attributes(span: trace.Span, galileo_span: WorkflowSpan) -> None: - """Set OpenTelemetry attributes for WorkflowSpan.""" - # Handle input - Union[str, Sequence[Message]] - if isinstance(galileo_span.input, str): - input_messages = [{"role": "user", "content": galileo_span.input}] - else: - # Sequence[Message] - serialize each message - input_messages = [] - for msg in list(galileo_span.input): - if hasattr(msg, "model_dump"): - input_messages.append(msg.model_dump(exclude_none=True)) - else: - input_messages.append(msg) - span.set_attribute("gen_ai.input.messages", json.dumps(input_messages)) - - # Handle output - Union[str, Message, Sequence[Document], None] - if galileo_span.output is None: - return - - output_value = galileo_span.output - # Type annotation to handle flexible content types (string or dict) - # Content can be: str (simple output), dict (documents), or dict (Message model_dump) - output_messages: list[dict[str, Any]] = [] - - if isinstance(output_value, str): - output_messages = [{"role": "assistant", "content": output_value}] - elif hasattr(output_value, "model_dump"): - # Single Message - output_messages = [output_value.model_dump(exclude_none=True)] - else: - # Sequence[Document] - wrap in assistant message - # Use document_adapter for consistency with _set_retriever_span_attributes - output_messages = [ - { - "role": "assistant", - "content": {"documents": document_adapter.dump_python(list(output_value), mode="json")}, - } - ] - - span.set_attribute("gen_ai.output.messages", json.dumps(output_messages)) - - @contextmanager def start_splunk_ao_span(galileo_span: GalileoSpan) -> Generator[trace.Span, Any, None]: tracer_provider = _TRACE_PROVIDER_CONTEXT_VAR.get() @@ -462,18 +351,14 @@ def start_splunk_ao_span(galileo_span: GalileoSpan) -> Generator[trace.Span, Any galileo_span, WorkflowSpan | AgentSpan ) with tracer.start_as_current_span(galileo_span.name) as span: - yield span - if is_conversation_root: - # OTel semantic-convention attributes are boolean; the native route's - # string-valued user_metadata bridge is an interim compatibility path. - span.set_attribute(GEN_AI_CONVERSATION_ROOT, value=True) - # Set dataset attributes for ground truth/reference output support - _apply_dataset_attributes( - span, galileo_span.dataset_input, galileo_span.dataset_output, galileo_span.dataset_metadata - ) - if isinstance(galileo_span, RetrieverSpan): - _set_retriever_span_attributes(span, galileo_span) - elif isinstance(galileo_span, ToolSpan): - _set_tool_span_attributes(span, galileo_span) - elif isinstance(galileo_span, WorkflowSpan): - _set_workflow_span_attributes(span, galileo_span) + try: + yield span + finally: + try: + attributes = build_span_attributes(galileo_span, _session_id_context.get(None)) + if is_conversation_root: + attributes[GEN_AI_CONVERSATION_ROOT] = True + for key, value in attributes.items(): + span.set_attribute(key, value) + except Exception: + logger.warning("Failed to finalize Splunk AO span attributes", exc_info=True) diff --git a/tests/test_attribute_mapping.py b/tests/test_attribute_mapping.py new file mode 100644 index 00000000..3e4cc859 --- /dev/null +++ b/tests/test_attribute_mapping.py @@ -0,0 +1,519 @@ +import json +from uuid import uuid4 + +import pytest + +from galileo_core.schemas.logging.llm import Message, MessageRole, ToolCall, ToolCallFunction +from galileo_core.schemas.logging.span import ( + AgentSpan, + AgentType, + LlmMetrics, + LlmSpan, + RetrieverSpan, + ToolSpan, + WorkflowSpan, +) +from galileo_core.schemas.shared.content_parts import FileContentPart, TextContentPart +from galileo_core.schemas.shared.document import Document +from splunk_ao.converter.attribute_mapping import ( + CONTENT_ALIAS_BY_GEN_AI, + SPLUNK_ALIAS_BY_GEN_AI, + build_span_attributes, + normalize_attributes_for_export, +) +from splunk_ao.logger.control import ControlResult, ControlSpan +from splunk_ao.schema import DataContentBlock, LoggedLlmSpan, LoggedMessage, TextContentBlock + + +def _text_message(role: str, content: str, *, finish_reason: str | None = None) -> dict: + message = {"role": role, "parts": [{"type": "text", "content": content}]} + if finish_reason is not None: + message["finish_reason"] = finish_reason + return message + + +def test_llm_mapping_covers_content_request_response_usage_and_units() -> None: + span = LlmSpan( + input="prompt", + output="answer", + model="gpt-5-nano", + temperature=0.0, + finish_reason="stop", + tools=[{"type": "function", "name": "search"}], + metrics=LlmMetrics( + num_input_tokens=12, + num_output_tokens=8, + num_total_tokens=20, + time_to_first_token_ns=250_000_000, + input_cost=0.01, + output_cost=0.02, + total_cost=0.03, + cache_read_input_tokens=4, + ), + ) + + attrs = build_span_attributes(span) + + assert attrs["gen_ai.operation.name"] == "chat" + assert json.loads(attrs["gen_ai.input.messages"]) == [_text_message("user", "prompt")] + assert json.loads(attrs["gen_ai.output.messages"]) == [_text_message("assistant", "answer", finish_reason="stop")] + assert attrs["gen_ai.request.model"] == "gpt-5-nano" + assert attrs["gen_ai.request.temperature"] == 0.0 + assert attrs["gen_ai.response.finish_reasons"] == ("stop",) + assert attrs["gen_ai.usage.input_tokens"] == 12 + assert attrs["gen_ai.usage.output_tokens"] == 8 + assert attrs["splunk_ao.llm.usage.total_tokens"] == 20 + assert attrs["gen_ai.response.time_to_first_chunk"] == 0.25 + assert attrs["splunk_ao.llm.time_to_first_token_ns"] == 250_000_000 + assert attrs["splunk_ao.llm.cost.total_usd"] == 0.03 + assert attrs["gen_ai.usage.cache_read.input_tokens"] == 4 + assert json.loads(attrs["gen_ai.tool.definitions"])[0]["name"] == "search" + + +def test_llm_output_uses_unknown_when_finish_reason_is_absent() -> None: + attrs = build_span_attributes(LlmSpan(input="prompt", output="answer")) + + assert json.loads(attrs["gen_ai.output.messages"]) == [ + _text_message("assistant", "answer", finish_reason="unknown") + ] + + +def test_llm_messages_preserve_tool_calls_and_tool_responses() -> None: + span = LlmSpan( + input=[ + Message( + role=MessageRole.assistant, + content="", + tool_calls=[ + ToolCall(id="call-1", function=ToolCallFunction(name="weather", arguments='{"city":"Paris"}')) + ], + ), + Message(role=MessageRole.tool, content='{"temperature":21}', tool_call_id="call-1"), + ], + output="It is 21 degrees.", + finish_reason="stop", + ) + + messages = json.loads(build_span_attributes(span)["gen_ai.input.messages"]) + + assert messages == [ + { + "role": "assistant", + "parts": [{"type": "tool_call", "id": "call-1", "name": "weather", "arguments": {"city": "Paris"}}], + }, + {"role": "tool", "parts": [{"type": "tool_call_response", "id": "call-1", "response": {"temperature": 21}}]}, + ] + + +def test_llm_content_parts_preserve_available_nonstandard_fields() -> None: + file_id = uuid4() + stored_file_span = LlmSpan( + input=[Message(role=MessageRole.user, content=[FileContentPart(file_id=file_id)])], output="answer" + ) + ingest_span = LoggedLlmSpan( + input=[ + LoggedMessage( + role=MessageRole.user, + content=[ + TextContentBlock(text="inspect this", index=2, metadata={"language": "en"}), + DataContentBlock( + modality="document", + mime_type="application/pdf", + base64="ZG9jdW1lbnQ=", + index=3, + metadata={"source": "upload"}, + ), + ], + ) + ], + output="answer", + ) + + stored_parts = json.loads(build_span_attributes(stored_file_span)["gen_ai.input.messages"])[0]["parts"] + ingest_parts = json.loads(build_span_attributes(ingest_span)["gen_ai.input.messages"])[0]["parts"] + + assert stored_parts == [{"type": "file", "file_id": str(file_id)}] + assert ingest_parts == [ + {"type": "text", "content": "inspect this", "index": 2, "metadata": {"language": "en"}}, + { + "type": "data", + "modality": "document", + "mime_type": "application/pdf", + "base64": "ZG9jdW1lbnQ=", + "index": 3, + "metadata": {"source": "upload"}, + }, + ] + assert "modality" not in stored_parts[0] + assert "content" not in stored_parts[0] + + +def test_llm_tool_definitions_flatten_openai_functions_and_preserve_flat_definitions() -> None: + span = LlmSpan( + input="prompt", + output="answer", + tools=[ + { + "type": "function", + "function": { + "name": "weather", + "description": "Get the weather", + "parameters": {"type": "object", "properties": {"city": {"type": "string"}}}, + }, + }, + {"type": "function", "name": "search", "custom": "preserved"}, + ], + ) + + definitions = json.loads(build_span_attributes(span)["gen_ai.tool.definitions"]) + + assert definitions == [ + { + "type": "function", + "name": "weather", + "description": "Get the weather", + "parameters": {"type": "object", "properties": {"city": {"type": "string"}}}, + }, + {"type": "function", "name": "search", "custom": "preserved"}, + ] + + +def test_common_mapping_uses_canonical_keys_and_omits_external_id() -> None: + session_id = uuid4() + span = ToolSpan( + name="search", + input='{"q":"x"}', + output="result", + status_code=500, + user_metadata={"team": "checkout"}, + tags=["production"], + external_id="not-an-otel-attribute", + dataset_input="question", + dataset_output="expected", + dataset_metadata={"split": "test"}, + session_id=session_id, + step_number=0, + ) + + attrs = build_span_attributes(span) + + assert attrs["gen_ai.operation.name"] == "execute_tool" + assert attrs["gen_ai.tool.name"] == "search" + assert json.loads(attrs["gen_ai.tool.call.arguments"]) == {"q": "x"} + assert json.loads(attrs["gen_ai.tool.call.result"]) == {"value": "result"} + assert attrs["gen_ai.conversation.id"] == str(session_id) + assert json.loads(attrs["splunk_ao.metadata"]) == {"team": "checkout"} + assert attrs["splunk_ao.tags"] == ("production",) + assert attrs["splunk_ao.status_code"] == 500 + assert attrs["error.type"] == "500" + assert attrs["splunk_ao.step_number"] == 0 + assert attrs["splunk_ao.dataset.input"] == "question" + assert attrs["splunk_ao.dataset.output"] == "expected" + assert json.loads(attrs["splunk_ao.dataset.metadata"]) == {"split": "test"} + assert not any("external_id" in key for key in attrs) + + +def test_explicit_session_context_precedes_span_session() -> None: + span = ToolSpan(name="search", session_id=uuid4()) + + assert build_span_attributes(span, session_id="context-session")["gen_ai.conversation.id"] == "context-session" + + +def test_retriever_mapping_uses_query_and_documents() -> None: + span = RetrieverSpan( + name="vector-search", input="what is RAG?", output=[Document(content="doc", metadata={"source": "kb"})] + ) + + attrs = build_span_attributes(span) + + assert attrs["gen_ai.operation.name"] == "retrieval" + assert attrs["gen_ai.retrieval.query.text"] == "what is RAG?" + assert json.loads(attrs["gen_ai.retrieval.documents"]) == [{"content": "doc", "metadata": {"source": "kb"}}] + assert attrs["splunk_ao.retrieval.documents.count"] == 1 + assert attrs["db.operation"] == "search" + assert "gen_ai.output.messages" not in attrs + + +@pytest.mark.parametrize( + ("span", "operation_key", "operation", "name_key", "name"), + [ + ( + WorkflowSpan(name="workflow", input="question", output="answer"), + "gen_ai.operation.name", + "invoke_workflow", + "gen_ai.workflow.name", + "workflow", + ), + ( + AgentSpan(name="planner", input="question", output="answer", agent_type=AgentType.planner), + "gen_ai.operation.name", + "invoke_agent", + "gen_ai.agent.name", + "planner", + ), + ], +) +def test_orchestration_mapping(span, operation_key: str, operation: str, name_key: str, name: str) -> None: + attrs = build_span_attributes(span) + + assert attrs[operation_key] == operation + assert attrs[name_key] == name + assert json.loads(attrs["gen_ai.input.messages"]) == [_text_message("user", "question")] + assert json.loads(attrs["gen_ai.output.messages"]) == [ + _text_message("assistant", "answer", finish_reason="unknown") + ] + + +def test_orchestration_extracts_serialized_langgraph_messages_with_multimodal_parts() -> None: + file_id = uuid4() + span = WorkflowSpan( + name="travel-planner", + input=json.dumps( + { + "messages": [ + { + "role": "user", + "content": [ + {"type": "text", "text": "Plan this trip"}, + {"type": "image", "url": "https://example.com/map.png"}, + {"type": "file", "file_id": str(file_id)}, + ], + } + ], + "destination": "Paris", + } + ), + output=json.dumps( + { + "update": { + "messages": [ + { + "role": "assistant", + "content": [ + {"type": "text", "text": "Here is the plan"}, + {"type": "data", "modality": "audio", "base64": "YXVkaW8="}, + ], + } + ] + } + } + ), + ) + + attrs = build_span_attributes(span) + + assert json.loads(attrs["gen_ai.input.messages"]) == [ + { + "role": "user", + "parts": [ + {"type": "text", "content": "Plan this trip"}, + {"type": "image", "url": "https://example.com/map.png"}, + {"type": "file", "file_id": str(file_id)}, + ], + } + ] + assert json.loads(attrs["gen_ai.output.messages"]) == [ + { + "role": "assistant", + "parts": [ + {"type": "text", "content": "Here is the plan"}, + {"type": "data", "modality": "audio", "base64": "YXVkaW8="}, + ], + "finish_reason": "unknown", + } + ] + + +def test_orchestration_accepts_native_content_part_sequences() -> None: + file_id = uuid4() + span = AgentSpan( + name="multimodal-agent", + agent_type=AgentType.planner, + input=[TextContentPart(text="Inspect this"), FileContentPart(file_id=file_id)], + output=[FileContentPart(file_id=file_id)], + ) + + attrs = build_span_attributes(span) + + assert json.loads(attrs["gen_ai.input.messages"]) == [ + { + "role": "user", + "parts": [{"type": "text", "content": "Inspect this"}, {"type": "file", "file_id": str(file_id)}], + } + ] + assert json.loads(attrs["gen_ai.output.messages"]) == [ + {"role": "assistant", "parts": [{"type": "file", "file_id": str(file_id)}], "finish_reason": "unknown"} + ] + + +def test_orchestration_output_omits_repeated_input_history() -> None: + user_message = {"role": "user", "content": "Plan a trip"} + assistant_message = {"role": "assistant", "content": "Where would you like to go?"} + span = AgentSpan( + name="planner", + agent_type=AgentType.planner, + input=json.dumps({"messages": [user_message]}), + output=json.dumps({"messages": [user_message, assistant_message]}), + ) + + attrs = build_span_attributes(span) + + assert json.loads(attrs["gen_ai.input.messages"]) == [_text_message("user", "Plan a trip")] + assert json.loads(attrs["gen_ai.output.messages"]) == [ + _text_message("assistant", "Where would you like to go?", finish_reason="unknown") + ] + + +def test_orchestration_preserves_schema_valid_parts_and_tool_calls() -> None: + span = WorkflowSpan( + name="tool-workflow", + input=json.dumps( + { + "messages": [ + { + "role": "assistant", + "parts": [{"type": "text", "content": "Checking weather"}], + "tool_calls": [ + {"id": "call-1", "function": {"name": "weather", "arguments": '{"city":"Paris"}'}} + ], + }, + {"role": "tool", "content": '{"temperature":21}', "tool_call_id": "call-1"}, + ] + } + ), + ) + + messages = json.loads(build_span_attributes(span)["gen_ai.input.messages"]) + + assert messages == [ + { + "role": "assistant", + "parts": [ + {"type": "text", "content": "Checking weather"}, + {"type": "tool_call", "id": "call-1", "name": "weather", "arguments": {"city": "Paris"}}, + ], + }, + {"role": "tool", "parts": [{"type": "tool_call_response", "id": "call-1", "response": {"temperature": 21}}]}, + ] + + +def test_orchestration_does_not_label_arbitrary_state_as_messages() -> None: + span = WorkflowSpan( + name="state-machine", + input=json.dumps({"current_agent": "coordinator", "travellers": 2}), + output=json.dumps({"next_agent": "flight_specialist"}), + ) + + attrs = build_span_attributes(span) + + assert "gen_ai.input.messages" not in attrs + assert "gen_ai.output.messages" not in attrs + + +def test_orchestration_keeps_non_json_strings_as_text_messages() -> None: + span = WorkflowSpan(name="workflow", input="{not-json", output="plain response") + + attrs = build_span_attributes(span) + + assert json.loads(attrs["gen_ai.input.messages"]) == [_text_message("user", "{not-json")] + assert json.loads(attrs["gen_ai.output.messages"]) == [ + _text_message("assistant", "plain response", finish_reason="unknown") + ] + + +def test_control_mapping_preserves_structured_content() -> None: + span = ControlSpan( + name="guardrail", input="question", output=ControlResult(action="observe", matched=True, confidence=0.9) + ) + + attrs = build_span_attributes(span) + output = json.loads(attrs["gen_ai.output.messages"]) + + assert attrs["splunk_ao.operation.name"] == "control" + assert json.loads(attrs["gen_ai.input.messages"]) == [_text_message("user", "question")] + assert output[0]["role"] == "assistant" + assert output[0]["finish_reason"] == "unknown" + assert json.loads(output[0]["parts"][0]["content"])["matched"] is True + + +def test_normalizer_duplicates_ordinary_attributes_and_relocates_all_content() -> None: + source = { + "gen_ai.request.model": "gpt-4o", + "gen_ai.usage.input_tokens": 42, + **{key: f"value-{index}" for index, key in enumerate(CONTENT_ALIAS_BY_GEN_AI)}, + } + + result = normalize_attributes_for_export(source) + + assert result["gen_ai.request.model"] == result["splunk_ao.request.model"] == "gpt-4o" + assert result["gen_ai.usage.input_tokens"] == result["splunk_ao.llm.usage.input_tokens"] == 42 + for source_key, destination_key in CONTENT_ALIAS_BY_GEN_AI.items(): + assert source_key not in result + assert result[destination_key] == source[source_key] + + +def test_normalizer_gen_ai_wins_collisions_and_sdk_marker_is_authoritative() -> None: + result = normalize_attributes_for_export( + { + "gen_ai.request.model": "canonical", + "splunk_ao.request.model": "stale", + "gen_ai.system": "legacy-provider", + "gen_ai.provider.name": "current-provider", + "splunk_ao.provider.name": "stale-provider", + "splunk_ao.system": "other-sdk", + } + ) + + assert result["splunk_ao.request.model"] == "canonical" + assert result["splunk_ao.provider.name"] == "current-provider" + assert result["splunk_ao.system"] == "splunk_ao_python" + + +def test_normalizer_does_not_use_deprecated_gen_ai_system_as_provider() -> None: + result = normalize_attributes_for_export({"gen_ai.system": "legacy-provider"}) + + assert result["gen_ai.system"] == "legacy-provider" + assert "splunk_ao.provider.name" not in result + + +def test_normalizer_converts_first_chunk_seconds_to_splunk_nanoseconds() -> None: + result = normalize_attributes_for_export({"gen_ai.response.time_to_first_chunk": 0.125}) + + assert result["gen_ai.response.time_to_first_chunk"] == 0.125 + assert result["splunk_ao.llm.time_to_first_token_ns"] == 125_000_000 + + +def test_normalizer_preserves_exact_source_nanoseconds() -> None: + result = normalize_attributes_for_export( + {"gen_ai.response.time_to_first_chunk": 0.123456789, "splunk_ao.llm.time_to_first_token_ns": 123_456_789} + ) + + assert result["splunk_ao.llm.time_to_first_token_ns"] == 123_456_789 + + +def test_normalizer_is_idempotent() -> None: + source = { + "gen_ai.operation.name": "chat", + "gen_ai.input.messages": json.dumps([_text_message("user", "question")]), + "custom.attribute": "unchanged", + } + + once = normalize_attributes_for_export(source) + + assert normalize_attributes_for_export(once) == once + + +def test_normalizer_can_be_disabled_for_developer_comparison() -> None: + source = { + "gen_ai.request.model": "gpt-4o", + "gen_ai.input.messages": json.dumps([_text_message("user", "question")]), + "splunk_ao.system": "source-value", + } + + assert normalize_attributes_for_export(source, enabled=False) == source + + +def test_every_alias_uses_an_explicit_destination_namespace() -> None: + assert SPLUNK_ALIAS_BY_GEN_AI + assert all(source.startswith("gen_ai.") for source in SPLUNK_ALIAS_BY_GEN_AI) + assert all(destination.startswith("splunk_ao.") for destination in SPLUNK_ALIAS_BY_GEN_AI.values()) diff --git a/tests/test_exporter_config.py b/tests/test_exporter_config.py index 12049145..7b8dc705 100644 --- a/tests/test_exporter_config.py +++ b/tests/test_exporter_config.py @@ -5,6 +5,7 @@ from splunk_ao.deployment import StandaloneConfig from splunk_ao.exporter.config import RoutingAttrs, routing_resource_attributes +from splunk_ao.exporter.span_transform import NormalizingSpanExporter from splunk_ao.exporter.standalone import build_standalone_exporter, resolve_standalone_exporter_config from splunk_ao.logger import logger as logger_module from splunk_ao.logger.logger import SplunkAOLogger @@ -132,7 +133,8 @@ def exporter_factory(**kwargs: Any) -> object: make_standalone_cfg(), make_routing(project_name="p"), _exporter_factory=exporter_factory ) - assert exporter is expected_exporter + assert isinstance(exporter, NormalizingSpanExporter) + assert exporter.delegate is expected_exporter assert captured == { "endpoint": "https://api.demo.galileocloud.io/otel/v1/traces", "headers": {"Splunk-AO-API-Key": "key", "project": "p"}, diff --git a/tests/test_exporter_o11y.py b/tests/test_exporter_o11y.py index f34c6802..47672ee0 100644 --- a/tests/test_exporter_o11y.py +++ b/tests/test_exporter_o11y.py @@ -7,6 +7,7 @@ from splunk_ao.deployment import O11yConfig from splunk_ao.exporter.config import RoutingAttrs from splunk_ao.exporter.o11y import build_o11y_exporter, resolve_o11y_exporter_config +from splunk_ao.exporter.span_transform import NormalizingSpanExporter from splunk_ao.shared.exceptions import MissingConfigurationError @@ -98,7 +99,8 @@ def exporter_factory(**kwargs: Any) -> object: _exporter_factory=exporter_factory, ) - assert exporter is expected_exporter + assert isinstance(exporter, NormalizingSpanExporter) + assert exporter.delegate is expected_exporter assert captured == { "endpoint": "https://ingest.us1.observability.splunkcloud.com/v2/trace/otlp", "headers": {"X-SF-Token": "tok", "projectid": "pid", "logstreamid": "lsid"}, diff --git a/tests/test_otel.py b/tests/test_otel.py index 25c97214..dd17adad 100644 --- a/tests/test_otel.py +++ b/tests/test_otel.py @@ -6,6 +6,7 @@ from galileo_core.schemas.logging.llm import Message, MessageRole from galileo_core.schemas.logging.span import AgentSpan, LlmSpan, RetrieverSpan, ToolSpan, WorkflowSpan from galileo_core.schemas.shared.document import Document +from splunk_ao.converter import build_span_attributes from splunk_ao.decorator import ( _dataset_input_context, _dataset_metadata_context, @@ -21,8 +22,6 @@ _TRACE_PROVIDER_CONTEXT_VAR, SplunkAOOTLPExporter, SplunkAOSpanProcessor, - _set_tool_span_attributes, - _set_workflow_span_attributes, start_splunk_ao_span, ) @@ -260,13 +259,13 @@ def test_processor_on_start_sets_content_not_routing_attributes(self, mock_proce assert mock_span.set_attribute.call_count == 1 actual_calls = {(args[0], args[1]) for args, _ in mock_span.set_attribute.call_args_list} - assert ("splunk_ao.session.id", "test-session") in actual_calls + assert ("gen_ai.conversation.id", "test-session") in actual_calls routing_keys = {"splunk_ao.project.name", "splunk_ao.logstream.name", "splunk_ao.experiment.id"} assert not routing_keys.intersection(key for key, _ in actual_calls) class TestSetToolSpanAttributes: - """Test suite for _set_tool_span_attributes function.""" + """Test the canonical tool-span builder.""" def test_tool_span_with_all_fields(self): """Test setting attributes when all ToolSpan fields are populated.""" @@ -278,41 +277,27 @@ def test_tool_span_with_all_fields(self): tool_call_id="call-123", status_code=200, ) - mock_otel_span = Mock() - - # When: setting tool span attributes - _set_tool_span_attributes(mock_otel_span, tool_span) + attrs = build_span_attributes(tool_span) - # Then: all attributes are set correctly - calls = {args[0]: args[1] for args, _ in mock_otel_span.set_attribute.call_args_list} - assert calls["gen_ai.operation.name"] == "execute_tool" - assert calls["gen_ai.tool.name"] == "test-tool" - assert calls["gen_ai.tool.call.arguments"] == "tool input data" - assert calls["gen_ai.tool.call.result"] == "tool output result" - assert calls["gen_ai.input.messages"] == json.dumps([{"role": "tool", "content": "tool input data"}]) - assert calls["gen_ai.output.messages"] == json.dumps([{"role": "tool", "content": "tool output result"}]) - assert calls["gen_ai.tool.call.id"] == "call-123" - assert mock_otel_span.set_attribute.call_count == 7 + assert attrs["gen_ai.operation.name"] == "execute_tool" + assert attrs["gen_ai.tool.name"] == "test-tool" + assert json.loads(attrs["gen_ai.tool.call.arguments"]) == {"value": "tool input data"} + assert json.loads(attrs["gen_ai.tool.call.result"]) == {"value": "tool output result"} + assert attrs["gen_ai.tool.call.id"] == "call-123" + assert "gen_ai.input.messages" not in attrs + assert "gen_ai.output.messages" not in attrs def test_tool_span_with_only_input(self): """Test setting attributes when only input is provided.""" # Given: a ToolSpan with only input (output and tool_call_id are None) tool_span = ToolSpan(name="test-tool", input="tool input only", output=None, tool_call_id=None, status_code=200) - mock_otel_span = Mock() + attrs = build_span_attributes(tool_span) - # When: setting tool span attributes - _set_tool_span_attributes(mock_otel_span, tool_span) - - # Then: operation name, tool name, and input attributes are set, but not output or tool_call_id - calls = {args[0]: args[1] for args, _ in mock_otel_span.set_attribute.call_args_list} - assert calls["gen_ai.operation.name"] == "execute_tool" - assert calls["gen_ai.tool.name"] == "test-tool" - assert calls["gen_ai.tool.call.arguments"] == "tool input only" - assert calls["gen_ai.input.messages"] == json.dumps([{"role": "tool", "content": "tool input only"}]) - assert "gen_ai.tool.call.result" not in calls - assert "gen_ai.output.messages" not in calls - assert "gen_ai.tool.call.id" not in calls - assert mock_otel_span.set_attribute.call_count == 4 + assert attrs["gen_ai.operation.name"] == "execute_tool" + assert attrs["gen_ai.tool.name"] == "test-tool" + assert json.loads(attrs["gen_ai.tool.call.arguments"]) == {"value": "tool input only"} + assert "gen_ai.tool.call.result" not in attrs + assert "gen_ai.tool.call.id" not in attrs def test_tool_span_with_output_no_tool_call_id(self): """Test setting attributes when output is provided but tool_call_id is None.""" @@ -320,21 +305,13 @@ def test_tool_span_with_output_no_tool_call_id(self): tool_span = ToolSpan( name="test-tool", input="tool input", output="tool output", tool_call_id=None, status_code=200 ) - mock_otel_span = Mock() + attrs = build_span_attributes(tool_span) - # When: setting tool span attributes - _set_tool_span_attributes(mock_otel_span, tool_span) - - # Then: operation name, tool name, input, and output attributes are set, but not tool_call_id - calls = {args[0]: args[1] for args, _ in mock_otel_span.set_attribute.call_args_list} - assert calls["gen_ai.operation.name"] == "execute_tool" - assert calls["gen_ai.tool.name"] == "test-tool" - assert calls["gen_ai.tool.call.arguments"] == "tool input" - assert calls["gen_ai.tool.call.result"] == "tool output" - assert calls["gen_ai.input.messages"] == json.dumps([{"role": "tool", "content": "tool input"}]) - assert calls["gen_ai.output.messages"] == json.dumps([{"role": "tool", "content": "tool output"}]) - assert "gen_ai.tool.call.id" not in calls - assert mock_otel_span.set_attribute.call_count == 6 + assert attrs["gen_ai.operation.name"] == "execute_tool" + assert attrs["gen_ai.tool.name"] == "test-tool" + assert json.loads(attrs["gen_ai.tool.call.arguments"]) == {"value": "tool input"} + assert json.loads(attrs["gen_ai.tool.call.result"]) == {"value": "tool output"} + assert "gen_ai.tool.call.id" not in attrs class TestStartGalileoSpan: @@ -374,10 +351,8 @@ def test_start_splunk_ao_span_dispatches_tool_span(self): assert "gen_ai.system" not in calls assert calls["gen_ai.operation.name"] == "execute_tool" assert calls["gen_ai.tool.name"] == "my-tool" - assert calls["gen_ai.tool.call.arguments"] == "tool input data" - assert calls["gen_ai.tool.call.result"] == "tool output result" - assert calls["gen_ai.input.messages"] == json.dumps([{"role": "tool", "content": "tool input data"}]) - assert calls["gen_ai.output.messages"] == json.dumps([{"role": "tool", "content": "tool output result"}]) + assert json.loads(calls["gen_ai.tool.call.arguments"]) == {"value": "tool input data"} + assert json.loads(calls["gen_ai.tool.call.result"]) == {"value": "tool output result"} assert calls["gen_ai.tool.call.id"] == "call-789" def test_start_splunk_ao_span_tool_span_with_none_output(self): @@ -401,12 +376,107 @@ def test_start_splunk_ao_span_tool_span_with_none_output(self): assert "gen_ai.system" not in calls assert calls["gen_ai.operation.name"] == "execute_tool" assert calls["gen_ai.tool.name"] == "minimal-tool" - assert calls["gen_ai.tool.call.arguments"] == "just input" - assert calls["gen_ai.input.messages"] == json.dumps([{"role": "tool", "content": "just input"}]) + assert json.loads(calls["gen_ai.tool.call.arguments"]) == {"value": "just input"} assert "gen_ai.tool.call.result" not in calls - assert "gen_ai.output.messages" not in calls assert "gen_ai.tool.call.id" not in calls + @pytest.mark.parametrize( + "galileo_span", + [ + LlmSpan(input="prompt", output="answer", model="gpt-4o"), + ToolSpan(name="search", input="query", output="result"), + RetrieverSpan(name="retrieval", input="query", output=[Document(content="result", metadata={})]), + WorkflowSpan(name="workflow", input="question", output="answer"), + AgentSpan(name="agent", input="question", output="answer"), + ], + ) + def test_start_span_applies_canonical_builder_for_every_supported_type(self, galileo_span): + expected = build_span_attributes(galileo_span, session_id="session-id") + mock_otel_span = Mock() + mock_tracer = Mock() + mock_tracer.start_as_current_span.return_value.__enter__ = Mock(return_value=mock_otel_span) + mock_tracer.start_as_current_span.return_value.__exit__ = Mock(return_value=False) + mock_provider = Mock() + mock_provider.get_tracer.return_value = mock_tracer + _TRACE_PROVIDER_CONTEXT_VAR.set(mock_provider) + token = _session_id_context.set("session-id") + + try: + with start_splunk_ao_span(galileo_span): + pass + finally: + _session_id_context.reset(token) + + calls = {args[0]: args[1] for args, _ in mock_otel_span.set_attribute.call_args_list} + assert {key: calls[key] for key in expected} == expected + + def test_start_span_applies_canonical_attributes_when_body_raises(self): + galileo_span = ToolSpan(name="search", input="query", output="result") + mock_otel_span = Mock() + mock_tracer = Mock() + mock_tracer.start_as_current_span.return_value.__enter__ = Mock(return_value=mock_otel_span) + mock_tracer.start_as_current_span.return_value.__exit__ = Mock(return_value=False) + mock_provider = Mock() + mock_provider.get_tracer.return_value = mock_tracer + _TRACE_PROVIDER_CONTEXT_VAR.set(mock_provider) + + with pytest.raises(RuntimeError, match="failure"), start_splunk_ao_span(galileo_span): + raise RuntimeError("failure") + + calls = {args[0]: args[1] for args, _ in mock_otel_span.set_attribute.call_args_list} + assert calls["gen_ai.operation.name"] == "execute_tool" + assert calls["gen_ai.tool.name"] == "search" + + @patch("splunk_ao.otel.logger.warning") + @patch("splunk_ao.otel.build_span_attributes", side_effect=ValueError("invalid partial span")) + def test_start_span_finalization_does_not_mask_body_exception(self, _mock_build, mock_warning): + galileo_span = ToolSpan(name="search", input="query", output="result") + mock_tracer = Mock() + mock_tracer.start_as_current_span.return_value.__enter__ = Mock(return_value=Mock()) + mock_tracer.start_as_current_span.return_value.__exit__ = Mock(return_value=False) + mock_provider = Mock() + mock_provider.get_tracer.return_value = mock_tracer + _TRACE_PROVIDER_CONTEXT_VAR.set(mock_provider) + + with pytest.raises(RuntimeError, match="user failure"), start_splunk_ao_span(galileo_span): + raise RuntimeError("user failure") + + mock_warning.assert_called_once_with("Failed to finalize Splunk AO span attributes", exc_info=True) + + @patch("splunk_ao.otel.logger.warning") + @patch("splunk_ao.otel.build_span_attributes", side_effect=ValueError("invalid span")) + def test_start_span_finalization_failure_does_not_fail_successful_body(self, _mock_build, mock_warning): + galileo_span = ToolSpan(name="search", input="query", output="result") + mock_tracer = Mock() + mock_tracer.start_as_current_span.return_value.__enter__ = Mock(return_value=Mock()) + mock_tracer.start_as_current_span.return_value.__exit__ = Mock(return_value=False) + mock_provider = Mock() + mock_provider.get_tracer.return_value = mock_tracer + _TRACE_PROVIDER_CONTEXT_VAR.set(mock_provider) + + with start_splunk_ao_span(galileo_span): + pass + + mock_warning.assert_called_once_with("Failed to finalize Splunk AO span attributes", exc_info=True) + + @patch("splunk_ao.otel.logger.warning") + @patch("splunk_ao.otel.build_span_attributes", return_value={"gen_ai.operation.name": "execute_tool"}) + def test_start_span_contains_attribute_write_failure(self, _mock_build, mock_warning): + galileo_span = ToolSpan(name="search", input="query", output="result") + mock_otel_span = Mock() + mock_otel_span.set_attribute.side_effect = ValueError("attribute rejected") + mock_tracer = Mock() + mock_tracer.start_as_current_span.return_value.__enter__ = Mock(return_value=mock_otel_span) + mock_tracer.start_as_current_span.return_value.__exit__ = Mock(return_value=False) + mock_provider = Mock() + mock_provider.get_tracer.return_value = mock_tracer + _TRACE_PROVIDER_CONTEXT_VAR.set(mock_provider) + + with start_splunk_ao_span(galileo_span): + pass + + mock_warning.assert_called_once_with("Failed to finalize Splunk AO span attributes", exc_info=True) + @pytest.mark.parametrize( "galileo_span", [ @@ -489,56 +559,36 @@ def test_start_splunk_ao_span_does_not_mark_ineligible_root(self, galileo_span): class TestWorkflowSpanAttributes: """Test suite for WorkflowSpan OpenTelemetry attribute mapping.""" - @pytest.fixture - def mock_dependencies(self): - """Set up mocks for testing workflow span attributes.""" - with patch("splunk_ao.otel.trace") as mock_trace_module, patch("splunk_ao.otel.json") as mock_json_module: - mock_span = Mock() - mock_json_module.dumps.return_value = '"test"' - yield {"span": mock_span, "trace": mock_trace_module, "json": mock_json_module} - - def test_workflow_span_with_string_input_output(self, mock_dependencies): + def test_workflow_span_with_string_input_output(self): """Test WorkflowSpan with string input and output.""" # Given: a WorkflowSpan with string input and output workflow_span = WorkflowSpan(name="test-workflow", input="input text", output="output text", status_code=200) - mock_span = mock_dependencies["span"] - mock_json = mock_dependencies["json"] - - # When: setting workflow span attributes - _set_workflow_span_attributes(mock_span, workflow_span) - - # Then: input and output should be wrapped in message format - assert mock_span.set_attribute.call_count == 2 - - # Check first call (input) - input_call = mock_span.set_attribute.call_args_list[0] - assert input_call[0][0] == "gen_ai.input.messages" - mock_json.dumps.assert_any_call([{"role": "user", "content": "input text"}]) + attrs = build_span_attributes(workflow_span) - # Check second call (output) - output_call = mock_span.set_attribute.call_args_list[1] - assert output_call[0][0] == "gen_ai.output.messages" - mock_json.dumps.assert_any_call([{"role": "assistant", "content": "output text"}]) + assert attrs["gen_ai.operation.name"] == "invoke_workflow" + assert attrs["gen_ai.workflow.name"] == "test-workflow" + assert json.loads(attrs["gen_ai.input.messages"]) == [ + {"role": "user", "parts": [{"type": "text", "content": "input text"}]} + ] + assert json.loads(attrs["gen_ai.output.messages"]) == [ + {"role": "assistant", "parts": [{"type": "text", "content": "output text"}], "finish_reason": "unknown"} + ] - def test_workflow_span_with_message_input_output(self, mock_dependencies): + def test_workflow_span_with_message_input_output(self): """Test WorkflowSpan with Message input and output.""" # Given: a WorkflowSpan with Message input and output input_msg = Message(role=MessageRole.user, content="user question") workflow_span = WorkflowSpan(name="test-workflow", input=[input_msg], output=input_msg, status_code=200) - mock_span = mock_dependencies["span"] - mock_dependencies["json"] - - # When: setting workflow span attributes - _set_workflow_span_attributes(mock_span, workflow_span) + attrs = build_span_attributes(workflow_span) - # Then: input and output should serialize Message objects - assert mock_span.set_attribute.call_count == 2 - input_call = mock_span.set_attribute.call_args_list[0] - assert input_call[0][0] == "gen_ai.input.messages" - output_call = mock_span.set_attribute.call_args_list[1] - assert output_call[0][0] == "gen_ai.output.messages" + assert json.loads(attrs["gen_ai.input.messages"]) == [ + {"role": "user", "parts": [{"type": "text", "content": "user question"}]} + ] + assert json.loads(attrs["gen_ai.output.messages"]) == [ + {"role": "user", "parts": [{"type": "text", "content": "user question"}], "finish_reason": "unknown"} + ] - def test_workflow_span_with_document_sequence_output(self, mock_dependencies): + def test_workflow_span_with_document_sequence_output(self): """Test WorkflowSpan with Document sequence output.""" # Given: a WorkflowSpan with string input and Document sequence output documents = [ @@ -551,37 +601,28 @@ def test_workflow_span_with_document_sequence_output(self, mock_dependencies): workflow_span = WorkflowSpan.model_construct( name="test-workflow", input="query", output=documents, status_code=200 ) - mock_span = mock_dependencies["span"] - mock_dependencies["json"] + attrs = build_span_attributes(workflow_span) - # When: setting workflow span attributes - _set_workflow_span_attributes(mock_span, workflow_span) + output = json.loads(attrs["gen_ai.output.messages"]) + documents = json.loads(output[0]["parts"][0]["content"]) + assert [document["content"] for document in documents] == ["doc1 content", "doc2 content"] - # Then: output should be wrapped in assistant message with documents - assert mock_span.set_attribute.call_count == 2 - output_call = mock_span.set_attribute.call_args_list[1] - assert output_call[0][0] == "gen_ai.output.messages" - - def test_workflow_span_with_none_output(self, mock_dependencies): + def test_workflow_span_with_none_output(self): """Test WorkflowSpan with None output (should not set output attribute).""" # Given: a WorkflowSpan with None output workflow_span = WorkflowSpan(name="test-workflow", input="input text", output=None, status_code=200) - mock_span = mock_dependencies["span"] - - # When: setting workflow span attributes - _set_workflow_span_attributes(mock_span, workflow_span) + attrs = build_span_attributes(workflow_span) - # Then: only input attribute should be set, not output - assert mock_span.set_attribute.call_count == 1 - input_call = mock_span.set_attribute.call_args_list[0] - assert input_call[0][0] == "gen_ai.input.messages" + assert json.loads(attrs["gen_ai.input.messages"]) == [ + {"role": "user", "parts": [{"type": "text", "content": "input text"}]} + ] + assert "gen_ai.output.messages" not in attrs - def test_workflow_span_in_start_splunk_ao_span(self, mock_dependencies): + def test_workflow_span_in_start_splunk_ao_span(self): """Test that WorkflowSpan is handled in start_splunk_ao_span context manager.""" # Given: a WorkflowSpan workflow_span = WorkflowSpan(name="test-workflow", input="input", output="output", status_code=200) - mock_span = mock_dependencies["span"] - mock_dependencies["json"] + mock_span = Mock() # Setup the mock tracer mock_tracer = Mock() diff --git a/tests/test_otel_native_paths.py b/tests/test_otel_native_paths.py index 661f62ca..77be8f35 100644 --- a/tests/test_otel_native_paths.py +++ b/tests/test_otel_native_paths.py @@ -295,8 +295,11 @@ def test_exporter_preserves_every_unaffected_span_field() -> None: assert getattr(exported, field) == getattr(source, field) assert exported.resource.schema_url == source.resource.schema_url assert exported.attributes["gen_ai.request.model"] == "gpt-4o" + assert exported.attributes["splunk_ao.request.model"] == "gpt-4o" assert exported.attributes["gen_ai.provider.name"] == "openai" + assert exported.attributes["splunk_ao.provider.name"] == "openai" assert exported.attributes["gen_ai.system"] == "legacy-upstream-provider" + assert exported.attributes["splunk_ao.system"] == "splunk_ao_python" assert exported.attributes["custom.attribute"] == "preserved" exporter.shutdown() @@ -338,7 +341,7 @@ def test_processor_does_not_put_routing_on_span_attributes() -> None: calls = {args[0]: args[1] for args, _ in span.set_attribute.call_args_list} assert not ROUTING_KEYS.intersection(calls) - assert calls["splunk_ao.session.id"] == "session-id" + assert calls["gen_ai.conversation.id"] == "session-id" assert calls["splunk_ao.dataset.input"] == "question" processor.shutdown() diff --git a/tests/test_span_transform.py b/tests/test_span_transform.py new file mode 100644 index 00000000..bda12811 --- /dev/null +++ b/tests/test_span_transform.py @@ -0,0 +1,153 @@ +from collections.abc import Sequence + +import pytest +from opentelemetry.sdk.resources import Resource +from opentelemetry.sdk.trace import ReadableSpan +from opentelemetry.sdk.trace.export import SpanExporter, SpanExportResult +from opentelemetry.sdk.util.instrumentation import InstrumentationScope +from opentelemetry.trace import SpanContext, SpanKind, TraceFlags +from opentelemetry.trace.status import Status, StatusCode + +from splunk_ao.exporter.span_transform import NormalizingSpanExporter, copy_span_for_export + + +class RecordingExporter(SpanExporter): + def __init__(self) -> None: + self.batches: list[Sequence[ReadableSpan]] = [] + self.flushes: list[int] = [] + self.shutdowns = 0 + + def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult: + self.batches.append(spans) + return SpanExportResult.SUCCESS + + def force_flush(self, timeout_millis: int = 30000) -> bool: + self.flushes.append(timeout_millis) + return True + + def shutdown(self) -> None: + self.shutdowns += 1 + + +def make_span(attributes: dict[str, object] | None = None) -> ReadableSpan: + return ReadableSpan( + name="span", + context=SpanContext( + trace_id=0x1234567890ABCDEF1234567890ABCDEF, + span_id=0x1234567890ABCDEF, + is_remote=False, + trace_flags=TraceFlags.SAMPLED, + ), + resource=Resource( + {"service.name": "checkout", "splunk_ao.project.name": "stale"}, + schema_url="https://opentelemetry.io/schemas/1.38.0", + ), + attributes=attributes or {"gen_ai.request.model": "gpt-4o"}, + kind=SpanKind.CLIENT, + status=Status(StatusCode.OK), + start_time=1, + end_time=2, + instrumentation_scope=InstrumentationScope("instrumentation", "1.0"), + ) + + +def test_copy_span_for_export_is_immutable_and_combines_normalization_with_routing() -> None: + source = make_span( + { + "gen_ai.request.model": "gpt-4o", + "gen_ai.input.messages": "input-json", + "splunk_ao.project.name": "stale-span-routing", + } + ) + source_attributes = dict(source.attributes or {}) + source_resource = source.resource + + exported = copy_span_for_export( + source, Resource({"splunk_ao.project.id": "project-id", "splunk_ao.logstream.id": "log-stream-id"}) + ) + + assert exported is not source + assert dict(source.attributes or {}) == source_attributes + assert source.resource is source_resource + assert exported.attributes["splunk_ao.request.model"] == "gpt-4o" + assert exported.attributes["splunk_ao.input.messages"] == "input-json" + assert "gen_ai.input.messages" not in exported.attributes + assert "splunk_ao.project.name" not in exported.attributes + assert exported.resource.attributes["splunk_ao.project.id"] == "project-id" + assert exported.resource.attributes["splunk_ao.logstream.id"] == "log-stream-id" + assert "splunk_ao.project.name" not in exported.resource.attributes + assert exported.resource.attributes["service.name"] == "checkout" + + +def test_copy_span_preserves_unaffected_readable_span_fields() -> None: + source = make_span() + exported = copy_span_for_export(source) + + for field in ( + "name", + "context", + "parent", + "events", + "links", + "kind", + "status", + "start_time", + "end_time", + "instrumentation_info", + "instrumentation_scope", + ): + assert getattr(exported, field) == getattr(source, field) + assert exported.resource.schema_url == source.resource.schema_url + + +def test_exporter_normalizes_once_and_delegates_lifecycle_once() -> None: + delegate = RecordingExporter() + exporter = NormalizingSpanExporter(delegate, Resource({"splunk_ao.project.name": "project"})) + + assert exporter.export((make_span(),)) == SpanExportResult.SUCCESS + assert exporter.force_flush(1234) is True + exporter.shutdown() + + assert len(delegate.batches) == 1 + assert delegate.batches[0][0].attributes["splunk_ao.request.model"] == "gpt-4o" + assert delegate.batches[0][0].resource.attributes["splunk_ao.project.name"] == "project" + assert delegate.flushes == [1234] + assert delegate.shutdowns == 1 + + +@pytest.mark.parametrize("value", ["0", "false", "FALSE", "no", "off"]) +def test_private_environment_switch_disables_only_attribute_normalization( + monkeypatch: pytest.MonkeyPatch, value: str +) -> None: + monkeypatch.setenv("SPLUNK_AO_DEV_ENABLE_ATTRIBUTE_NORMALIZATION", value) + delegate = RecordingExporter() + exporter = NormalizingSpanExporter(delegate, Resource({"splunk_ao.project.name": "project"})) + source = make_span({"gen_ai.input.messages": "input-json"}) + + exporter.export((source,)) + exported = delegate.batches[0][0] + + assert exported.attributes == source.attributes + assert "splunk_ao.system" not in exported.attributes + assert exported.resource.attributes["splunk_ao.project.name"] == "project" + + +def test_private_environment_switch_is_read_at_exporter_construction(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("SPLUNK_AO_DEV_ENABLE_ATTRIBUTE_NORMALIZATION", "false") + delegate = RecordingExporter() + exporter = NormalizingSpanExporter(delegate) + monkeypatch.setenv("SPLUNK_AO_DEV_ENABLE_ATTRIBUTE_NORMALIZATION", "true") + + exporter.export((make_span(),)) + + assert "splunk_ao.request.model" not in delegate.batches[0][0].attributes + + +def test_default_normalization_is_enabled(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("SPLUNK_AO_DEV_ENABLE_ATTRIBUTE_NORMALIZATION", raising=False) + delegate = RecordingExporter() + exporter = NormalizingSpanExporter(delegate) + + exporter.export((make_span(),)) + + assert delegate.batches[0][0].attributes["splunk_ao.system"] == "splunk_ao_python"