diff --git a/.github/workflows/pytest.yml b/.github/workflows/pytest.yml index 402d83e5300..a7c62ec355e 100644 --- a/.github/workflows/pytest.yml +++ b/.github/workflows/pytest.yml @@ -243,20 +243,13 @@ jobs: tests/live_gateway/mcp/test_request_trace_policy.py \ tests/unit/mcpgateway/services/test_a2a_service.py \ tests/live_gateway/a2a/test_a2a_trace_context_e2e.py \ - --basetemp="${RUNNER_TEMP}/observability-tests" -v --tb=short - uv run --extra plugins --extra observability pytest \ - tests/unit/mcpgateway/services/test_gateway_service.py \ - -k health_completion_metrics -v --tb=short + -v --tb=short - - name: Preserve observability failure artifacts - if: failure() - uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4.6.2 - with: - name: observability-failure-artifacts - path: | - ${{ runner.temp }}/observability-tests/**/*.log - ${{ runner.temp }}/observability-tests/**/collector.jsonl - retention-days: 7 + - name: 🧪 Legacy Instana (< 3.12) live-gateway regression + run: | + LEGACY_INSTANA_E2E=1 uv run --extra observability --with "instana==3.11.1" pytest \ + tests/live_gateway/mcp/test_legacy_instana_e2e.py \ + -v --tb=short # ----------------------------------------------------------- # 4️⃣ Upload coverage artifacts (XML + HTML) diff --git a/mcpgateway/observability.py b/mcpgateway/observability.py index c5ca8258dd1..d5de217f243 100644 --- a/mcpgateway/observability.py +++ b/mcpgateway/observability.py @@ -16,6 +16,7 @@ import inspect import logging import os +import sys from typing import Any, Callable, cast, Dict, Mapping, MutableMapping, Optional from urllib.parse import urlparse @@ -42,6 +43,7 @@ OTEL_AVAILABLE = True except ImportError: # OpenTelemetry not installed - set to None for graceful degradation + otel_baggage = None trace = None otel_extract = None otel_inject = None @@ -95,7 +97,6 @@ def create(attrs: Dict[str, Any]) -> Dict[str, Any]: # type: ignore[override] # Provide minimal module shims so tests can patch ConsoleSpanExporter path try: # Standard - import sys import types if ("pytest" in sys.modules) or (os.getenv("MCP_TESTING") == "1"): @@ -346,6 +347,139 @@ class BaggageSpanAttributePolicy: _BAGGAGE_SPAN_ATTRIBUTE_POLICY = BaggageSpanAttributePolicy() +_LEGACY_INSTANA_COMPAT_MARKER = "_contextforge_otel_context_compat" + + +def _legacy_instana_span_context(context: Any, span_context_cls: Any) -> Any: + """Translate an OpenTelemetry ``Context`` into a legacy Instana ``SpanContext``. + + Mirrors the parent conversion that Instana >= 3.12 performs for ``context=``. + An explicit context without a valid span yields an invalid Instana span + context, so the new span becomes a root (standard OTel semantics) instead of + silently inheriting the ambient current span. + + Args: + context: OpenTelemetry context supplied by the caller. + span_context_cls: ``instana.span_context.SpanContext``. + + Returns: + An Instana ``SpanContext`` describing the parent. + """ + # Third-Party + from instana.util.ids import hex_id, hex_id_limited # pylint: disable=import-outside-toplevel,import-error + from opentelemetry import trace as otel_trace_api # pylint: disable=import-outside-toplevel + + parent = otel_trace_api.get_current_span(context).get_span_context() + if isinstance(parent, span_context_cls): + return parent + if parent is None or not parent.is_valid: + return span_context_cls(trace_id=otel_trace_api.INVALID_TRACE_ID, span_id=otel_trace_api.INVALID_SPAN_ID, is_remote=False) + hex_trace_id = hex_id(parent.trace_id) + is_long = len(hex_trace_id) > 16 + return span_context_cls( + trace_id=int(hex_id_limited(parent.trace_id), 16) if is_long else parent.trace_id, + span_id=parent.span_id, + is_remote=parent.is_remote, + trace_flags=parent.trace_flags, + trace_state=parent.trace_state, + long_trace_id=hex_trace_id if is_long else None, + ) + + +def _wrap_legacy_instana_method(original: Callable[..., Any], span_context_cls: Any) -> Callable[..., Any]: + """Wrap a legacy Instana tracer method so it accepts the OTel ``context=`` keyword. + + Args: + original: Unbound ``InstanaTracer`` method that only understands ``span_context``. + span_context_cls: ``instana.span_context.SpanContext``. + + Returns: + Wrapper translating ``context`` into ``span_context``; all other arguments, + return values, and exceptions pass through unchanged. + """ + + def wrapper(self: Any, *args: Any, **kwargs: Any) -> Any: + """Translate ``context`` to ``span_context`` and delegate. + + Args: + self: Tracer instance. + *args: Positional arguments forwarded unchanged. + **kwargs: Keyword arguments; ``context`` is translated. + + Returns: + Whatever the original method returns. + """ + if "context" in kwargs: + context = kwargs.pop("context") + if context is not None and kwargs.get("span_context") is None: + kwargs["span_context"] = _legacy_instana_span_context(context, span_context_cls) + return original(self, *args, **kwargs) + + wrapper.__name__ = getattr(original, "__name__", "wrapper") + wrapper.__qualname__ = getattr(original, "__qualname__", wrapper.__name__) + wrapper.__doc__ = getattr(original, "__doc__", None) + wrapper.__wrapped__ = original # type: ignore[attr-defined] + setattr(wrapper, _LEGACY_INSTANA_COMPAT_MARKER, True) + return wrapper + + +def _accepts_kwarg(func: Any, name: str) -> Optional[bool]: + """Return whether ``func`` accepts keyword ``name``; ``None`` if not introspectable. + + Args: + func: Callable to inspect. + name: Keyword argument name. + + Returns: + True/False, or None when the signature cannot be determined. + """ + try: + parameters = inspect.signature(func).parameters + except (TypeError, ValueError): + return None + return name in parameters or any(p.kind == inspect.Parameter.VAR_KEYWORD for p in parameters.values()) + + +def ensure_legacy_instana_context_compat() -> bool: + """Make legacy Instana tracers (< 3.12) accept the standard OTel ``context=`` keyword. + + MCP SDK 2.x, FastAPI native telemetry, and ContextForge call + ``start_span(context=...)`` / ``start_as_current_span(context=...)``. Legacy + ``InstanaTracer`` only accepts ``span_context=`` and raises ``TypeError``. + The adapter patches the ``InstanaTracer`` class once, so tracers cached by + third-party modules before ContextForge initializes are covered too. + + It is a no-op when Instana was not loaded by the process (this function never + imports Instana itself), when the tracer already supports ``context`` + (Instana >= 3.12), or when it was already applied. + + Returns: + True if the legacy tracer is (now) adapted, False otherwise. + """ + tracer_module = sys.modules.get("instana.tracer") + span_context_module = sys.modules.get("instana.span_context") + tracer_cls = getattr(tracer_module, "InstanaTracer", None) + span_context_cls = getattr(span_context_module, "SpanContext", None) + if tracer_cls is None or span_context_cls is None: + return False + + adapted = False + for method_name in ("start_span", "start_as_current_span"): + method = tracer_cls.__dict__.get(method_name) + if method is None: + continue + if getattr(method, _LEGACY_INSTANA_COMPAT_MARKER, False): + adapted = True + continue + # Only adapt the exact legacy capability: explicit `span_context`, no `context`. + if _accepts_kwarg(method, "context") is not False or _accepts_kwarg(method, "span_context") is not True: + continue + setattr(tracer_cls, method_name, _wrap_legacy_instana_method(method, span_context_cls)) + adapted = True + logger.info("Adapted legacy InstanaTracer.%s to accept the OpenTelemetry `context` argument", method_name) + return adapted + + def _supports_exporter_kwarg(exporter_cls: Any, kwarg: str) -> bool: """Return whether an exporter constructor accepts a keyword argument. @@ -1196,6 +1330,12 @@ def init_telemetry() -> Optional[Any]: """ # pylint: disable=global-statement global _TRACER + # Process-wide: FastAPI and MCP SDK telemetry use the global (possibly legacy + # Instana) tracer even when ContextForge's own OTEL integration is disabled. + try: + ensure_legacy_instana_context_compat() + except Exception as exc: # pylint: disable=broad-exception-caught + logger.warning("Failed to adapt legacy Instana tracer: %s", exc) cfg = get_settings() # Check if observability is disabled (default: disabled) diff --git a/tests/live_gateway/README.md b/tests/live_gateway/README.md index ed1b02e5611..ec4a1b89e7f 100644 --- a/tests/live_gateway/README.md +++ b/tests/live_gateway/README.md @@ -34,6 +34,7 @@ Specific subsuites need additional services on top: | `mcp/` | gateway with MCP transports registered | `make testing-up` (default profile) | | `mcp/test_oauth_status_live.py` | Postgres reachable at `localhost:5433` (the compose default) | `make testing-up` | | `mcp/test_private_key_jwt_e2e.py` | stub AS and upstream reachable via `host.docker.internal` (the compose default) | `make testing-up` | +| `mcp/test_legacy_instana_e2e.py` | self-contained: boots its own gateway under legacy Instana AutoTrace; set `LEGACY_INSTANA_E2E=1` to fail instead of skip | `LEGACY_INSTANA_E2E=1 uv run --extra observability --with "instana==3.11.1" pytest tests/live_gateway/mcp/test_legacy_instana_e2e.py` | | `sso/` | Keycloak (jwks tests) and/or Entra ID (entra tests) | `docker compose --profile sso up -d` for Keycloak; `AZURE_*` env vars for Entra | | `e2e_rust/` | gateway built with the Rust transport (edge or full mode) | `make testing-up` with the Rust profile, or rebuild compose images with Rust enabled | | `a2a/` | self-contained: each test boots its own gateway subprocess; needs the `observability` extra | `uv run --extra plugins --extra observability pytest tests/live_gateway/a2a/` | diff --git a/tests/live_gateway/mcp/test_legacy_instana_e2e.py b/tests/live_gateway/mcp/test_legacy_instana_e2e.py new file mode 100644 index 00000000000..93bd2404dd8 --- /dev/null +++ b/tests/live_gateway/mcp/test_legacy_instana_e2e.py @@ -0,0 +1,226 @@ +# -*- coding: utf-8 -*- +"""Location: ./tests/live_gateway/mcp/test_legacy_instana_e2e.py +Copyright contributors to the MCP-CONTEXT-FORGE project +SPDX-License-Identifier: Apache-2.0 + +Black-box regression for legacy Instana AutoTrace (``instana < 3.12``, issue #7208). + +Legacy ``InstanaTracer`` rejects the OpenTelemetry ``context=`` keyword used by +FastAPI native telemetry (``start_span``), the MCP SDK 2.x (``start_as_current_span``), +and ContextForge. This test boots an isolated gateway subprocess under the real +Instana AutoTrace bootstrap (``AUTOWRAPT_BOOTSTRAP=instana``), registers a real +streamable-HTTP MCP upstream, and verifies ``/health`` plus an authenticated MCP +``tools/call`` succeed with no escaping context-keyword ``TypeError``. + +Run (opt-in; fails rather than skips when ``LEGACY_INSTANA_E2E=1``):: + + LEGACY_INSTANA_E2E=1 uv run --extra observability \\ + --with "instana==3.11.1" \\ + pytest tests/live_gateway/mcp/test_legacy_instana_e2e.py -v + +No Instana backend is needed; the agent host points at an unused loopback port. +The gateway uses a private SQLite DB and per-run ephemeral signing keys. +""" + +# Standard +import asyncio +from importlib.metadata import PackageNotFoundError, version +import os +from pathlib import Path +import re +import secrets +import socket +import subprocess +import sys +import threading +import time + +# Third-Party +import httpx +import pytest + +# First-Party +from tests.helpers.auth import make_auth_headers, make_test_jwt + +pytestmark = pytest.mark.e2e +_REPO_ROOT = Path(__file__).resolve().parents[3] +_OPT_IN = os.getenv("LEGACY_INSTANA_E2E", "").lower() in {"1", "true", "yes"} +_CONTEXT_TYPE_ERROR = re.compile(r"TypeError:.*unexpected keyword argument '?context'?") + + +def _unavailable(reason): + """Fail when explicitly enabled; otherwise skip.""" + if _OPT_IN: + pytest.fail(reason) + pytest.skip(reason) + + +def _free_port(): + """Return an ephemeral loopback port.""" + with socket.socket() as listener: + listener.bind(("127.0.0.1", 0)) + return listener.getsockname()[1] + + +@pytest.fixture(scope="module") +def upstream_mcp_server(): + """Serve a real streamable-HTTP MCP upstream exposing an ``echo`` tool.""" + # Third-Party + from mcp.server.mcpserver import MCPServer + import uvicorn + + app = MCPServer("legacy-instana-upstream") + + @app.tool() + def echo(text: str) -> str: + """Return the supplied text.""" + return f"echo:{text}" + + port = _free_port() + server = uvicorn.Server(uvicorn.Config(app.streamable_http_app(host="127.0.0.1", stateless_http=True, json_response=True), host="127.0.0.1", port=port, log_level="error")) + worker = threading.Thread(target=server.run, daemon=True) + worker.start() + deadline = time.monotonic() + 15 + while not server.started and time.monotonic() < deadline: + time.sleep(0.05) + assert server.started, "upstream MCP server failed to start" + yield f"http://127.0.0.1:{port}/mcp" + server.should_exit = True + worker.join(timeout=5) + + +@pytest.fixture(scope="module") +def legacy_instana_gateway(tmp_path_factory, upstream_mcp_server): + """Boot a gateway under legacy Instana AutoTrace and register the upstream.""" + try: + instana_version = version("instana") + except PackageNotFoundError: + _unavailable("instana is not installed; run with --with 'instana==3.11.1'") + if tuple(int(part) for part in instana_version.split(".")[:2]) >= (3, 12): + _unavailable(f"instana {instana_version} supports `context=`; a legacy (< 3.12) sensor is required") + + tmp_path = tmp_path_factory.mktemp("legacy-instana-gateway") + gateway_port = _free_port() + signing_key = secrets.token_urlsafe(48) + env = { + "PATH": os.environ["PATH"], + "PYTHONPATH": str(_REPO_ROOT), + "AUTOWRAPT_BOOTSTRAP": "instana", + "INSTANA_SERVICE_NAME": "contextforge-legacy-instana-e2e", + "INSTANA_AGENT_HOST": "127.0.0.1", + "INSTANA_AGENT_PORT": str(_free_port()), + # gevent is only present in dev envs (via locust); production images lack it. + # Instana would otherwise gevent.monkey.patch_all(), which breaks anyio. + "INSTANA_GEVENT_MONKEY_OPTIONS": "no-socket,no-dns,no-time,no-select,no-thread,no-os,no-ssl,no-subprocess,no-sys,no-aggressive,no-event,no-builtins,no-signal,no-queue,no-contextvars", + "DATABASE_URL": f"sqlite:///{tmp_path / 'gateway.db'}", + "CACHE_TYPE": "memory", + "REDIS_URL": "", + "JWT_SECRET_KEY": signing_key, + "AUTH_ENCRYPTION_SECRET": secrets.token_urlsafe(48), + "PLATFORM_ADMIN_PASSWORD": secrets.token_urlsafe(32), + "DEFAULT_USER_PASSWORD": secrets.token_urlsafe(32), + "PLATFORM_ADMIN_EMAIL": "admin@example.com", + "AUTH_REQUIRED": "true", + "REQUIRE_USER_IN_DB": "false", + "MCPGATEWAY_ADMIN_API_ENABLED": "true", + "MCPGATEWAY_UI_ENABLED": "false", + "SSRF_ALLOW_LOCALHOST": "true", + "PLUGINS_ENABLED": "false", + "OBSERVABILITY_ENABLED": "false", + # ContextForge's own request-root middleware runs on the Instana global provider. + "OTEL_ENABLE_OBSERVABILITY": "true", + "OTEL_TRACES_EXPORTER": "console", + "LOG_LEVEL": "INFO", + "PYTHONUNBUFFERED": "1", + } + token = make_test_jwt("admin@example.com", is_admin=True, teams=None, secret=signing_key, algorithm="HS256") + log_path = tmp_path / "gateway.log" + with log_path.open("w") as log_file: + process = subprocess.Popen( + [sys.executable, "-m", "uvicorn", "mcpgateway.main:app", "--host", "127.0.0.1", "--port", str(gateway_port)], + cwd=tmp_path, + env=env, + stdout=log_file, + stderr=subprocess.STDOUT, + ) + try: + base_url = f"http://127.0.0.1:{gateway_port}" + with httpx.Client(base_url=base_url, headers=make_auth_headers(token), timeout=15, trust_env=False) as client: + deadline = time.monotonic() + 90 + while time.monotonic() < deadline: + assert process.poll() is None, log_path.read_text()[-5000:] + try: + client.get("/health") + break + except httpx.TransportError: + time.sleep(0.2) + else: + pytest.fail(f"Gateway startup timed out: {log_path.read_text()[-5000:]}") + + response = client.post( + "/gateways", + json={"name": "legacy-instana-upstream", "url": upstream_mcp_server, "transport": "STREAMABLEHTTP", "visibility": "public"}, + ) + assert response.status_code in (200, 201), response.text + gateway_id = response.json()["id"] + + tool_name = None + deadline = time.monotonic() + 30 + while tool_name is None and time.monotonic() < deadline: + tools = client.get("/tools", params={"limit": 0, "gateway_id": gateway_id}) + if tools.status_code == 200: + tool_name = next((tool["name"] for tool in tools.json() if tool.get("name", "").endswith("echo")), None) + if tool_name is None: + time.sleep(0.5) + assert tool_name, f"upstream echo tool was not discovered: {log_path.read_text()[-3000:]}" + + yield {"client": client, "base_url": base_url, "token": token, "tool_name": tool_name, "log_path": log_path} + finally: + process.terminate() + try: + process.wait(timeout=15) + except subprocess.TimeoutExpired: + process.kill() + process.wait(timeout=5) + + +def test_gateway_runs_under_legacy_instana(legacy_instana_gateway): + """The gateway process is actually instrumented by the legacy Instana sensor and adapter.""" + log = legacy_instana_gateway["log_path"].read_text() + assert "Adapted legacy InstanaTracer.start_span" in log, log[-3000:] + assert "Adapted legacy InstanaTracer.start_as_current_span" in log, log[-3000:] + + +def test_health_ok_under_legacy_instana(legacy_instana_gateway): + """FastAPI native telemetry calls ``start_span(context=...)`` on every request.""" + response = legacy_instana_gateway["client"].get("/health") + assert response.status_code == 200, response.text + + +def test_mcp_tools_call_under_legacy_instana(legacy_instana_gateway): + """Authenticated streamable-HTTP MCP ``tools/call`` completes with no context TypeError.""" + # Third-Party + import httpx2 + from mcp import ClientSession + from mcp.client.streamable_http import create_mcp_http_client, streamable_http_client + + async def _call(): + http_client = create_mcp_http_client(headers=make_auth_headers(legacy_instana_gateway["token"]), timeout=httpx2.Timeout(15.0)) + async with streamable_http_client(f"{legacy_instana_gateway['base_url']}/mcp/", http_client=http_client) as (read_stream, write_stream): + async with ClientSession(read_stream, write_stream, read_timeout_seconds=15.0) as session: + await session.initialize() + return await session.call_tool(legacy_instana_gateway["tool_name"], {"text": "instana"}) + + result = asyncio.run(_call()) + assert not result.is_error, result + assert any("echo:instana" in getattr(item, "text", "") for item in result.content), result + + log = legacy_instana_gateway["log_path"].read_text() + assert not _CONTEXT_TYPE_ERROR.search(log), log[-5000:] + + +def test_unauthenticated_mcp_rejected_under_legacy_instana(legacy_instana_gateway): + """Deny path: unauthenticated MCP requests are still rejected.""" + with httpx.Client(base_url=legacy_instana_gateway["base_url"], timeout=15, trust_env=False) as anonymous: + response = anonymous.post("/mcp/", json={"jsonrpc": "2.0", "id": 1, "method": "tools/list"}, headers={"Accept": "application/json, text/event-stream"}) + assert response.status_code == 401, response.text diff --git a/tests/unit/mcpgateway/test_observability.py b/tests/unit/mcpgateway/test_observability.py index 705af2d52c3..98a23573391 100644 --- a/tests/unit/mcpgateway/test_observability.py +++ b/tests/unit/mcpgateway/test_observability.py @@ -832,3 +832,243 @@ async def test_middleware_envelope_publication_failure_is_nonfatal(self): await middleware(scope, receive, send) assert captured["headers"] == original_headers + + +INBOUND_TRACEPARENT = "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01" # pragma: allowlist secret + + +@pytest.fixture +def fake_legacy_instana(monkeypatch): + """Install an ``instana`` stand-in whose tracer mirrors the Instana 3.11 signatures. + + The fake exposes the same ``span_context``-only signatures as Instana < 3.12, + backed by the real OpenTelemetry context API, so no Instana dependency is needed. + """ + pytest.importorskip("opentelemetry.trace") + # Standard + from contextlib import contextmanager + import sys + import types + + # Third-Party + from opentelemetry import trace as otel_trace_api + + class SpanContext(otel_trace_api.SpanContext): + """Legacy Instana span context (extra kwargs accepted and ignored).""" + + def __new__(cls, trace_id, span_id, is_remote, trace_flags=None, trace_state=None, **_kwargs): + return super().__new__(cls, trace_id, span_id, is_remote, trace_flags, trace_state) + + class FakeSpan(otel_trace_api.NonRecordingSpan): + def __init__(self, ctx, parent_id): + super().__init__(ctx) + self.parent_id = parent_id + + counter = {"id": 0x1000} + + class InstanaTracer: + """Legacy tracer: only ``span_context``; ``context=`` raises TypeError.""" + + def start_span(self, name, span_context=None, kind=None, attributes=None, links=None, start_time=None, record_exception=True, set_status_on_exception=True): + parent = span_context if span_context else otel_trace_api.get_current_span().get_span_context() + counter["id"] += 1 + if parent is not None and parent.is_valid: + ctx = SpanContext(parent.trace_id, counter["id"], False) + return FakeSpan(ctx, parent.span_id) + return FakeSpan(SpanContext(counter["id"], counter["id"], False), None) + + @contextmanager + def start_as_current_span(self, name, span_context=None, kind=None, attributes=None, links=None, start_time=None, record_exception=True, set_status_on_exception=True, end_on_exit=True): + span = self.start_span(name, span_context=span_context, kind=kind) + with otel_trace_api.use_span(span, end_on_exit=end_on_exit) as current: + yield current + + tracer_mod = types.ModuleType("instana.tracer") + tracer_mod.InstanaTracer = InstanaTracer + span_context_mod = types.ModuleType("instana.span_context") + span_context_mod.SpanContext = SpanContext + ids_mod = types.ModuleType("instana.util.ids") + ids_mod.hex_id = lambda value: format(int(value), "016x") if int(value) < (1 << 64) else format(int(value), "032x") + ids_mod.hex_id_limited = lambda value: format(int(value), "032x")[-16:] + monkeypatch.setitem(sys.modules, "instana.tracer", tracer_mod) + monkeypatch.setitem(sys.modules, "instana.span_context", span_context_mod) + monkeypatch.setitem(sys.modules, "instana.util.ids", ids_mod) + return InstanaTracer + + +class TestLegacyInstanaContextCompat: + """Process-wide adapter for legacy Instana tracers rejecting ``context=``.""" + + @staticmethod + def _inbound_context(): + # Third-Party + from opentelemetry.propagate import extract + + return extract({"traceparent": INBOUND_TRACEPARENT}) + + def test_cached_instance_start_span_accepts_context(self, fake_legacy_instana): + """Instances created before adaptation (e.g. FastAPI/MCP SDK caches) are covered.""" + cached = fake_legacy_instana() + with pytest.raises(TypeError, match="unexpected keyword argument 'context'"): + cached.start_span("pre", context=self._inbound_context()) + + assert observability.ensure_legacy_instana_context_compat() is True + span = cached.start_span("GET /health", context=self._inbound_context(), kind=None) + assert span.get_span_context().trace_id == 0x0AF7651916CD43DD8448EB211C80319C & ((1 << 64) - 1) + assert span.parent_id == 0xB7AD6B7169203331 + + def test_start_as_current_span_accepts_context_and_restores(self, fake_legacy_instana): + """``start_as_current_span(context=...)`` parents correctly and restores the outer context.""" + # Third-Party + from opentelemetry import context as otel_context_api + from opentelemetry import trace as otel_trace_api + + observability.ensure_legacy_instana_context_compat() + tracer = fake_legacy_instana() + before = otel_context_api.get_current() + with tracer.start_as_current_span("mcp", context=self._inbound_context()) as span: + assert span.parent_id == 0xB7AD6B7169203331 + assert otel_trace_api.get_current_span() is span + child = tracer.start_span("child") + assert child.parent_id == span.get_span_context().span_id + assert child.get_span_context().trace_id == span.get_span_context().trace_id + assert otel_context_api.get_current() is before + + def test_context_restored_when_body_raises(self, fake_legacy_instana): + """Application failures inside the span propagate and the context is restored.""" + # Third-Party + from opentelemetry import context as otel_context_api + + observability.ensure_legacy_instana_context_compat() + before = otel_context_api.get_current() + with pytest.raises(RuntimeError, match="app"): + with fake_legacy_instana().start_as_current_span("x", context=self._inbound_context()): + raise RuntimeError("app") + assert otel_context_api.get_current() is before + + def test_explicit_empty_context_creates_root(self, fake_legacy_instana): + """An explicit context without a span roots a new trace rather than inheriting the ambient span.""" + # Third-Party + from opentelemetry.context import Context + + observability.ensure_legacy_instana_context_compat() + tracer = fake_legacy_instana() + with tracer.start_as_current_span("outer") as outer: + span = tracer.start_span("root", context=Context()) + outer_context = outer.get_span_context() + root_context = span.get_span_context() + # Semantic check only: a new trace, not the ambient one (Instana may encode the + # missing parent as None or 0, so the parent_id representation is not asserted). + assert root_context.is_valid + assert root_context.trace_id != outer_context.trace_id + assert span.parent_id != outer_context.span_id + + def test_context_none_uses_current(self, fake_legacy_instana): + """``context=None`` keeps the legacy ambient-parent behavior.""" + observability.ensure_legacy_instana_context_compat() + tracer = fake_legacy_instana() + with tracer.start_as_current_span("outer") as outer: + span = tracer.start_span("inner", context=None) + assert span.parent_id == outer.get_span_context().span_id + + def test_idempotent(self, fake_legacy_instana): + """Repeated initialization wraps each method exactly once.""" + assert observability.ensure_legacy_instana_context_compat() is True + first = (fake_legacy_instana.start_span, fake_legacy_instana.start_as_current_span) + assert observability.ensure_legacy_instana_context_compat() is True + assert (fake_legacy_instana.start_span, fake_legacy_instana.start_as_current_span) == first + assert first[0].__wrapped__.__name__ == "start_span" + assert not hasattr(first[0].__wrapped__, "__wrapped__") + + def test_unrelated_type_error_propagates(self, fake_legacy_instana): + """Unrelated TypeErrors from the tracer are not swallowed or retried.""" + observability.ensure_legacy_instana_context_compat() + with pytest.raises(TypeError): + fake_legacy_instana().start_span("x", context=self._inbound_context(), bogus=True) + + def test_noop_for_modern_instana(self, monkeypatch): + """Instana >= 3.12 (``context`` already supported) is left untouched.""" + # Standard + import sys + import types + + class InstanaTracer: + def start_span(self, name, context=None, kind=None): + return (name, context) + + def start_as_current_span(self, name, context=None, kind=None): + return (name, context) + + tracer_mod = types.ModuleType("instana.tracer") + tracer_mod.InstanaTracer = InstanaTracer + span_context_mod = types.ModuleType("instana.span_context") + span_context_mod.SpanContext = object + monkeypatch.setitem(sys.modules, "instana.tracer", tracer_mod) + monkeypatch.setitem(sys.modules, "instana.span_context", span_context_mod) + original = (InstanaTracer.__dict__["start_span"], InstanaTracer.__dict__["start_as_current_span"]) + + assert observability.ensure_legacy_instana_context_compat() is False + assert (InstanaTracer.__dict__["start_span"], InstanaTracer.__dict__["start_as_current_span"]) == original + + def test_noop_without_instana(self, monkeypatch): + """No Instana loaded (ordinary OTel tracers): no-op and Instana is not imported.""" + # Standard + import sys + + monkeypatch.delitem(sys.modules, "instana.tracer", raising=False) + monkeypatch.delitem(sys.modules, "instana.span_context", raising=False) + assert observability.ensure_legacy_instana_context_compat() is False + assert "instana.tracer" not in sys.modules + + def test_noop_for_sdk_tracer(self): + """Standard OpenTelemetry SDK tracers are never patched.""" + trace_sdk = pytest.importorskip("opentelemetry.sdk.trace") + tracer_cls = getattr(trace_sdk, "Tracer", None) + if tracer_cls is None: + tracer_mod = getattr(trace_sdk, "tracer", None) + if tracer_mod is not None: + tracer_cls = getattr(tracer_mod, "Tracer", None) + if tracer_cls is None: + pytest.skip("OpenTelemetry SDK Tracer not exposed in this version") + before = tracer_cls.__dict__["start_span"] + observability.ensure_legacy_instana_context_compat() + assert tracer_cls.__dict__["start_span"] is before + + def test_init_telemetry_applies_adapter_even_when_disabled(self, fake_legacy_instana, monkeypatch): + """Adapter runs process-wide even when ContextForge OTEL is disabled (FastAPI/MCP still trace).""" + monkeypatch.setenv("OTEL_ENABLE_OBSERVABILITY", "false") + get_settings.cache_clear() + try: + assert init_telemetry() is None + finally: + get_settings.cache_clear() + assert getattr(fake_legacy_instana.start_span, observability._LEGACY_INSTANA_COMPAT_MARKER, False) + + def test_init_telemetry_tolerates_adapter_failure(self, monkeypatch): + """Adapter errors are logged, never fatal.""" + monkeypatch.setattr(observability, "ensure_legacy_instana_context_compat", MagicMock(side_effect=RuntimeError("boom"))) + monkeypatch.setenv("OTEL_ENABLE_OBSERVABILITY", "false") + get_settings.cache_clear() + try: + assert init_telemetry() is None + finally: + get_settings.cache_clear() + + def test_uninspectable_method_skipped(self, monkeypatch): + """Methods whose signature cannot be introspected are left alone.""" + # Standard + import sys + import types + + class InstanaTracer: + start_span = print # builtin: no introspectable span_context signature + start_as_current_span = print + + tracer_mod = types.ModuleType("instana.tracer") + tracer_mod.InstanaTracer = InstanaTracer + span_context_mod = types.ModuleType("instana.span_context") + span_context_mod.SpanContext = object + monkeypatch.setitem(sys.modules, "instana.tracer", tracer_mod) + monkeypatch.setitem(sys.modules, "instana.span_context", span_context_mod) + assert observability.ensure_legacy_instana_context_compat() is False + assert InstanaTracer.start_span is print