opentelemetry-sdk: Add ability to refresh process sensitive Resource attributes

.changelog/5280.added

@@ -0,0 +1 @@
+`opentelemetry-sdk`: Add ability to refresh process dependent Resource attributes

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

@@ -8,6 +8,7 @@
 import concurrent.futures
 import json
 import logging
+import os
 import threading
 import traceback
 import warnings
@@ -21,7 +22,7 @@
     cast,
     overload,
 )
-from weakref import WeakSet
+from weakref import WeakMethod, WeakSet
 
 from typing_extensions import deprecated
 
@@ -56,7 +57,10 @@
 from opentelemetry.sdk.environment_variables._internal import (
     parse_boolean_environment_variable,
 )
-from opentelemetry.sdk.resources import Resource
+from opentelemetry.sdk.resources import (
+    Resource,
+    _get_process_dependent_resource,
+)
 from opentelemetry.sdk.util import ns_to_iso_str
 from opentelemetry.sdk.util._configurator import RuleBasedConfigurator
 from opentelemetry.sdk.util.instrumentation import (
@@ -690,6 +694,9 @@ def _is_enabled(self) -> bool:
     def _set_logger_config(self, logger_config: _LoggerConfig) -> None:
         self._logger_config = logger_config
 
+    def _set_resource(self, resource: Resource) -> None:
+        self._resource = resource
+
     @property
     def instrumentation_scope(self):
         return self._instrumentation_scope
@@ -809,11 +816,31 @@ def __init__(
         self._logger_cache_lock = Lock()
         self._active_loggers: WeakSet[Logger] = WeakSet()
         self._active_loggers_lock = Lock()
+        if hasattr(os, "register_at_fork"):
+            weak_at_fork = WeakMethod(self._handle_fork)
+
+            def _after_in_child() -> None:
+                if at_fork := weak_at_fork():
+                    at_fork()
+
+            os.register_at_fork(after_in_child=_after_in_child)
+
+    def _handle_fork(self) -> None:
+        self._logger_cache_lock = Lock()
+        self._active_loggers_lock = Lock()
+        self._update_resource(_get_process_dependent_resource())
 
     @property
     def resource(self):
         return self._resource
 
+    def _update_resource(self, resource: Resource) -> None:
+        with self._active_loggers_lock:
+            self._resource = self._resource.merge(resource)
+            for logger in list(self._active_loggers):
+                # pylint: disable-next=protected-access
+                logger._set_resource(self._resource)
+
     def _get_logger_no_cache(
         self,
         name: str,
@@ -894,7 +921,7 @@ def _set_logger_configurator(
         """
         self._logger_configurator = logger_configurator
         with self._active_loggers_lock:
-            for logger in self._active_loggers:
+            for logger in list(self._active_loggers):
                 # pylint: disable-next=protected-access
                 logger._set_logger_config(
                     self._apply_logger_configurator(

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

@@ -1,6 +1,7 @@
 # Copyright The OpenTelemetry Authors
 # SPDX-License-Identifier: Apache-2.0
 
+import os
 import weakref
 from atexit import register, unregister
 from collections.abc import Callable, Sequence
@@ -51,7 +52,10 @@
 from opentelemetry.sdk.metrics._internal.sdk_configuration import (
     SdkConfiguration,
 )
-from opentelemetry.sdk.resources import Resource
+from opentelemetry.sdk.resources import (
+    Resource,
+    _get_process_dependent_resource,
+)
 from opentelemetry.sdk.util._configurator import RuleBasedConfigurator
 from opentelemetry.sdk.util.instrumentation import (
     InstrumentationScope,
@@ -526,6 +530,21 @@ def __init__(
             )
             metric_reader._set_meter_provider(self)
 
+        if hasattr(os, "register_at_fork"):
+            weak_at_fork = weakref.WeakMethod(self._handle_fork)
+
+            def _after_in_child() -> None:
+                if at_fork := weak_at_fork():
+                    at_fork()
+
+            os.register_at_fork(after_in_child=_after_in_child)
+
+    def _handle_fork(self) -> None:
+        self._lock = Lock()
+        self._meter_lock = Lock()
+        type(self)._all_metric_readers_lock = Lock()
+        self._update_resource(_get_process_dependent_resource())
+
     def _set_meter_configurator(
         self, *, meter_configurator: _MeterConfiguratorT
     ):
@@ -543,6 +562,12 @@ def _set_meter_configurator(
                     self._apply_meter_configurator(instrumentation_scope)
                 )
 
+    def _update_resource(self, resource: Resource) -> None:
+        with self._meter_lock:
+            self._sdk_config.resource = self._sdk_config.resource.merge(
+                resource
+            )
+
     def _apply_meter_configurator(
         self, instrumentation_scope: InstrumentationScope
     ) -> _MeterConfig:

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

@@ -293,6 +293,21 @@ def detect(self) -> "Resource":
         """Don't call `Resource.create` here to avoid an infinite loop, instead instantiate `Resource` directly"""
         raise NotImplementedError()
 
+    # pylint: disable-next=no-self-use
+    def is_process_dependent(self) -> bool:
+        """Return whether this detector depends on the current process identity.
+
+        Process dependent detectors may return resource attributes that become
+        stale after a process identity change, such as :func:`os.fork`.
+        Detectors returning ``True`` should be re-run after such changes so the
+        resulting :class:`Resource` describes the current process.
+
+        Returns:
+            ``True`` if this detector should be re-run after process identity
+            changes otherwise ``False``.
+        """
+        return False
+
 
 class OTELResourceDetector(ResourceDetector):
     # pylint: disable=no-self-use
@@ -339,6 +354,10 @@ def __init__(
         super().__init__(raise_on_error=raise_on_error)
         self._include_command_args = include_command_args
 
+    # pylint: disable=no-self-use
+    def is_process_dependent(self) -> bool:
+        return True
+
     def detect(self) -> "Resource":
         _runtime_version = ".".join(
             map(
@@ -491,8 +510,19 @@ class ServiceInstanceIdResourceDetector(ResourceDetector):
     UUID for service.instance.id to uniquely identify each service instance.
     The ID is shared across all detector instances within the same process and
     regenerated automatically when the process PID changes (e.g. after a fork).
+
+    Note: because this detector is process dependent, providers refresh it
+    automatically after a fork and merge the newly detected value on top of
+    the existing resource. This means that if a user explicitly sets
+    `service.instance.id` (e.g. via OTEL_RESOURCE_ATTRIBUTES or
+    Resource.create(attributes=...)), that value will be overwritten with a
+    newly generated UUID the next time the process forks. This differs from
+    Resource.create(), which does preserve user supplied overrides.
     """
 
+    def is_process_dependent(self) -> bool:
+        return True
+
     def detect(self) -> "Resource":
         # pylint: disable-next=global-statement
         global _service_instance_id, _service_instance_id_pid
@@ -577,6 +607,17 @@ def _build_resource_detectors() -> list["ResourceDetector"]:
     return detectors
 
 
+def _get_process_dependent_resource() -> Resource:  # pyright: ignore[reportUnusedFunction]
+    return get_aggregated_resources(
+        [
+            detector
+            for detector in _build_resource_detectors()
+            if detector.is_process_dependent()
+        ],
+        Resource.get_empty(),
+    )
+
+
 def get_aggregated_resources(
     detectors: Sequence["ResourceDetector"],
     initial_resource: Resource | None = None,

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

@@ -50,7 +50,10 @@
 from opentelemetry.sdk.environment_variables._internal import (
     parse_boolean_environment_variable,
 )
-from opentelemetry.sdk.resources import Resource
+from opentelemetry.sdk.resources import (
+    Resource,
+    _get_process_dependent_resource,
+)
 from opentelemetry.sdk.trace import sampling
 from opentelemetry.sdk.trace._tracer_metrics import create_tracer_metrics
 from opentelemetry.sdk.trace.id_generator import IdGenerator, RandomIdGenerator
@@ -1141,6 +1144,9 @@ def __init__(
     def _set_tracer_config(self, tracer_config: _TracerConfig):
         self._tracer_config = tracer_config
 
+    def _set_resource(self, resource: Resource) -> None:
+        self.resource = resource
+
     def _is_enabled(self) -> bool:
         """If the tracer is not enabled, start_span will create a NonRecordingSpan"""
         return self._tracer_config.is_enabled
@@ -1345,6 +1351,18 @@ def __init__(
         )
         self._tracers_lock = threading.Lock()
         self._tracers: dict[InstrumentationScope, Tracer] = {}
+        if hasattr(os, "register_at_fork"):
+            weak_at_fork = weakref.WeakMethod(self._handle_fork)
+
+            def _after_in_child() -> None:
+                if at_fork := weak_at_fork():
+                    at_fork()
+
+            os.register_at_fork(after_in_child=_after_in_child)
+
+    def _handle_fork(self) -> None:
+        self._tracers_lock = threading.Lock()
+        self._update_resource(_get_process_dependent_resource())
 
     def _set_tracer_configurator(
         self, *, tracer_configurator: _TracerConfiguratorT
@@ -1367,6 +1385,12 @@ def _set_tracer_configurator(
     def resource(self) -> Resource:
         return self._resource
 
+    def _update_resource(self, resource: Resource) -> None:
+        with self._tracers_lock:
+            self._resource = self._resource.merge(resource)
+            for tracer in self._tracers.values():
+                tracer._set_resource(self._resource)  # pylint: disable=protected-access
+
     def _apply_tracer_configurator(
         self, instrumentation_scope: InstrumentationScope
     ):

opentelemetry-sdk/tests/logs/scripts/logger_provider_resource_after_fork.py

@@ -0,0 +1,82 @@
+# Copyright The OpenTelemetry Authors
+# SPDX-License-Identifier: Apache-2.0
+
+import json
+import os
+
+from opentelemetry._logs import LogRecord
+from opentelemetry.sdk._logs import LoggerProvider
+from opentelemetry.sdk._logs.export import (
+    InMemoryLogRecordExporter,
+    SimpleLogRecordProcessor,
+)
+from opentelemetry.sdk.resources import PROCESS_PID
+
+
+# pylint: disable-next=too-many-locals
+def main() -> None:
+    exporter = InMemoryLogRecordExporter()
+    logger_provider = LoggerProvider(shutdown_on_exit=False)
+    logger_provider.add_log_record_processor(
+        SimpleLogRecordProcessor(exporter)
+    )
+    logger = logger_provider.get_logger("cached")
+    parent_pid = os.getpid()
+    parent_resource_pid = logger_provider.resource.attributes[PROCESS_PID]
+    parent_logger_pid = logger.resource.attributes[PROCESS_PID]
+
+    pid = os.fork()
+    if not pid:
+        child_pid = os.getpid()
+        new_logger = logger_provider.get_logger("new")
+        logger.emit(LogRecord(observed_timestamp=0, body="cached"))
+        new_logger.emit(LogRecord(observed_timestamp=0, body="new"))
+        finished_logs = exporter.get_finished_logs()
+        print(
+            json.dumps(
+                {
+                    "child_pid": child_pid,
+                    "provider_pid": logger_provider.resource.attributes[
+                        PROCESS_PID
+                    ],
+                    "cached_logger_pid": logger.resource.attributes[
+                        PROCESS_PID
+                    ],
+                    "new_logger_pid": new_logger.resource.attributes[
+                        PROCESS_PID
+                    ],
+                    "exported_resource_pids": [
+                        log.resource.attributes[PROCESS_PID]
+                        for log in finished_logs
+                    ],
+                    "log_bodies": sorted(
+                        log.log_record.body for log in finished_logs
+                    ),
+                }
+            ),
+            flush=True,
+        )
+        # pylint: disable-next=protected-access
+        os._exit(0)
+
+    os.waitpid(pid, 0)
+    print(
+        json.dumps(
+            {
+                "parent_pid": parent_pid,
+                "parent_resource_pid": parent_resource_pid,
+                "parent_logger_pid": parent_logger_pid,
+                "parent_resource_pid_after_fork": logger_provider.resource.attributes[
+                    PROCESS_PID
+                ],
+                "parent_logger_pid_after_fork": logger.resource.attributes[
+                    PROCESS_PID
+                ],
+            }
+        ),
+        flush=True,
+    )
+
+
+if __name__ == "__main__":
+    main()