feat: Add werkzeug instrumentation and unittests

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

Author: CagriYonca

src/instana/__init__.py

@@ -184,6 +184,7 @@ def boot_agent() -> None:
             sqlalchemy,  # noqa: F401
             starlette,  # noqa: F401
             urllib3,  # noqa: F401
+            werkzeug,  # noqa: F401
             gevent,  # noqa: F401
         )
         from instana.instrumentation.aiohttp import (

src/instana/instrumentation/werkzeug.py

@@ -0,0 +1,275 @@
+# (c) Copyright IBM Corp. 2026
+
+"""
+Instana Werkzeug Instrumentation
+
+This module provides automatic instrumentation for Werkzeug-based applications.
+Werkzeug is a comprehensive WSGI web application library used by Flask and other frameworks.
+
+This module automatically patches Werkzeug applications when imported via autowrapt.
+"""
+
+try:
+    from typing import (
+        TYPE_CHECKING,
+        Any,
+        Callable,
+        Iterable,
+        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 instana.span.span import InstanaSpan
+
+    def _create_span_with_context(environ: dict[str, Any]) -> tuple["InstanaSpan", Any]:
+        """
+        Create and configure a span with context for the request.
+
+        Args:
+            environ: WSGI environment dictionary
+
+        Returns:
+            Tuple of (span, context_token)
+        """
+        tracer = get_tracer()
+        parent_context = tracer.extract(Format.HTTP_HEADERS, environ)
+        span = tracer.start_span("wsgi", context=parent_context)
+
+        ctx = trace.set_span_in_context(span)
+        token = context.attach(ctx)
+
+        extract_custom_headers(span, environ, format=True)
+        _set_request_attributes(span, environ)
+
+        return span, token
+
+    def _build_start_response(
+        span: "InstanaSpan",
+        start_response: Callable,
+    ) -> Callable:
+        """Create an instrumented start_response callable."""
+
+        def new_start_response(
+            status: str,
+            headers: list[tuple[str, str]],
+            exc_info: Optional[tuple[Any, Any, Any]] = None,
+        ) -> Callable:
+            """Modified start_response with trace context injection."""
+            try:
+                extract_custom_headers(span, headers)
+                tracer = get_tracer()
+                tracer.inject(
+                    span.context,
+                    Format.HTTP_HEADERS,
+                    headers,
+                )
+
+                status_code = _parse_status_code(status)
+                if status_code is not None:
+                    if status_code >= 500:
+                        span.mark_as_errored()
+                    span.set_attribute(
+                        SpanAttributes.HTTP_STATUS_CODE,
+                        status_code,
+                    )
+
+                return start_response(
+                    status,
+                    _normalize_headers(headers),
+                    exc_info,
+                )
+            except Exception:
+                logger.debug("Error in Werkzeug start_response wrapper", exc_info=True)
+                return start_response(status, headers, exc_info)
+
+        return new_start_response
+
+    def _normalize_headers(
+        headers: list[tuple[str, Any]],
+    ) -> list[tuple[str, str]]:
+        """
+        Ensure all header values are strings for WSGI compliance.
+
+        Args:
+            headers: List of (name, value) tuples
+
+        Returns:
+            List of (name, str_value) tuples
+        """
+        return [
+            (name, value if isinstance(value, str) else str(value))
+            for name, value in headers
+        ]
+
+    def _parse_status_code(status: str) -> Optional[int]:
+        """Safely parse the HTTP status code from a WSGI status string."""
+        try:
+            return int(status.split()[0])
+        except (AttributeError, IndexError, TypeError, ValueError):
+            return None
+
+    def _end_span_after_iterating(
+        iterable: Iterable[bytes],
+        span: "InstanaSpan",
+        token: Any,
+    ) -> Iterable[bytes]:
+        """
+        Generator that yields from the iterable and ensures span cleanup.
+
+        Args:
+            iterable: The response iterable from the application
+            span: The active span
+            token: The context token
+
+        Yields:
+            Response chunks from the iterable
+        """
+        try:
+            yield from iterable
+        finally:
+            # Ensure iterable cleanup (important for generators)
+            if hasattr(iterable, "close"):
+                try:
+                    iterable.close()  # type: ignore
+                except Exception:
+                    logger.debug("Error closing iterable", exc_info=True)
+
+            # End span and detach token after iteration completes
+            if span and span.is_recording():
+                span.end()
+            if token:
+                context.detach(token)  # type: ignore
+
+    def _scrub_query_params(query_string: str) -> Optional[str]:
+        """
+        Scrub secrets from query string parameters.
+
+        Args:
+            query_string: The query string to scrub
+
+        Returns:
+            Scrubbed query string or None if agent not available
+        """
+        if agent is not None:
+            return strip_secrets_from_query(
+                query_string,
+                agent.options.secrets_matcher,  # type: ignore
+                agent.options.secrets_list,  # type: ignore
+            )
+        return None
+
+    def _set_request_attributes(span: "InstanaSpan", environ: dict[str, Any]) -> None:
+        """
+        Extract and set HTTP attributes from the WSGI environ.
+
+        Args:
+            span: The active span
+            environ: WSGI environment dictionary
+        """
+        try:
+            # Set HTTP method
+            if "REQUEST_METHOD" in environ:
+                span.set_attribute(
+                    SpanAttributes.HTTP_METHOD, environ["REQUEST_METHOD"]
+                )
+
+            # Set HTTP path
+            if "PATH_INFO" in environ:
+                span.set_attribute("http.path", environ["PATH_INFO"])
+
+            # Set HTTP query parameters (with secrets scrubbed)
+            if environ.get("QUERY_STRING"):
+                scrubbed_params = _scrub_query_params(environ["QUERY_STRING"])
+                if scrubbed_params is not None:
+                    span.set_attribute("http.params", scrubbed_params)
+
+            # Set HTTP host
+            if "HTTP_HOST" in environ:
+                span.set_attribute(
+                    SpanAttributes.HTTP_HOST,
+                    environ["HTTP_HOST"],
+                )
+
+            # Set HTTP URL (without query string to avoid exposing secrets)
+            if "wsgi.url_scheme" in environ:
+                scheme = environ["wsgi.url_scheme"]
+                host = environ.get("HTTP_HOST", "")
+                script_name = environ.get("SCRIPT_NAME", "")
+                path = environ.get("PATH_INFO", "")
+
+                url = f"{scheme}://{host}{script_name}{path}"
+                span.set_attribute(SpanAttributes.HTTP_URL, url)
+
+        except Exception:
+            logger.debug("Error setting request attributes", exc_info=True)
+
+    # Autowrapt patching for automatic instrumentation
+    class _TracedWSGIApp:
+        """Wrapper that traces WSGI applications."""
+
+        def __init__(self, app: Callable) -> None:
+            self.app = app
+
+        def __call__(self, environ: dict[str, Any], start_response: Callable) -> Any:
+            try:
+                span, token = _create_span_with_context(environ)
+                wrapped_start_response = _build_start_response(span, start_response)
+            except Exception:
+                logger.debug("werkzeug setup failed", exc_info=True)
+                return self.app(environ, start_response)
+
+            try:
+                iterable = self.app(environ, wrapped_start_response)
+                return _end_span_after_iterating(iterable, span, token)
+            except Exception as exc:
+                try:
+                    if span and span.is_recording():
+                        span.record_exception(exc)
+                        span.end()
+                    if token:
+                        context.detach(token)
+                except Exception:
+                    logger.debug("werkzeug cleanup failed", exc_info=True)
+                raise
+
+    @wrapt.patch_function_wrapper("werkzeug.serving", "run_simple")
+    def run_simple_with_instana(
+        wrapped: Callable,
+        instance: Any,
+        args: tuple,
+        kwargs: dict[str, Any],
+    ) -> Any:
+        """Patch werkzeug.serving.run_simple to wrap WSGI applications."""
+        try:
+            # run_simple(hostname, port, application, ...)
+            if len(args) >= 3:
+                hostname, port, application = args[0], args[1], args[2]
+                instrumented_app = _TracedWSGIApp(application)
+                logger.debug(f"Werkzeug app wrapped: {hostname}:{port}")
+                args = (hostname, port, instrumented_app) + args[3:]
+            elif "application" in kwargs:
+                application = kwargs["application"]
+                instrumented_app = _TracedWSGIApp(application)
+                kwargs["application"] = instrumented_app
+                logger.debug("Werkzeug app wrapped (kwargs)")
+        except Exception:
+            logger.debug("Failed to wrap Werkzeug app", exc_info=True)
+
+        return wrapped(*args, **kwargs)
+
+    logger.debug("Instrumenting werkzeug")
+
+except ImportError:
+    pass
+
+# Made with Bob