fix(examples-py): validate Elasticsearch, Google ADK, and Google GenAI OTel 2.x spans

python/tracing/elasticsearch/01_sync_client.py

@@ -0,0 +1,52 @@
+"""Trace synchronous Elasticsearch index, search, and error operations."""
+
+from __future__ import annotations
+
+from elasticsearch import Elasticsearch, NotFoundError
+from respan import workflow
+
+from _shared import example_attributes, local_elasticsearch, make_respan
+
+EXAMPLE_NAME = "sync-client"
+
+
+@workflow(name="elasticsearch_sync_client")
+def run_sync_client(prompt: str, endpoint: str) -> dict[str, object]:
+    client = Elasticsearch(endpoint)
+    try:
+        indexed = client.index(
+            index="audit-index",
+            id="doc-1",
+            document={"title": prompt, "category": "observability"},
+            refresh=True,
+        )
+        searched = client.search(
+            index="audit-index",
+            query={"match": {"title": "Tracing"}},
+        )
+        missing_status = 0
+        try:
+            client.get(index="audit-index", id="missing")
+        except NotFoundError as exc:
+            missing_status = exc.status_code
+        return {
+            "indexed": indexed["result"],
+            "hits": searched["hits"]["total"]["value"],
+            "missing_status": missing_status,
+        }
+    finally:
+        client.close()
+
+
+def main() -> None:
+    respan = make_respan(EXAMPLE_NAME)
+    try:
+        with local_elasticsearch() as endpoint, example_attributes(EXAMPLE_NAME):
+            result = run_sync_client("Tracing Elasticsearch", endpoint)
+            print(result)
+    finally:
+        respan.shutdown()
+
+
+if __name__ == "__main__":
+    main()

python/tracing/elasticsearch/02_async_client.py

@@ -0,0 +1,48 @@
+"""Trace asynchronous Elasticsearch index and search operations."""
+
+from __future__ import annotations
+
+import asyncio
+
+from elasticsearch import AsyncElasticsearch
+from respan import workflow
+
+from _shared import example_attributes, local_elasticsearch, make_respan
+
+EXAMPLE_NAME = "async-client"
+
+
+@workflow(name="elasticsearch_async_client")
+async def run_async_client(prompt: str, endpoint: str) -> dict[str, object]:
+    client = AsyncElasticsearch(endpoint)
+    try:
+        indexed = await client.index(
+            index="audit-index",
+            id="doc-1",
+            document={"title": prompt, "category": "async-observability"},
+            refresh=True,
+        )
+        searched = await client.search(
+            index="audit-index",
+            query={"match": {"title": "Async"}},
+        )
+        return {
+            "indexed": indexed["result"],
+            "hits": searched["hits"]["total"]["value"],
+        }
+    finally:
+        await client.close()
+
+
+def main() -> None:
+    respan = make_respan(EXAMPLE_NAME)
+    try:
+        with local_elasticsearch() as endpoint, example_attributes(EXAMPLE_NAME):
+            result = asyncio.run(run_async_client("Async Elasticsearch", endpoint))
+            print(result)
+    finally:
+        respan.shutdown()
+
+
+if __name__ == "__main__":
+    main()

python/tracing/elasticsearch/README.md

@@ -0,0 +1,12 @@
+# Elasticsearch tracing examples
+
+These examples use the real synchronous and asynchronous Elasticsearch Python
+clients against a deterministic local HTTP server, so no Elasticsearch cluster
+or credentials are required. Traces are exported with the `RESPAN_API_KEY` from
+the repository root `.env` file.
+
+Run both examples with one marker:
+
+```bash
+RESPAN_EXAMPLE_RUN_ID=my-audit-marker python run_all.py
+```

python/tracing/elasticsearch/_shared.py

