feat(tracing): resource renaming

Author: florentinl

ddtrace/_trace/processor/__init__.py

@@ -2,6 +2,7 @@
 from collections import defaultdict
 from itertools import chain
 import logging
+import re
 from threading import RLock
 from typing import Any
 from typing import DefaultDict
@@ -15,6 +16,7 @@
 from ddtrace._trace.span import Span
 from ddtrace._trace.span import _get_64_highest_order_bits_as_hex
 from ddtrace.constants import _APM_ENABLED_METRIC_KEY as MK_APM_ENABLED
+from ddtrace.ext import http
 from ddtrace.internal import gitmetadata
 from ddtrace.internal import telemetry
 from ddtrace.internal.constants import COMPONENT
@@ -497,3 +499,68 @@ def reset(
                 "spans_created": defaultdict(int),
                 "spans_finished": defaultdict(int),
             }
+
+
+class ResourceRenamingProcessor(SpanProcessor):
+    def __init__(self):
+        self._URL_PATH_EXTRACTION_RE = re.compile(
+            r"^(?P<protocol>[a-z]+://(?P<host>[^?/]+))?(?P<path>/[^?]*)(?P<query>(\?).*)?$"
+        )
+
+        self._INT_RE = re.compile(r"^[1-9][0-9]+$")
+        self._INT_ID_RE = re.compile(r"^[0-9._-]{3,}$")
+        self._HEX_RE = re.compile(r"^[A-Fa-f0-9]{6,}$")
+        self._HEX_ID_RE = re.compile(r"^[A-Fa-f0-9._-]{6,}$")
+        self._STR_RE = re.compile(r"^(.{20,}|.*[%&'()*+,:=@].*)$")
+
+    def _compute_simplified_endpoint_path_element(self, elem: str) -> str:
+        """Applies the parameter replacement rules to a single path element."""
+        if self._INT_RE.fullmatch(elem):
+            return "{param:int}"
+        if self._INT_ID_RE.fullmatch(elem):
+            return "{param:int_id}"
+        if self._HEX_RE.fullmatch(elem):
+            return "{param:hex}"
+        if self._HEX_ID_RE.fullmatch(elem):
+            return "{param:hex_id}"
+        if self._STR_RE.fullmatch(elem):
+            return "{param:str}"
+        return elem
+
+    def _compute_simplified_endpoint(self, url: Optional[str]) -> str:
+        """Extracts and simplifies the path from an HTTP URL according to the RFC."""
+        if not url:
+            return "/"
+
+        match = self._URL_PATH_EXTRACTION_RE.match(url)
+        if not match:
+            return ""
+        path = match.group("path")
+        if not path or path == "/":
+            return "/"
+
+        elements: List[str] = []
+        for part in path.split("/"):
+            if part:
+                elements.append(part)
+                if len(elements) >= 8:
+                    break
+
+        if not elements:
+            return "/"
+
+        elements = [self._compute_simplified_endpoint_path_element(elem) for elem in elements]
+        return "/" + "/".join(elements)
+
+    def on_span_start(self, span):
+        pass
+
+    def on_span_finish(self, span):
+        route = span.get_tag(http.ROUTE)
+
+        if not route or config._trace_resource_renaming_always_simplified_endpoint:
+            url = span.get_tag(http.URL)
+            endpoint = self._compute_simplified_endpoint(url)
+            span.set_tag_str(http.ENDPOINT, endpoint)
+        else:
+            span.set_tag_str(http.ENDPOINT, endpoint)

ddtrace/_trace/tracer.py

@@ -16,6 +16,7 @@
 from typing import cast
 
 from ddtrace._trace.context import Context
+from ddtrace._trace.processor import ResourceRenamingProcessor
 from ddtrace._trace.processor import SpanAggregator
 from ddtrace._trace.processor import SpanProcessor
 from ddtrace._trace.processor import TopLevelSpanProcessor
@@ -84,6 +85,9 @@ def _default_span_processors_factory(
             SpanStatsProcessorV06(),
         )
 
+    if config._trace_resource_renaming_enabled:
+        span_processors.append(ResourceRenamingProcessor())
+
     span_processors.append(profiling_span_processor)
 
     return span_processors

ddtrace/ext/http.py

@@ -18,6 +18,7 @@
 CLIENT_IP = "http.client_ip"
 ROUTE = "http.route"
 REFERRER_HOSTNAME = "http.referrer_hostname"
+ENDPOINT = "http.endpoint"
 
 # HTTP headers
 REFERER_HEADER = "referer"

ddtrace/internal/processor/stats.py

@@ -48,6 +48,8 @@ def _is_measured(span: Span) -> bool:
     str,  # type
     int,  # http status code
     bool,  # synthetics request
+    str,  # http method
+    str,  # http endpoint
 ]
 
 
@@ -73,8 +75,10 @@ def _span_aggr_key(span: Span) -> SpanAggrKey:
     resource = span.resource or ""
     _type = span.span_type or ""
     status_code = span.get_tag("http.status_code") or 0
+    method = span.get_tag("http.method") or ""
+    endpoint = span.get_tag("http.endpoint") or ""
     synthetics = span.context.dd_origin == "synthetics"
-    return span.name, service, resource, _type, int(status_code), synthetics
+    return (span.name, service, resource, _type, int(status_code), synthetics, method, endpoint)
 
 
 class SpanStatsProcessorV06(PeriodicService, SpanProcessor):
@@ -157,12 +161,14 @@ def _serialize_buckets(self) -> List[Dict]:
             serialized_bucket_keys.append(bucket_time_ns)
 
             for aggr_key, stat_aggr in bucket.items():
-                name, service, resource, _type, http_status, synthetics = aggr_key
+                name, service, resource, _type, http_status, synthetics, http_method, http_endpoint = aggr_key
                 serialized_bucket = {
                     "Name": compat.ensure_text(name),
                     "Resource": compat.ensure_text(resource),
                     "Synthetics": synthetics,
                     "HTTPStatusCode": http_status,
+                    "HTTPMethod": http_method,
+                    "HTTPEndpoint": http_endpoint,
                     "Hits": stat_aggr.hits,
                     "TopLevelHits": stat_aggr.top_level_hits,
                     "Duration": stat_aggr.duration,

ddtrace/settings/_config.py

@@ -661,6 +661,14 @@ def __init__(self):
         self._inferred_proxy_services_enabled = _get_config("DD_TRACE_INFERRED_PROXY_SERVICES_ENABLED", False, asbool)
         self._trace_safe_instrumentation_enabled = _get_config("DD_TRACE_SAFE_INSTRUMENTATION_ENABLED", False, asbool)
 
+        # Resource renaming
+        self._trace_resource_renaming_enabled = _get_config(
+            "DD_TRACE_RESOURCE_RENAMING", default=False, modifier=asbool
+        )
+        self._trace_resource_renaming_always_simplified_endpoint = _get_config(
+            "DD_TRACE_RESOURCE_RENAMING_ALWAYS_SIMPLIFIED_ENDPOINT", default=False, modifier=asbool
+        )
+
     def __getattr__(self, name) -> Any:
         if name in self._config:
             return self._config[name].value()