-
Notifications
You must be signed in to change notification settings - Fork 38
feat: Add twisted instrumentation and unit/integration tests #888
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Merged
+1,163
−0
Merged
Changes from all commits
Commits
File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -105,3 +105,4 @@ uv.lock | |
|
|
||
| # Sandbox | ||
| sandbox/ | ||
| .bob/ | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Empty file.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,150 @@ | ||
| # (c) Copyright IBM Corp. 2026 | ||
| """Instana instrumentation for the Twisted HTTP client (``twisted.web.client.Agent``). | ||
|
|
||
| Wraps ``Agent.request`` to create an exit span for every outgoing HTTP request, | ||
| propagate Instana correlation headers, scrub query-parameter secrets, and record | ||
| the response status code (or exception) when the returned ``Deferred`` resolves. | ||
| """ | ||
|
|
||
| try: | ||
| from typing import TYPE_CHECKING, Callable, Union | ||
|
|
||
| import wrapt | ||
| 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 | ||
|
|
||
| if TYPE_CHECKING: | ||
| from twisted.internet.defer import Deferred | ||
| from twisted.web.iweb import IResponse | ||
|
|
||
| from instana.span.span import InstanaSpan | ||
|
|
||
| @wrapt.patch_function_wrapper("twisted.web.client", "Agent.request") | ||
| def request_with_instana( | ||
| wrapped: "Callable[..., Deferred]", | ||
| instance: object, | ||
| argv: tuple[object, ...], | ||
| kwargs: dict[str, object], | ||
| ) -> "Deferred": | ||
| """Wrapt wrapper for ``Agent.request`` that adds an exit span. | ||
|
|
||
| Starts a ``twisted-client`` span, injects Instana trace-correlation | ||
| headers into the outgoing request, and attaches ``finish_tracing`` as | ||
| both a callback and errback on the returned ``Deferred`` so the span is | ||
| always closed. Falls back to the unwrapped call on any instrumentation | ||
| error to keep the application path safe. | ||
| """ | ||
| 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 = {} | ||
| 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 client 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: | ||
| status_code = result.code | ||
| span.set_attribute(SpanAttributes.HTTP_STATUS_CODE, status_code) | ||
|
|
||
| # Capture response headers | ||
| headers_dict = { | ||
| k.decode("latin-1"): v[0].decode("utf-8") | ||
| for k, v in result.headers.getAllRawHeaders() | ||
| } | ||
| extract_custom_headers(span, headers_dict) | ||
|
|
||
| if status_code >= 500: | ||
| span.mark_as_errored({ | ||
| "http.error": result.phrase.decode("latin-1") | ||
| }) | ||
| except Exception: | ||
| logger.debug("twisted client finish_tracing", exc_info=True) | ||
| finally: | ||
| if span.is_recording(): | ||
| span.end() | ||
|
|
||
| return result | ||
|
|
||
| logger.debug("Instrumenting twisted client") | ||
| except ImportError: | ||
| pass |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,175 @@ | ||
| # (c) Copyright IBM Corp. 2026 | ||
| """Instana instrumentation for the Twisted HTTP server (``twisted.web.resource.Resource``). | ||
|
|
||
| Wraps ``Resource.render`` to create an entry span for every incoming HTTP | ||
| request, extract Instana trace-correlation headers, scrub query-parameter | ||
| secrets, inject correlation headers into the response, and close the span | ||
| when the Twisted request lifecycle ends via ``notifyFinish``. | ||
| """ | ||
|
|
||
| try: | ||
| from typing import TYPE_CHECKING, Callable, Optional | ||
|
|
||
| import wrapt | ||
| from opentelemetry import context, trace | ||
| 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 | ||
|
|
||
| if TYPE_CHECKING: | ||
| from twisted.python.failure import Failure | ||
| from twisted.web.http import Request | ||
| from twisted.web.resource import Resource | ||
|
|
||
| @wrapt.patch_function_wrapper("twisted.web.resource", "Resource.render") | ||
| def render_with_instana( | ||
| wrapped: "Callable[..., Optional[bytes]]", | ||
| instance: "Resource", | ||
| argv: tuple[object, ...], | ||
| kwargs: dict[str, object], | ||
| ) -> Optional[bytes]: | ||
| """Wrapt wrapper for ``Resource.render`` that adds an entry span. | ||
|
|
||
| Extracts any existing Instana trace context from the incoming request | ||
| headers and starts a ``twisted-server`` span as a child. The span is | ||
| set as the active context for the synchronous duration of ``wrapped()`` | ||
| so that downstream exit instrumentation (e.g. ``twisted-client``) can | ||
| find it. ``finish_tracing`` is registered on the ``notifyFinish`` | ||
| deferred to close the span once the full response has been written. | ||
| Falls back to the unwrapped call on any instrumentation error. | ||
| """ | ||
| request = argv[0] | ||
| span = None | ||
| token = None | ||
| try: | ||
| tracer = get_tracer() | ||
|
|
||
| # Extract parent context from incoming request headers | ||
| headers_dict = {} | ||
| parent_context = None | ||
| if request.requestHeaders: | ||
| headers_dict = { | ||
| 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 during the synchronous | ||
| # wrapped() call. We detach unconditionally in the finally | ||
| # block below once wrapped() has returned. | ||
| ctx = trace.set_span_in_context(span) | ||
| token = context.attach(ctx) | ||
|
|
||
| # 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 | ||
| request._instana_finished = False | ||
|
|
||
| finish_deferred = request.notifyFinish() | ||
| finish_deferred.addBoth(finish_tracing, request) | ||
|
|
||
| return wrapped(*argv, **kwargs) | ||
| except Exception: | ||
| if span is not None and span.is_recording(): | ||
| span.end() | ||
| logger.debug("twisted server render_with_instana", exc_info=True) | ||
| finally: | ||
| if token is not None: | ||
| context.detach(token) | ||
|
|
||
| return wrapped(*argv, **kwargs) | ||
|
|
||
| def finish_tracing( | ||
| result: "Optional[Failure]", request: "Request" | ||
| ) -> "Optional[Failure]": | ||
| """Finish tracing when the Twisted request lifecycle completes.""" | ||
| if request._instana_finished: | ||
| return result | ||
|
|
||
| request._instana_finished = True | ||
| span = request._instana | ||
| try: | ||
| status_code = request.code | ||
| if isinstance(status_code, int): | ||
| span.set_attribute(SpanAttributes.HTTP_STATUS_CODE, status_code) | ||
|
|
||
| # 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 isinstance(status_code, int) and status_code >= 500: | ||
| span.mark_as_errored({ | ||
| "http.error": request.code_message.decode("latin-1") | ||
| }) | ||
| except Exception: | ||
| logger.debug("twisted server finish_tracing", exc_info=True) | ||
| finally: | ||
| if span.is_recording(): | ||
| span.end() | ||
|
|
||
| return result | ||
|
|
||
| logger.debug("Instrumenting twisted server") | ||
| except ImportError: | ||
| pass | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.