fix(sdk): regenerate service.instance.id post-fork in MeterProvider and TracerProvider

CHANGELOG.md

@@ -12,6 +12,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## Unreleased
 
+- `opentelemetry-sdk`: Add `service.instance.id` to default resource so every process gets a unique instance identity at startup
+  ([#5000](https://github.com/open-telemetry/opentelemetry-python/pull/5000))
+- `opentelemetry-sdk`: Regenerate a single shared `service.instance.id` post-fork, applied to every registered `MeterProvider` and `TracerProvider` (including providers created after the fork), so prefork workers get distinct but cross-signal-consistent resource identities
+  ([#5000](https://github.com/open-telemetry/opentelemetry-python/pull/5000))
 - `opentelemetry-sdk`: Add file configuration support with YAML/JSON loading, environment variable substitution, and schema validation against the vendored OTel config JSON schema
   ([#4898](https://github.com/open-telemetry/opentelemetry-python/pull/4898))
 - Fix intermittent CI failures in `getting-started` and `tracecontext` jobs caused by GitHub git CDN SHA propagation lag by installing contrib packages from the already-checked-out local copy instead of a second git clone

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

@@ -0,0 +1,98 @@
+# Copyright The OpenTelemetry Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Shared post-fork handling of ``service.instance.id`` for SDK providers.
+
+When a prefork server (gunicorn, uWSGI, ...) forks worker processes, every
+worker inherits the same :class:`~opentelemetry.sdk.resources.Resource` -
+including the same ``service.instance.id`` - from the master process. Exporting
+telemetry from several workers under one identity collides in the backend
+(last-write-wins instead of correct aggregation).
+
+This module owns a *single* :func:`os.register_at_fork` hook and a registry of
+providers. After a fork, **one** fresh ``service.instance.id`` is generated and
+applied to *every* registered provider, so all signals (metrics and traces) in
+a given worker share the same, unique instance id. Providers created *after* the
+fork adopt the same id at construction time via :func:`register_provider`.
+"""
+
+from __future__ import annotations
+
+import os
+import weakref
+from threading import Lock
+from typing import Optional, Protocol, runtime_checkable
+from uuid import uuid4
+
+from opentelemetry.sdk.resources import SERVICE_INSTANCE_ID, Resource
+
+
+@runtime_checkable
+class _ForkAwareProvider(Protocol):
+    def _reset_service_instance_id(self, service_instance_id: str) -> None: ...
+
+
+_lock = Lock()
+# WeakSet so providers that go out of scope are dropped automatically, matching
+# the WeakMethod/WeakSet pattern used elsewhere in the SDK for fork hooks.
+_providers: "weakref.WeakSet[_ForkAwareProvider]" = weakref.WeakSet()
+# The instance id for the current process generation. ``None`` until the first
+# fork, so providers in a never-forked process keep the id already present in
+# their resource (the one from the default resource detector).
+_service_instance_id: Optional[str] = None
+
+
+def reset_service_instance_id(resource: Resource, service_instance_id: str) -> Resource:
+    """Return ``resource`` with ``service.instance.id`` replaced.
+
+    All other resource attributes are preserved via :meth:`Resource.merge`.
+    """
+    return resource.merge(Resource({SERVICE_INSTANCE_ID: service_instance_id}))
+
+
+def register_provider(provider: _ForkAwareProvider) -> None:
+    """Register ``provider`` so it is refreshed on the next fork.
+
+    If the current process is already a forked child (i.e. a fork happened
+    before this provider was created), the provider immediately adopts the
+    child's shared instance id so late-created providers stay consistent with
+    the ones that existed at fork time.
+    """
+    with _lock:
+        _providers.add(provider)
+        service_instance_id = _service_instance_id
+    if service_instance_id is not None:
+        provider._reset_service_instance_id(service_instance_id)
+
+
+def _after_in_child() -> None:
+    # ``_lock`` is held across the fork (acquired in ``before``), so we own
+    # exclusive access to the registry here and release it once done.
+    global _service_instance_id  # noqa: PLW0603
+    service_instance_id = str(uuid4())
+    _service_instance_id = service_instance_id
+    providers = list(_providers)
+    _lock.release()
+    for provider in providers:
+        provider._reset_service_instance_id(service_instance_id)
+
+
+if hasattr(os, "register_at_fork"):
+    # Acquiring ``_lock`` in ``before`` guarantees no other thread holds it
+    # while we fork, avoiding a deadlock in the single-threaded child.
+    os.register_at_fork(
+        before=_lock.acquire,
+        after_in_parent=_lock.release,
+        after_in_child=_after_in_child,
+    )

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

@@ -34,6 +34,7 @@
 )
 from opentelemetry.metrics import UpDownCounter as APIUpDownCounter
 from opentelemetry.metrics import _Gauge as APIGauge
+from opentelemetry.sdk._shared_internal import _fork
 from opentelemetry.sdk.environment_variables import (
     OTEL_METRICS_EXEMPLAR_FILTER,
     OTEL_SDK_DISABLED,
@@ -456,6 +457,11 @@ def __init__(
         self._shutdown_once = Once()
         self._shutdown = False
 
+        # Register with the shared fork hook so every provider in a forked
+        # worker gets the same fresh service.instance.id (see
+        # opentelemetry.sdk._shared_internal._fork).
+        _fork.register_provider(self)
+
         for metric_reader in self._sdk_config.metric_readers:
             with self._all_metric_readers_lock:
                 if metric_reader in self._all_metric_readers:
@@ -471,6 +477,19 @@ def __init__(
                 self._measurement_consumer.collect
             )
 
+    def _reset_service_instance_id(self, service_instance_id: str) -> None:
+        """Apply a post-fork ``service.instance.id`` shared across providers.
+
+        Called by the shared fork hook (see
+        ``opentelemetry.sdk._shared_internal._fork``) in each forked worker so
+        that metrics and traces report the same, unique instance id without
+        colliding with sibling workers. All other resource attributes are
+        preserved.
+        """
+        self._sdk_config.resource = _fork.reset_service_instance_id(
+            self._sdk_config.resource, service_instance_id
+        )
+
     def force_flush(self, timeout_millis: float = 10_000) -> bool:
         deadline_ns = time_ns() + timeout_millis * 10**6
 

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

@@ -72,6 +72,7 @@
 from types import ModuleType
 from typing import List, Optional, Set, cast
 from urllib import parse
+from uuid import uuid4
 
 from opentelemetry.attributes import BoundedAttributes
 from opentelemetry.sdk.environment_variables import (
@@ -323,6 +324,7 @@ def to_json(self, indent: Optional[int] = 4) -> str:
         TELEMETRY_SDK_LANGUAGE: "python",
         TELEMETRY_SDK_NAME: "opentelemetry",
         TELEMETRY_SDK_VERSION: _OPENTELEMETRY_SDK_VERSION,
+        SERVICE_INSTANCE_ID: str(uuid4()),
     }
 )
 

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

@@ -51,6 +51,7 @@
 from opentelemetry import trace as trace_api
 from opentelemetry.attributes import BoundedAttributes
 from opentelemetry.sdk import util
+from opentelemetry.sdk._shared_internal import _fork
 from opentelemetry.sdk.environment_variables import (
     OTEL_ATTRIBUTE_COUNT_LIMIT,
     OTEL_ATTRIBUTE_VALUE_LENGTH_LIMIT,
@@ -1366,6 +1367,24 @@ def __init__(
             _tracer_configurator or _default_tracer_configurator
         )
 
+        # Register with the shared fork hook so every provider in a forked
+        # worker gets the same fresh service.instance.id (see
+        # opentelemetry.sdk._shared_internal._fork).
+        _fork.register_provider(self)
+
+    def _reset_service_instance_id(self, service_instance_id: str) -> None:
+        """Apply a post-fork ``service.instance.id`` shared across providers.
+
+        Called by the shared fork hook (see
+        ``opentelemetry.sdk._shared_internal._fork``) in each forked worker so
+        that metrics and traces report the same, unique instance id without
+        colliding with sibling workers. All other resource attributes are
+        preserved.
+        """
+        self._resource = _fork.reset_service_instance_id(
+            self._resource, service_instance_id
+        )
+
     def _set_tracer_configurator(
         self, *, tracer_configurator: _TracerConfiguratorT
     ):

opentelemetry-sdk/tests/metrics/test_meter_provider_fork.py

@@ -0,0 +1,126 @@
+# Copyright The OpenTelemetry Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# pylint: disable=protected-access
+
+import multiprocessing
+import os
+import unittest
+from platform import system
+
+from opentelemetry.sdk.metrics import MeterProvider
+from opentelemetry.sdk.resources import Resource
+
+_fork_ctx = (
+    multiprocessing.get_context("fork") if system() != "Windows" else None
+)
+
+
+@unittest.skipUnless(
+    hasattr(os, "fork"),
+    "needs *nix",
+)
+class TestMeterProviderFork(unittest.TestCase):
+    def test_reset_service_instance_id_changes_service_instance_id(self):
+        """_reset_service_instance_id should assign the given service.instance.id."""
+        resource = Resource({"service.instance.id": "original-id"})
+        provider = MeterProvider(resource=resource)
+
+        original_id = provider._sdk_config.resource.attributes.get(
+            "service.instance.id"
+        )
+        self.assertEqual(original_id, "original-id")
+
+        provider._reset_service_instance_id("new-id")
+
+        new_id = provider._sdk_config.resource.attributes.get(
+            "service.instance.id"
+        )
+        self.assertEqual(new_id, "new-id")
+
+    def test_reset_service_instance_id_preserves_other_resource_attributes(
+        self,
+    ):
+        """_reset_service_instance_id should not affect other resource attributes."""
+        resource = Resource(
+            {
+                "service.name": "my-service",
+                "service.instance.id": "original-id",
+                "deployment.environment": "production",
+            }
+        )
+        provider = MeterProvider(resource=resource)
+
+        provider._reset_service_instance_id("new-id")
+
+        attrs = provider._sdk_config.resource.attributes
+        self.assertEqual(attrs.get("service.name"), "my-service")
+        self.assertEqual(attrs.get("deployment.environment"), "production")
+
+    def test_fork_produces_unique_service_instance_ids(self):
+        """Each forked worker should get a distinct service.instance.id."""
+        provider = MeterProvider()
+
+        parent_id = provider._sdk_config.resource.attributes.get(
+            "service.instance.id"
+        )
+        self.assertIsNotNone(parent_id)
+
+        def child(conn):
+            child_id = provider._sdk_config.resource.attributes.get(
+                "service.instance.id"
+            )
+            conn.send(child_id)
+            conn.close()
+
+        parent_conn, child_conn = _fork_ctx.Pipe()
+        process = _fork_ctx.Process(target=child, args=(child_conn,))
+        process.start()
+        child_id = parent_conn.recv()
+        process.join()
+
+        # Child should have a different service.instance.id than parent
+        self.assertNotEqual(parent_id, child_id)
+        self.assertIsNotNone(child_id)
+
+    def test_multiple_forks_produce_unique_service_instance_ids(self):
+        """Each of N forked workers should have a distinct service.instance.id."""
+        provider = MeterProvider()
+
+        def child(conn):
+            child_id = provider._sdk_config.resource.attributes.get(
+                "service.instance.id"
+            )
+            conn.send(child_id)
+            conn.close()
+
+        ids = set()
+        processes = []
+        conns = []
+
+        for _ in range(4):
+            parent_conn, child_conn = _fork_ctx.Pipe()
+            process = _fork_ctx.Process(target=child, args=(child_conn,))
+            processes.append(process)
+            conns.append(parent_conn)
+            process.start()
+
+        for conn in conns:
+            ids.add(conn.recv())
+
+        for process in processes:
+            process.join()
+
+        # All 4 workers should have distinct IDs
+        self.assertEqual(len(ids), 4)

opentelemetry-sdk/tests/resources/test_resources.py

@@ -45,6 +45,7 @@
     PROCESS_RUNTIME_DESCRIPTION,
     PROCESS_RUNTIME_NAME,
     PROCESS_RUNTIME_VERSION,
+    SERVICE_INSTANCE_ID,
     SERVICE_NAME,
     TELEMETRY_SDK_LANGUAGE,
     TELEMETRY_SDK_NAME,
@@ -88,6 +89,9 @@ def test_create(self):
             TELEMETRY_SDK_LANGUAGE: "python",
             TELEMETRY_SDK_VERSION: _OPENTELEMETRY_SDK_VERSION,
             SERVICE_NAME: "unknown_service",
+            SERVICE_INSTANCE_ID: _DEFAULT_RESOURCE.attributes[
+                SERVICE_INSTANCE_ID
+            ],
         }
 
         resource = Resource.create(attributes)
@@ -211,6 +215,9 @@ def test_immutability(self):
             TELEMETRY_SDK_LANGUAGE: "python",
             TELEMETRY_SDK_VERSION: _OPENTELEMETRY_SDK_VERSION,
             SERVICE_NAME: "unknown_service",
+            SERVICE_INSTANCE_ID: _DEFAULT_RESOURCE.attributes[
+                SERVICE_INSTANCE_ID
+            ],
         }
 
         attributes_copy = attributes.copy()