remote sampling - full implementation

Author: jj22ee

.github/component_owners.yml

@@ -36,6 +36,7 @@ components:
     - NathanielRN
     - Kausik-A
     - srprash
+    - jj22ee
 
   instrumentation/opentelemetry-instrumentation-tortoiseorm:
     - tonybaloney

sdk-extension/opentelemetry-sdk-extension-aws/README.rst

@@ -74,6 +74,30 @@ populate `resource` attributes by creating a `TraceProvider` using the `AwsEc2Re
 Refer to each detectors' docstring to determine any possible requirements for that
 detector.
 
+
+Usage (AWS X-Ray Remote Sampler)
+----------------------------
+
+
+Use the provided AWS X-Ray Remote Sampler by setting this sampler in your instrumented application:
+
+.. code-block:: python
+
+    from opentelemetry.sdk.extension.aws.trace.sampler import AwsXRayRemoteSampler
+    from opentelemetry import trace
+    from opentelemetry.sdk.resources import Resource
+    from opentelemetry.sdk.trace import TracerProvider
+    from opentelemetry.semconv.resource import ResourceAttributes
+    from opentelemetry.util.types import Attributes
+
+    resource = Resource.create(attributes={
+        ResourceAttributes.SERVICE_NAME: "myService",
+        ResourceAttributes.CLOUD_PLATFORM: "aws_ec2",
+    })
+    xraySampler = AwsXRayRemoteSampler(resource=resource, polling_interval=300)
+    trace.set_tracer_provider(TracerProvider(sampler=xraySampler))
+
+
 References
 ----------
 

sdk-extension/opentelemetry-sdk-extension-aws/pyproject.toml

@@ -26,7 +26,10 @@ classifiers = [
   "Programming Language :: Python :: 3.13",
 ]
 dependencies = [
-  "opentelemetry-sdk ~= 1.12",
+  "opentelemetry-api ~= 1.23",
+  "opentelemetry-sdk ~= 1.23",
+  "opentelemetry-instrumentation ~= 0.44b0",
+  "opentelemetry-semantic-conventions ~= 0.44b0",
 ]
 
 [project.entry-points.opentelemetry_id_generator]
@@ -39,6 +42,9 @@ aws_eks = "opentelemetry.sdk.extension.aws.resource.eks:AwsEksResourceDetector"
 aws_elastic_beanstalk = "opentelemetry.sdk.extension.aws.resource.beanstalk:AwsBeanstalkResourceDetector"
 aws_lambda = "opentelemetry.sdk.extension.aws.resource._lambda:AwsLambdaResourceDetector"
 
+[project.entry-points.opentelemetry_sampler]
+aws_xray_remote_sampler = "opentelemetry.sdk.extension.aws.trace.sampler:AwsXRayRemoteSampler"
+
 [project.urls]
 Homepage = "https://github.com/open-telemetry/opentelemetry-python-contrib/tree/main/sdk-extension/opentelemetry-sdk-extension-aws"
 Repository = "https://github.com/open-telemetry/opentelemetry-python-contrib"

sdk-extension/opentelemetry-sdk-extension-aws/src/opentelemetry/sdk/extension/aws/trace/sampler/__init__.py

@@ -0,0 +1,20 @@
+# 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.
+
+# pylint: disable=no-name-in-module
+from opentelemetry.sdk.extension.aws.trace.sampler.aws_xray_remote_sampler import (
+    AwsXRayRemoteSampler,
+)
+
+__all__ = ["AwsXRayRemoteSampler"]

sdk-extension/opentelemetry-sdk-extension-aws/src/opentelemetry/sdk/extension/aws/trace/sampler/_aws_xray_sampling_client.py

@@ -0,0 +1,141 @@
+# 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.
+
+# Includes work from:
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+import json
+from logging import getLogger
+from typing import List, Optional
+
+import requests
+
+# pylint: disable=no-name-in-module
+from opentelemetry.instrumentation.utils import suppress_instrumentation
+from opentelemetry.sdk.extension.aws.trace.sampler._sampling_rule import (
+    _SamplingRule,
+)
+from opentelemetry.sdk.extension.aws.trace.sampler._sampling_target import (
+    _SamplingTargetResponse,
+)
+
+_logger = getLogger(__name__)
+DEFAULT_SAMPLING_PROXY_ENDPOINT = "http://127.0.0.1:2000"
+
+
+class _AwsXRaySamplingClient:
+    def __init__(
+        self,
+        endpoint: str = DEFAULT_SAMPLING_PROXY_ENDPOINT,
+        log_level: Optional[str] = None,
+    ):
+        # Override default log level
+        if log_level is not None:
+            _logger.setLevel(log_level)
+
+        self.__get_sampling_rules_endpoint = endpoint + "/GetSamplingRules"
+        self.__get_sampling_targets_endpoint = endpoint + "/SamplingTargets"
+
+        self.__session = requests.Session()
+
+    def get_sampling_rules(self) -> List[_SamplingRule]:
+        sampling_rules: List["_SamplingRule"] = []
+        headers = {"content-type": "application/json"}
+
+        with suppress_instrumentation():
+            try:
+                xray_response = self.__session.post(
+                    url=self.__get_sampling_rules_endpoint,
+                    headers=headers,
+                    timeout=20,
+                )
+                sampling_rules_response = xray_response.json()
+                if (
+                    sampling_rules_response is None
+                    or "SamplingRuleRecords" not in sampling_rules_response
+                ):
+                    _logger.error(
+                        "SamplingRuleRecords is missing in getSamplingRules response: %s",
+                        sampling_rules_response,
+                    )
+                    return []
+                sampling_rules_records = sampling_rules_response[
+                    "SamplingRuleRecords"
+                ]
+                for record in sampling_rules_records:
+                    if "SamplingRule" not in record:
+                        _logger.error(
+                            "SamplingRule is missing in SamplingRuleRecord"
+                        )
+                    else:
+                        sampling_rules.append(
+                            _SamplingRule(**record["SamplingRule"])
+                        )
+
+            except requests.exceptions.RequestException as req_err:
+                _logger.error("Request error occurred: %s", req_err)
+            except json.JSONDecodeError as json_err:
+                _logger.error("Error in decoding JSON response: %s", json_err)
+            # pylint: disable=broad-exception-caught
+            except Exception as err:
+                _logger.error(
+                    "Error occurred when attempting to fetch rules: %s", err
+                )
+
+            return sampling_rules
+
+    def get_sampling_targets(
+        self, statistics: List["dict[str, str | float | int]"]
+    ) -> _SamplingTargetResponse:
+        sampling_targets_response = _SamplingTargetResponse(
+            LastRuleModification=None,
+            SamplingTargetDocuments=None,
+            UnprocessedStatistics=None,
+        )
+        headers = {"content-type": "application/json"}
+
+        with suppress_instrumentation():
+            try:
+                xray_response = self.__session.post(
+                    url=self.__get_sampling_targets_endpoint,
+                    headers=headers,
+                    timeout=20,
+                    json={"SamplingStatisticsDocuments": statistics},
+                )
+                xray_response_json = xray_response.json()
+                if (
+                    xray_response_json is None
+                    or "SamplingTargetDocuments" not in xray_response_json
+                    or "LastRuleModification" not in xray_response_json
+                ):
+                    _logger.debug(
+                        "getSamplingTargets response is invalid. Unable to update targets."
+                    )
+                    return sampling_targets_response
+
+                sampling_targets_response = _SamplingTargetResponse(
+                    **xray_response_json
+                )
+            except requests.exceptions.RequestException as req_err:
+                _logger.debug("Request error occurred: %s", req_err)
+            except json.JSONDecodeError as json_err:
+                _logger.debug("Error in decoding JSON response: %s", json_err)
+            # pylint: disable=broad-exception-caught
+            except Exception as err:
+                _logger.debug(
+                    "Error occurred when attempting to fetch targets: %s", err
+                )
+
+            return sampling_targets_response

sdk-extension/opentelemetry-sdk-extension-aws/src/opentelemetry/sdk/extension/aws/trace/sampler/_clock.py

@@ -0,0 +1,37 @@
+# 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.
+
+# Includes work from:
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+import datetime
+
+
+class _Clock:
+    def __init__(self):
+        self.__datetime = datetime.datetime
+
+    def now(self) -> datetime.datetime:
+        return self.__datetime.now()
+
+    # pylint: disable=no-self-use
+    def from_timestamp(self, timestamp: float) -> datetime.datetime:
+        return datetime.datetime.fromtimestamp(timestamp)
+
+    def time_delta(self, seconds: float) -> datetime.timedelta:
+        return datetime.timedelta(seconds=seconds)
+
+    def max(self) -> datetime.datetime:
+        return datetime.datetime.max

sdk-extension/opentelemetry-sdk-extension-aws/src/opentelemetry/sdk/extension/aws/trace/sampler/_fallback_sampler.py

@@ -0,0 +1,78 @@
+# 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.
+
+# Includes work from:
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+from typing import Optional, Sequence
+
+# pylint: disable=no-name-in-module
+from opentelemetry.context import Context
+from opentelemetry.sdk.extension.aws.trace.sampler._clock import _Clock
+from opentelemetry.sdk.extension.aws.trace.sampler._rate_limiting_sampler import (
+    _RateLimitingSampler,
+)
+from opentelemetry.sdk.trace.sampling import (
+    Decision,
+    Sampler,
+    SamplingResult,
+    TraceIdRatioBased,
+)
+from opentelemetry.trace import Link, SpanKind
+from opentelemetry.trace.span import TraceState
+from opentelemetry.util.types import Attributes
+
+
+class _FallbackSampler(Sampler):
+    def __init__(self, clock: _Clock):
+        self.__rate_limiting_sampler = _RateLimitingSampler(1, clock)
+        self.__fixed_rate_sampler = TraceIdRatioBased(0.05)
+
+    # pylint: disable=no-self-use
+    def should_sample(
+        self,
+        parent_context: Optional["Context"],
+        trace_id: int,
+        name: str,
+        kind: Optional[SpanKind] = None,
+        attributes: Optional[Attributes] = None,
+        links: Optional[Sequence["Link"]] = None,
+        trace_state: Optional["TraceState"] = None,
+    ) -> "SamplingResult":
+        sampling_result = self.__rate_limiting_sampler.should_sample(
+            parent_context,
+            trace_id,
+            name,
+            kind=kind,
+            attributes=attributes,
+            links=links,
+            trace_state=trace_state,
+        )
+        if sampling_result.decision is not Decision.DROP:
+            return sampling_result
+        return self.__fixed_rate_sampler.should_sample(
+            parent_context,
+            trace_id,
+            name,
+            kind=kind,
+            attributes=attributes,
+            links=links,
+            trace_state=trace_state,
+        )
+
+    # pylint: disable=no-self-use
+    def get_description(self) -> str:
+        description = "FallbackSampler{fallback sampling with sampling config of 1 req/sec and 5% of additional requests}"
+        return description