From 67f0028bc8c631860e3ee165b6c826792a7b86d3 Mon Sep 17 00:00:00 2001 From: Pablo Pardo Garcia Date: Wed, 19 Aug 2026 18:36:05 +0200 Subject: [PATCH] feat: stamp service.instance.id on spans and share it with the heartbeat MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit One uuid per client, minted at init before the Resource is built: stamped as the service.instance.id resource attribute on every span and injected into HeartbeatSender so heartbeat payloads carry the same identity. The backend can now join heartbeats to traces and count replicas. Stamped unconditionally (heartbeat off, scoped clients, disabled). A forked child keeps re-arming its heartbeat with a fresh id; its spans keep the ancestor id (immutable Resource) — documented in init(). --- src/rius/client.py | 21 ++++++++- src/rius/heartbeat.py | 7 ++- src/rius/semconv.py | 6 +++ tests/test_heartbeat.py | 6 +++ tests/test_instance_id.py | 91 +++++++++++++++++++++++++++++++++++++++ 5 files changed, 129 insertions(+), 2 deletions(-) create mode 100644 tests/test_instance_id.py 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()