Reintroduce weakref, I accidentlly undid that change in my last PR

Author: DylanRussell

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

@@ -50,12 +50,6 @@ class LogExportResult(enum.Enum):
     FAILURE = 1
 
 
-class BatchLogExportStrategy(enum.Enum):
-    EXPORT_ALL = 0
-    EXPORT_WHILE_BATCH_EXCEEDS_THRESHOLD = 1
-    EXPORT_AT_LEAST_ONE_BATCH = 2
-
-
 class LogExporter(abc.ABC):
     """Interface for exporting logs.
     Interface to be implemented by services that want to export logs received

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

@@ -19,6 +19,7 @@
 import logging
 import os
 import threading
+import weakref
 from abc import ABC
 from typing import (
     TYPE_CHECKING,
@@ -38,7 +39,7 @@
 if TYPE_CHECKING:
     from opentelemetry.sdk._logs import LogData
     from opentelemetry.sdk._logs.export import LogExporter
-    from opentelemetry.sdk.trace import Span
+    from opentelemetry.sdk.trace import ReadableSpan
     from opentelemetry.sdk.trace.export import SpanExporter
 
 
@@ -49,7 +50,7 @@ class BatchExportStrategy(enum.Enum):
 
 
 class BatchProcessor(ABC):
-    _queue: Deque["Union[LogData, Span]"]
+    _queue: Deque["Union[LogData, ReadableSpan]"]
 
     def __init__(
         self,
@@ -84,7 +85,8 @@ def __init__(
         self._worker_awaken = threading.Event()
         self._worker_thread.start()
         if hasattr(os, "register_at_fork"):
-            os.register_at_fork(after_in_child=self._at_fork_reinit)  # pylint: disable=protected-access
+            weak_reinit = weakref.WeakMethod(self._at_fork_reinit)
+            os.register_at_fork(after_in_child=lambda: weak_reinit()())  # pyright: ignore[reportOptionalCall] pylint: disable=unnecessary-lambda
         self._pid = os.getpid()
 
     def _should_export_batch(
@@ -156,7 +158,7 @@ def _export(self, batch_strategy: BatchExportStrategy) -> None:
                     )
                 detach(token)
 
-    def emit(self, data: "Union[LogData, Span]") -> None:
+    def emit(self, data: "Union[LogData, ReadableSpan]") -> None:
         if self._shutdown:
             self._logger.info("Shutdown called, ignoring %s.", self._exporting)
             return

opentelemetry-sdk/tests/shared_internal/test_batch_processor.py

@@ -13,10 +13,12 @@
 # limitations under the License.
 
 # pylint: disable=protected-access
+import gc
 import multiprocessing
 import os
 import time
 import unittest
+import weakref
 from concurrent.futures import ThreadPoolExecutor
 from platform import system
 from sys import version_info
@@ -194,3 +196,18 @@ def child(conn):
         batch_processor.force_flush()
         # Single export for the telemetry we emitted at the start of the test.
         assert exporter.export.call_count == 1
+
+    def test_record_processor_is_garbage_collected(
+        self, batch_processor_class, telemetry
+    ):
+        exporter = Mock()
+        processor = batch_processor_class(exporter)
+        weak_ref = weakref.ref(processor)
+        processor.shutdown()
+
+        # When the processor is garbage collected
+        del processor
+        gc.collect()
+
+        # Then the reference to the processor should no longer exist
+        assert weak_ref() is None