Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions src/instana/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down
1 change: 1 addition & 0 deletions src/instana/instrumentation/twisted/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
# (c) Copyright IBM Corp. 2026
133 changes: 133 additions & 0 deletions src/instana/instrumentation/twisted/client.py
Original file line number Diff line number Diff line change
@@ -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
130 changes: 130 additions & 0 deletions src/instana/instrumentation/twisted/server.py
Original file line number Diff line number Diff line change
@@ -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(

Check failure on line 21 in src/instana/instrumentation/twisted/server.py

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

Refactor this function to reduce its Cognitive Complexity from 21 to the 15 allowed.

See more on https://sonarcloud.io/project/issues?id=instana_python-sensor&issues=AZ9bBvd3S1IsCbTKd_Ng&open=AZ9bBvd3S1IsCbTKd_Ng&pullRequest=888
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
4 changes: 4 additions & 0 deletions src/instana/span/kind.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,8 @@
"httpx",
"tornado-client",
"tornado-server",
"twisted-client",
"twisted-server",
"urllib3",
"wsgi",
"asgi",
Expand All @@ -31,6 +33,7 @@
"rabbitmq",
"rpc-server",
"tornado-server",
"twisted-server",
"gcps-consumer",
"asgi",
"kafka-consumer",
Expand All @@ -57,6 +60,7 @@
"sqlalchemy",
"s3",
"tornado-client",
"twisted-client",
"urllib3",
"pymongo",
"gcs",
Expand Down
30 changes: 30 additions & 0 deletions tests/apps/twisted_server/__init__.py
Original file line number Diff line number Diff line change
@@ -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")
Loading
Loading