diff --git a/CHANGELOG.md b/CHANGELOG.md index 540f7b9d347..a3b4b93030e 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/_shared_internal/_fork.py b/opentelemetry-sdk/src/opentelemetry/sdk/_shared_internal/_fork.py new file mode 100644 index 00000000000..e01b6d882c1 --- /dev/null +++ b/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, + ) diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/metrics/_internal/__init__.py b/opentelemetry-sdk/src/opentelemetry/sdk/metrics/_internal/__init__.py index a2adaa36a98..1402993b6cd 100644 --- a/opentelemetry-sdk/src/opentelemetry/sdk/metrics/_internal/__init__.py +++ b/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 diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/resources/__init__.py b/opentelemetry-sdk/src/opentelemetry/sdk/resources/__init__.py index a04d27e9ab1..addb321be79 100644 --- a/opentelemetry-sdk/src/opentelemetry/sdk/resources/__init__.py +++ b/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()), } ) diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/trace/__init__.py b/opentelemetry-sdk/src/opentelemetry/sdk/trace/__init__.py index e0b639d81cf..2fb92dd12ad 100644 --- a/opentelemetry-sdk/src/opentelemetry/sdk/trace/__init__.py +++ b/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 ): diff --git a/opentelemetry-sdk/tests/metrics/test_meter_provider_fork.py b/opentelemetry-sdk/tests/metrics/test_meter_provider_fork.py new file mode 100644 index 00000000000..252fa8ab9de --- /dev/null +++ b/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) diff --git a/opentelemetry-sdk/tests/resources/test_resources.py b/opentelemetry-sdk/tests/resources/test_resources.py index c083eff1460..ea1d02afab8 100644 --- a/opentelemetry-sdk/tests/resources/test_resources.py +++ b/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() diff --git a/opentelemetry-sdk/tests/test_fork_service_instance_id.py b/opentelemetry-sdk/tests/test_fork_service_instance_id.py new file mode 100644 index 00000000000..6f9ad0ef888 --- /dev/null +++ b/opentelemetry-sdk/tests/test_fork_service_instance_id.py @@ -0,0 +1,93 @@ +# 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 SERVICE_INSTANCE_ID +from opentelemetry.sdk.trace import TracerProvider + +_fork_ctx = ( + multiprocessing.get_context("fork") if system() != "Windows" else None +) + + +def _instance_id(provider): + resource = ( + provider._sdk_config.resource + if isinstance(provider, MeterProvider) + else provider._resource + ) + return resource.attributes.get(SERVICE_INSTANCE_ID) + + +@unittest.skipUnless(hasattr(os, "fork"), "needs *nix") +class TestForkServiceInstanceId(unittest.TestCase): + def test_meter_and_tracer_share_instance_id_after_fork(self): + """Both providers must report the SAME, fresh id in a forked worker.""" + meter_provider = MeterProvider() + tracer_provider = TracerProvider() + + parent_id = _instance_id(meter_provider) + # Before forking, both providers already share the default id. + self.assertEqual(parent_id, _instance_id(tracer_provider)) + + def child(conn): + conn.send( + ( + _instance_id(meter_provider), + _instance_id(tracer_provider), + ) + ) + conn.close() + + parent_conn, child_conn = _fork_ctx.Pipe() + process = _fork_ctx.Process(target=child, args=(child_conn,)) + process.start() + child_meter_id, child_tracer_id = parent_conn.recv() + process.join() + + # Same id across signals, unique at the instance level. + self.assertEqual(child_meter_id, child_tracer_id) + self.assertNotEqual(child_meter_id, parent_id) + + def test_provider_created_after_fork_adopts_child_id(self): + """A provider built post-fork must adopt the child's shared id.""" + meter_provider = MeterProvider() + parent_id = _instance_id(meter_provider) + + def child(conn): + # Created only in the child, after the fork hook has run. + late_tracer_provider = TracerProvider() + conn.send( + ( + _instance_id(meter_provider), + _instance_id(late_tracer_provider), + ) + ) + conn.close() + + parent_conn, child_conn = _fork_ctx.Pipe() + process = _fork_ctx.Process(target=child, args=(child_conn,)) + process.start() + forked_meter_id, late_tracer_id = parent_conn.recv() + process.join() + + self.assertEqual(forked_meter_id, late_tracer_id) + self.assertNotEqual(late_tracer_id, parent_id) diff --git a/opentelemetry-sdk/tests/trace/test_tracer_provider_fork.py b/opentelemetry-sdk/tests/trace/test_tracer_provider_fork.py new file mode 100644 index 00000000000..69f8a1b87af --- /dev/null +++ b/opentelemetry-sdk/tests/trace/test_tracer_provider_fork.py @@ -0,0 +1,114 @@ +# 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.resources import Resource +from opentelemetry.sdk.trace import TracerProvider + +_fork_ctx = ( + multiprocessing.get_context("fork") if system() != "Windows" else None +) + + +@unittest.skipUnless( + hasattr(os, "fork"), + "needs *nix", +) +class TestTracerProviderFork(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 = TracerProvider(resource=resource) + + original_id = provider._resource.attributes.get("service.instance.id") + self.assertEqual(original_id, "original-id") + + provider._reset_service_instance_id("new-id") + + new_id = provider._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 = TracerProvider(resource=resource) + + provider._reset_service_instance_id("new-id") + + attrs = provider._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 = TracerProvider() + + parent_id = provider._resource.attributes.get("service.instance.id") + self.assertIsNotNone(parent_id) + + def child(conn): + child_id = provider._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() + + 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 = TracerProvider() + + def child(conn): + child_id = provider._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() + + self.assertEqual(len(ids), 4)