addressing comments

Author: jj22ee

aws-opentelemetry-distro/src/amazon/opentelemetry/distro/sampler/_rule_cache.py

@@ -1,14 +1,13 @@
 # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 # SPDX-License-Identifier: Apache-2.0
-import copy
 import datetime
 from logging import getLogger
 from threading import Lock
 from typing import Optional, Sequence
 
 from amazon.opentelemetry.distro.sampler._fallback_sampler import _FallbackSampler
-from amazon.opentelemetry.distro.sampler._rule import _Rule
 from amazon.opentelemetry.distro.sampler._sampling_rule import _SamplingRule
+from amazon.opentelemetry.distro.sampler._sampling_rule_applier import _SamplingRuleApplier
 from opentelemetry.context import Context
 from opentelemetry.sdk.resources import Resource
 from opentelemetry.sdk.trace.sampling import SamplingResult
@@ -22,9 +21,8 @@
 
 
 class _RuleCache:
-    rules: [_Rule] = []
-
     def __init__(self, resource: Resource, fallback_sampler: _FallbackSampler, date_time: datetime, lock: Lock):
+        self.__rule_appliers: [_SamplingRuleApplier] = []
         self.__cache_lock = lock
         self.__resource = resource
         self._fallback_sampler = fallback_sampler
@@ -41,9 +39,9 @@ def should_sample(
         links: Sequence[Link] = None,
         trace_state: TraceState = None,
     ) -> SamplingResult:
