Add filters for minimum severity level and trace based sampling

Author: rads-1996

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

@@ -21,6 +21,7 @@
     LogRecordDroppedAttributesWarning,
     LogRecordLimits,
     LogRecordProcessor,
+    FilteringLogRecordProcessor,
     ReadableLogRecord,
     ReadWriteLogRecord,
 )
@@ -32,6 +33,7 @@
     "LogLimits",
     "LogRecordLimits",
     "LogRecordProcessor",
+    "FilteringLogRecordProcessor",
     "LogDroppedAttributesWarning",
     "LogRecordDroppedAttributesWarning",
     "ReadableLogRecord",

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

@@ -55,6 +55,7 @@
 from opentelemetry.trace import (
     format_span_id,
     format_trace_id,
+    get_current_span,
 )
 from opentelemetry.util.types import AnyValue, _ExtendedAttributes
 
@@ -317,6 +318,35 @@ def force_flush(self, timeout_millis: int = 30000) -> bool:
         """
 
 
+class FilteringLogRecordProcessor(LogRecordProcessor):
+    """A processor that drops records based on minimum severity and/or trace based sampling parameters provided by the user."""
+
+    def __init__(
+        self,
+        log_record_processor: LogRecordProcessor,
+        *,
+        minimum_severity_level: SeverityNumber = SeverityNumber.UNSPECIFIED,
+        enable_trace_based_sampling: bool = False,
+    ):
+        self._log_record_processor = log_record_processor
+        self._minimum_severity_level = minimum_severity_level
+        self._enable_trace_based_sampling = enable_trace_based_sampling
+
+    def on_emit(self, log_record: ReadWriteLogRecord):
+        record = log_record.log_record
+        if is_less_than_min_severity(record, self._minimum_severity_level):
+            return
+        if should_drop_logs_for_trace_based(record, self._enable_trace_based_sampling):
+            return
+        self._log_record_processor.on_emit(log_record)
+
+    def shutdown(self):
+        self._log_record_processor.shutdown()
+
+    def force_flush(self, timeout_millis: int = 30000) -> bool:
+        return self._log_record_processor.force_flush(timeout_millis)
+
+
 # Temporary fix until https://github.com/PyCQA/pylint/issues/4098 is resolved
 # pylint:disable=no-member
 class SynchronousMultiLogRecordProcessor(LogRecordProcessor):
@@ -835,3 +865,29 @@ def std_to_otel(levelno: int) -> SeverityNumber:
     if levelno > 53:
         return SeverityNumber.FATAL4
     return _STD_TO_OTEL[levelno]
+
+
+def is_less_than_minimum_severity_level(
+    record: LogRecord, minimum_severity_level: SeverityNumber
+) -> bool:
+    if record.severity_number is not None:
+        if (
+            minimum_severity_level is not None
+            and minimum_severity_level != SeverityNumber.UNSPECIFIED
+            and record.severity_number.value < minimum_severity_level.value
+        ):
+            return True
+    return False
+
+
+def should_drop_logs_for_trace_based_sampling(
+    record: LogRecord, enable_trace_based_sampling: bool
+) -> bool:
+    if enable_trace_based_sampling:
+        if record.context is not None:
+            span = get_current_span(record.context)
+            span_context = span.get_span_context()
+            if span_context.is_valid and not span_context.trace_flags.sampled:
+                return True
+    return False
+

opentelemetry-sdk/tests/logs/test_logs.py

@@ -27,6 +27,7 @@
 from opentelemetry.sdk._logs._internal import (
     NoOpLogger,
     SynchronousMultiLogRecordProcessor,
+    FilteringLogRecordProcessor,
 )
 from opentelemetry.sdk.environment_variables import OTEL_SDK_DISABLED
 from opentelemetry.sdk.resources import Resource
@@ -214,3 +215,243 @@ def test_can_emit_with_keywords_arguments(self):
         self.assertEqual(result_log_record.attributes, {"some": "attributes"})
         self.assertEqual(result_log_record.event_name, "event_name")
         self.assertEqual(log_data.resource, logger.resource)
+
+    def test_emit_logrecord_with_min_severity_filtering(self):
+        logger_processor_mock = Mock()
+        filtering_processor = FilteringLogRecordProcessor(
+            logger_processor_mock,
+            minimum_severity_level=SeverityNumber.DEBUG4,
+        )
+        logger, _ = self._get_logger(filtering_processor)
+
+        log_record_info = LogRecord(
+            observed_timestamp=0,
+            body="info log line",
+            severity_number=SeverityNumber.DEBUG,
+            severity_text="DEBUG",
+        )
+
+        logger.emit(log_record_info)
+        logger_processor_mock.on_emit.assert_not_called()
+
+        logger_processor_mock.reset_mock()
+
+        log_record_error = LogRecord(
+            observed_timestamp=0,
+            body="error log line",
+            severity_number=SeverityNumber.ERROR,
+            severity_text="ERROR",
+        )
+
+        logger.emit(log_record_error)
+
+        logger_processor_mock.on_emit.assert_called_once()
+        log_data = logger_processor_mock.on_emit.call_args.args[0]
+        self.assertTrue(isinstance(log_data.log_record, LogRecord))
+        self.assertEqual(log_data.log_record.severity_number, SeverityNumber.ERROR)
+
+    def test_emit_logrecord_with_min_severity_unspecified(self):
+        logger_processor_mock = Mock()
+        filtering_processor = FilteringLogRecordProcessor(
+            logger_processor_mock,
+            minimum_severity_level=SeverityNumber.UNSPECIFIED,
+        )
+        logger, _ = self._get_logger(filtering_processor)
+        log_record = LogRecord(
+            observed_timestamp=0,
+            body="debug log line",
+            severity_number=SeverityNumber.DEBUG,
+            severity_text="DEBUG",
+        )
+        logger.emit(log_record)
+        logger_processor_mock.on_emit.assert_called_once()
+
+    def test_emit_logrecord_with_trace_based_filtering(self):
+        logger_processor_mock = Mock()
+        filtering_processor = FilteringLogRecordProcessor(
+            logger_processor_mock,
+            enable_trace_based_sampling=True,
+        )
+        logger, _ = self._get_logger(filtering_processor)
+
+        mock_span_context = Mock()
+        mock_span_context.is_valid = True
+        mock_span_context.trace_flags.sampled = False
+
+        mock_span = Mock()
+        mock_span.get_span_context.return_value = mock_span_context
+
+        mock_context = Mock()
+
+        with patch(
+            "opentelemetry.sdk._logs._internal.get_current_span",
+            return_value=mock_span,
+        ):
+            log_record = LogRecord(
+                observed_timestamp=0,
+                body="should be dropped",
+                severity_number=SeverityNumber.INFO,
+                severity_text="INFO",
+                context=mock_context,
+            )
+
+            logger.emit(log_record)
+            logger_processor_mock.on_emit.assert_not_called()
+
+        logger_processor_mock.reset_mock()
+
+        mock_span_context = Mock()
+        mock_span_context.is_valid = True
+        mock_span_context.trace_flags.sampled = True
+
+        mock_span = Mock()
+        mock_span.get_span_context.return_value = mock_span_context
+
+        with patch(
+            "opentelemetry.sdk._logs._internal.get_current_span",
+            return_value=mock_span,
+        ):
+            log_record = LogRecord(
+                observed_timestamp=0,
+                body="should pass",
+                severity_number=SeverityNumber.INFO,
+                severity_text="INFO",
+                context=mock_context,
+            )
+
+            logger.emit(log_record)
+            logger_processor_mock.on_emit.assert_called_once()
+
+    def test_emit_logrecord_trace_filtering_disabled(self):
+        logger_processor_mock = Mock()
+        filtering_processor = FilteringLogRecordProcessor(
+            logger_processor_mock,
+            enable_trace_based_sampling=False,
+        )
+        logger, _ = self._get_logger(filtering_processor)
+
+        mock_span_context = Mock()
+        mock_span_context.is_valid = False
+        mock_span_context.trace_flags.sampled = False
+
+        mock_span = Mock()
+        mock_span.get_span_context.return_value = mock_span_context
+
+        mock_context = Mock()
+
+        with patch(
+            "opentelemetry.sdk._logs._internal.get_current_span",
+            return_value=mock_span,
+        ):
+            log_record = LogRecord(
+                observed_timestamp=0,
+                body="should be emitted when filtering disabled",
+                severity_number=SeverityNumber.INFO,
+                severity_text="INFO",
+                context=mock_context,
+            )
+
+            logger.emit(log_record)
+            logger_processor_mock.on_emit.assert_called_once()
+
+    def test_emit_logrecord_trace_filtering_edge_cases(self):
+        logger_processor_mock = Mock()
+        filtering_processor = FilteringLogRecordProcessor(
+            logger_processor_mock,
+            enable_trace_based_sampling=True,
+        )
+        logger, _ = self._get_logger(filtering_processor)
+
+        mock_span_context = Mock()
+        mock_span_context.is_valid = False
+        mock_span_context.trace_flags.sampled = True
+
+        mock_span = Mock()
+        mock_span.get_span_context.return_value = mock_span_context
+
+        mock_context = Mock()
+
+        with patch(
+            "opentelemetry.sdk._logs._internal.get_current_span",
+            return_value=mock_span,
+        ):
+            log_record = LogRecord(
+                observed_timestamp=0,
+                body="invalid but sampled",
+                severity_number=SeverityNumber.INFO,
+                severity_text="INFO",
+                context=mock_context,
+            )
+
+            logger.emit(log_record)
+            logger_processor_mock.on_emit.assert_called_once()
+
+        logger_processor_mock.reset_mock()
+
+        mock_span_context = Mock()
+        mock_span_context.is_valid = True
+        mock_span_context.trace_flags.sampled = False
+
+        mock_span = Mock()
+        mock_span.get_span_context.return_value = mock_span_context
+
+        with patch(
+            "opentelemetry.sdk._logs._internal.get_current_span",
+            return_value=mock_span,
+        ):
+            log_record = LogRecord(
+                observed_timestamp=0,
+                body="valid but not sampled",
+                severity_number=SeverityNumber.INFO,
+                severity_text="INFO",
+                context=mock_context,
+            )
+
+            logger.emit(log_record)
+            logger_processor_mock.on_emit.assert_not_called()
+
+    def test_emit_both_min_severity_and_trace_based_filtering(self):
+        logger_processor_mock = Mock()
+        filtering_processor = FilteringLogRecordProcessor(
+            logger_processor_mock,
+            minimum_severity_level=SeverityNumber.WARN,
+            enable_trace_based_sampling=True,
+        )
+        logger, _ = self._get_logger(filtering_processor)
+
+        mock_span_context = Mock()
+        mock_span_context.is_valid = True
+        mock_span_context.trace_flags.sampled = True
+
+        mock_span = Mock()
+        mock_span.get_span_context.return_value = mock_span_context
+
+        mock_context = Mock()
+
+        with patch(
+            "opentelemetry.sdk._logs._internal.get_current_span",
+            return_value=mock_span,
+        ):
+            log_record_info = LogRecord(
+                observed_timestamp=0,
+                body="info log line",
+                severity_number=SeverityNumber.INFO,
+                severity_text="INFO",
+                context=mock_context,
+            )
+
+            logger.emit(log_record_info)
+            logger_processor_mock.on_emit.assert_not_called()
+
+            logger_processor_mock.reset_mock()
+
+            log_record_error = LogRecord(
+                observed_timestamp=0,
+                body="error log line",
+                severity_number=SeverityNumber.ERROR,
+                severity_text="ERROR",
+                context=mock_context,
+            )
+
+            logger.emit(log_record_error)
+            logger_processor_mock.on_emit.assert_called_once()