Implement filtering logic for min_severity and trace_based parameters

CHANGELOG.md

@@ -16,6 +16,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   ([#4755](https://github.com/open-telemetry/opentelemetry-python/pull/4755))
 - logs: extend Logger.emit to accept separated keyword arguments
   ([#4737](https://github.com/open-telemetry/opentelemetry-python/pull/4737))
+- Add `minimum_severity` and `trace_based` logger parameters to filter logs
+  ([#4765](https://github.com/open-telemetry/opentelemetry-python/pull/4765))
 
 ## Version 1.37.0/0.58b0 (2025-09-11)
 

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

@@ -675,6 +675,8 @@ def __init__(
             ConcurrentMultiLogRecordProcessor,
         ],
         instrumentation_scope: InstrumentationScope,
+        min_severity_level: SeverityNumber = SeverityNumber.UNSPECIFIED,
+        trace_based: bool = False,
     ):
         super().__init__(
             instrumentation_scope.name,
@@ -685,6 +687,8 @@ def __init__(
         self._resource = resource
         self._multi_log_record_processor = multi_log_record_processor
         self._instrumentation_scope = instrumentation_scope
+        self._min_severity_level = min_severity_level
+        self._trace_based = trace_based
 
     @property
     def resource(self):
@@ -744,6 +748,10 @@ def emit(
             record = LogRecord._from_api_log_record(
                 record=record, resource=self._resource
             )
+        if is_less_than_min_severity(record, self._min_severity_level):
+            return
+        if should_drop_logs_for_unsampled_traces(record, self._trace_based):
+            return
 
         log_data = LogData(record, self._instrumentation_scope)
 
@@ -758,6 +766,8 @@ def __init__(
         multi_log_record_processor: SynchronousMultiLogRecordProcessor
         | ConcurrentMultiLogRecordProcessor
         | None = None,
+        min_severity_level: SeverityNumber = SeverityNumber.UNSPECIFIED,
+        trace_based: bool = False,
     ):
         if resource is None:
             self._resource = Resource.create({})
@@ -773,6 +783,8 @@ def __init__(
             self._at_exit_handler = atexit.register(self.shutdown)
         self._logger_cache = {}
         self._logger_cache_lock = Lock()
+        self._min_severity_level = min_severity_level
+        self._trace_based = trace_based
 
     @property
     def resource(self):
@@ -794,6 +806,8 @@ def _get_logger_no_cache(
                 schema_url,
                 attributes,
             ),
+            self._min_severity_level,
+            self._trace_based,
         )
 
     def _get_logger_cached(
@@ -920,3 +934,54 @@ def std_to_otel(levelno: int) -> SeverityNumber:
     if levelno > 53:
         return SeverityNumber.FATAL4
     return _STD_TO_OTEL[levelno]
+
+
+def is_less_than_min_severity(
+    record: LogRecord, min_severity: SeverityNumber
+) -> bool:
+    """Checks if the log record's severity number is less than the minimum severity level.
+
+    Args:
+        record: The log record to be processed.
+        min_severity: The minimum severity level.
+
+    Returns:
+        True if the log record's severity number is less than the minimum
+        severity level, False otherwise. Log records with an unspecified severity (i.e. `0`)
+        are not affected by this parameter and therefore bypass minimum severity filtering.
+    """
+    if record.severity_number is not None:
+        if (
+            min_severity is not None
+            and min_severity != SeverityNumber.UNSPECIFIED
+            and record.severity_number.value < min_severity.value
+        ):
+            return True
+    return False
+
+
+def should_drop_logs_for_unsampled_traces(
+    record: LogRecord, trace_based_flag: bool
+) -> bool:
+    """Determines whether the logger should drop log records associated with unsampled traces.
+
+    If `trace_based` is `true`, log records associated with unsampled traces are dropped by the `Logger`.
+    A log record is considered associated with an unsampled trace if it has a valid `SpanId` and its
+    `TraceFlags` indicate that the trace is unsampled. A log record that isn't associated with a trace
+    context is not affected by this parameter and therefore bypasses trace-based filtering.
+
+    Args:
+        record: The log record to be processed.
+        trace_based_flag: A boolean flag indicating whether trace-based filtering is enabled. If not explicitly set,
+        the `trace_based` parameter is set to `false`
+
+    Returns:
+        True if the log record should be dropped due to being associated with an unsampled trace.
+    """
+    if trace_based_flag:
+        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

@@ -74,6 +74,10 @@ def test_get_logger(self):
         self.assertEqual(
             logger._instrumentation_scope.attributes, {"key": "value"}
         )
+        self.assertEqual(
+            logger._min_severity_level, SeverityNumber.UNSPECIFIED
+        )
+        self.assertFalse(logger._trace_based)
 
     @patch.dict("os.environ", {OTEL_SDK_DISABLED: "true"})
     def test_get_logger_with_sdk_disabled(self):
@@ -83,7 +87,9 @@ def test_get_logger_with_sdk_disabled(self):
 
     @patch.object(Resource, "create")
     def test_logger_provider_init(self, resource_patch):
-        logger_provider = LoggerProvider()
+        logger_provider = LoggerProvider(
+            min_severity_level=SeverityNumber.DEBUG4, trace_based=True
+        )
         resource_patch.assert_called_once()
         self.assertIsNotNone(logger_provider._resource)
         self.assertTrue(
@@ -92,6 +98,10 @@ def test_logger_provider_init(self, resource_patch):
                 SynchronousMultiLogRecordProcessor,
             )
         )
+        self.assertEqual(
+            logger_provider._min_severity_level, SeverityNumber.DEBUG4
+        )
+        self.assertTrue(logger_provider._trace_based)
         self.assertIsNotNone(logger_provider._at_exit_handler)
 
 
@@ -171,3 +181,210 @@ def test_can_emit_with_keywords_arguments(self):
         self.assertEqual(log_record.attributes, {"some": "attributes"})
         self.assertEqual(log_record.event_name, "event_name")
         self.assertEqual(log_record.resource, logger.resource)
+
+    def test_emit_logrecord_with_min_severity_filtering(self):
+        """Test that logs below minimum severity are filtered out"""
+        logger, log_record_processor_mock = self._get_logger()
+        logger._min_severity_level = SeverityNumber.DEBUG4
+
+        log_record_info = LogRecord(
+            observed_timestamp=0,
+            body="info log line",
+            severity_number=SeverityNumber.DEBUG,
+            severity_text="DEBUG",
+        )
+
+        logger.emit(log_record_info)
+        log_record_processor_mock.on_emit.assert_not_called()
+
+        log_record_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)
+
+        log_record_processor_mock.on_emit.assert_called_once()
+        log_data = log_record_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):
+        """Test that when min severity is UNSPECIFIED, all logs are emitted"""
+        logger, log_record_processor_mock = self._get_logger()
+        log_record = LogRecord(
+            observed_timestamp=0,
+            body="debug log line",
+            severity_number=SeverityNumber.DEBUG,
+            severity_text="DEBUG",
+        )
+        logger.emit(log_record)
+        log_record_processor_mock.on_emit.assert_called_once()
+
+    def test_emit_logrecord_with_trace_based_filtering(self):
+        """Test that logs are filtered based on trace sampling state"""
+        logger, log_record_processor_mock = self._get_logger()
+        logger._trace_based = True
+
+        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)
+            log_record_processor_mock.on_emit.assert_not_called()
+
+        log_record_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
+
+    def test_emit_logrecord_trace_filtering_disabled(self):
+        """Test that when trace-based filtering is disabled, all logs are emitted"""
+        logger, log_record_processor_mock = self._get_logger()
+
+        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)
+            log_record_processor_mock.on_emit.assert_called_once()
+
+    def test_emit_logrecord_trace_filtering_edge_cases(self):
+        """Test edge cases for trace-based filtering"""
+        logger, log_record_processor_mock = self._get_logger()
+        logger._trace_based = True
+
+        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)
+            log_record_processor_mock.on_emit.assert_called_once()
+
+        log_record_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)
+            log_record_processor_mock.on_emit.assert_not_called()
+
+    def test_emit_both_min_severity_and_trace_based_filtering(self):
+        """Test that both min severity and trace-based filtering work together"""
+        logger, log_record_processor_mock = self._get_logger()
+        logger._min_severity_level = SeverityNumber.WARN
+        logger._trace_based = True
+
+        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)
+            log_record_processor_mock.on_emit.assert_not_called()
+
+            log_record_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)
+            log_record_processor_mock.on_emit.assert_called_once()