-        for rule in self.rules:
-            if rule.matches(self.__resource, attributes):
-                return rule.should_sample(
+        for rule_applier in self.__rule_appliers:
+            if rule_applier.matches(self.__resource, attributes):
+                return rule_applier.should_sample(
                     parent_context,
                     trace_id,
                     name,
@@ -59,29 +57,29 @@ def should_sample(
 
     def update_sampling_rules(self, new_sampling_rules: [_SamplingRule]) -> None:
         new_sampling_rules.sort()
-        temp_rules = []
+        temp_rule_appliers = []
         for sampling_rule in new_sampling_rules:
             if sampling_rule.RuleName == "":
-                _logger.info("sampling rule without rule name is not supported")
+                _logger.debug("sampling rule without rule name is not supported")
                 continue
             if sampling_rule.Version != 1:
-                _logger.info("sampling rule without Version 1 is not supported: RuleName: %s", sampling_rule.RuleName)
+                _logger.debug("sampling rule without Version 1 is not supported: RuleName: %s", sampling_rule.RuleName)
                 continue
-            temp_rules.append(_Rule(copy.deepcopy(sampling_rule)))
+            temp_rule_appliers.append(_SamplingRuleApplier(sampling_rule))
 
         self.__cache_lock.acquire()
 
-        # map list of rules by each rule's sampling_rule name
-        rule_map = {rule.sampling_rule.RuleName: rule for rule in self.rules}
+        # map list of rule appliers by each applier's sampling_rule name
+        rule_applier_map = {rule.sampling_rule.RuleName: rule for rule in self.__rule_appliers}
 
-        # If a sampling rule has not changed, keep its respective rule in the cache.
-        for index, new_rule in enumerate(temp_rules):
-            rule_name_to_check = new_rule.sampling_rule.RuleName
-            if rule_name_to_check in rule_map:
-                previous_rule = rule_map[rule_name_to_check]
-                if new_rule.sampling_rule == previous_rule.sampling_rule:
-                    temp_rules[index] = previous_rule
-        self.rules = temp_rules
+        # If a sampling rule has not changed, keep its respective applier in the cache.
+        for index, new_applier in enumerate(temp_rule_appliers):
+            rule_name_to_check = new_applier.sampling_rule.RuleName
+            if rule_name_to_check in rule_applier_map:
+                old_applier = rule_applier_map[rule_name_to_check]
+                if new_applier.sampling_rule == old_applier.sampling_rule:
+                    temp_rule_appliers[index] = old_applier
+        self.__rule_appliers = temp_rule_appliers
         self._last_modified = datetime.datetime.now()
 
         self.__cache_lock.release()

aws-opentelemetry-distro/src/amazon/opentelemetry/distro/sampler/_rule.py renamed to aws-opentelemetry-distro/src/amazon/opentelemetry/distro/sampler/_sampling_rule_applier.py

@@ -1,20 +1,21 @@
 # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 # SPDX-License-Identifier: Apache-2.0
 from typing import Optional, Sequence
+from urllib.parse import urlparse
 
 from amazon.opentelemetry.distro.sampler._matcher import _Matcher, cloud_platform_mapping
 from amazon.opentelemetry.distro.sampler._sampling_rule import _SamplingRule
 from opentelemetry.context import Context
 from opentelemetry.sdk.resources import Resource
 from opentelemetry.sdk.trace.sampling import ALWAYS_ON, SamplingResult
-from opentelemetry.semconv.resource import ResourceAttributes
+from opentelemetry.semconv.resource import CloudPlatformValues, ResourceAttributes
 from opentelemetry.semconv.trace import SpanAttributes
 from opentelemetry.trace import Link, SpanKind
 from opentelemetry.trace.span import TraceState
 from opentelemetry.util.types import Attributes
 
 
-class _Rule:
+class _SamplingRuleApplier:
     def __init__(self, sampling_rule: _SamplingRule):
         self.sampling_rule = sampling_rule
         # TODO add self.next_target_fetch_time from maybe time.process_time() or cache's datetime object
@@ -39,48 +40,49 @@ def should_sample(
         )
 
     def matches(self, resource: Resource, attributes: Attributes) -> bool:
-        http_target = None
-        http_url = None
-        http_method = None
-        http_host = None
+        url_path = None
+        url_full = None
+        http_request_method = None
+        server_address = None
         service_name = None
 
         if attributes is not None:
-            http_target = attributes.get(SpanAttributes.HTTP_TARGET, None)
-            http_method = attributes.get(SpanAttributes.HTTP_METHOD, None)
-            http_url = attributes.get(SpanAttributes.HTTP_URL, None)
-            http_host = attributes.get(SpanAttributes.HTTP_HOST, None)
-        # NOTE: The above span attribute keys are deprecated in favor of:
-        # URL_PATH/URL_QUERY, HTTP_REQUEST_METHOD, URL_FULL, SERVER_ADDRESS/SERVER_PORT
-        # For now, the old attribute keys are kept for consistency with other centralized samplers
+            url_path = attributes.get(SpanAttributes.URL_PATH, None)
+            url_full = attributes.get(SpanAttributes.URL_FULL, None)
+            http_request_method = attributes.get(SpanAttributes.HTTP_REQUEST_METHOD, None)
+            server_address = attributes.get(SpanAttributes.SERVER_ADDRESS, None)
 
         # Resource shouldn't be none as it should default to empty resource
         if resource is not None:
             service_name = resource.attributes.get(ResourceAttributes.SERVICE_NAME, "")
 
         # target may be in url
-        if http_target is None and http_url is not None:
-            scheme_end_index = http_url.find("://")
-            # Per spec, http.url is always populated with scheme://host/target. If scheme doesn't
-            # match, assume it's bad instrumentation and ignore.
+        if url_path is None and url_full is not None:
+            scheme_end_index = url_full.find("://")
+            # For network calls, URL usually has `scheme://host[:port][path][?query][#fragment]` format
+            # Per spec, http.url is always populated with scheme://host/target.
+            # If scheme doesn't match, assume it's bad instrumentation and ignore.
             if scheme_end_index > -1:
-                path_index = http_url.find("/", scheme_end_index + len("://"))
-                if path_index == -1:
-                    http_target = "/"
-                else:
-                    http_target = http_url[path_index:]
+                # urlparse("scheme://netloc/path;parameters?query#fragment")
+                url_path = urlparse(url_full).path
+                if url_path == "":
+                    url_path = "/"
+        elif url_path is None and url_full is None:
+            # When missing, the URL Path is assumed to be /
+            url_path = "/"
 
         return (
             _Matcher.attribute_match(attributes, self.sampling_rule.Attributes)
-            and _Matcher.wild_card_match(http_target, self.sampling_rule.URLPath)
-            and _Matcher.wild_card_match(http_method, self.sampling_rule.HTTPMethod)
-            and _Matcher.wild_card_match(http_host, self.sampling_rule.Host)
+            and _Matcher.wild_card_match(url_path, self.sampling_rule.URLPath)
+            and _Matcher.wild_card_match(http_request_method, self.sampling_rule.HTTPMethod)
+            and _Matcher.wild_card_match(server_address, self.sampling_rule.Host)
             and _Matcher.wild_card_match(service_name, self.sampling_rule.ServiceName)
-            and _Matcher.wild_card_match(self.get_service_type(resource), self.sampling_rule.ServiceType)
+            and _Matcher.wild_card_match(self.__get_service_type(resource), self.sampling_rule.ServiceType)
+            and _Matcher.wild_card_match(self.__get_arn(resource, attributes), self.sampling_rule.ResourceARN)
         )
 
     # pylint: disable=no-self-use
-    def get_service_type(self, resource: Resource) -> str:
+    def __get_service_type(self, resource: Resource) -> str:
         if resource is None:
             return ""
 
@@ -89,3 +91,17 @@ def get_service_type(self, resource: Resource) -> str:
             return ""
 
         return cloud_platform_mapping.get(cloud_platform, "")
+
+    # pylint: disable=no-self-use
+    def __get_arn(self, resource: Resource, attributes: Attributes) -> str:
+        if resource is not None:
+            arn = resource.attributes.get(ResourceAttributes.AWS_ECS_CONTAINER_ARN, None)
+            if arn is not None:
+                return arn
+        if attributes is not None and self.__get_service_type(resource=resource) == cloud_platform_mapping.get(
+            CloudPlatformValues.AWS_LAMBDA.value
+        ):
+            arn = attributes.get(SpanAttributes.CLOUD_RESOURCE_ID, None)
+            if arn is not None:
+                return arn
+        return ""

aws-opentelemetry-distro/src/amazon/opentelemetry/distro/sampler/aws_xray_remote_sampler.py

@@ -91,7 +91,7 @@ def should_sample(
     ) -> SamplingResult:
 
         if self.__rule_cache.expired():
-            _logger.info("Rule cache is expired so using fallback sampling strategy")
+            _logger.debug("Rule cache is expired so using fallback sampling strategy")
             return self.__fallback_sampler.should_sample(
                 parent_context, trace_id, name, kind=kind, attributes=attributes, links=links, trace_state=trace_state
             )

aws-opentelemetry-distro/tests/amazon/opentelemetry/distro/sampler/test_rule_cache.py

@@ -12,12 +12,12 @@
 class TestRuleCache(TestCase):
     def test_cache_update_rules_and_sorts_rules(self):
         cache = _RuleCache(None, None, datetime, Lock())
-        self.assertTrue(len(cache.rules) == 0)
+        self.assertTrue(len(cache._RuleCache__rule_appliers) == 0)
 
         rule1 = _SamplingRule(Priority=200, RuleName="only_one_rule", Version=1)
         rules = [rule1]
         cache.update_sampling_rules(rules)
-        self.assertTrue(len(cache.rules) == 1)
+        self.assertTrue(len(cache._RuleCache__rule_appliers) == 1)
 
         rule1 = _SamplingRule(Priority=200, RuleName="abcdef", Version=1)
         rule2 = _SamplingRule(Priority=100, RuleName="abc", Version=1)
@@ -28,13 +28,13 @@ def test_cache_update_rules_and_sorts_rules(self):
         rules = [rule1, rule2, rule3, rule4, rule5, rule6]
         cache.update_sampling_rules(rules)
 
-        self.assertTrue(len(cache.rules) == 6)
-        self.assertEqual(cache.rules[0].sampling_rule.RuleName, "abcdef")
-        self.assertEqual(cache.rules[1].sampling_rule.RuleName, "A")
-        self.assertEqual(cache.rules[2].sampling_rule.RuleName, "Abc")
-        self.assertEqual(cache.rules[3].sampling_rule.RuleName, "ab")
-        self.assertEqual(cache.rules[4].sampling_rule.RuleName, "abc")
-        self.assertEqual(cache.rules[5].sampling_rule.RuleName, "abcdef")
+        self.assertTrue(len(cache._RuleCache__rule_appliers) == 6)
+        self.assertEqual(cache._RuleCache__rule_appliers[0].sampling_rule.RuleName, "abcdef")
+        self.assertEqual(cache._RuleCache__rule_appliers[1].sampling_rule.RuleName, "A")
+        self.assertEqual(cache._RuleCache__rule_appliers[2].sampling_rule.RuleName, "Abc")
+        self.assertEqual(cache._RuleCache__rule_appliers[3].sampling_rule.RuleName, "ab")
+        self.assertEqual(cache._RuleCache__rule_appliers[4].sampling_rule.RuleName, "abc")
+        self.assertEqual(cache._RuleCache__rule_appliers[5].sampling_rule.RuleName, "abcdef")
 
     def test_rule_cache_expiration_logic(self):
         dt = datetime
@@ -54,34 +54,34 @@ def test_update_cache_with_only_one_rule_changed(self):
         rules = [rule1, rule2, rule3]
         cache.update_sampling_rules(rules)
 
-        cache_rules_copy = cache.rules
+        cache_rules_copy = cache._RuleCache__rule_appliers
 
         new_rule3 = _SamplingRule(Priority=5, RuleName="Abc", Version=1)
         rules = [rule1, rule2, new_rule3]
         cache.update_sampling_rules(rules)
 
-        self.assertTrue(len(cache.rules) == 3)
-        self.assertEqual(cache.rules[0].sampling_rule.RuleName, "abcdef")
-        self.assertEqual(cache.rules[1].sampling_rule.RuleName, "Abc")
-        self.assertEqual(cache.rules[2].sampling_rule.RuleName, "ab")
+        self.assertTrue(len(cache._RuleCache__rule_appliers) == 3)
+        self.assertEqual(cache._RuleCache__rule_appliers[0].sampling_rule.RuleName, "abcdef")
+        self.assertEqual(cache._RuleCache__rule_appliers[1].sampling_rule.RuleName, "Abc")
+        self.assertEqual(cache._RuleCache__rule_appliers[2].sampling_rule.RuleName, "ab")
 
         # Compare that only rule1 and rule2 objects have not changed due to new_rule3 even after sorting
-        self.assertTrue(cache_rules_copy[0] is cache.rules[0])
-        self.assertTrue(cache_rules_copy[1] is cache.rules[2])
-        self.assertTrue(cache_rules_copy[2] is not cache.rules[1])
+        self.assertTrue(cache_rules_copy[0] is cache._RuleCache__rule_appliers[0])
+        self.assertTrue(cache_rules_copy[1] is cache._RuleCache__rule_appliers[2])
+        self.assertTrue(cache_rules_copy[2] is not cache._RuleCache__rule_appliers[1])
 
     def test_update_rules_removes_older_rule(self):
         cache = _RuleCache(None, None, datetime, Lock())
-        self.assertTrue(len(cache.rules) == 0)
+        self.assertTrue(len(cache._RuleCache__rule_appliers) == 0)
 
         rule1 = _SamplingRule(Priority=200, RuleName="first_rule", Version=1)
         rules = [rule1]
         cache.update_sampling_rules(rules)
-        self.assertTrue(len(cache.rules) == 1)
-        self.assertEqual(cache.rules[0].sampling_rule.RuleName, "first_rule")
+        self.assertTrue(len(cache._RuleCache__rule_appliers) == 1)
+        self.assertEqual(cache._RuleCache__rule_appliers[0].sampling_rule.RuleName, "first_rule")
 
         rule1 = _SamplingRule(Priority=200, RuleName="second_rule", Version=1)
         rules = [rule1]
         cache.update_sampling_rules(rules)
-        self.assertTrue(len(cache.rules) == 1)
-        self.assertEqual(cache.rules[0].sampling_rule.RuleName, "second_rule")
+        self.assertTrue(len(cache._RuleCache__rule_appliers) == 1)
+        self.assertEqual(cache._RuleCache__rule_appliers[0].sampling_rule.RuleName, "second_rule")