Skip to content
Open
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@

# pylint: disable=unused-import

import os
from abc import ABC, abstractmethod
from threading import Lock
from time import time_ns
Expand Down Expand Up @@ -76,8 +77,25 @@ def __init__(
self._async_instruments: List[
"opentelemetry.sdk.metrics._internal.instrument._Asynchronous"
] = []
if hasattr(os, "register_at_fork"):
os.register_at_fork(
after_in_child=self._at_fork_reinit
) # pylint: disable=protected-access

def _at_fork_reinit(self):
"""Reinitialize lock in child process after fork"""
self._lock._at_fork_reinit()
# Lazy reinitialization of storages on first use post fork. This is
# done to avoid the overhead of reinitializing the storages on
# every fork.
self._needs_storage_reinit = True
self._async_instruments.clear()

def consume_measurement(self, measurement: Measurement) -> None:
if getattr(self, '_needs_storage_reinit', False):
self._reinit_storages()
self._needs_storage_reinit = False

should_sample_exemplar = (
self._sdk_config.exemplar_filter.should_sample(
measurement.value,
Expand Down Expand Up @@ -105,6 +123,11 @@ def collect(
metric_reader: "opentelemetry.sdk.metrics.MetricReader",
timeout_millis: float = 10_000,
) -> Optional[Iterable[Metric]]:

if getattr(self, '_needs_storage_reinit', False):
self._reinit_storages()
self._needs_storage_reinit = False

with self._lock:
metric_reader_storage = self._reader_storages[metric_reader]
# for now, just use the defaults
Expand Down Expand Up @@ -143,3 +166,11 @@ def collect(
result = self._reader_storages[metric_reader].collect()

return result

def _reinit_storages(self):
# Reinitialize the storages. Use to reinitialize the storages after a
# fork to avoid duplicate data points.
with self._lock:
for storage in self._reader_storages.values():
storage._lock._at_fork_reinit()
storage._instrument_view_instrument_matches.clear()