diff --git a/src/rius/client.py b/src/rius/client.py index 233e085..f69b22c 100644 --- a/src/rius/client.py +++ b/src/rius/client.py @@ -4,6 +4,7 @@ import logging import threading +import uuid from collections.abc import Callable, Sequence from typing import Any @@ -26,7 +27,7 @@ from .instrumentation import enable_instrumentations from .masking import MaskingSpanExporter from .pending import PendingSpanProcessor -from .semconv import TRACER_NAME +from .semconv import SERVICE_INSTANCE_ID, TRACER_NAME from .session import SessionSpanProcessor logger = logging.getLogger(__name__) @@ -171,6 +172,14 @@ def init( (``RIUS_HEARTBEAT``; on by default, set False to opt out). Pings ``/v1/heartbeat`` from init until process exit so the platform can tell a live-but-idle agent from a vanished one. + Each ``init()`` mints one instance id, sent both in heartbeat + payloads and on every span as the ``service.instance.id`` + resource attribute, so the platform can join the two and count + replicas. Fork caveat: a child forked after ``init()`` heartbeats + under a fresh id, but its spans keep the parent's (the OTel + Resource is immutable), identifying the pre-fork process family; + for exact per-worker span identity, call ``init()`` after the + fork (e.g. in gunicorn's ``post_fork``). heartbeat_interval: Seconds between pings (default 15, clamped to ``[5, 300]``; the backend derives staleness from this). agent_name: Identity heartbeats group under; defaults to @@ -260,11 +269,20 @@ def _do_init( partial_spans_delay=partial_spans_delay, session_id=session_id, ) + # One identity per client lifetime, shared by spans (resource) and + # heartbeats (payload instance_id) so the backend can join them. Minted + # here — before the Resource — because the sender is constructed much + # later. Fork caveat: the Resource is immutable, so a forked child's + # spans keep this (ancestor) id while its heartbeat re-arms with a fresh + # one; deployments needing exact per-worker span identity should init() + # after fork. + instance_id = str(uuid.uuid4()) # telemetry.sdk.* is reserved for the OTel SDK itself (Resource.create fills # it); we identify as a distribution via telemetry.distro.*. resource = Resource.create( { "service.name": config.service_name, + SERVICE_INSTANCE_ID: instance_id, "telemetry.distro.name": "glassflow-rius", "telemetry.distro.version": __version__, } @@ -346,6 +364,7 @@ def _do_init( headers=config.headers, interval=config.heartbeat_interval, agent_name=config.agent_name, + instance_id=instance_id, tracker=tracker, transport=heartbeat_transport, ) diff --git a/src/rius/heartbeat.py b/src/rius/heartbeat.py index c77e4e7..6700936 100644 --- a/src/rius/heartbeat.py +++ b/src/rius/heartbeat.py @@ -156,6 +156,7 @@ def __init__( headers: dict[str, str], interval: float, agent_name: str, + instance_id: str, tracker: OpenRootSpanTracker, transport: Callable[[dict[str, Any]], None] | None = None, ping_timeout: float = _PING_TIMEOUT_S, @@ -171,7 +172,11 @@ def __init__( self._send: Callable[[dict[str, Any], float], None] = lambda p, _t: transport(p) else: self._send = _http_transport(url, headers) - self._instance_id = str(uuid.uuid4()) + # Injected: the same id rides spans as the service.instance.id + # resource attribute, which is what lets the backend join heartbeats + # to traces. Only _reset_in_child mints a fresh one (a forked child + # is a new process lifetime). + self._instance_id = instance_id self._stop_event = threading.Event() self._stopped = False self._lock = threading.Lock() diff --git a/src/rius/semconv.py b/src/rius/semconv.py index 53e5585..7be0df0 100644 --- a/src/rius/semconv.py +++ b/src/rius/semconv.py @@ -17,6 +17,12 @@ # coordination, tracked separately. TRACER_NAME = "glassflow" +# --- Resource attribute keys --- +# OTel standard identity of one process lifetime (one uuid per client, minted +# at init). The heartbeat payload's instance_id carries the SAME value, which +# is what lets the backend join heartbeats to traces and count replicas. +SERVICE_INSTANCE_ID = "service.instance.id" + # --- Attribute keys --- # OpenInference OPENINFERENCE_SPAN_KIND = "openinference.span.kind" diff --git a/tests/test_heartbeat.py b/tests/test_heartbeat.py index 2f65e84..28e61d2 100644 --- a/tests/test_heartbeat.py +++ b/tests/test_heartbeat.py @@ -141,6 +141,7 @@ def _sender( headers={}, interval=interval, agent_name="checkout-agent", + instance_id=str(uuid.uuid4()), tracker=tracker or OpenRootSpanTracker(), transport=sent.append, ) @@ -201,6 +202,7 @@ def transport(payload: dict[str, Any]) -> None: headers={}, interval=3600.0, agent_name="a", + instance_id=str(uuid.uuid4()), tracker=OpenRootSpanTracker(), transport=transport, ) @@ -313,6 +315,7 @@ def broken(_: dict[str, Any]) -> None: headers={}, interval=3600.0, agent_name="a", + instance_id=str(uuid.uuid4()), tracker=OpenRootSpanTracker(), transport=broken, ) @@ -351,6 +354,7 @@ def log_message(self, *args: Any) -> None: # silence test output headers={"Authorization": "Bearer test-key"}, interval=3600.0, agent_name="e2e-agent", + instance_id=str(uuid.uuid4()), tracker=OpenRootSpanTracker(), ) sender._send_ping() # noqa: SLF001 @@ -436,6 +440,7 @@ def test_stop_with_dead_slow_endpoint_returns_quickly() -> None: headers={}, interval=3600.0, agent_name="a", + instance_id=str(uuid.uuid4()), tracker=OpenRootSpanTracker(), ping_timeout=0.3, final_ping_timeout=0.3, @@ -457,6 +462,7 @@ def test_final_ping_timeout_defaults_shorter_than_ping_timeout() -> None: headers={}, interval=3600.0, agent_name="a", + instance_id=str(uuid.uuid4()), tracker=OpenRootSpanTracker(), ) assert sender._final_ping_timeout < sender._ping_timeout # noqa: SLF001 diff --git a/tests/test_instance_id.py b/tests/test_instance_id.py new file mode 100644 index 0000000..e6d12af --- /dev/null +++ b/tests/test_instance_id.py @@ -0,0 +1,91 @@ +"""Instance identity: one id per client, on spans AND heartbeats (RIUS-436). + +The heartbeat identifies a process lifetime with an ``instance_id``; spans +must carry the same identity as the standard ``service.instance.id`` +resource attribute, so the backend can join heartbeats to traces and +distinguish replicas sharing one ``agent_name``. The id belongs to the +client, not the heartbeat feature: it is stamped even with heartbeat off. +""" + +from __future__ import annotations + +import uuid +from typing import Any + +from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter + +from rius import init +from rius.semconv import SERVICE_INSTANCE_ID + + +def _resource_instance_id(client: Any) -> object: + return client._provider.resource.attributes.get(SERVICE_INSTANCE_ID) # noqa: SLF001 + + +def test_resource_contains_service_instance_id() -> None: + client = init(set_global=False, service_name="svc", span_exporter=InMemorySpanExporter()) + try: + instance_id = _resource_instance_id(client) + assert isinstance(instance_id, str) + uuid.UUID(instance_id) # a valid UUID + finally: + client.shutdown() + + +def test_exported_spans_carry_the_instance_id() -> None: + exporter = InMemorySpanExporter() + client = init(set_global=False, service_name="svc", span_exporter=exporter) + try: + client.get_tracer().start_span("root").end() + assert client.flush() + (span,) = exporter.get_finished_spans() + assert span.resource.attributes[SERVICE_INSTANCE_ID] == _resource_instance_id(client) + finally: + client.shutdown() + + +def test_heartbeat_instance_id_matches_resource() -> None: + sent: list[dict[str, Any]] = [] + client = init( + set_global=False, + service_name="svc", + heartbeat=True, + heartbeat_transport=sent.append, + span_exporter=InMemorySpanExporter(), + ) + try: + client._heartbeat._send_ping() # noqa: SLF001 — deterministic ping + assert sent[-1]["instance_id"] == _resource_instance_id(client) + finally: + client.shutdown() + + +def test_instance_id_present_with_heartbeat_disabled() -> None: + client = init( + set_global=False, + service_name="svc", + heartbeat=False, + span_exporter=InMemorySpanExporter(), + ) + try: + assert isinstance(_resource_instance_id(client), str) + finally: + client.shutdown() + + +def test_instance_id_present_when_disabled() -> None: + client = init(set_global=False, service_name="svc", disabled=True) + try: + assert isinstance(_resource_instance_id(client), str) + finally: + client.shutdown() + + +def test_distinct_clients_get_distinct_ids() -> None: + a = init(set_global=False, service_name="svc", span_exporter=InMemorySpanExporter()) + b = init(set_global=False, service_name="svc", span_exporter=InMemorySpanExporter()) + try: + assert _resource_instance_id(a) != _resource_instance_id(b) + finally: + a.shutdown() + b.shutdown()