@@ -0,0 +1,135 @@
+"""Shared local Elasticsearch server and Respan setup for the examples."""
+
+from __future__ import annotations
+
+import json
+import os
+import threading
+from contextlib import contextmanager
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from pathlib import Path
+from typing import Iterator
+from uuid import uuid4
+
+from dotenv import load_dotenv
+from respan import Respan, propagate_attributes
+from respan_instrumentation_elasticsearch import ElasticsearchInstrumentor
+
+PROJECT_ROOT = Path(__file__).resolve().parents[3]
+
+
+def example_run_id() -> str:
+    return os.getenv("RESPAN_EXAMPLE_RUN_ID") or f"elasticsearch-{uuid4().hex[:10]}"
+
+
+def make_respan(example_name: str) -> Respan:
+    load_dotenv(PROJECT_ROOT / ".env", override=True)
+    api_key = os.getenv("RESPAN_API_KEY")
+    if not api_key:
+        raise RuntimeError("RESPAN_API_KEY must be set in respan-example-projects/.env")
+    return Respan(
+        api_key=api_key,
+        base_url=os.getenv("RESPAN_BASE_URL", "https://api.respan.ai/api"),
+        app_name="elasticsearch-examples",
+        instrumentations=[ElasticsearchInstrumentor()],
+        is_batching_enabled=False,
+        metadata={
+            "integration": "elasticsearch",
+            "example": example_name,
+            "run_id": example_run_id(),
+        },
+    )
+
+
+def example_attributes(example_name: str):
+    run_id = example_run_id()
+    return propagate_attributes(
+        custom_identifier=f"{run_id}:{example_name}",
+        trace_group_identifier=f"{run_id}:{example_name}",
+        metadata={
+            "integration": "elasticsearch",
+            "example": example_name,
+            "run_id": run_id,
+        },
+    )
+
+
+class _ElasticsearchHandler(BaseHTTPRequestHandler):
+    server_version = "Elasticsearch/9.5.0"
+
+    def log_message(self, format: str, *args: object) -> None:
+        return None
+
+    def _send_json(self, status: int, payload: dict[str, object]) -> None:
+        body = json.dumps(payload).encode("utf-8")
+        self.send_response(status)
+        self.send_header("Content-Type", "application/json")
+        self.send_header("Content-Length", str(len(body)))
+        self.send_header("X-Elastic-Product", "Elasticsearch")
+        self.end_headers()
+        self.wfile.write(body)
+
+    def do_PUT(self) -> None:
+        self._send_json(
+            201,
+            {
+                "_index": "audit-index",
+                "_id": "doc-1",
+                "_version": 1,
+                "result": "created",
+                "_shards": {"total": 1, "successful": 1, "failed": 0},
+                "_seq_no": 0,
+                "_primary_term": 1,
+            },
+        )
+
+    def do_POST(self) -> None:
+        self._send_json(
+            200,
+            {
+                "took": 1,
+                "timed_out": False,
+                "_shards": {"total": 1, "successful": 1, "skipped": 0, "failed": 0},
+                "hits": {
+                    "total": {"value": 1, "relation": "eq"},
+                    "max_score": 1.0,
+                    "hits": [
+                        {
+                            "_index": "audit-index",
+                            "_id": "doc-1",
+                            "_score": 1.0,
+                            "_source": {"title": "Tracing Elasticsearch"},
+                        }
+                    ],
+                },
+            },
+        )
+
+    def do_GET(self) -> None:
+        self._send_json(
+            404,
+            {
+                "_index": "audit-index",
+                "_id": "missing",
+                "found": False,
+                "error": {
+                    "type": "document_missing_exception",
+                    "reason": "document [missing] is absent",
+                },
+                "status": 404,
+            },
+        )
+
+
+@contextmanager
+def local_elasticsearch() -> Iterator[str]:
+    server = ThreadingHTTPServer(("127.0.0.1", 0), _ElasticsearchHandler)
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    try:
+        host, port = server.server_address
+        yield f"http://{host}:{port}"
+    finally:
+        server.shutdown()
+        server.server_close()
+        thread.join(timeout=5)

