Fix memory leaks with weakref module

Author: aabmass

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

@@ -18,6 +18,7 @@
 from threading import Lock
 from time import time_ns
 from typing import Optional, Sequence
+import weakref
 
 # This kind of import is needed to avoid Sphinx errors.
 import opentelemetry.sdk.metrics
@@ -393,7 +394,7 @@ class MeterProvider(APIMeterProvider):
     """
 
     _all_metric_readers_lock = Lock()
-    _all_metric_readers = set()
+    _all_metric_readers = weakref.WeakSet()
 
     def __init__(
         self,

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

@@ -22,6 +22,7 @@
 from threading import Event, Lock, RLock, Thread
 from time import time_ns
 from typing import IO, Callable, Dict, Iterable, Optional
+import weakref
 
 from typing_extensions import final
 
@@ -490,8 +491,9 @@ def __init__(
             )
             self._daemon_thread.start()
             if hasattr(os, "register_at_fork"):
+                weak_at_fork = weakref.WeakMethod(self._at_fork_reinit)
                 os.register_at_fork(
-                    after_in_child=self._at_fork_reinit
+                    after_in_child=lambda: weak_at_fork()()
                 )  # pylint: disable=protected-access
         elif self._export_interval_millis <= 0:
             raise ValueError(

reproduce.py

@@ -0,0 +1,58 @@
+import gc
+import logging
+import weakref
+
+from opentelemetry.exporter.otlp.proto.http.metric_exporter import (
+    OTLPMetricExporter,
+)
+from opentelemetry.sdk.metrics import MeterProvider
+from opentelemetry.sdk.metrics._internal.export import (
+    PeriodicExportingMetricReader,
+)
+
+logging.basicConfig(level=logging.DEBUG)
+
+
+def create_and_clean():
+    # setup_otlp_exporter
+    otlp_exporter = OTLPMetricExporter(
+        endpoint="http://localhost:4318/v1/metrics"
+    )
+    otlp_exporter_weakref = weakref.ref(otlp_exporter)
+
+    reader = PeriodicExportingMetricReader(
+        otlp_exporter, export_interval_millis=5000
+    )
+    reader_weakref = weakref.ref(reader)
+
+    provider = MeterProvider(metric_readers=[reader])
+    provider_weakref = weakref.ref(provider)
+
+    provider.shutdown()
+    return otlp_exporter_weakref, reader_weakref, provider_weakref
+
+
+def check_referrers(wr, name):
+    if wr() is not None:
+        logging.warning("%s was not properly garbage collected", name)
+        referrers = gc.get_referrers(wr())
+        logging.debug(f"Direct referrers to %s: {len(referrers)}", name)
+        for ref in referrers:
+            logging.debug(f"%s referrer {str(ref)} type: {type(ref)}", name)
+    else:
+        logging.info("%s was properly garbage collected", name)
+
+
+def main():
+    otlp_exporter_weakref, reader_weakref, provider_weakref = (
+        create_and_clean()
+    )
+    gc.collect()
+
+    check_referrers(otlp_exporter_weakref, "OTLP EXPORTER")
+    check_referrers(reader_weakref, "READER")
+    check_referrers(provider_weakref, "PROVIDER")
+
+
+if __name__ == "__main__":
+    main()