diff --git a/src/glassflow/client.py b/src/glassflow/client.py index 5778657..c815929 100644 --- a/src/glassflow/client.py +++ b/src/glassflow/client.py @@ -108,6 +108,7 @@ def init( agent_name: str | None = None, heartbeat_transport: Callable[[dict[str, Any]], None] | None = None, partial_spans: bool | None = None, + partial_spans_delay: float | None = None, set_global: bool = True, ) -> GlassflowClient: """Initialize the SDK: build a tracer provider that exports OTLP traces. @@ -168,6 +169,7 @@ def init( agent_name=agent_name, heartbeat_transport=heartbeat_transport, partial_spans=partial_spans, + partial_spans_delay=partial_spans_delay, set_global=set_global, ) @@ -189,6 +191,7 @@ def _do_init( agent_name: str | None, heartbeat_transport: Callable[[dict[str, Any]], None] | None, partial_spans: bool | None, + partial_spans_delay: float | None, set_global: bool, ) -> GlassflowClient: global _current_client @@ -204,6 +207,7 @@ def _do_init( heartbeat_interval=heartbeat_interval, agent_name=agent_name, partial_spans=partial_spans, + partial_spans_delay=partial_spans_delay, ) # telemetry.sdk.* is reserved for the OTel SDK itself (Resource.create fills # it); we identify as a distribution via telemetry.distro.*. @@ -227,7 +231,9 @@ def _do_init( if config.partial_spans: # Pending snapshots ride the SAME batch pipeline as final spans # (exporter, retries, masking); see pending.py for the contract. - provider.add_span_processor(PendingSpanProcessor(batch_processor)) + provider.add_span_processor( + PendingSpanProcessor(batch_processor, delay=config.partial_spans_delay) + ) provider.add_span_processor(batch_processor) if set_global and not config.disabled: diff --git a/src/glassflow/config.py b/src/glassflow/config.py index 68b62bf..2c994f0 100644 --- a/src/glassflow/config.py +++ b/src/glassflow/config.py @@ -25,6 +25,7 @@ ENV_HEARTBEAT_INTERVAL = "GLASSFLOW_HEARTBEAT_INTERVAL" ENV_AGENT_NAME = "GLASSFLOW_AGENT_NAME" ENV_PARTIAL_SPANS = "GLASSFLOW_PARTIAL_SPANS" +ENV_PARTIAL_SPANS_DELAY = "GLASSFLOW_PARTIAL_SPANS_DELAY" # The backend expresses staleness as multiples of the interval, so the clamp # bounds are part of the heartbeat wire contract. @@ -32,6 +33,12 @@ HEARTBEAT_INTERVAL_MAX = 300.0 DEFAULT_HEARTBEAT_INTERVAL = 15.0 +# Debounce for partial spans (GLA2-244): 0 = emit immediately at span start; +# N>0 = emit only if the span is still open after N seconds. Beyond 60s a +# "live" view stops being live, so larger values are clamped. +PARTIAL_SPANS_DELAY_MIN = 0.0 +PARTIAL_SPANS_DELAY_MAX = 60.0 + _TRUENESS = frozenset({"1", "true", "yes", "on"}) @@ -71,6 +78,7 @@ class GlassflowConfig: heartbeat_interval: float = DEFAULT_HEARTBEAT_INTERVAL agent_name: str = DEFAULT_SERVICE_NAME partial_spans: bool = False + partial_spans_delay: float = 0.0 @property def traces_endpoint(self) -> str: @@ -92,6 +100,21 @@ def _clamp_sample_rate(value: float) -> float: return clamped +def _clamp_partial_spans_delay(value: float) -> float: + """Clamp to [0, 60] — out-of-range degrades, never crashes init().""" + if PARTIAL_SPANS_DELAY_MIN <= value <= PARTIAL_SPANS_DELAY_MAX: + return value + clamped = min(max(value, PARTIAL_SPANS_DELAY_MIN), PARTIAL_SPANS_DELAY_MAX) + logger.warning( + "partial_spans_delay %s is outside [%s, %s]; clamped to %s", + value, + PARTIAL_SPANS_DELAY_MIN, + PARTIAL_SPANS_DELAY_MAX, + clamped, + ) + return clamped + + def _clamp_heartbeat_interval(value: float) -> float: """Clamp to the contract bounds — out-of-range degrades, never crashes init().""" if HEARTBEAT_INTERVAL_MIN <= value <= HEARTBEAT_INTERVAL_MAX: @@ -120,6 +143,7 @@ def resolve_config( heartbeat_interval: float | None = None, agent_name: str | None = None, partial_spans: bool | None = None, + partial_spans_delay: float | None = None, ) -> GlassflowConfig: """Resolve SDK configuration from arguments, environment, then defaults. @@ -155,6 +179,11 @@ def resolve_config( sampled span at span START (``GLASSFLOW_PARTIAL_SPANS``), so in-flight work is visible and crashes leave a record. Off by default until the backend's unfinished-spans storage ships. + partial_spans_delay: Debounce for pending snapshots + (``GLASSFLOW_PARTIAL_SPANS_DELAY``), clamped to ``[0, 60]`` + seconds. ``0`` (default) emits at span start; ``N`` emits only if + the span is still open after N seconds — spans that finish + sooner cost no network at all. Returns: The resolved, immutable ``GlassflowConfig``. @@ -180,6 +209,11 @@ def resolve_config( resolved_partial_spans = ( _env_bool(ENV_PARTIAL_SPANS, default=False) if partial_spans is None else partial_spans ) + resolved_partial_spans_delay = _clamp_partial_spans_delay( + _env_float(ENV_PARTIAL_SPANS_DELAY, default=0.0) + if partial_spans_delay is None + else partial_spans_delay + ) resolved_headers = dict(headers or {}) has_auth = any(key.lower() == "authorization" for key in resolved_headers) @@ -198,4 +232,5 @@ def resolve_config( heartbeat_interval=resolved_heartbeat_interval, agent_name=resolved_agent_name, partial_spans=resolved_partial_spans, + partial_spans_delay=resolved_partial_spans_delay, ) diff --git a/src/glassflow/pending.py b/src/glassflow/pending.py index 8d38e7e..37a6f1b 100644 --- a/src/glassflow/pending.py +++ b/src/glassflow/pending.py @@ -18,14 +18,27 @@ - Identity/taxonomy attributes only (``PENDING_IDENTITY_ATTRIBUTES`` / ``_PREFIXES``); NEVER content, whatever instrumentation set it. -v1 emits immediately on start (Logfire-style). The emission seam -(:meth:`PendingSpanProcessor._emit`) exists so a debounce ("only emit if -still open after N seconds" — the volume escape valve) can be added later -without changing the wire contract. +Debounce (GLA2-244): with ``partial_spans_delay > 0`` the snapshot is held +for N seconds and only emitted if the span is STILL OPEN then — a span that +finishes first costs zero network. Most agent spans live milliseconds, so a +small delay cuts pending volume drastically while keeping the live view +useful (anything worth watching live is open longer than the delay). The +snapshot is still built at ``on_start`` and held, never rebuilt at emit +time: content set during the delay (``set_input`` etc.) can never leak onto +a pending. A delayed pending is byte-identical to an immediate one — zero +wire/backend/UI impact. """ from __future__ import annotations +import heapq +import itertools +import logging +import os +import threading +import time +import weakref +from collections.abc import Callable from typing import Any from opentelemetry import context as otel_context @@ -38,6 +51,136 @@ PENDING_IDENTITY_PREFIXES, ) +logger = logging.getLogger(__name__) + +_SpanKey = tuple[int, int] # (trace_id, span_id) + +# ``os.register_at_fork`` callbacks can never be unregistered, so the hook is +# installed once at module level over a weak set of live schedulers — the same +# pattern as the heartbeat sender. A forked child re-arms its scheduler thread +# with an EMPTY registry: the parent's open spans are not the child's. +_active_schedulers: weakref.WeakSet[PendingScheduler] = weakref.WeakSet() +_fork_hook_installed = False +_fork_lock = threading.Lock() + + +def _reset_schedulers_in_child() -> None: # pragma: no cover - exercised via fork + for scheduler in list(_active_schedulers): + scheduler._at_fork_reinit() + + +def _install_fork_hook() -> None: + global _fork_hook_installed + with _fork_lock: + if _fork_hook_installed or not hasattr(os, "register_at_fork"): + return + os.register_at_fork(after_in_child=_reset_schedulers_in_child) + _fork_hook_installed = True + + +class PendingScheduler: + """Delays snapshot emission; a span ending first cancels its snapshot. + + ONE daemon thread regardless of span volume: deadlines live in a heap, + snapshots in a key->snapshot registry. ``cancel`` just drops the registry + entry (heap entries for cancelled keys are discarded lazily), so both + ``schedule`` and ``cancel`` are O(log n) / O(1) — safe on the span hot + path. ``clock`` and ``start_thread`` are injectable for tests. + """ + + def __init__( + self, + *, + emit: Callable[[ReadableSpan], None], + delay: float, + clock: Callable[[], float] = time.monotonic, + start_thread: bool = True, + ) -> None: + self._emit_fn = emit + self._delay = delay + self._clock = clock + self._cond = threading.Condition() + self._heap: list[tuple[float, int, _SpanKey]] = [] + self._snapshots: dict[_SpanKey, ReadableSpan] = {} + self._counter = itertools.count() # heap tiebreaker + self._stopped = False + self._thread: threading.Thread | None = None + if start_thread: + _active_schedulers.add(self) + _install_fork_hook() + self._start_thread() + + def _start_thread(self) -> None: + self._thread = threading.Thread( + target=self._run, name="glassflow-pending-scheduler", daemon=True + ) + self._thread.start() + + def schedule(self, key: _SpanKey, snapshot: ReadableSpan) -> None: + with self._cond: + if self._stopped: + return + self._snapshots[key] = snapshot + heapq.heappush(self._heap, (self._clock() + self._delay, next(self._counter), key)) + self._cond.notify_all() + + def cancel(self, key: _SpanKey) -> None: + """Span ended before its deadline: the pending never hits the wire.""" + with self._cond: + self._snapshots.pop(key, None) + + def pop_due(self) -> None: + """Emit every snapshot whose deadline has passed (thread and tests).""" + due: list[ReadableSpan] = [] + with self._cond: + now = self._clock() + while self._heap and self._heap[0][0] <= now: + _, _, key = heapq.heappop(self._heap) + snapshot = self._snapshots.pop(key, None) + if snapshot is not None: # None = cancelled, discard lazily + due.append(snapshot) + for snapshot in due: + try: + self._emit_fn(snapshot) + except Exception: # noqa: BLE001 - never propagate into the SDK + logger.debug("pending snapshot emission failed", exc_info=True) + + def shutdown(self) -> None: + """Drop everything not yet due: the final spans are being flushed at + this moment, so any pending emitted now would be instantly superseded.""" + with self._cond: + self._stopped = True + self._snapshots.clear() + self._heap.clear() + self._cond.notify_all() + if self._thread is not None: + self._thread.join(timeout=1.0) + + def _at_fork_reinit(self) -> None: # pragma: no cover - exercised via fork + # Fresh lock (the parent's may be held mid-fork), empty registry, new + # thread: parent spans do not exist in the child. + self._cond = threading.Condition() + self._heap = [] + self._snapshots = {} + if not self._stopped: + self._start_thread() + + def _run(self) -> None: + while True: + with self._cond: + if self._stopped: + return + # discard cancelled heads so the timeout tracks a LIVE deadline + while self._heap and self._heap[0][2] not in self._snapshots: + heapq.heappop(self._heap) + timeout = None + if self._heap: + timeout = max(0.0, self._heap[0][0] - self._clock()) + self._cond.wait(timeout) + if self._stopped: + return + self.pop_due() + def _identity_attributes(attributes: Any) -> dict[str, Any]: """Filter a span's start-time attributes down to the pending allowlist.""" @@ -50,6 +193,10 @@ def _identity_attributes(attributes: Any) -> dict[str, Any]: } +def _span_key(context: Any) -> _SpanKey: + return (context.trace_id, context.span_id) + + class PendingSpanProcessor(SpanProcessor): """Exports a pending snapshot of every sampled span at ``on_start``. @@ -57,15 +204,26 @@ class PendingSpanProcessor(SpanProcessor): (``delegate.on_end``), so pendings share the exporter, batching, retry, and masking pipeline with final spans — nothing bespoke on the wire path. ``on_start`` stays an in-memory enqueue: the never-block guarantee holds. + + With ``delay > 0`` (GLA2-244) emission is debounced through a + :class:`PendingScheduler`; ``delay == 0`` keeps the emit-immediately + behavior with no scheduler thread at all. """ - def __init__(self, delegate: SpanProcessor) -> None: + def __init__(self, delegate: SpanProcessor, *, delay: float = 0.0) -> None: self._delegate = delegate + self._scheduler: PendingScheduler | None = None + if delay > 0: + self._scheduler = PendingScheduler(emit=delegate.on_end, delay=delay) def on_start(self, span: Span, parent_context: otel_context.Context | None = None) -> None: if not span.is_recording(): return - self._emit(self._snapshot(span)) + snapshot = self._snapshot(span) + if self._scheduler is not None: + self._scheduler.schedule(_span_key(span.get_span_context()), snapshot) + else: + self._emit(snapshot) @staticmethod def _snapshot(span: Span) -> ReadableSpan: @@ -87,15 +245,22 @@ def _snapshot(span: Span) -> ReadableSpan: ) def _emit(self, snapshot: ReadableSpan) -> None: - # The debounce seam: a future timer wraps THIS call (delay + cancel on - # early end), leaving the snapshot construction and wire shape alone. self._delegate.on_end(snapshot) - def on_end(self, span: ReadableSpan) -> None: # pragma: no cover - no-op - pass + def on_end(self, span: ReadableSpan) -> None: + # The debounce cancellation hook: a span that ends within the delay + # never sends its pending at all. + if self._scheduler is not None and span.context is not None: + self._scheduler.cancel(_span_key(span.context)) - def shutdown(self) -> None: # pragma: no cover - delegate owns the exporter - pass + def shutdown(self) -> None: + if self._scheduler is not None: + self._scheduler.shutdown() def force_flush(self, timeout_millis: int = 30000) -> bool: + # Deliberately NOT a drop (deviation from the ticket's prose, kept to + # its ACs): flush() happens mid-operation — killing scheduled pendings + # here would silently disable liveness for spans that stay open. The + # batch delegate flushes its own queue; not-yet-due pendings simply + # emit later if their spans are still open. return True diff --git a/tests/test_pending_delay.py b/tests/test_pending_delay.py new file mode 100644 index 0000000..2c9fb5f --- /dev/null +++ b/tests/test_pending_delay.py @@ -0,0 +1,141 @@ +"""Debounced partial spans (GLA2-244): delay emission, cancel on fast finish. + +Timing is injected everywhere (fake monotonic clocks, bounded Event waits) — +no test sleeps. +""" + +from __future__ import annotations + +import threading + +from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter + +from glassflow import init +from glassflow.pending import PendingScheduler +from glassflow.semconv import GLASSFLOW_SPAN_PENDING + +# --- config resolution ------------------------------------------------------- + + +def test_delay_defaults_to_zero() -> None: + from glassflow.config import resolve_config + + assert resolve_config().partial_spans_delay == 0.0 + + +def test_delay_env_var_and_clamp(monkeypatch) -> None: + from glassflow.config import resolve_config + + monkeypatch.setenv("GLASSFLOW_PARTIAL_SPANS_DELAY", "2.5") + assert resolve_config().partial_spans_delay == 2.5 + # explicit argument wins; out-of-range clamps instead of crashing + assert resolve_config(partial_spans_delay=9999).partial_spans_delay == 60.0 + assert resolve_config(partial_spans_delay=-1).partial_spans_delay == 0.0 + + +# --- scheduler core (pure logic, no thread) ----------------------------------- + + +class _Clock: + def __init__(self) -> None: + self.now = 100.0 + + def __call__(self) -> float: + return self.now + + +def _scheduler(emitted: list, clock: _Clock, delay: float = 5.0) -> PendingScheduler: + return PendingScheduler(emit=emitted.append, delay=delay, clock=clock, start_thread=False) + + +def test_snapshot_not_due_before_delay() -> None: + emitted: list = [] + clock = _Clock() + s = _scheduler(emitted, clock) + s.schedule(("t", 1), "snapshot-1") + s.pop_due() + assert emitted == [] + + +def test_snapshot_emitted_once_after_delay() -> None: + emitted: list = [] + clock = _Clock() + s = _scheduler(emitted, clock) + s.schedule(("t", 1), "snapshot-1") + clock.now += 5.0 + s.pop_due() + s.pop_due() # idempotent: never emitted twice + assert emitted == ["snapshot-1"] + + +def test_cancel_before_due_means_no_emission() -> None: + emitted: list = [] + clock = _Clock() + s = _scheduler(emitted, clock) + s.schedule(("t", 1), "snapshot-1") + s.cancel(("t", 1)) + clock.now += 60.0 + s.pop_due() + assert emitted == [] + + +def test_shutdown_drops_scheduled_pendings() -> None: + emitted: list = [] + clock = _Clock() + s = _scheduler(emitted, clock) + s.schedule(("t", 1), "snapshot-1") + s.shutdown() + clock.now += 60.0 + s.pop_due() + assert emitted == [] + # post-shutdown schedules are ignored, not errors + s.schedule(("t", 2), "snapshot-2") + clock.now += 60.0 + s.pop_due() + assert emitted == [] + + +def test_thread_emits_when_due() -> None: + """One real-thread smoke test: emission signals an Event (bounded wait).""" + done = threading.Event() + s = PendingScheduler(emit=lambda _snap: done.set(), delay=0.01, start_thread=True) + s.schedule(("t", 1), "snapshot-1") + assert done.wait(timeout=5.0), "scheduler thread never emitted the due snapshot" + s.shutdown() + + +# --- end-to-end through init() ------------------------------------------------ + + +def _memory_client(**kwargs: object): + exporter = InMemorySpanExporter() + client = init( + span_exporter=exporter, + set_global=False, + service_name="test-svc", + instruments=[], + **kwargs, # type: ignore[arg-type] + ) + return client, exporter + + +def test_fast_span_produces_no_pending_on_the_wire() -> None: + """The whole point: a span finishing within the delay costs zero network.""" + client, exporter = _memory_client(partial_spans=True, partial_spans_delay=30.0) + with client.get_tracer().start_as_current_span("quick"): + pass + client.flush() + spans = exporter.get_finished_spans() + assert len(spans) == 1 + assert GLASSFLOW_SPAN_PENDING not in spans[0].attributes + + +def test_delay_zero_keeps_immediate_emission() -> None: + client, exporter = _memory_client(partial_spans=True, partial_spans_delay=0.0) + with client.get_tracer().start_as_current_span("op"): + pass + client.flush() + markers = [ + bool(s.attributes.get(GLASSFLOW_SPAN_PENDING)) for s in exporter.get_finished_spans() + ] + assert sorted(markers) == [False, True]