python/tracing/elasticsearch/requirements.txt

@@ -0,0 +1,4 @@
+elasticsearch[async]>=8.13.0,<10.0.0
+python-dotenv>=1.0.0
+respan-ai>=2.16.1
+respan-instrumentation-elasticsearch>=0.1.0

python/tracing/elasticsearch/run_all.py

@@ -0,0 +1,24 @@
+"""Run every Elasticsearch tracing example with one audit marker."""
+
+from __future__ import annotations
+
+import os
+import subprocess
+import sys
+from pathlib import Path
+from uuid import uuid4
+
+SCRIPTS = ("01_sync_client.py", "02_async_client.py")
+
+
+def main() -> None:
+    directory = Path(__file__).resolve().parent
+    env = os.environ.copy()
+    env.setdefault("RESPAN_EXAMPLE_RUN_ID", f"elasticsearch-{uuid4().hex[:10]}")
+    print(f"RESPAN_EXAMPLE_RUN_ID={env['RESPAN_EXAMPLE_RUN_ID']}", flush=True)
+    for script in SCRIPTS:
+        subprocess.run([sys.executable, str(directory / script)], env=env, check=True)
+
+
+if __name__ == "__main__":
+    main()

python/tracing/google-adk/01_hello_world.py

@@ -6,14 +6,14 @@
 from google.adk.agents import Agent
 from respan import workflow
 
-from _shared import create_gateway_model, create_respan, run_agent_once
+from _shared import create_gateway_model, create_respan, example_attributes, run_agent_once
 
 SCRIPT_NAME = Path(__file__).name
 APP_NAME = SCRIPT_NAME.removesuffix(".py")
 
 
 @workflow(name=SCRIPT_NAME)
-async def run_hello_world() -> str:
+async def run_hello_world(prompt: str) -> str:
     agent = Agent(
         name="hello_world_agent",
         model=create_gateway_model(),
@@ -22,7 +22,7 @@ async def run_hello_world() -> str:
     output = await run_agent_once(
         agent=agent,
         app_name=APP_NAME,
-        prompt="Say hello from a traced Google ADK agent.",
+        prompt=prompt,
     )
     print(output)
     return output
@@ -31,7 +31,8 @@ async def run_hello_world() -> str:
 async def main() -> None:
     respan = create_respan(APP_NAME)
     try:
-        await run_hello_world()
+        with example_attributes(APP_NAME):
+            await run_hello_world("Say hello from a traced Google ADK agent.")
     finally:
         respan.shutdown()
 

python/tracing/google-adk/02_tool_use.py

@@ -6,7 +6,7 @@
 from google.adk.agents import Agent
 from respan import workflow
 
-from _shared import create_gateway_model, create_respan, run_agent_once
+from _shared import create_gateway_model, create_respan, example_attributes, run_agent_once
 
 SCRIPT_NAME = Path(__file__).name
 APP_NAME = SCRIPT_NAME.removesuffix(".py")
@@ -18,7 +18,7 @@ def get_weather(city: str) -> str:
 
 
 @workflow(name=SCRIPT_NAME)
-async def run_tool_use() -> str:
+async def run_tool_use(prompt: str) -> str:
     agent = Agent(
         name="weather_agent",
         model=create_gateway_model(),
@@ -31,7 +31,7 @@ async def run_tool_use() -> str:
     output = await run_agent_once(
         agent=agent,
         app_name=APP_NAME,
-        prompt="Use get_weather for San Francisco and summarize the result.",
+        prompt=prompt,
     )
     print(output)
     return output
@@ -40,7 +40,10 @@ async def run_tool_use() -> str:
 async def main() -> None:
     respan = create_respan(APP_NAME)
     try:
-        await run_tool_use()
+        with example_attributes(APP_NAME):
+            await run_tool_use(
+                "Use get_weather for San Francisco and summarize the result."
+            )
     finally:
         respan.shutdown()