feat: stamp service.instance.id on spans and share it with the heartbeat

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
             ``<endpoint>/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,
         )

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()

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"

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

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()