HYBIM-887 feat(otel): normalize native and user-wired paths

README.md

@@ -82,6 +82,11 @@ export SPLUNK_AO_PROJECT="your-project-name"
 export SPLUNK_AO_LOG_STREAM="your-log-stream-name"
 ```
 
+When using `SplunkAOSpanProcessor`, routing is captured when its exporter is
+constructed and remains fixed for that exporter's lifetime. This matches the
+OpenTelemetry Resource model and keeps request headers consistent for batched
+spans. Use a separate processor and exporter for each additional destination.
+
 Set `SPLUNK_AO_LOGGING_DISABLED=true` to disable telemetry collection and
 export.
 

examples/agent/pydantic-ai-support-agent/src/main.py

@@ -2,20 +2,19 @@
 
 from dataclasses import dataclass
 from datetime import datetime
-from typing import Optional
 
-from splunk_ao.otel import SplunkAOSpanProcessor, add_splunk_ao_span_processor
 from opentelemetry import trace
 from opentelemetry.sdk.trace import TracerProvider
 from pydantic import BaseModel
 from pydantic_ai import Agent, RunContext
 
+from splunk_ao.otel import SplunkAOSpanProcessor, add_splunk_ao_span_processor
+
 # Set up Splunk AO observability
 provider = TracerProvider()
 trace.set_tracer_provider(provider)
 add_splunk_ao_span_processor(
-    tracer_provider=provider,
-    processor=SplunkAOSpanProcessor(project="pydantic-ai-support", logstream="default"),
+    tracer_provider=provider, processor=SplunkAOSpanProcessor(project="pydantic-ai-support", agentstream="default")
 )
 
 # Enable instrumentation on all PydanticAI agents
@@ -53,42 +52,15 @@ class SupportTicket:
 
 # --- Mock Database ---
 CUSTOMERS_DB: dict[str, Customer] = {
-    "C001": Customer(
-        "C001",
-        "Alice Johnson",
-        "alice@example.com",
-        "enterprise",
-        datetime(2022, 1, 15),
-    ),
+    "C001": Customer("C001", "Alice Johnson", "alice@example.com", "enterprise", datetime(2022, 1, 15)),
     "C002": Customer("C002", "Bob Smith", "bob@example.com", "premium", datetime(2023, 6, 20)),
     "C003": Customer("C003", "Carol White", "carol@example.com", "standard", datetime(2024, 3, 10)),
 }
 
 ORDERS_DB: dict[str, Order] = {
-    "ORD-1001": Order(
-        "ORD-1001",
-        "C001",
-        "Enterprise License",
-        5000.00,
-        "delivered",
-        datetime(2024, 11, 1),
-    ),
-    "ORD-1002": Order(
-        "ORD-1002",
-        "C001",
-        "Support Package",
-        1200.00,
-        "shipped",
-        datetime(2024, 12, 15),
-    ),
-    "ORD-1003": Order(
-        "ORD-1003",
-        "C002",
-        "Premium Subscription",
-        299.99,
-        "pending",
-        datetime(2025, 1, 5),
-    ),
+    "ORD-1001": Order("ORD-1001", "C001", "Enterprise License", 5000.00, "delivered", datetime(2024, 11, 1)),
+    "ORD-1002": Order("ORD-1002", "C001", "Support Package", 1200.00, "shipped", datetime(2024, 12, 15)),
+    "ORD-1003": Order("ORD-1003", "C002", "Premium Subscription", 299.99, "pending", datetime(2025, 1, 5)),
 }
 
 TICKETS_DB: dict[str, SupportTicket] = {}
@@ -104,7 +76,7 @@ class SupportDeps:
 # --- Response Model ---
 class SupportResponse(BaseModel):
     message: str
-    ticket_id: Optional[str] = None
+    ticket_id: str | None = None
     refund_processed: bool = False
 
 
@@ -128,9 +100,7 @@ async def get_customer_info(ctx: RunContext[SupportDeps]) -> str:
     customer = CUSTOMERS_DB.get(ctx.deps.customer_id)
     if not customer:
         return "Customer not found."
-    return (
-        f"Customer: {customer.name}\n" f"Email: {customer.email}\n" f"Tier: {customer.tier}\n" f"Account since: {customer.account_created.strftime('%Y-%m-%d')}"
-    )
+    return f"Customer: {customer.name}\nEmail: {customer.email}\nTier: {customer.tier}\nAccount since: {customer.account_created.strftime('%Y-%m-%d')}"
 
 
 @support_agent.tool
@@ -184,11 +154,7 @@ async def create_support_ticket(ctx: RunContext[SupportDeps], subject: str, prio
     TICKET_COUNTER += 1
     ticket_id = f"TKT-{TICKET_COUNTER}"
     ticket = SupportTicket(
-        id=ticket_id,
-        customer_id=ctx.deps.customer_id,
-        subject=subject,
-        priority=priority,
-        status="open",
+        id=ticket_id, customer_id=ctx.deps.customer_id, subject=subject, priority=priority, status="open"
     )
     TICKETS_DB[ticket_id] = ticket
     return f"Support ticket created: {ticket_id} (Priority: {priority})"

splunk-ao-a2a/README.md

@@ -39,18 +39,23 @@ Splunk AO observability for [A2A (Agent-to-Agent)](https://github.com/google/A2A
 pip install splunk-ao-a2a
 ```
 
