Merge branch 'main' into simplify-client-py-6721

Author: xrmx

CHANGELOG.md

@@ -12,6 +12,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## Unreleased
 
+- Add experimental composite samplers
+  ([#4714](https://github.com/open-telemetry/opentelemetry-python/pull/4714))
+- Filter duplicate logs out of some internal `logger`'s logs on the export logs path that might otherwise endlessly log or cause a recursion depth exceeded issue in cases where logging itself results in an exception.
+  ([#4695](https://github.com/open-telemetry/opentelemetry-python/pull/4695)).
 - docs: linked the examples with their github source code location and added Prometheus example
   ([#4728](https://github.com/open-telemetry/opentelemetry-python/pull/4728))
 

docs/getting_started/tests/test_metrics.py

@@ -0,0 +1,103 @@
+# 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 json
+import os
+import subprocess
+import sys
+import unittest
+
+
+class TestMetrics(unittest.TestCase):
+    def test_metrics(self):
+        """Test that metrics example produces expected values"""
+        # Run the metrics example
+        test_script = f"{os.path.dirname(os.path.realpath(__file__))}/../metrics_example.py"
+
+        result = subprocess.run(
+            [sys.executable, test_script],
+            capture_output=True,
+            text=True,
+            timeout=10,
+            check=True,
+        )
+
+        # Script should run successfully
+        self.assertEqual(result.returncode, 0)
+
+        # Parse the JSON output
+        output_data = json.loads(result.stdout)
+
+        # Get the metrics from the JSON structure
+        metrics = output_data["resource_metrics"][0]["scope_metrics"][0][
+            "metrics"
+        ]
+
+        # Create a lookup dict for easier testing
+        metrics_by_name = {metric["name"]: metric for metric in metrics}
+
+        # Test Counter: should be 1 (called counter.add(1))
+        counter_value = metrics_by_name["counter"]["data"]["data_points"][0][
+            "value"
+        ]
+        self.assertEqual(counter_value, 1, "Counter should have value 1")
+
+        # Test UpDownCounter: should be -4 (1 + (-5) = -4)
+        updown_value = metrics_by_name["updown_counter"]["data"][
+            "data_points"
+        ][0]["value"]
+        self.assertEqual(
+            updown_value, -4, "UpDownCounter should have value -4"
+        )
+
+        # Test Histogram: should have count=1, sum=99.9
+        histogram_data = metrics_by_name["histogram"]["data"]["data_points"][0]
+        self.assertEqual(
+            histogram_data["count"], 1, "Histogram should have count 1"
+        )
+        self.assertEqual(
+            histogram_data["sum"], 99.9, "Histogram should have sum 99.9"
+        )
+
+        # Test Gauge: should be 1 (last value set)
+        gauge_value = metrics_by_name["gauge"]["data"]["data_points"][0][
+            "value"
+        ]
+        self.assertEqual(gauge_value, 1, "Gauge should have value 1")
+
+        # Test Observable Counter: should be 1 (from callback)
+        obs_counter_value = metrics_by_name["observable_counter"]["data"][
+            "data_points"
+        ][0]["value"]
+        self.assertEqual(
+            obs_counter_value, 1, "Observable counter should have value 1"
+        )
+
+        # Test Observable UpDownCounter: should be -10 (from callback)
+        obs_updown_value = metrics_by_name["observable_updown_counter"][
+            "data"
+        ]["data_points"][0]["value"]
+        self.assertEqual(
+            obs_updown_value,
+            -10,
+            "Observable updown counter should have value -10",
+        )
+
+        # Test Observable Gauge: should be 9 (from callback)
+        obs_gauge_value = metrics_by_name["observable_gauge"]["data"][
+            "data_points"
+        ][0]["value"]
+        self.assertEqual(
+            obs_gauge_value, 9, "Observable gauge should have value 9"
+        )

docs/getting_started/tests/test_otlpcollector.py

@@ -0,0 +1,43 @@
+# 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.
+
+# oltpcollector_example.py
+import os
+import subprocess
+import sys
+import unittest
+
+
+class TestOTLPCollector(unittest.TestCase):
+    def test_otlpcollector(self):
+        """Test that OTLP collector example outputs 'Hello world!'"""
+        dirpath = os.path.dirname(os.path.realpath(__file__))
+        test_script = f"{dirpath}/../otlpcollector_example.py"
+
+        # Run the script with a short timeout since it will retry forever
+        with subprocess.Popen(
+            [sys.executable, test_script],
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            text=True,
+        ) as process:
+            # Wait 2 seconds then kill it (enough time to print "Hello world!")
+            try:
+                stdout, _ = process.communicate(timeout=2)
+            except subprocess.TimeoutExpired:
+                process.kill()
+                stdout, _ = process.communicate()
+
+        # Check that it printed the expected message
+        self.assertIn("Hello world!", stdout)

exporter/opentelemetry-exporter-otlp-proto-grpc/src/opentelemetry/exporter/otlp/proto/grpc/exporter.py

@@ -59,6 +59,7 @@
     KeyValue,
 )
 from opentelemetry.proto.resource.v1.resource_pb2 import Resource  # noqa: F401
+from opentelemetry.sdk._shared_internal import DuplicateFilter
 from opentelemetry.sdk.environment_variables import (
     OTEL_EXPORTER_OTLP_CERTIFICATE,
     OTEL_EXPORTER_OTLP_CLIENT_CERTIFICATE,
@@ -87,6 +88,8 @@
 )
 _MAX_RETRYS = 6
 logger = getLogger(__name__)
+# This prevents logs generated when a log fails to be written to generate another log which fails to be written etc. etc.
+logger.addFilter(DuplicateFilter())
 SDKDataT = TypeVar("SDKDataT")
 ResourceDataT = TypeVar("ResourceDataT")
 TypingResourceT = TypeVar("TypingResourceT")

exporter/opentelemetry-exporter-otlp-proto-http/src/opentelemetry/exporter/otlp/proto/http/_log_exporter/__init__.py

@@ -38,6 +38,7 @@
     LogExporter,
     LogExportResult,
 )
+from opentelemetry.sdk._shared_internal import DuplicateFilter
 from opentelemetry.sdk.environment_variables import (
     OTEL_EXPORTER_OTLP_CERTIFICATE,
     OTEL_EXPORTER_OTLP_CLIENT_CERTIFICATE,
@@ -57,6 +58,8 @@
 from opentelemetry.util.re import parse_env_headers
 
 _logger = logging.getLogger(__name__)
+# This prevents logs generated when a log fails to be written to generate another log which fails to be written etc. etc.
+_logger.addFilter(DuplicateFilter())
 
 
 DEFAULT_COMPRESSION = Compression.NoCompression

opentelemetry-sdk/src/opentelemetry/sdk/_logs/_internal/export/__init__.py

@@ -27,7 +27,7 @@
     set_value,
 )
 from opentelemetry.sdk._logs import LogData, LogRecord, LogRecordProcessor
-from opentelemetry.sdk._shared_internal import BatchProcessor
+from opentelemetry.sdk._shared_internal import BatchProcessor, DuplicateFilter
 from opentelemetry.sdk.environment_variables import (
     OTEL_BLRP_EXPORT_TIMEOUT,
     OTEL_BLRP_MAX_EXPORT_BATCH_SIZE,
@@ -43,6 +43,7 @@
     "Unable to parse value for %s as integer. Defaulting to %s."
 )
 _logger = logging.getLogger(__name__)
+_logger.addFilter(DuplicateFilter())
 
 
 class LogExportResult(enum.Enum):

opentelemetry-sdk/src/opentelemetry/sdk/_shared_internal/__init__.py

@@ -39,6 +39,29 @@
 from opentelemetry.util._once import Once
 
 
+class DuplicateFilter(logging.Filter):
+    """Filter that can be applied to internal `logger`'s.
+
+    Currently applied to `logger`s on the export logs path that could otherwise cause endless logging of errors or a
+    recursion depth exceeded issue in cases where logging itself results in an exception."""
+
+    def filter(self, record):
+        current_log = (
+            record.module,
+            record.levelno,
+            record.msg,
+            # We need to pick a time longer than the OTLP LogExporter timeout
+            # which defaults to 10 seconds, but not pick something so long that
+            # it filters out useful logs.
+            time.time() // 20,
+        )
+        if current_log != getattr(self, "last_log", None):
+            self.last_log = current_log  # pylint: disable=attribute-defined-outside-init
+            return True
+        # False means python's `logging` module will no longer process this log.
+        return False
+
+
 class BatchExportStrategy(enum.Enum):
     EXPORT_ALL = 0
     EXPORT_WHILE_BATCH_EXCEEDS_THRESHOLD = 1
@@ -89,6 +112,7 @@ def __init__(
             daemon=True,
         )
         self._logger = logging.getLogger(__name__)
+        self._logger.addFilter(DuplicateFilter())
         self._exporting = exporting
 
         self._shutdown = False

opentelemetry-sdk/src/opentelemetry/sdk/trace/_sampling_experimental/__init__.py

@@ -0,0 +1,31 @@
+# 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.
+
+__all__ = [
+    "ComposableSampler",
+    "SamplingIntent",
+    "composable_always_off",
+    "composable_always_on",
+    "composable_parent_threshold",
+    "composable_traceid_ratio_based",
+    "composite_sampler",
+]
+
+
+from ._always_off import composable_always_off
+from ._always_on import composable_always_on
+from ._composable import ComposableSampler, SamplingIntent
+from ._parent_threshold import composable_parent_threshold
+from ._sampler import composite_sampler
+from ._traceid_ratio import composable_traceid_ratio_based

opentelemetry-sdk/src/opentelemetry/sdk/trace/_sampling_experimental/_always_off.py

@@ -0,0 +1,55 @@
+# 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.
+
+from __future__ import annotations
+
+from typing import Sequence
+
+from opentelemetry.context import Context
+from opentelemetry.trace import Link, SpanKind, TraceState
+from opentelemetry.util.types import Attributes
+
+from ._composable import ComposableSampler, SamplingIntent
+from ._util import INVALID_THRESHOLD
+
+_intent = SamplingIntent(threshold=INVALID_THRESHOLD, threshold_reliable=False)
+
+
+class _ComposableAlwaysOffSampler(ComposableSampler):
+    def sampling_intent(
+        self,
+        parent_ctx: Context | None,
+        name: str,
+        span_kind: SpanKind | None,
+        attributes: Attributes,
+        links: Sequence[Link] | None,
+        trace_state: TraceState | None = None,
+    ) -> SamplingIntent:
+        return _intent
+
+    def get_description(self) -> str:
+        return "ComposableAlwaysOff"
+
+
+_always_off = _ComposableAlwaysOffSampler()
+
+
+def composable_always_off() -> ComposableSampler:
+    """Returns a composable sampler that does not sample any span.
+
+    - Always returns a SamplingIntent with no threshold, indicating all spans should be dropped
+    - Sets threshold_reliable to false
+    - Does not add any attributes
+    """
+    return _always_off

opentelemetry-sdk/src/opentelemetry/sdk/trace/_sampling_experimental/_always_on.py

@@ -0,0 +1,55 @@
+# 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.
+
+from __future__ import annotations
+
+from typing import Sequence
+
+from opentelemetry.context import Context
+from opentelemetry.trace import Link, SpanKind, TraceState
+from opentelemetry.util.types import Attributes
+
+from ._composable import ComposableSampler, SamplingIntent
+from ._util import MIN_THRESHOLD
+
+_intent = SamplingIntent(threshold=MIN_THRESHOLD)
+
+
+class _ComposableAlwaysOnSampler(ComposableSampler):
+    def sampling_intent(
+        self,
+        parent_ctx: Context | None,
+        name: str,
+        span_kind: SpanKind | None,
+        attributes: Attributes,
+        links: Sequence[Link] | None,
+        trace_state: TraceState | None = None,
+    ) -> SamplingIntent:
+        return _intent
+
+    def get_description(self) -> str:
+        return "ComposableAlwaysOn"
+
+
+_always_on = _ComposableAlwaysOnSampler()
+
+
+def composable_always_on() -> ComposableSampler:
+    """Returns a composable sampler that samples all spans.
+
+    - Always returns a SamplingIntent with threshold set to sample all spans (threshold = 0)
+    - Sets threshold_reliable to true
+    - Does not add any attributes
+    """
+    return _always_on