fix(sdk): add max_spans limit to InMemorySpanExporter (#5296)

* fix(sdk): add max_spans limit to InMemorySpanExporter

* test(sdk): improve test coverage and modernize type hints in InMemorySpanExporter

- Update InMemorySpanExporter type hint from typing.Optional[int] to int | None.
- Refactor test_max_spans_limit to test distinct spans and assert order.

* Apply suggestion from @xrmx

* Create 5296.added

* Rename 5296.added to .changelog/5296.added

---------

Co-authored-by: Riccardo Magliocchetti <riccardo.magliocchetti@gmail.com>

Author: shayannab

.changelog/5296.added

@@ -0,0 +1 @@
+`opentelemetry-sdk`: Make it possible to limit the size of stored spans in InMemorySpanExporter

opentelemetry-sdk/src/opentelemetry/sdk/trace/export/in_memory_span_exporter.py

@@ -3,6 +3,7 @@
 
 import collections.abc
 import threading
+from collections import deque
 
 from opentelemetry.sdk.trace import ReadableSpan
 from opentelemetry.sdk.trace.export import SpanExporter, SpanExportResult
@@ -12,12 +13,17 @@ class InMemorySpanExporter(SpanExporter):
     """Implementation of :class:`.SpanExporter` that stores spans in memory.
 
     This class can be used for testing purposes. It stores the exported spans
-    in a list in memory that can be retrieved using the
+    in a deque in memory that can be retrieved using the
     :func:`.get_finished_spans` method.
+
+    Args:
+        max_spans: Maximum number of spans to store in memory. When the limit
+            is reached, the oldest spans are dropped automatically. If ``None``
+            (default), there is no limit.
     """
 
-    def __init__(self) -> None:
-        self._finished_spans: list[ReadableSpan] = []
+    def __init__(self, max_spans: int | None = None) -> None:
+        self._finished_spans: deque[ReadableSpan] = deque(maxlen=max_spans)
         self._stopped = False
         self._lock = threading.Lock()
 

opentelemetry-sdk/tests/trace/export/test_in_memory_span_exporter.py

@@ -62,3 +62,22 @@ def test_return_code(self):
         # after shutdown export should fail
         ret = memory_exporter.export(span_list)
         self.assertEqual(ret, export.SpanExportResult.FAILURE)
+
+    def test_max_spans_limit(self):
+        """Test that max_spans limits memory usage by dropping oldest spans."""
+        exporter = InMemorySpanExporter(max_spans=2)
+        span1 = trace._Span("span1", mock.Mock(spec=trace_api.SpanContext))
+        span2 = trace._Span("span2", mock.Mock(spec=trace_api.SpanContext))
+        span3 = trace._Span("span3", mock.Mock(spec=trace_api.SpanContext))
+        exporter.export([span1, span2, span3])
+        span_list = exporter.get_finished_spans()
+        self.assertEqual(len(span_list), 2)
+        self.assertSequenceEqual((span2, span3), span_list)
+
+    def test_max_spans_none_is_unlimited(self):
+        """Test that max_spans=None (default) imposes no limit."""
+        exporter = InMemorySpanExporter(max_spans=None)
+        span = trace._Span("name", mock.Mock(spec=trace_api.SpanContext))
+        exporter.export([span] * 100)
+        span_list = exporter.get_finished_spans()
+        self.assertEqual(len(span_list), 100)