-**Requirements:** Python 3.11+, a [Splunk AO API key](https://www.splunk.com/), and [a2a-sdk](https://pypi.org/project/a2a-sdk/) 0.3+
+**Requirements:** Python 3.11+, Splunk AO standalone or Splunk Observability Cloud credentials, and [a2a-sdk](https://pypi.org/project/a2a-sdk/) 0.3+
 
 ## Quick Start
 
 ```python
-from splunk_ao.otel import SplunkAOSpanProcessor, add_splunk_ao_span_processor
+from splunk_ao.otel import add_splunk_ao_span_processor
 from splunk_ao_a2a import A2AInstrumentor
 from opentelemetry.sdk.trace import TracerProvider
 
 provider = TracerProvider()
-add_splunk_ao_span_processor(provider, SplunkAOSpanProcessor())
+add_splunk_ao_span_processor(provider)
 A2AInstrumentor().instrument(tracer_provider=provider, agent_name="orchestrator")
+
+try:
+    run_application()
+finally:
+    provider.shutdown()
 ```
 
 Once instrumented, all `a2a-sdk` client and server interactions produce OTel spans automatically.
@@ -63,14 +68,29 @@ Once instrumented, all `a2a-sdk` client and server interactions produce OTel spa
 | `agent_name` | Name of this agent, set on spans as `gen_ai.agent.name`. |
 | `capture_content` | Set to `False` to disable capturing message content (e.g. for PII compliance). |
 
-Environment variables for the Splunk AO exporter:
+For standalone Splunk AO:
 
 | Environment Variable | Description |
 |---------------------|-------------|
 | `SPLUNK_AO_API_KEY` | Splunk AO API key (required) |
 | `SPLUNK_AO_CONSOLE_URL` | Splunk AO console URL (required for self-hosted deployments, e.g. `http://localhost:8088`) |
-| `SPLUNK_AO_PROJECT` | Project name (alternative to `SplunkAOSpanProcessor(project=...)`) |
-| `SPLUNK_AO_AGENT_STREAM` | Agent stream name (alternative to `SplunkAOSpanProcessor(logstream=...)`) |
+| `SPLUNK_AO_API_URL` | Explicit API URL (optional; otherwise derived from the console URL) |
+| `SPLUNK_AO_PROJECT` / `SPLUNK_AO_PROJECT_ID` | Project name or ID |
+| `SPLUNK_AO_AGENT_STREAM` / `SPLUNK_AO_AGENT_STREAM_ID` | Agent-stream name or ID |
+
+For Splunk Observability Cloud:
+
+| Environment Variable | Description |
+|---------------------|-------------|
+| `SPLUNK_AO_REALM` | Observability Cloud realm (required) |
+| `SPLUNK_AO_SF_TOKEN` | SignalFlow ingest token used for OTLP export (required) |
+| `SPLUNK_AO_PROJECT` / `SPLUNK_AO_PROJECT_ID` | Optional project routing |
+| `SPLUNK_AO_AGENT_STREAM` / `SPLUNK_AO_AGENT_STREAM_ID` | Optional agent-stream routing |
+
+Applications should configure project and agent-stream routing. If it is
+accidentally absent, export remains non-blocking and ingestion may assign the
+trace to the unknown-project bucket.
+The same Python setup works for both deployments; only the environment changes.
 
 ## Features
 
@@ -120,7 +140,7 @@ from a2a.types import (
     AgentCapabilities, AgentCard, AgentSkill, Message, Role,
     TaskState, TaskStatus, TaskStatusUpdateEvent, TextPart,
 )
-from splunk_ao.otel import SplunkAOSpanProcessor, add_splunk_ao_span_processor
+from splunk_ao.otel import add_splunk_ao_span_processor
 from splunk_ao_a2a import A2AInstrumentor
 from langchain.agents import create_agent
 from langchain_core.tools import tool
@@ -133,7 +153,7 @@ from typing_extensions import TypedDict
 
 # ---- Only 4 lines needed for full distributed tracing ----
 provider = TracerProvider()
-add_splunk_ao_span_processor(provider, SplunkAOSpanProcessor())
+add_splunk_ao_span_processor(provider)
 A2AInstrumentor().instrument(tracer_provider=provider, agent_name="orchestrator")
 LangchainInstrumentor().instrument(tracer_provider=provider)
 

splunk-ao-a2a/examples/.env.example

@@ -1,5 +1,13 @@
+# Standalone Splunk AO
 SPLUNK_AO_API_KEY=your-splunk-ao-key
 SPLUNK_AO_CONSOLE_URL=https://<splunk-ao-console-host>
 SPLUNK_AO_PROJECT=a2a-distributed-tracing-demo
 SPLUNK_AO_LOG_STREAM=dev
+
+# O11y cloud
+# SPLUNK_AO_REALM=us1
+# SPLUNK_AO_SF_TOKEN=your-sf-ingest-token
+# SPLUNK_AO_PROJECT=a2a-distributed-tracing-demo
+# SPLUNK_AO_LOG_STREAM=dev
+
 OPENAI_API_KEY=your-openai-key

splunk-ao-a2a/src/splunk_ao_a2a/instrumentor.py

@@ -32,11 +32,11 @@ class A2AInstrumentor(BaseInstrumentor):  # type: ignore[misc]
     Example::
 
         from opentelemetry.sdk.trace import TracerProvider
-        from splunk_ao.otel import SplunkAOSpanProcessor, add_splunk_ao_span_processor
+        from splunk_ao.otel import add_splunk_ao_span_processor
         from splunk_ao_a2a import A2AInstrumentor
 
         provider = TracerProvider()
-        add_splunk_ao_span_processor(provider, SplunkAOSpanProcessor())
+        add_splunk_ao_span_processor(provider)
         A2AInstrumentor().instrument(tracer_provider=provider, agent_name="my-agent")
 
         # To disable message content capture (e.g. for PII compliance):

splunk-ao-a2a/tests/test_splunk_ao_compatibility.py

@@ -0,0 +1,80 @@
+from collections.abc import Sequence
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+from opentelemetry.sdk.trace import ReadableSpan, TracerProvider
+from opentelemetry.sdk.trace.export import SimpleSpanProcessor, SpanExporter, SpanExportResult
+
+from splunk_ao.deployment import DeploymentMode, StandaloneConfig
+from splunk_ao.otel import SplunkAOOTLPExporter, SplunkAOSpanProcessor, add_splunk_ao_span_processor
+from splunk_ao_a2a import _spans
+from splunk_ao_a2a._constants import INSTRUMENTOR_NAME, INSTRUMENTOR_VERSION
+
+
+class RecordingExporter(SpanExporter):
+    def __init__(self) -> None:
+        self.spans: list[ReadableSpan] = []
+
+    def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult:
+        self.spans.extend(spans)
+        return SpanExportResult.SUCCESS
+
+    def shutdown(self) -> None:
+        pass
+
+
+def test_a2a_native_span_uses_user_wired_deployment_aware_processor() -> None:
+    delegate = RecordingExporter()
+    captured_config: dict[str, object] = {}
+
+    def exporter_factory(**kwargs: object) -> RecordingExporter:
+        captured_config.update(kwargs)
+        return delegate
+
+    config = MagicMock()
+    config.resolve_deployment.return_value = DeploymentMode.STANDALONE
+    standalone = StandaloneConfig(
+        api_key="standalone-key",
+        console_url="https://console.example.com",
+        api_url="https://api.example.com",
+    )
+
+    with (
+        patch("splunk_ao.otel.SplunkAOConfig.get", return_value=config),
+        patch("splunk_ao.otel.StandaloneConfig.from_env", return_value=standalone),
+    ):
+        exporter = SplunkAOOTLPExporter(
+            project="a2a-project",
+            agentstream="a2a-agent-stream",
+            _exporter_factory=exporter_factory,
+        )
+        processor = SplunkAOSpanProcessor(SpanProcessor=SimpleSpanProcessor, _exporter=exporter)
+        provider = TracerProvider()
+        assert add_splunk_ao_span_processor(provider, processor) is processor
+        tracer = provider.get_tracer(INSTRUMENTOR_NAME, INSTRUMENTOR_VERSION)
+        try:
+            with tracer.start_as_current_span("a2a.client.send_message") as span:
+                _spans.set_client_attributes(
+                    span,
+                    SimpleNamespace(context_id="context-id", task_id="task-id"),
+                    "SendMessage",
+                    "orchestrator",
+                )
+        finally:
+            provider.shutdown()
+
+    assert captured_config == {
+        "endpoint": "https://api.example.com/otel/v1/traces",
+        "headers": {
+            "Splunk-AO-API-Key": "standalone-key",
+            "project": "a2a-project",
+            "logstream": "a2a-agent-stream",
+        },
+    }
+    exported = delegate.spans[0]
+    assert exported.instrumentation_scope.name == INSTRUMENTOR_NAME
+    assert exported.attributes["gen_ai.system"] == "a2a"
+    assert exported.attributes["a2a.rpc.method"] == "SendMessage"
+    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