Improve DuplicateFilter behavior and log partial success unconditionally.

Author: michaelsafyan

CHANGELOG.md

@@ -12,7 +12,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## Unreleased
 
-- OTLP exporters now log partial success responses at `debug` level when `OTEL_LOG_LEVEL` is set to `debug` or `verbose`.
+- OTLP exporters now log partial success responses.
   ([#4805](https://github.com/open-telemetry/opentelemetry-python/pull/4805))
 - docs: Added sqlcommenter example
   ([#4734](https://github.com/open-telemetry/opentelemetry-python/pull/4734))

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

@@ -11,7 +11,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import sys
+
 from os import environ
 from typing import Dict, Literal, Optional, Sequence, Tuple, Union
 from typing import Sequence as TypingSequence
@@ -113,11 +113,6 @@ def _translate_data(
     ) -> ExportLogsServiceRequest:
         return encode_logs(data)
 
-    def _log_partial_success(self, partial_success):
-        # Override that skips the "logging" module due to the possibility
-        # of circular logic (logging -> OTLP logs export).
-        sys.stderr.write(f"Partial success:\n{partial_success}\n")
-
     def export(  # type: ignore [reportIncompatibleMethodOverride]
         self,
         batch: Sequence[ReadableLogRecord],

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

@@ -374,12 +374,9 @@ def _translate_data(
     ) -> ExportServiceRequestT:
         pass
 
-    def _log_partial_success(self, partial_success):
-        logger.debug("Partial success:\n%s", partial_success)
-
     def _process_response(self, response):
         if response.HasField("partial_success"):
-            self._log_partial_success(response.partial_success)
+            logger.debug("Partial success:\n%s", response.partial_success)
 
     def _export(
         self,

exporter/opentelemetry-exporter-otlp-proto-grpc/tests/logs/test_otlp_logs_exporter.py

@@ -15,7 +15,6 @@
 # pylint: disable=too-many-lines
 
 import time
-from io import StringIO
 from os.path import dirname
 from unittest import TestCase
 from unittest.mock import Mock, patch
@@ -29,9 +28,7 @@
     OTLPLogExporter,
 )
 from opentelemetry.proto.collector.logs.v1.logs_service_pb2 import (
-    ExportLogsPartialSuccess,
     ExportLogsServiceRequest,
-    ExportLogsServiceResponse,
 )
 from opentelemetry.proto.common.v1.common_pb2 import AnyValue, KeyValue
 from opentelemetry.proto.common.v1.common_pb2 import (
@@ -319,26 +316,6 @@ def export_log_and_deserialize(self, log_data):
         )
         return log_records
 
-    @patch("sys.stderr", new_callable=StringIO)
-    def test_partial_success_recorded_directly_to_stderr(self, mock_stderr):
-        # pylint: disable=protected-access
-        exporter = OTLPLogExporter()
-        exporter._client = Mock()
-        exporter._client.Export.return_value = ExportLogsServiceResponse(
-            partial_success=ExportLogsPartialSuccess(
-                rejected_log_records=1,
-                error_message="Log record dropped",
-            )
-        )
-
-        exporter.export([self.log_data_1])
-
-        self.assertIn("Partial success:\n", mock_stderr.getvalue())
-        self.assertIn("rejected_log_records: 1\n", mock_stderr.getvalue())
-        self.assertIn(
-            'error_message: "Log record dropped"\n', mock_stderr.getvalue()
-        )
-
     def test_exported_log_without_trace_id(self):
         log_records = self.export_log_and_deserialize(self.log_data_4)
         if log_records:

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

@@ -46,17 +46,36 @@ class DuplicateFilter(logging.Filter):
     recursion depth exceeded issue in cases where logging itself results in an exception."""
 
     def filter(self, record):
-        current_log = (
+        # 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_boundary = time.time() // 20
+        current_log_by_line = (
+            record.module,
+            record.pathname,
+            record.lineno,
+            time_boundary,
+        )
+        previous_log_by_line = getattr(self, "last_source", None)
+        previous_log_by_line_count = getattr(self, "last_source_count", 1)
+        if current_log_by_line == previous_log_by_line:
+            recurring_count = previous_log_by_line_count + 1
+            self.last_source_count = recurring_count  # pylint: disable=attribute-defined-outside-init
+            if recurring_count > 3:
+                # Don't allow further processing, since the same line is repeatedly emitting logs.
+                return False
+        else:
+            self.last_source = current_log_by_line  # pylint: disable=attribute-defined-outside-init
+            self.last_source_count = 1  # pylint: disable=attribute-defined-outside-init
+
+        current_log_by_content = (
             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,
+            time_boundary,
         )
-        if current_log != getattr(self, "last_log", None):
-            self.last_log = current_log  # pylint: disable=attribute-defined-outside-init
+        if current_log_by_content != getattr(self, "last_log", None):
+            self.last_log = current_log_by_content  # pylint: disable=attribute-defined-outside-init
             return True
         # False means python's `logging` module will no longer process this log.
         return False