AWS X-Ray Remote Sampler Part 1 - Initial Classes and Rules Poller Implementation

.github/component_owners.yml

@@ -36,9 +36,8 @@ components:
  - NathanielRN
 
  sdk-extension/opentelemetry-sdk-extension-aws:
- - NathanielRN
- - Kausik-A
  - srprash
+ - jj22ee
 
  instrumentation/opentelemetry-instrumentation-tortoiseorm:
  - tonybaloney

CHANGELOG.md

@@ -13,6 +13,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- `opentelemetry-sdk-extension-aws` Add AWS X-Ray Remote Sampler with initial Rules Poller implementation
+ ([#3366](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3366))
+
 ### Fixed
 
 - `opentelemetry-instrumentation` Catch `ModuleNotFoundError` when the library is not installed

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/_sampling_rule.py

@@ -0,0 +1,80 @@
+# 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
+
+
+# Disable snake_case naming style so this class can match the sampling rules response from X-Ray
+# pylint: disable=invalid-name
+class _SamplingRule:
+ def __init__(
+ self,
+ Attributes: Optional["dict[str, str]"] = None,
+ FixedRate: Optional[float] = None,
+ HTTPMethod: Optional[str] = None,
+ Host: Optional[str] = None,
+ Priority: Optional[int] = None,
+ ReservoirSize: Optional[int] = None,
+ ResourceARN: Optional[str] = None,
+ RuleARN: Optional[str] = None,
+ RuleName: Optional[str] = None,
+ ServiceName: Optional[str] = None,
+ ServiceType: Optional[str] = None,
+ URLPath: Optional[str] = None,
+ Version: Optional[int] = None,
+ ):
+ self.Attributes = Attributes if Attributes is not None else {}
+ self.FixedRate = FixedRate if FixedRate is not None else 0.0
+ self.HTTPMethod = HTTPMethod if HTTPMethod is not None else ""
+ self.Host = Host if Host is not None else ""
+ # Default to value with lower priority than default rule
+ self.Priority = Priority if Priority is not None else 10001
+ self.ReservoirSize = ReservoirSize if ReservoirSize is not None else 0
+ self.ResourceARN = ResourceARN if ResourceARN is not None else ""
+ self.RuleARN = RuleARN if RuleARN is not None else ""
+ self.RuleName = RuleName if RuleName is not None else ""
+ self.ServiceName = ServiceName if ServiceName is not None else ""
+ self.ServiceType = ServiceType if ServiceType is not None else ""
+ self.URLPath = URLPath if URLPath is not None else ""
+ self.Version = Version if Version is not None else 0
+
+ def __lt__(self, other: "_SamplingRule") -> bool:
+ if self.Priority == other.Priority:
+ # String order priority example:
+ # "A","Abc","a","ab","abc","abcdef"
+ return self.RuleName < other.RuleName
+ return self.Priority < other.Priority
+
+ def __eq__(self, other: object) -> bool:
+ if not isinstance(other, _SamplingRule):
+ return False
+ return (
+ self.FixedRate == other.FixedRate
+ and self.HTTPMethod == other.HTTPMethod
+ and self.Host == other.Host
+ and self.Priority == other.Priority
+ and self.ReservoirSize == other.ReservoirSize
+ and self.ResourceARN == other.ResourceARN
+ and self.RuleARN == other.RuleARN
+ and self.RuleName == other.RuleName
+ and self.ServiceName == other.ServiceName
+ and self.ServiceType == other.ServiceType
+ and self.URLPath == other.URLPath
+ and self.Version == other.Version
+ and self.Attributes == other.Attributes
+ )