feat: Add twisted instrumentation and unit/integration tests

Signed-off-by: Cagri Yonca <cagri@ibm.com>

Author: CagriYonca

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:

src/instana/instrumentation/twisted/__init__.py

@@ -0,0 +1,130 @@
+# (c) Copyright IBM Corp. 2026
+
+try:
+    from typing import TYPE_CHECKING, Any, Callable, Union
+
+    import wrapt
+
+    if TYPE_CHECKING:
+        from twisted.internet.defer import Deferred
+        from twisted.web.client import Agent
+        from twisted.web.iweb import IResponse
+
+        from instana.span.span import InstanaSpan
+
+    from opentelemetry.context import get_current
+    from opentelemetry.semconv.trace import SpanAttributes
+    from twisted.python.failure import Failure
+    from twisted.web.http_headers import Headers as TwistedHeaders
+
+    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
+            if headers is None or not isinstance(headers, TwistedHeaders):
+                headers = TwistedHeaders({})
+
+            # Capture outgoing request headers
+            headers_dict = {
+                k.decode("latin-1"): v[0].decode("utf-8")
+                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.encode("latin-1"), [value.encode("utf-8")])
+
+            # 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: "Union[IResponse, Failure]", span: "InstanaSpan"
+    ) -> "Union[IResponse, Failure]":
+        """Callback/errback attached to the Agent.request Deferred."""
+        try:
+            if isinstance(result, Failure):
+                span.record_exception(result.value)
+            else:
+                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("utf-8")
+                    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

src/instana/instrumentation/twisted/client.py

@@ -0,0 +1,134 @@
+# (c) Copyright IBM Corp. 2026
+
+try:
+    from typing import TYPE_CHECKING, Any, Callable, Optional
+
+    import wrapt
+    from opentelemetry import context, trace
+
+    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("utf-8")
+                    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)
+
+            # Set span as current so downstream code (e.g. twisted-client) can find it
+            ctx = trace.set_span_in_context(span)
+            token = context.attach(ctx)
+            request._instana_token = token
+
+            # Extract the URL components
+            host = request.getHeader("host") or ""
+            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.encode("latin-1"), value.encode("utf-8"))
+
+            # 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("utf-8")
+                        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:
+                    context.detach(request._instana_token)
+                    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

src/instana/instrumentation/twisted/server.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",

src/instana/span/kind.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")