From e558a6fd2be46920ccfd40cfe5a16c642d59cfd5 Mon Sep 17 00:00:00 2001 From: Cagri Yonca Date: Mon, 13 Jul 2026 14:10:05 +0200 Subject: [PATCH] feat: Add twisted instrumentation and unit/integration tests Signed-off-by: Cagri Yonca --- src/instana/__init__.py | 6 + .../instrumentation/twisted/__init__.py | 1 + src/instana/instrumentation/twisted/client.py | 133 +++++ src/instana/instrumentation/twisted/server.py | 130 +++++ src/instana/span/kind.py | 4 + tests/apps/twisted_server/__init__.py | 30 + tests/apps/twisted_server/app.py | 122 +++++ tests/frameworks/test_twisted_client.py | 516 ++++++++++++++++++ tests/frameworks/test_twisted_server.py | 387 +++++++++++++ tests/requirements.txt | 1 + 10 files changed, 1330 insertions(+) create mode 100644 src/instana/instrumentation/twisted/__init__.py create mode 100644 src/instana/instrumentation/twisted/client.py create mode 100644 src/instana/instrumentation/twisted/server.py create mode 100644 tests/apps/twisted_server/__init__.py create mode 100644 tests/apps/twisted_server/app.py create mode 100644 tests/frameworks/test_twisted_client.py create mode 100644 tests/frameworks/test_twisted_server.py diff --git a/src/instana/__init__.py b/src/instana/__init__.py index 347f97ed..a8ecf07d 100644 --- a/src/instana/__init__.py +++ b/src/instana/__init__.py @@ -213,6 +213,12 @@ def boot_agent() -> None: from instana.instrumentation.tornado import ( server as tornado_server, # noqa: F401 ) + from instana.instrumentation.twisted import ( + client as twisted_client, # noqa: F401 + ) + from instana.instrumentation.twisted import ( + server as twisted_server, # noqa: F401 + ) def _start_profiler() -> None: diff --git a/src/instana/instrumentation/twisted/__init__.py b/src/instana/instrumentation/twisted/__init__.py new file mode 100644 index 00000000..a0d9d901 --- /dev/null +++ b/src/instana/instrumentation/twisted/__init__.py @@ -0,0 +1 @@ +# (c) Copyright IBM Corp. 2026 diff --git a/src/instana/instrumentation/twisted/client.py b/src/instana/instrumentation/twisted/client.py new file mode 100644 index 00000000..beb95c56 --- /dev/null +++ b/src/instana/instrumentation/twisted/client.py @@ -0,0 +1,133 @@ +# (c) Copyright IBM Corp. 2026 + +try: + from typing import TYPE_CHECKING, Any, Callable + + import wrapt + + if TYPE_CHECKING: + from twisted.internet.defer import Deferred + from twisted.web.client import Agent + + from instana.span.span import InstanaSpan + + from opentelemetry.context import get_current + from opentelemetry.semconv.trace import SpanAttributes + + from instana.log import logger + from instana.propagators.format import Format + from instana.singletons import agent, get_tracer + from instana.span.span import get_current_span + from instana.util.secrets import strip_secrets_from_query + from instana.util.traceutils import extract_custom_headers + + @wrapt.patch_function_wrapper("twisted.web.client", "Agent.request") + def request_with_instana( + wrapped: Callable[..., Any], + instance: "Agent", + argv: tuple[Any, ...], + kwargs: dict[str, Any], + ) -> "Deferred": + try: + parent_span = get_current_span() + + # If we're not tracing, just return + if not parent_span.is_recording(): + return wrapped(*argv, **kwargs) + + # argv: (method, url[, headers[, bodyProducer]]) + method = argv[0] + url = argv[1] + headers = argv[2] if len(argv) > 2 else kwargs.get("headers") + + method_str = ( + method.decode("latin-1") if isinstance(method, bytes) else str(method) + ) + url_str = url.decode("latin-1") if isinstance(url, bytes) else str(url) + + parent_context = get_current() + tracer = get_tracer() + span = tracer.start_span("twisted-client", context=parent_context) + + # Query param scrubbing + parts = url_str.split("?", 1) + span.set_attribute(SpanAttributes.HTTP_URL, parts[0]) + if len(parts) > 1 and parts[1]: + cleaned_qp = strip_secrets_from_query( + parts[1], + agent.options.secrets_matcher, + agent.options.secrets_list, + ) + span.set_attribute("http.params", cleaned_qp) + + span.set_attribute(SpanAttributes.HTTP_METHOD, method_str) + + # Build / augment headers with trace correlation + from twisted.web.http_headers import Headers as TwistedHeaders + + if headers is None or not isinstance(headers, TwistedHeaders): + headers = TwistedHeaders({}) + + # Capture outgoing request headers + headers_dict = { + k.decode("latin-1"): v[0].decode("latin-1") + for k, v in headers.getAllRawHeaders() + } + extract_custom_headers(span, headers_dict) + + # Inject Instana correlation headers + inject_carrier: dict[str, str] = {} + tracer.inject(span.context, Format.HTTP_HEADERS, inject_carrier) + for key, value in inject_carrier.items(): + headers.setRawHeaders(key, [value]) + + # Rebuild argv with the modified headers + new_argv = (argv[0], argv[1], headers) + argv[3:] + + deferred = wrapped(*new_argv, **kwargs) + + if deferred is not None: + deferred.addBoth(_finish_tracing, span) + + return deferred + except Exception: + logger.debug("twisted request_with_instana", exc_info=True) + return wrapped(*argv, **kwargs) + + def _finish_tracing(result: Any, span: "InstanaSpan") -> Any: + """Callback/errback attached to the Agent.request Deferred.""" + try: + from twisted.python.failure import Failure + + if isinstance(result, Failure): + span.record_exception(result.value) + if span.is_recording(): + span.end() + return result + + # result is a twisted.web.iweb.IResponse + response = result + status_code = response.code + span.set_attribute(SpanAttributes.HTTP_STATUS_CODE, status_code) + + if status_code >= 400: + span.mark_as_errored() + + # Capture response headers + headers_dict = { + k.decode("latin-1"): v[0].decode("latin-1") + for k, v in response.headers.getAllRawHeaders() + } + extract_custom_headers(span, headers_dict) + + except Exception: + logger.debug("twisted _finish_tracing", exc_info=True) + finally: + if span.is_recording(): + span.end() + + return result + + logger.debug("Instrumenting twisted client") +except ImportError: + pass diff --git a/src/instana/instrumentation/twisted/server.py b/src/instana/instrumentation/twisted/server.py new file mode 100644 index 00000000..8749528e --- /dev/null +++ b/src/instana/instrumentation/twisted/server.py @@ -0,0 +1,130 @@ +# (c) Copyright IBM Corp. 2026 + +try: + from typing import TYPE_CHECKING, Any, Callable, Optional + + import wrapt + + if TYPE_CHECKING: + from twisted.web.http import Request + from twisted.web.resource import Resource + + from opentelemetry.semconv.trace import SpanAttributes + + from instana.log import logger + from instana.propagators.format import Format + from instana.singletons import agent, get_tracer + from instana.util.secrets import strip_secrets_from_query + from instana.util.traceutils import extract_custom_headers + + @wrapt.patch_function_wrapper("twisted.web.resource", "Resource.render") + def render_with_instana( + wrapped: Callable[..., Any], + instance: "Resource", + argv: tuple[Any, ...], + kwargs: dict[str, Any], + ) -> Optional[bytes]: + try: + request: "Request" = argv[0] + tracer = get_tracer() + + # Extract parent context from incoming request headers + parent_context = None + if request.requestHeaders: + headers_dict: dict[str, str] = { + k.decode("latin-1"): v[0].decode("latin-1") + for k, v in request.requestHeaders.getAllRawHeaders() + } + parent_context = tracer.extract(Format.HTTP_HEADERS, headers_dict) + + span = tracer.start_span("twisted-server", context=parent_context) + + # Extract the URL components + raw_host = request.getHeader(b"host") or b"" + host = ( + raw_host.decode("latin-1") if isinstance(raw_host, bytes) else raw_host + ) + scheme = "https" if request.isSecure() else "http" + raw_path = request.path + path = ( + raw_path.decode("latin-1") if isinstance(raw_path, bytes) else raw_path + ) + + url = f"{scheme}://{host}{path}" + span.set_attribute(SpanAttributes.HTTP_URL, url) + + raw_method = request.method + method = ( + raw_method.decode("latin-1") + if isinstance(raw_method, bytes) + else raw_method + ) + span.set_attribute(SpanAttributes.HTTP_METHOD, method) + + # Query param scrubbing + raw_query = request.uri + query = ( + raw_query.decode("latin-1") + if isinstance(raw_query, bytes) + else raw_query + ) + if "?" in query: + qs = query.split("?", 1)[1] + if qs: + cleaned_qp = strip_secrets_from_query( + qs, + agent.options.secrets_matcher, + agent.options.secrets_list, + ) + span.set_attribute("http.params", cleaned_qp) + + # Request header tracking support + extract_custom_headers(span, headers_dict) + + # Inject correlation headers into response + response_headers = {} + tracer.inject(span.context, Format.HTTP_HEADERS, response_headers) + for key, value in response_headers.items(): + request.setHeader(key, value) + + # Store span on request for later retrieval + request._instana = span + + # Wrap request.finish to close the span when the response is complete + original_finish = request.finish + + def finish_with_instana() -> None: + try: + _span = request._instana + status_code = request.code + if isinstance(status_code, int): + _span.set_attribute( + SpanAttributes.HTTP_STATUS_CODE, status_code + ) + if status_code >= 500: + _span.mark_as_errored() + + # Capture response headers + response_hdrs = { + k.decode("latin-1"): v[0].decode("latin-1") + for k, v in request.responseHeaders.getAllRawHeaders() + } + extract_custom_headers(_span, response_hdrs) + + if _span.is_recording(): + _span.end() + except Exception: + logger.debug("twisted finish_with_instana", exc_info=True) + finally: + original_finish() + + request.finish = finish_with_instana + + return wrapped(*argv, **kwargs) + except Exception: + logger.debug("twisted render_with_instana", exc_info=True) + return wrapped(*argv, **kwargs) + + logger.debug("Instrumenting twisted server") +except ImportError: + pass diff --git a/src/instana/span/kind.py b/src/instana/span/kind.py index 1cc5e8dd..f7a074c3 100644 --- a/src/instana/span/kind.py +++ b/src/instana/span/kind.py @@ -16,6 +16,8 @@ "httpx", "tornado-client", "tornado-server", + "twisted-client", + "twisted-server", "urllib3", "wsgi", "asgi", @@ -31,6 +33,7 @@ "rabbitmq", "rpc-server", "tornado-server", + "twisted-server", "gcps-consumer", "asgi", "kafka-consumer", @@ -57,6 +60,7 @@ "sqlalchemy", "s3", "tornado-client", + "twisted-client", "urllib3", "pymongo", "gcs", diff --git a/tests/apps/twisted_server/__init__.py b/tests/apps/twisted_server/__init__.py new file mode 100644 index 00000000..affa8f49 --- /dev/null +++ b/tests/apps/twisted_server/__init__.py @@ -0,0 +1,30 @@ +# (c) Copyright IBM Corp. 2026 + +import os +import socket + +from ...helpers import testenv +from ..utils import launch_background_thread + +app_thread = None + + +def _get_free_port() -> int: + """Ask the OS for a free port.""" + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: + s.bind(("127.0.0.1", 0)) + return s.getsockname()[1] + + +if not any(( + app_thread, + os.environ.get("GEVENT_TEST"), + os.environ.get("CASSANDRA_TEST"), +)): + testenv["twisted_port"] = _get_free_port() + testenv["twisted_server"] = "http://127.0.0.1:" + str(testenv["twisted_port"]) + + # Background Twisted application + from .app import run_server + + app_thread = launch_background_thread(run_server, "Twisted") diff --git a/tests/apps/twisted_server/app.py b/tests/apps/twisted_server/app.py new file mode 100644 index 00000000..37d92ff2 --- /dev/null +++ b/tests/apps/twisted_server/app.py @@ -0,0 +1,122 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- + +# (c) Copyright IBM Corp. 2026 + +from twisted.internet import reactor +from twisted.web import server +from twisted.web.client import Agent, readBody +from twisted.web.http_headers import Headers +from twisted.web.resource import Resource + +from tests.helpers import testenv + + +class RootResource(Resource): + isLeaf = True + + def render_GET(self, request): + return b"Hello Twisted" + + def render_POST(self, request): + return b"Hello Twisted post" + + +class R301Resource(Resource): + isLeaf = True + + def render_GET(self, request): + request.setResponseCode(301) + request.setHeader(b"location", b"/") + return b"" + + +class R404Resource(Resource): + isLeaf = True + + def render_GET(self, request): + request.setResponseCode(404) + return b"Not Found" + + +class R500Resource(Resource): + isLeaf = True + + def render_GET(self, request): + request.setResponseCode(500) + return b"Internal Server Error" + + +class ResponseHeadersResource(Resource): + isLeaf = True + + def render_GET(self, request): + request.setHeader(b"X-Capture-This-Too", b"this too") + request.setHeader(b"X-Capture-That-Too", b"that too") + return b"Stan wuz here with headers!" + + +class FetchResource(Resource): + """GET /fetch?url= — makes an outbound Agent.request so + twisted-client instrumentation is exercised from within the reactor.""" + + isLeaf = True + + def render_GET(self, request): + target = request.args.get(b"url", [None])[0] + if not target: + request.setResponseCode(400) + return b"missing url param" + + agent_obj = Agent(reactor) + d = agent_obj.request(b"GET", target, Headers({}), None) + + def on_response(response): + return readBody(response) + + def on_body(body): + request.write(b"Fetched: " + body) + request.finish() + + def on_error(failure): + request.setResponseCode(502) + request.write(b"Fetch error: " + failure.getErrorMessage().encode()) + request.finish() + + d.addCallback(on_response) + d.addCallback(on_body) + d.addErrback(on_error) + return server.NOT_DONE_YET + + +class TwistedApp(Resource): + """Root resource that dispatches to child resources by path.""" + + def getChild(self, path, request): + if path == b"": + # / — serve root + return RootResource() + if path == b"301": + return R301Resource() + if path == b"404": + return R404Resource() + if path == b"500": + return R500Resource() + if path == b"response_headers": + return ResponseHeadersResource() + if path == b"fetch": + return FetchResource() + return Resource.getChild(self, path, request) + + def render_GET(self, request): + return b"Hello Twisted" + + def render_POST(self, request): + return b"Hello Twisted post" + + +def run_server(): + root = TwistedApp() + site = server.Site(root) + reactor.listenTCP(testenv["twisted_port"], site) + reactor.run(installSignalHandlers=False) diff --git a/tests/frameworks/test_twisted_client.py b/tests/frameworks/test_twisted_client.py new file mode 100644 index 00000000..253831e7 --- /dev/null +++ b/tests/frameworks/test_twisted_client.py @@ -0,0 +1,516 @@ +# (c) Copyright IBM Corp. 2026 + +import time +import threading +from typing import Generator, Optional + +import pytest +import requests + +import tests.apps.twisted_server # noqa: F401 +from instana.singletons import agent, get_tracer +from instana.span.span import get_current_span +from instana.util.ids import hex_id +from tests.helpers import get_first_span_by_name, testenv + + +class TestTwistedClient: + @pytest.fixture(autouse=True) + def _resource(self) -> Generator[None, None, None]: + """Clear all spans before a test run""" + self.tracer = get_tracer() + self.recorder = self.tracer.span_processor + self.recorder.clear_spans() + yield + + def _make_request( + self, + path: str, + method: str = "GET", + headers: Optional[dict] = None, + params: Optional[dict] = None, + allow_redirects: bool = True, + ): + """Run a Twisted Agent request from within a test span.""" + from twisted.internet import reactor + from twisted.web.client import Agent + from twisted.web.http_headers import Headers + + result_holder = {} + error_holder = {} + + def run_in_reactor(): + def on_response(response): + result_holder["response"] = response + + def on_error(failure): + error_holder["failure"] = failure + + twisted_headers = Headers({}) + if headers: + for k, v in headers.items(): + twisted_headers.setRawHeaders(k, [v]) + + agent_obj = Agent(reactor) + + url = (testenv["twisted_server"] + path).encode("utf-8") + if params: + from urllib.parse import urlencode + + url = ( + testenv["twisted_server"] + path + "?" + urlencode(params) + ).encode("utf-8") + + d = agent_obj.request(method.encode("utf-8"), url, twisted_headers, None) + d.addCallbacks(on_response, on_error) + return d + + event = threading.Event() + + def run(): + with self.tracer.start_as_current_span("test"): + d = run_in_reactor() + + def done(result): + event.set() + return result + + d.addBoth(done) + + reactor.callFromThread(run) + event.wait(timeout=5) + + return result_holder.get("response"), error_holder.get("failure") + + def test_get(self) -> None: + with self.tracer.start_as_current_span("test"): + response = requests.get(testenv["twisted_server"] + "/") + + time.sleep(0.5) + spans = self.recorder.queued_spans() + assert len(spans) == 3 + + server_span = get_first_span_by_name(spans, "twisted-server") + client_span = get_first_span_by_name(spans, "urllib3") + test_span = get_first_span_by_name(spans, "sdk") + + assert server_span + assert client_span + assert test_span + + assert not get_current_span().is_recording() + + # Same traceId + traceId = test_span.t + assert traceId == client_span.t + assert traceId == server_span.t + + # Parent relationships + assert client_span.p == test_span.s + assert server_span.p == client_span.s + + # Error logging + assert not test_span.ec + assert not client_span.ec + assert not server_span.ec + + assert server_span.data["http"]["status"] == 200 + assert testenv["twisted_server"] + "/" == server_span.data["http"]["url"] + assert not server_span.data["http"].get("params") + assert server_span.data["http"]["method"] == "GET" + + assert "X-INSTANA-T" in response.headers + assert response.headers["X-INSTANA-T"] == hex_id(traceId) + assert "X-INSTANA-S" in response.headers + assert response.headers["X-INSTANA-S"] == hex_id(server_span.s) + assert "X-INSTANA-L" in response.headers + assert response.headers["X-INSTANA-L"] == "1" + assert "Server-Timing" in response.headers + assert response.headers["Server-Timing"] == f"intid;desc={hex_id(traceId)}" + + def test_post(self) -> None: + with self.tracer.start_as_current_span("test"): + _ = requests.post(testenv["twisted_server"] + "/", data={"hello": "post"}) + + time.sleep(0.5) + spans = self.recorder.queued_spans() + assert len(spans) == 3 + + server_span = get_first_span_by_name(spans, "twisted-server") + client_span = get_first_span_by_name(spans, "urllib3") + test_span = get_first_span_by_name(spans, "sdk") + + assert server_span + assert client_span + assert test_span + + assert not get_current_span().is_recording() + + # Same traceId + traceId = test_span.t + assert traceId == client_span.t + assert traceId == server_span.t + + # Parent relationships + assert client_span.p == test_span.s + assert server_span.p == client_span.s + + # Error logging + assert not test_span.ec + assert not client_span.ec + assert not server_span.ec + + assert server_span.data["http"]["status"] == 200 + assert testenv["twisted_server"] + "/" == server_span.data["http"]["url"] + assert not server_span.data["http"].get("params") + assert server_span.data["http"]["method"] == "POST" + + def test_get_301(self) -> None: + with self.tracer.start_as_current_span("test"): + _ = requests.get(testenv["twisted_server"] + "/301", allow_redirects=False) + + time.sleep(0.5) + spans = self.recorder.queued_spans() + assert len(spans) == 3 + + server_span = get_first_span_by_name(spans, "twisted-server") + client_span = get_first_span_by_name(spans, "urllib3") + test_span = get_first_span_by_name(spans, "sdk") + + assert server_span + assert client_span + assert test_span + + assert not get_current_span().is_recording() + + # Same traceId + traceId = test_span.t + assert traceId == client_span.t + assert traceId == server_span.t + + # Parent relationships + assert client_span.p == test_span.s + assert server_span.p == client_span.s + + assert server_span.data["http"]["status"] == 301 + assert testenv["twisted_server"] + "/301" == server_span.data["http"]["url"] + assert server_span.data["http"]["method"] == "GET" + + def test_get_404(self) -> None: + with self.tracer.start_as_current_span("test"): + _ = requests.get(testenv["twisted_server"] + "/404") + + time.sleep(0.5) + spans = self.recorder.queued_spans() + assert len(spans) == 3 + + server_span = get_first_span_by_name(spans, "twisted-server") + client_span = get_first_span_by_name(spans, "urllib3") + test_span = get_first_span_by_name(spans, "sdk") + + assert server_span + assert client_span + assert test_span + + assert server_span.data["http"]["status"] == 404 + assert testenv["twisted_server"] + "/404" == server_span.data["http"]["url"] + + def test_get_500(self) -> None: + with self.tracer.start_as_current_span("test"): + _ = requests.get(testenv["twisted_server"] + "/500") + + time.sleep(0.5) + spans = self.recorder.queued_spans() + assert len(spans) == 3 + + server_span = get_first_span_by_name(spans, "twisted-server") + client_span = get_first_span_by_name(spans, "urllib3") + test_span = get_first_span_by_name(spans, "sdk") + + assert server_span + assert client_span + assert test_span + + assert not get_current_span().is_recording() + + # Same traceId + traceId = test_span.t + assert traceId == client_span.t + assert traceId == server_span.t + + assert server_span.data["http"]["status"] == 500 + assert testenv["twisted_server"] + "/500" == server_span.data["http"]["url"] + + # Error logging + assert not test_span.ec + assert client_span.ec == 1 + assert server_span.ec == 1 + + def test_get_with_params_to_scrub(self) -> None: + with self.tracer.start_as_current_span("test"): + _ = requests.get(testenv["twisted_server"] + "/", params={"secret": "yeah"}) + + time.sleep(0.5) + spans = self.recorder.queued_spans() + assert len(spans) == 3 + + server_span = get_first_span_by_name(spans, "twisted-server") + client_span = get_first_span_by_name(spans, "urllib3") + test_span = get_first_span_by_name(spans, "sdk") + + assert server_span + assert client_span + assert test_span + + assert not get_current_span().is_recording() + + # Same traceId + traceId = test_span.t + assert traceId == client_span.t + assert traceId == server_span.t + + assert server_span.data["http"]["status"] == 200 + assert testenv["twisted_server"] + "/" == server_span.data["http"]["url"] + assert server_span.data["http"]["params"] == "secret=" + + def test_request_header_capture(self) -> None: + original_extra_http_headers = agent.options.extra_http_headers + agent.options.extra_http_headers = ["X-Capture-This", "X-Capture-That"] + + with self.tracer.start_as_current_span("test"): + _ = requests.get( + testenv["twisted_server"] + "/", + headers={"X-Capture-This": "this", "X-Capture-That": "that"}, + ) + + agent.options.extra_http_headers = original_extra_http_headers + + time.sleep(0.5) + spans = self.recorder.queued_spans() + assert len(spans) == 3 + + server_span = get_first_span_by_name(spans, "twisted-server") + client_span = get_first_span_by_name(spans, "urllib3") + test_span = get_first_span_by_name(spans, "sdk") + + assert server_span + assert client_span + assert test_span + + assert "X-Capture-This" in server_span.data["http"]["header"] + assert server_span.data["http"]["header"]["X-Capture-This"] == "this" + assert "X-Capture-That" in server_span.data["http"]["header"] + assert server_span.data["http"]["header"]["X-Capture-That"] == "that" + + def test_response_header_capture(self) -> None: + original_extra_http_headers = agent.options.extra_http_headers + agent.options.extra_http_headers = ["X-Capture-This-Too", "X-Capture-That-Too"] + + with self.tracer.start_as_current_span("test"): + _ = requests.get(testenv["twisted_server"] + "/response_headers") + + agent.options.extra_http_headers = original_extra_http_headers + + time.sleep(0.5) + spans = self.recorder.queued_spans() + assert len(spans) == 3 + + server_span = get_first_span_by_name(spans, "twisted-server") + client_span = get_first_span_by_name(spans, "urllib3") + test_span = get_first_span_by_name(spans, "sdk") + + assert server_span + assert client_span + assert test_span + + assert "X-Capture-This-Too" in server_span.data["http"]["header"] + assert server_span.data["http"]["header"]["X-Capture-This-Too"] == "this too" + assert "X-Capture-That-Too" in server_span.data["http"]["header"] + assert server_span.data["http"]["header"]["X-Capture-That-Too"] == "that too" + + def test_agent_request_without_active_span(self) -> None: + """Agent.request with no active span must skip client instrumentation + (exercises the early-return branch in request_with_instana).""" + import threading + + from twisted.internet import reactor + from twisted.web.client import Agent + from twisted.web.http_headers import Headers + + result_holder: dict = {} + event = threading.Event() + + def do_request(): + # No active span — parent_span.is_recording() will be False + agent_obj = Agent(reactor) + d = agent_obj.request( + b"GET", + (testenv["twisted_server"] + "/").encode(), + Headers({}), + None, + ) + + def on_response(response): + result_holder["code"] = response.code + event.set() + + def on_error(failure): + result_holder["error"] = str(failure) + event.set() + + d.addCallbacks(on_response, on_error) + + reactor.callFromThread(do_request) + event.wait(timeout=5) + + time.sleep(0.5) + spans = self.recorder.queued_spans() + + # No twisted-client span should be created (no active parent span) + from tests.helpers import get_first_span_by_name as _get + + client_span = _get(spans, "twisted-client") + assert client_span is None + # The server still produces its own root span + assert result_holder.get("code") == 200 + + def test_agent_request_network_failure(self) -> None: + """Agent.request to an unreachable host exercises the Failure errback + path in _finish_tracing (client.py:102-106).""" + import threading + + from twisted.internet import reactor + from twisted.web.client import Agent + from twisted.web.http_headers import Headers + + event = threading.Event() + + def do_request(): + with self.tracer.start_as_current_span("test"): + agent_obj = Agent(reactor) + # Port 19999 is not listening — connection refused → Failure + d = agent_obj.request( + b"GET", + b"http://127.0.0.1:19999/", + Headers({}), + None, + ) + + def done(_): + event.set() + + d.addBoth(done) + + reactor.callFromThread(do_request) + event.wait(timeout=5) + + time.sleep(0.5) + spans = self.recorder.queued_spans() + + twisted_client_span = get_first_span_by_name(spans, "twisted-client") + test_span = get_first_span_by_name(spans, "sdk") + + assert twisted_client_span + assert test_span + # Failure path must mark the span errored + assert twisted_client_span.ec == 1 + + def test_agent_request_200(self) -> None: + """Twisted Agent.request 200 — exercises _finish_tracing success path + (lines 109-111, 126-129 in client.py).""" + response, failure = self._make_request("/") + + assert failure is None + assert response is not None + assert response.code == 200 + + time.sleep(0.5) + spans = self.recorder.queued_spans() + + twisted_client_span = get_first_span_by_name(spans, "twisted-client") + test_span = get_first_span_by_name(spans, "sdk") + + assert twisted_client_span + assert test_span + + # Same trace + assert twisted_client_span.t == test_span.t + assert twisted_client_span.p == test_span.s + + assert twisted_client_span.data["http"]["status"] == 200 + assert twisted_client_span.data["http"]["method"] == "GET" + assert not twisted_client_span.ec + + def test_agent_request_500_marks_errored(self) -> None: + """Twisted Agent.request 500 — exercises `status_code >= 400` branch + (lines 113-114 in client.py).""" + response, failure = self._make_request("/500") + + assert failure is None + assert response is not None + assert response.code == 500 + + time.sleep(0.5) + spans = self.recorder.queued_spans() + + twisted_client_span = get_first_span_by_name(spans, "twisted-client") + test_span = get_first_span_by_name(spans, "sdk") + + assert twisted_client_span + assert test_span + + assert twisted_client_span.data["http"]["status"] == 500 + assert twisted_client_span.ec == 1 + + def test_agent_request_response_headers(self) -> None: + """Twisted Agent.request captures response headers — exercises + extract_custom_headers in _finish_tracing (lines 117-121 in client.py).""" + from instana.singletons import agent as instana_agent + + original = instana_agent.options.extra_http_headers + instana_agent.options.extra_http_headers = [ + "X-Capture-This-Too", + "X-Capture-That-Too", + ] + + try: + response, failure = self._make_request("/response_headers") + finally: + instana_agent.options.extra_http_headers = original + + assert failure is None + assert response is not None + + time.sleep(0.5) + spans = self.recorder.queued_spans() + + twisted_client_span = get_first_span_by_name(spans, "twisted-client") + assert twisted_client_span + + assert "X-Capture-This-Too" in twisted_client_span.data["http"]["header"] + assert ( + twisted_client_span.data["http"]["header"]["X-Capture-This-Too"] + == "this too" + ) + assert "X-Capture-That-Too" in twisted_client_span.data["http"]["header"] + assert ( + twisted_client_span.data["http"]["header"]["X-Capture-That-Too"] + == "that too" + ) + + def test_agent_request_query_params_scrubbed(self) -> None: + """Twisted Agent.request with query params — exercises param scrubbing + (lines 56-61 in client.py) and sets http.params on the span.""" + response, failure = self._make_request("/", params={"secret": "topsecret"}) + + assert failure is None + assert response is not None + + time.sleep(0.5) + spans = self.recorder.queued_spans() + + twisted_client_span = get_first_span_by_name(spans, "twisted-client") + assert twisted_client_span + + assert twisted_client_span.data["http"]["params"] == "secret=" + assert twisted_client_span.data["http"]["url"].endswith("/") diff --git a/tests/frameworks/test_twisted_server.py b/tests/frameworks/test_twisted_server.py new file mode 100644 index 00000000..efd8749b --- /dev/null +++ b/tests/frameworks/test_twisted_server.py @@ -0,0 +1,387 @@ +# (c) Copyright IBM Corp. 2026 + +import time +from typing import Generator + +import pytest +import requests + +import tests.apps.twisted_server # noqa: F401 +from instana.singletons import agent, get_tracer +from instana.span.span import get_current_span +from instana.util.ids import hex_id +from tests.helpers import get_first_span_by_name, testenv + + +class TestTwistedServer: + @pytest.fixture(autouse=True) + def _resource(self) -> Generator[None, None, None]: + """Clear all spans before a test run""" + self.tracer = get_tracer() + self.recorder = self.tracer.span_processor + self.recorder.clear_spans() + yield + + def test_get(self) -> None: + with self.tracer.start_as_current_span("test"): + response = requests.get(testenv["twisted_server"] + "/") + + time.sleep(0.5) + spans = self.recorder.queued_spans() + assert len(spans) == 3 + + twisted_span = get_first_span_by_name(spans, "twisted-server") + urllib3_span = get_first_span_by_name(spans, "urllib3") + test_span = get_first_span_by_name(spans, "sdk") + + assert twisted_span + assert urllib3_span + assert test_span + + assert not get_current_span().is_recording() + + # Same traceId + traceId = test_span.t + assert traceId == urllib3_span.t + assert traceId == twisted_span.t + + # Parent relationships + assert urllib3_span.p == test_span.s + assert twisted_span.p == urllib3_span.s + + # Synthetic + assert not twisted_span.sy + assert not urllib3_span.sy + assert not test_span.sy + + # Error logging + assert not test_span.ec + assert not urllib3_span.ec + assert not twisted_span.ec + + assert twisted_span.data["http"]["status"] == 200 + assert testenv["twisted_server"] + "/" == twisted_span.data["http"]["url"] + assert not twisted_span.data["http"].get("params") + assert twisted_span.data["http"]["method"] == "GET" + assert not twisted_span.stack + + assert "X-INSTANA-T" in response.headers + assert response.headers["X-INSTANA-T"] == hex_id(traceId) + assert "X-INSTANA-S" in response.headers + assert response.headers["X-INSTANA-S"] == hex_id(twisted_span.s) + assert "X-INSTANA-L" in response.headers + assert response.headers["X-INSTANA-L"] == "1" + assert "Server-Timing" in response.headers + assert response.headers["Server-Timing"] == f"intid;desc={hex_id(traceId)}" + + def test_post(self) -> None: + with self.tracer.start_as_current_span("test"): + response = requests.post( + testenv["twisted_server"] + "/", data={"hello": "post"} + ) + + time.sleep(0.5) + spans = self.recorder.queued_spans() + assert len(spans) == 3 + + twisted_span = get_first_span_by_name(spans, "twisted-server") + urllib3_span = get_first_span_by_name(spans, "urllib3") + test_span = get_first_span_by_name(spans, "sdk") + + assert twisted_span + assert urllib3_span + assert test_span + + assert not get_current_span().is_recording() + + assert twisted_span.n == "twisted-server" + assert urllib3_span.n == "urllib3" + assert test_span.n == "sdk" + + # Same traceId + traceId = test_span.t + assert traceId == urllib3_span.t + assert traceId == twisted_span.t + + # Parent relationships + assert urllib3_span.p == test_span.s + assert twisted_span.p == urllib3_span.s + + # Error logging + assert not test_span.ec + assert not urllib3_span.ec + assert not twisted_span.ec + + assert twisted_span.data["http"]["status"] == 200 + assert testenv["twisted_server"] + "/" == twisted_span.data["http"]["url"] + assert not twisted_span.data["http"].get("params") + assert twisted_span.data["http"]["method"] == "POST" + assert not twisted_span.stack + + assert "X-INSTANA-T" in response.headers + assert response.headers["X-INSTANA-T"] == hex_id(traceId) + assert "X-INSTANA-S" in response.headers + assert response.headers["X-INSTANA-S"] == hex_id(twisted_span.s) + assert "X-INSTANA-L" in response.headers + assert response.headers["X-INSTANA-L"] == "1" + assert "Server-Timing" in response.headers + assert response.headers["Server-Timing"] == f"intid;desc={hex_id(traceId)}" + + def test_synthetic_request(self) -> None: + with self.tracer.start_as_current_span("test"): + _ = requests.get( + testenv["twisted_server"] + "/", + headers={"X-INSTANA-SYNTHETIC": "1"}, + ) + + time.sleep(0.5) + spans = self.recorder.queued_spans() + assert len(spans) == 3 + + twisted_span = get_first_span_by_name(spans, "twisted-server") + urllib3_span = get_first_span_by_name(spans, "urllib3") + test_span = get_first_span_by_name(spans, "sdk") + + assert twisted_span.sy + assert not urllib3_span.sy + assert not test_span.sy + + def test_get_301(self) -> None: + with self.tracer.start_as_current_span("test"): + # Don't follow redirects so we capture the 301 span + _ = requests.get( + testenv["twisted_server"] + "/301", + allow_redirects=False, + ) + + time.sleep(0.5) + spans = self.recorder.queued_spans() + assert len(spans) == 3 + + twisted_span = get_first_span_by_name(spans, "twisted-server") + urllib3_span = get_first_span_by_name(spans, "urllib3") + test_span = get_first_span_by_name(spans, "sdk") + + assert twisted_span + assert urllib3_span + assert test_span + + assert not get_current_span().is_recording() + + assert twisted_span.n == "twisted-server" + assert urllib3_span.n == "urllib3" + assert test_span.n == "sdk" + + # Same traceId + traceId = test_span.t + assert traceId == urllib3_span.t + assert traceId == twisted_span.t + + # Parent relationships + assert urllib3_span.p == test_span.s + assert twisted_span.p == urllib3_span.s + + # Error logging + assert not test_span.ec + assert not urllib3_span.ec + assert not twisted_span.ec + + assert twisted_span.data["http"]["status"] == 301 + assert testenv["twisted_server"] + "/301" == twisted_span.data["http"]["url"] + assert not twisted_span.data["http"].get("params") + assert twisted_span.data["http"]["method"] == "GET" + assert not twisted_span.stack + + def test_get_404(self) -> None: + with self.tracer.start_as_current_span("test"): + _ = requests.get(testenv["twisted_server"] + "/404") + + time.sleep(0.5) + spans = self.recorder.queued_spans() + assert len(spans) == 3 + + twisted_span = get_first_span_by_name(spans, "twisted-server") + urllib3_span = get_first_span_by_name(spans, "urllib3") + test_span = get_first_span_by_name(spans, "sdk") + + assert twisted_span + assert urllib3_span + assert test_span + + assert not get_current_span().is_recording() + + assert twisted_span.data["http"]["status"] == 404 + assert testenv["twisted_server"] + "/404" == twisted_span.data["http"]["url"] + assert twisted_span.data["http"]["method"] == "GET" + assert not twisted_span.stack + + def test_get_500(self) -> None: + with self.tracer.start_as_current_span("test"): + _ = requests.get(testenv["twisted_server"] + "/500") + + time.sleep(0.5) + spans = self.recorder.queued_spans() + assert len(spans) == 3 + + twisted_span = get_first_span_by_name(spans, "twisted-server") + urllib3_span = get_first_span_by_name(spans, "urllib3") + test_span = get_first_span_by_name(spans, "sdk") + + assert twisted_span + assert urllib3_span + assert test_span + + assert not get_current_span().is_recording() + + # Error logging + assert not test_span.ec + assert urllib3_span.ec == 1 + assert twisted_span.ec == 1 + + assert twisted_span.data["http"]["status"] == 500 + assert testenv["twisted_server"] + "/500" == twisted_span.data["http"]["url"] + assert twisted_span.data["http"]["method"] == "GET" + assert not twisted_span.stack + + def test_get_with_params_to_scrub(self) -> None: + with self.tracer.start_as_current_span("test"): + _ = requests.get( + testenv["twisted_server"] + "/", + params={"secret": "yeah"}, + ) + + time.sleep(0.5) + spans = self.recorder.queued_spans() + assert len(spans) == 3 + + assert not get_current_span().is_recording() + + twisted_span = get_first_span_by_name(spans, "twisted-server") + urllib3_span = get_first_span_by_name(spans, "urllib3") + test_span = get_first_span_by_name(spans, "sdk") + + assert twisted_span + assert urllib3_span + assert test_span + + assert twisted_span.n == "twisted-server" + assert urllib3_span.n == "urllib3" + assert test_span.n == "sdk" + + # Same traceId + traceId = test_span.t + assert traceId == urllib3_span.t + assert traceId == twisted_span.t + + # Parent relationships + assert urllib3_span.p == test_span.s + assert twisted_span.p == urllib3_span.s + + # Error logging + assert not test_span.ec + assert not urllib3_span.ec + assert not twisted_span.ec + + assert twisted_span.data["http"]["status"] == 200 + assert testenv["twisted_server"] + "/" == twisted_span.data["http"]["url"] + assert twisted_span.data["http"]["params"] == "secret=" + assert twisted_span.data["http"]["method"] == "GET" + assert not twisted_span.stack + + def test_request_header_capture(self) -> None: + original_extra_http_headers = agent.options.extra_http_headers + agent.options.extra_http_headers = ["X-Capture-This", "X-Capture-That"] + + with self.tracer.start_as_current_span("test"): + _ = requests.get( + testenv["twisted_server"] + "/", + params={"secret": "iloveyou"}, + headers={"X-Capture-This": "this", "X-Capture-That": "that"}, + ) + + agent.options.extra_http_headers = original_extra_http_headers + + time.sleep(0.5) + spans = self.recorder.queued_spans() + assert len(spans) == 3 + + twisted_span = get_first_span_by_name(spans, "twisted-server") + urllib3_span = get_first_span_by_name(spans, "urllib3") + test_span = get_first_span_by_name(spans, "sdk") + + assert twisted_span + assert urllib3_span + assert test_span + + assert not get_current_span().is_recording() + + assert twisted_span.data["http"]["status"] == 200 + assert testenv["twisted_server"] + "/" == twisted_span.data["http"]["url"] + assert twisted_span.data["http"]["params"] == "secret=" + assert twisted_span.data["http"]["method"] == "GET" + assert not twisted_span.stack + + assert "X-Capture-This" in twisted_span.data["http"]["header"] + assert twisted_span.data["http"]["header"]["X-Capture-This"] == "this" + assert "X-Capture-That" in twisted_span.data["http"]["header"] + assert twisted_span.data["http"]["header"]["X-Capture-That"] == "that" + + def test_response_header_capture(self) -> None: + original_extra_http_headers = agent.options.extra_http_headers + agent.options.extra_http_headers = ["X-Capture-This-Too", "X-Capture-That-Too"] + + with self.tracer.start_as_current_span("test"): + _ = requests.get( + testenv["twisted_server"] + "/response_headers", + params={"secret": "itsasecret"}, + ) + + agent.options.extra_http_headers = original_extra_http_headers + + time.sleep(0.5) + spans = self.recorder.queued_spans() + assert len(spans) == 3 + + twisted_span = get_first_span_by_name(spans, "twisted-server") + urllib3_span = get_first_span_by_name(spans, "urllib3") + test_span = get_first_span_by_name(spans, "sdk") + + assert twisted_span + assert urllib3_span + assert test_span + + assert not get_current_span().is_recording() + + assert twisted_span.data["http"]["status"] == 200 + assert ( + testenv["twisted_server"] + "/response_headers" + == twisted_span.data["http"]["url"] + ) + assert twisted_span.data["http"]["params"] == "secret=" + assert twisted_span.data["http"]["method"] == "GET" + assert not twisted_span.stack + + assert "X-Capture-This-Too" in twisted_span.data["http"]["header"] + assert twisted_span.data["http"]["header"]["X-Capture-This-Too"] == "this too" + assert "X-Capture-That-Too" in twisted_span.data["http"]["header"] + assert twisted_span.data["http"]["header"]["X-Capture-That-Too"] == "that too" + + def test_no_tracing_context(self) -> None: + """Requests without an active parent span still produce a root twisted-server span.""" + # No start_as_current_span wrapper — simulates an uninstrumented caller + response = requests.get(testenv["twisted_server"] + "/") + + time.sleep(0.5) + spans = self.recorder.queued_spans() + # Only the twisted-server span — no parent sdk or urllib3 span attached + assert len(spans) >= 1 + + twisted_span = get_first_span_by_name(spans, "twisted-server") + assert twisted_span + assert twisted_span.data["http"]["status"] == 200 + # No parent — this is a root span + assert not twisted_span.p + + assert "X-INSTANA-T" in response.headers + assert "X-INSTANA-S" in response.headers + assert "Server-Timing" in response.headers diff --git a/tests/requirements.txt b/tests/requirements.txt index 4dc2717e..b9420362 100644 --- a/tests/requirements.txt +++ b/tests/requirements.txt @@ -36,6 +36,7 @@ spyne>=2.14.0; python_version < "3.12" sqlalchemy>=2.0.0 starlette>=0.38.2; python_version == "3.13" tornado>=6.4.1 +twisted>=24.0.0 tracerite<=1.1.1; python_version < "3.9" uvicorn>=0.13.4 urllib3>=1.26.5