Move changes to the http package.

Author: JacksonWeber

instrumentation/opentelemetry-instrumentation-requests/src/opentelemetry/instrumentation/requests/__init__.py

@@ -124,10 +124,6 @@ def response_hook(span, request_obj, response):
     _StabilityMode,
 )
 from opentelemetry.instrumentation.instrumentor import BaseInstrumentor
-from opentelemetry.instrumentation.requests.constants import (
-    BOT_PATTERNS,
-    TEST_PATTERNS,
-)
 from opentelemetry.instrumentation.requests.package import _instruments
 from opentelemetry.instrumentation.requests.version import __version__
 from opentelemetry.instrumentation.utils import (
@@ -138,7 +134,6 @@ def response_hook(span, request_obj, response):
 from opentelemetry.propagate import inject
 from opentelemetry.semconv._incubating.attributes.user_agent_attributes import (
     USER_AGENT_SYNTHETIC_TYPE,
-    UserAgentSyntheticTypeValues,
 )
 from opentelemetry.semconv.attributes.error_attributes import ERROR_TYPE
 from opentelemetry.semconv.attributes.network_attributes import (
@@ -153,6 +148,7 @@ def response_hook(span, request_obj, response):
 from opentelemetry.trace.span import Span
 from opentelemetry.util.http import (
     ExcludeList,
+    detect_synthetic_user_agent,
     get_excluded_urls,
     parse_excluded_urls,
     redact_url,
@@ -166,33 +162,6 @@ def response_hook(span, request_obj, response):
 _ResponseHookT = Optional[Callable[[Span, PreparedRequest, Response], None]]
 
 
-def _detect_synthetic_user_agent(user_agent: str) -> Optional[str]:
-    """
-    Detect synthetic user agent type based on user agent string contents.
-
-    Args:
-        user_agent: The user agent string to analyze
-
-    Returns:
-        UserAgentSyntheticTypeValues.TEST if user agent contains any pattern from TEST_PATTERNS
-        UserAgentSyntheticTypeValues.BOT if user agent contains any pattern from BOT_PATTERNS
-        None otherwise
-
-    Note: Test patterns take priority over bot patterns.
-    """
-    if not user_agent:
-        return None
-
-    user_agent_lower = user_agent.lower()
-
-    if any(test_pattern in user_agent_lower for test_pattern in TEST_PATTERNS):
-        return UserAgentSyntheticTypeValues.TEST.value
-    if any(bot_pattern in user_agent_lower for bot_pattern in BOT_PATTERNS):
-        return UserAgentSyntheticTypeValues.BOT.value
-
-    return None
-
-
 def _set_http_status_code_attribute(
     span,
     status_code,
@@ -283,7 +252,7 @@ def get_or_create_headers():
 
         # Check for synthetic user agent type
         user_agent = headers.get("User-Agent")
-        synthetic_type = _detect_synthetic_user_agent(user_agent)
+        synthetic_type = detect_synthetic_user_agent(user_agent)
         if synthetic_type:
             span_attributes[USER_AGENT_SYNTHETIC_TYPE] = synthetic_type
 

util/opentelemetry-util-http/src/opentelemetry/util/http/__init__.py

@@ -19,7 +19,7 @@
 from re import IGNORECASE as RE_IGNORECASE
 from re import compile as re_compile
 from re import search
-from typing import Callable, Iterable, overload
+from typing import Callable, Iterable, Optional, overload
 from urllib.parse import parse_qs, urlencode, urlparse, urlunparse
 
 from opentelemetry.semconv._incubating.attributes.http_attributes import (
@@ -34,6 +34,10 @@
     NET_HOST_NAME,
     NET_HOST_PORT,
 )
+from opentelemetry.semconv._incubating.attributes.user_agent_attributes import (
+    UserAgentSyntheticTypeValues,
+)
+from opentelemetry.util.http.constants import BOT_PATTERNS, TEST_PATTERNS
 
 OTEL_INSTRUMENTATION_HTTP_CAPTURE_HEADERS_SANITIZE_FIELDS = (
     "OTEL_INSTRUMENTATION_HTTP_CAPTURE_HEADERS_SANITIZE_FIELDS"
@@ -301,3 +305,30 @@ def redact_url(url: str) -> str:
     url = remove_url_credentials(url)
     url = redact_query_parameters(url)
     return url
+
+
+def detect_synthetic_user_agent(user_agent: Optional[str]) -> Optional[str]:
+    """
+    Detect synthetic user agent type based on user agent string contents.
+
+    Args:
+        user_agent: The user agent string to analyze
+
+    Returns:
+        UserAgentSyntheticTypeValues.TEST if user agent contains any pattern from TEST_PATTERNS
+        UserAgentSyntheticTypeValues.BOT if user agent contains any pattern from BOT_PATTERNS
+        None otherwise
+
+    Note: Test patterns take priority over bot patterns.
+    """
+    if not user_agent:
+        return None
+
+    user_agent_lower = user_agent.lower()
+
+    if any(test_pattern in user_agent_lower for test_pattern in TEST_PATTERNS):
+        return UserAgentSyntheticTypeValues.TEST.value
+    if any(bot_pattern in user_agent_lower for bot_pattern in BOT_PATTERNS):
+        return UserAgentSyntheticTypeValues.BOT.value
+
+    return None

instrumentation/opentelemetry-instrumentation-requests/src/opentelemetry/instrumentation/requests/constants.py renamed to util/opentelemetry-util-http/src/opentelemetry/util/http/constants.py

@@ -13,10 +13,10 @@
 # limitations under the License.
 
 """
-Constants for OpenTelemetry requests instrumentation.
+Constants for OpenTelemetry HTTP utilities.
 
 This module contains configuration constants and pattern definitions used
-by the requests instrumentation for various features like synthetic user
+by HTTP instrumentation utilities for various features like synthetic user
 agent detection.
 """
 

util/opentelemetry-util-http/tests/test_detect_synthetic_user_agent.py

@@ -0,0 +1,92 @@
+# Copyright The OpenTelemetry Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import unittest
+
+from opentelemetry.semconv._incubating.attributes.user_agent_attributes import (
+    UserAgentSyntheticTypeValues,
+)
+from opentelemetry.util.http import detect_synthetic_user_agent
+
+
+class TestDetectSyntheticUserAgent(unittest.TestCase):
+    def test_detect_bot_googlebot(self):
+        """Test detection of googlebot user agent."""
+        user_agent = "Mozilla/5.0 (compatible; Googlebot/2.1; +http://www.google.com/bot.html)"
+        result = detect_synthetic_user_agent(user_agent)
+        self.assertEqual(result, UserAgentSyntheticTypeValues.BOT.value)
+
+    def test_detect_bot_bingbot(self):
+        """Test detection of bingbot user agent."""
+        user_agent = "Mozilla/5.0 (compatible; bingbot/2.0; +http://www.bing.com/bingbot.htm)"
+        result = detect_synthetic_user_agent(user_agent)
+        self.assertEqual(result, UserAgentSyntheticTypeValues.BOT.value)
+
+    def test_detect_test_alwayson(self):
+        """Test detection of alwayson test user agent."""
+        user_agent = "AlwaysOn-Monitor/1.0"
+        result = detect_synthetic_user_agent(user_agent)
+        self.assertEqual(result, UserAgentSyntheticTypeValues.TEST.value)
+
+    def test_case_insensitive_detection(self):
+        """Test that detection is case insensitive."""
+        # Test uppercase patterns
+        user_agent_bot = "GOOGLEBOT/2.1"
+        result = detect_synthetic_user_agent(user_agent_bot)
+        self.assertEqual(result, UserAgentSyntheticTypeValues.BOT.value)
+
+        user_agent_test = "ALWAYSON-Monitor/1.0"
+        result = detect_synthetic_user_agent(user_agent_test)
+        self.assertEqual(result, UserAgentSyntheticTypeValues.TEST.value)
+
+    def test_normal_user_agent_not_detected(self):
+        """Test that normal browser user agents are not detected as synthetic."""
+        user_agent = (
+            "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
+        )
+        result = detect_synthetic_user_agent(user_agent)
+        self.assertIsNone(result)
+
+    def test_none_user_agent(self):
+        """Test that None user agent returns None."""
+        result = detect_synthetic_user_agent(None)
+        self.assertIsNone(result)
+
+    def test_empty_user_agent(self):
+        """Test that empty user agent returns None."""
+        result = detect_synthetic_user_agent("")
+        self.assertIsNone(result)
+
+    def test_substring_match(self):
+        """Test that substrings are detected correctly."""
+        # Test googlebot in middle of string
+        user_agent = "MyApp/1.0 googlebot crawler"
+        result = detect_synthetic_user_agent(user_agent)
+        self.assertEqual(result, UserAgentSyntheticTypeValues.BOT.value)
+
+        # Test alwayson in middle of string
+        user_agent = "TestFramework/1.0 alwayson monitoring"
+        result = detect_synthetic_user_agent(user_agent)
+        self.assertEqual(result, UserAgentSyntheticTypeValues.TEST.value)
+
+    def test_priority_test_over_bot(self):
+        """Test that test patterns take priority over bot patterns."""
+        user_agent = "alwayson-googlebot/1.0"
+        result = detect_synthetic_user_agent(user_agent)
+        # alwayson should be checked first and return 'test'
+        self.assertEqual(result, UserAgentSyntheticTypeValues.TEST.value)
+
+
+if __name__ == "__main__":
+    unittest.main()