From 806cbb7b0f9011b2e942737687a82e6bd24b3792 Mon Sep 17 00:00:00 2001 From: Pablo Pardo Garcia Date: Mon, 27 Jul 2026 12:59:41 +0200 Subject: [PATCH 1/2] feat: emit partial (pending) spans at span start (GLA2-195) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Opt-in (init(partial_spans=True) / GLASSFLOW_PARTIAL_SPANS, default OFF until the backend's unfinished-spans storage ships): every sampled span additionally exports a content-free snapshot at START — same trace/span/parent ids, name, and start timestamp as the final span, zero duration, marked glassflow.span.pending=true. The backend stores it as an unfinished row the real span replaces at end; a snapshot that is never replaced is the durable record of what a crashed agent was doing. Mechanics: PendingSpanProcessor's on_start builds a ReadableSpan snapshot and delegates it to the provider's existing BatchSpanProcessor (shared exporter/batching/retries/masking; on_start stays an in-memory enqueue — the never-block guarantee holds). Attributes are filtered by an identity ALLOWLIST (kind, operation, provider, tool name, plus the gen_ai.request.* prefix) so content can never ride a pending, whatever instrumentation set it. Sampled-out spans and disabled mode produce no snapshots. The marker key knowingly bends the convention-native rule: OTel has no pending-span mechanism to align with (spec #3732/#4646, semconv #2133 open, none planned); rationale recorded in semconv.py. Supporting change: the SDK's own span/generation helpers now attach identity attributes at span CREATION (kind_attributes/_creation_ attributes) instead of only immediately after — on_start-built snapshots would otherwise see empty attributes. Final spans are byte-identical (the same values were previously set post-creation). The _emit seam is the future debounce point (only emit if still open after N seconds — the network-volume escape valve for fast spans); designed for, deliberately not built (follow-up ticket). --- src/glassflow/client.py | 12 ++- src/glassflow/config.py | 11 +++ src/glassflow/generation.py | 21 ++++- src/glassflow/pending.py | 101 ++++++++++++++++++++++ src/glassflow/semconv.py | 38 +++++++++ src/glassflow/spans.py | 16 +++- tests/test_pending_spans.py | 165 ++++++++++++++++++++++++++++++++++++ 7 files changed, 358 insertions(+), 6 deletions(-) create mode 100644 src/glassflow/pending.py create mode 100644 tests/test_pending_spans.py diff --git a/src/glassflow/client.py b/src/glassflow/client.py index 9cfa15a..5778657 100644 --- a/src/glassflow/client.py +++ b/src/glassflow/client.py @@ -19,6 +19,7 @@ from .heartbeat import HeartbeatSender, OpenRootSpanTracker from .instrumentation import enable_instrumentations from .masking import MaskingSpanExporter +from .pending import PendingSpanProcessor from .semconv import TRACER_NAME logger = logging.getLogger(__name__) @@ -106,6 +107,7 @@ def init( heartbeat_interval: float | None = None, agent_name: str | None = None, heartbeat_transport: Callable[[dict[str, Any]], None] | None = None, + partial_spans: bool | None = None, set_global: bool = True, ) -> GlassflowClient: """Initialize the SDK: build a tracer provider that exports OTLP traces. @@ -165,6 +167,7 @@ def init( heartbeat_interval=heartbeat_interval, agent_name=agent_name, heartbeat_transport=heartbeat_transport, + partial_spans=partial_spans, set_global=set_global, ) @@ -185,6 +188,7 @@ def _do_init( heartbeat_interval: float | None, agent_name: str | None, heartbeat_transport: Callable[[dict[str, Any]], None] | None, + partial_spans: bool | None, set_global: bool, ) -> GlassflowClient: global _current_client @@ -199,6 +203,7 @@ def _do_init( heartbeat=heartbeat, heartbeat_interval=heartbeat_interval, agent_name=agent_name, + partial_spans=partial_spans, ) # telemetry.sdk.* is reserved for the OTel SDK itself (Resource.create fills # it); we identify as a distribution via telemetry.distro.*. @@ -218,7 +223,12 @@ def _do_init( exporter = MaskingSpanExporter( exporter, capture_content=config.capture_content, mask=mask ) - provider.add_span_processor(BatchSpanProcessor(exporter)) + batch_processor = BatchSpanProcessor(exporter) + 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(batch_processor) if set_global and not config.disabled: trace.set_tracer_provider(provider) diff --git a/src/glassflow/config.py b/src/glassflow/config.py index b62fc28..68b62bf 100644 --- a/src/glassflow/config.py +++ b/src/glassflow/config.py @@ -24,6 +24,7 @@ ENV_HEARTBEAT = "GLASSFLOW_HEARTBEAT" ENV_HEARTBEAT_INTERVAL = "GLASSFLOW_HEARTBEAT_INTERVAL" ENV_AGENT_NAME = "GLASSFLOW_AGENT_NAME" +ENV_PARTIAL_SPANS = "GLASSFLOW_PARTIAL_SPANS" # The backend expresses staleness as multiples of the interval, so the clamp # bounds are part of the heartbeat wire contract. @@ -69,6 +70,7 @@ class GlassflowConfig: heartbeat: bool = False heartbeat_interval: float = DEFAULT_HEARTBEAT_INTERVAL agent_name: str = DEFAULT_SERVICE_NAME + partial_spans: bool = False @property def traces_endpoint(self) -> str: @@ -117,6 +119,7 @@ def resolve_config( heartbeat: bool | None = None, heartbeat_interval: float | None = None, agent_name: str | None = None, + partial_spans: bool | None = None, ) -> GlassflowConfig: """Resolve SDK configuration from arguments, environment, then defaults. @@ -148,6 +151,10 @@ def resolve_config( agent_name: Identity heartbeats group under (``GLASSFLOW_AGENT_NAME``); defaults to ``service_name`` so the agents view and the traces view agree on what an "agent" is. + partial_spans: Export a content-free pending snapshot of every + 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. Returns: The resolved, immutable ``GlassflowConfig``. @@ -170,6 +177,9 @@ def resolve_config( else heartbeat_interval ) resolved_agent_name = agent_name or os.getenv(ENV_AGENT_NAME) or resolved_service_name + resolved_partial_spans = ( + _env_bool(ENV_PARTIAL_SPANS, default=False) if partial_spans is None else partial_spans + ) resolved_headers = dict(headers or {}) has_auth = any(key.lower() == "authorization" for key in resolved_headers) @@ -187,4 +197,5 @@ def resolve_config( heartbeat=resolved_heartbeat, heartbeat_interval=resolved_heartbeat_interval, agent_name=resolved_agent_name, + partial_spans=resolved_partial_spans, ) diff --git a/src/glassflow/generation.py b/src/glassflow/generation.py index 3f4e9f8..70ebdea 100644 --- a/src/glassflow/generation.py +++ b/src/glassflow/generation.py @@ -31,6 +31,7 @@ GEN_AI_USAGE_OUTPUT_TOKENS, TRACER_NAME, SpanKind, + kind_attributes, set_span_kind, ) @@ -234,6 +235,18 @@ def _configure( generation.set_input(input) +def _creation_attributes(model: str | None, provider: str | None, operation: str) -> dict[str, str]: + """Identity attributes for an LLM span at CREATION (pending snapshots + are built at on_start; anything set later is invisible to them).""" + attributes = kind_attributes(SpanKind.LLM) + attributes[GEN_AI_OPERATION_NAME] = operation + if model is not None: + attributes[GEN_AI_REQUEST_MODEL] = model + if provider is not None: + attributes[GEN_AI_PROVIDER_NAME] = provider + return attributes + + def start_generation( name: str, *, @@ -260,7 +273,9 @@ def start_generation( Returns: A ``Generation`` handle; call ``.end()`` when the call completes. """ - span = trace.get_tracer(TRACER_NAME, __version__).start_span(name) + span = trace.get_tracer(TRACER_NAME, __version__).start_span( + name, attributes=_creation_attributes(model, provider, operation) + ) generation = Generation(span) _configure( generation, @@ -294,7 +309,9 @@ def start_as_current_generation( metadata; the span ends when the block exits. """ tracer = trace.get_tracer(TRACER_NAME, __version__) - with tracer.start_as_current_span(name) as span: + with tracer.start_as_current_span( + name, attributes=_creation_attributes(model, provider, operation) + ) as span: generation = Generation(span) _configure( generation, diff --git a/src/glassflow/pending.py b/src/glassflow/pending.py new file mode 100644 index 0000000..8d38e7e --- /dev/null +++ b/src/glassflow/pending.py @@ -0,0 +1,101 @@ +"""Partial (pending) spans: a content-free snapshot exported at span start. + +Spans normally leave the process only when they END, so an in-flight agent +run is invisible and a crashed one exports nothing. With ``partial_spans`` +enabled, every sampled span additionally exports a snapshot at START; the +backend stores it as an unfinished row that the real span replaces at end +(same trace/span id and start timestamp — the identity the storage layer +keys replacement on), and a snapshot that is never replaced is the durable +record of what a crashed agent was doing. + +Wire contract (GLA2-195): + +- Same trace_id, span_id, parent, name, and start timestamp as the final + span; ``end_time == start_time`` (OTLP cannot represent an unfinished + span, so the snapshot is an ended zero-duration span with a marker). +- The ``glassflow.span.pending`` marker attribute (see ``semconv.py`` for + why a vendor-namespaced key is unavoidable here). +- 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. +""" + +from __future__ import annotations + +from typing import Any + +from opentelemetry import context as otel_context +from opentelemetry.sdk.trace import ReadableSpan, Span, SpanProcessor +from opentelemetry.trace import Status, StatusCode + +from .semconv import ( + GLASSFLOW_SPAN_PENDING, + PENDING_IDENTITY_ATTRIBUTES, + PENDING_IDENTITY_PREFIXES, +) + + +def _identity_attributes(attributes: Any) -> dict[str, Any]: + """Filter a span's start-time attributes down to the pending allowlist.""" + if not attributes: + return {} + return { + key: value + for key, value in attributes.items() + if key in PENDING_IDENTITY_ATTRIBUTES or key.startswith(PENDING_IDENTITY_PREFIXES) + } + + +class PendingSpanProcessor(SpanProcessor): + """Exports a pending snapshot of every sampled span at ``on_start``. + + Delegates the snapshot to the provider's existing batch processor + (``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. + """ + + def __init__(self, delegate: SpanProcessor) -> None: + self._delegate = delegate + + 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)) + + @staticmethod + def _snapshot(span: Span) -> ReadableSpan: + attributes = _identity_attributes(span.attributes) + attributes[GLASSFLOW_SPAN_PENDING] = True + return ReadableSpan( + name=span.name, + context=span.get_span_context(), + parent=span.parent, + resource=span.resource, + attributes=attributes, + events=(), + links=(), + kind=span.kind, + instrumentation_scope=span.instrumentation_scope, + status=Status(StatusCode.UNSET), + start_time=span.start_time, + end_time=span.start_time, # zero duration: unfinished, marked + ) + + 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 shutdown(self) -> None: # pragma: no cover - delegate owns the exporter + pass + + def force_flush(self, timeout_millis: int = 30000) -> bool: + return True diff --git a/src/glassflow/semconv.py b/src/glassflow/semconv.py index ab6f232..bc2cb89 100644 --- a/src/glassflow/semconv.py +++ b/src/glassflow/semconv.py @@ -40,6 +40,30 @@ # gen_ai.* naming style, precedent Langfuse's completion_start_time. GEN_AI_FIRST_TOKEN_EVENT = "gen_ai.first_token" +# --- Pending (partial) spans (GLA2-195) --- +# Marks the content-free snapshot exported at span START; the backend maps it +# to Finished=0 and the real span replaces it at end. This key knowingly bends +# the convention-native rule (no glassflow.* namespace): OpenTelemetry has NO +# pending-span mechanism to align with (spec #3732/#4646, semconv #2133 — all +# open, none planned), and the only shipping precedent (Logfire's +# logfire.span_type) is equally vendor-namespaced. +GLASSFLOW_SPAN_PENDING = "glassflow.span.pending" + +# Attributes allowed to ride a pending snapshot: identity/taxonomy known at +# span start. An ALLOWLIST on purpose — content exclusion must hold for +# third-party instrumentors' attribute families too, and a blocklist would +# have to enumerate all of them. +PENDING_IDENTITY_ATTRIBUTES = frozenset( + { + OPENINFERENCE_SPAN_KIND, + GEN_AI_OPERATION_NAME, + GEN_AI_PROVIDER_NAME, + GEN_AI_TOOL_NAME, + } +) +# gen_ai.request.* (model, temperature, ...) is identity, not content. +PENDING_IDENTITY_PREFIXES = (GEN_AI_REQUEST_PREFIX,) + # Attribute keys carrying user content — masked/stripped at export (see masking.py). CONTENT_ATTRIBUTES = frozenset( { @@ -107,6 +131,20 @@ class SpanKind(str, Enum): } +def kind_attributes(kind: SpanKind) -> dict[str, str]: + """Identity attributes for a span of ``kind``, for setting at CREATION. + + Pending snapshots (pending.py) are built at ``on_start``, so taxonomy set + via ``set_attribute`` afterwards is invisible to them — passing these at + span creation is what makes a pending span classifiable. + """ + attributes = {OPENINFERENCE_SPAN_KIND: kind.value} + operation = _OPERATION_BY_KIND.get(kind) + if operation is not None: + attributes[GEN_AI_OPERATION_NAME] = operation + return attributes + + def set_span_kind(span: Span, kind: SpanKind) -> None: """Stamp a span with its OpenInference kind and (if applicable) gen_ai operation.""" span.set_attribute(OPENINFERENCE_SPAN_KIND, kind.value) diff --git a/src/glassflow/spans.py b/src/glassflow/spans.py index e04a182..0acdbc6 100644 --- a/src/glassflow/spans.py +++ b/src/glassflow/spans.py @@ -24,7 +24,14 @@ from . import __version__ from ._serde import serialize -from .semconv import INPUT_VALUE, OUTPUT_VALUE, TRACER_NAME, SpanKind, set_span_kind +from .semconv import ( + INPUT_VALUE, + OUTPUT_VALUE, + TRACER_NAME, + SpanKind, + kind_attributes, + set_span_kind, +) class Observation: @@ -98,7 +105,10 @@ def start_span(name: str, *, kind: SpanKind = SpanKind.CHAIN, input: Any = None) current span and does not auto-record exceptions. Use ``start_as_current_span`` for block-scoped tracing. """ - span = trace.get_tracer(TRACER_NAME, __version__).start_span(name) + # kind at CREATION so pending snapshots (on_start) can classify the span + span = trace.get_tracer(TRACER_NAME, __version__).start_span( + name, attributes=kind_attributes(kind) + ) observation = Observation(span) _configure(observation, kind, input) return observation @@ -117,7 +127,7 @@ def start_as_current_span( (OpenTelemetry's ``start_as_current_span`` default), then re-raised. """ tracer = trace.get_tracer(TRACER_NAME, __version__) - with tracer.start_as_current_span(name) as span: + with tracer.start_as_current_span(name, attributes=kind_attributes(kind)) as span: observation = Observation(span) _configure(observation, kind, input) yield observation diff --git a/tests/test_pending_spans.py b/tests/test_pending_spans.py new file mode 100644 index 0000000..c1ad3df --- /dev/null +++ b/tests/test_pending_spans.py @@ -0,0 +1,165 @@ +"""Partial (pending) spans: a content-free snapshot exported at span start (GLA2-195). + +Wire contract under test: same trace/span/parent ids, same name and start +timestamp as the final span; zero duration; the pending marker attribute; +identity/taxonomy attributes only, never content. +""" + +from __future__ import annotations + +from opentelemetry.sdk.trace import SpanProcessor as _SpanProcessor +from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter + +from glassflow import init +from glassflow.semconv import GLASSFLOW_SPAN_PENDING + + +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 _split(spans): + pending = [s for s in spans if s.attributes.get(GLASSFLOW_SPAN_PENDING)] + final = [s for s in spans if not s.attributes.get(GLASSFLOW_SPAN_PENDING)] + return pending, final + + +def test_flag_off_by_default_behavior_unchanged() -> None: + client, exporter = _memory_client() + with client.get_tracer().start_as_current_span("op"): + pass + client.flush() + spans = exporter.get_finished_spans() + assert len(spans) == 1 + assert GLASSFLOW_SPAN_PENDING not in spans[0].attributes + + +def test_pending_snapshot_mirrors_identity_of_final_span() -> None: + client, exporter = _memory_client(partial_spans=True) + with client.get_tracer().start_as_current_span("op"): + pass + client.flush() + (pending,), (final,) = _split(exporter.get_finished_spans()) + + assert pending.attributes[GLASSFLOW_SPAN_PENDING] is True + # identical identity: the ClickHouse sort key must match for replacement + assert pending.context.trace_id == final.context.trace_id + assert pending.context.span_id == final.context.span_id + assert pending.name == final.name + assert pending.start_time == final.start_time + # zero duration: OTLP cannot represent an unfinished span + assert pending.end_time == pending.start_time + assert final.end_time > final.start_time + + +def test_pending_preserves_parent_linkage() -> None: + client, exporter = _memory_client(partial_spans=True) + tracer = client.get_tracer() + with tracer.start_as_current_span("root"), tracer.start_as_current_span("child"): + pass + client.flush() + pending, final = _split(exporter.get_finished_spans()) + pending_child = next(s for s in pending if s.name == "child") + final_child = next(s for s in final if s.name == "child") + assert pending_child.parent is not None + assert pending_child.parent.span_id == final_child.parent.span_id + + +def test_pending_carries_identity_attributes_but_never_content() -> None: + client, exporter = _memory_client(partial_spans=True) + with client.get_tracer().start_as_current_span( + "chat", + attributes={ + "openinference.span.kind": "LLM", + "gen_ai.request.model": "gpt-4o", + "gen_ai.request.temperature": 0.2, + "input.value": "SECRET-CONTENT", + }, + ): + pass + client.flush() + (pending,), (final,) = _split(exporter.get_finished_spans()) + assert pending.attributes["openinference.span.kind"] == "LLM" + assert pending.attributes["gen_ai.request.model"] == "gpt-4o" + assert pending.attributes["gen_ai.request.temperature"] == 0.2 + assert "input.value" not in pending.attributes, "content must NEVER ride a pending span" + assert final.attributes["input.value"] == "SECRET-CONTENT" + + +def test_sampled_out_spans_produce_no_pending() -> None: + client, exporter = _memory_client(partial_spans=True, sample_rate=0.0) + with client.get_tracer().start_as_current_span("op"): + pass + client.flush() + assert exporter.get_finished_spans() == () + + +def test_disabled_kills_pendings_too() -> None: + exporter = InMemorySpanExporter() + client = init( + span_exporter=exporter, set_global=False, disabled=True, partial_spans=True, instruments=[] + ) + with client.get_tracer().start_as_current_span("op"): + pass + client.flush() + assert exporter.get_finished_spans() == () + + +def test_env_var_enables_partial_spans(monkeypatch) -> None: + from glassflow.config import resolve_config + + monkeypatch.setenv("GLASSFLOW_PARTIAL_SPANS", "true") + assert resolve_config().partial_spans is True + # explicit argument wins over the environment + assert resolve_config(partial_spans=False).partial_spans is False + + +class _StartAttributeRecorder(_SpanProcessor): + """Span processor recording each span's attributes as seen at on_start.""" + + def __init__(self) -> None: + self.seen: dict[str, dict] = {} + + def on_start(self, span, parent_context=None) -> None: # noqa: ANN001 + self.seen[span.name] = dict(span.attributes or {}) + + def on_end(self, span) -> None: # noqa: ANN001 + pass + + def shutdown(self) -> None: + pass + + def force_flush(self, timeout_millis: int = 30000) -> bool: + return True + + +def test_sdk_helpers_expose_identity_attributes_at_span_start() -> None: + """Pending snapshots are built at on_start, so the SDK's own APIs must + attach kind/model/provider at CREATION — set_attribute after the fact is + invisible to the snapshot.""" + from opentelemetry import trace as otel_trace + + import glassflow + + recorder = _StartAttributeRecorder() + otel_trace.get_tracer_provider().add_span_processor(recorder) # type: ignore[attr-defined] + + with glassflow.start_as_current_span("kindly", kind=glassflow.SpanKind.RETRIEVER): + pass + with glassflow.start_as_current_generation("genny", model="gpt-4o", provider="openai"): + pass + + assert recorder.seen["kindly"]["openinference.span.kind"] == "RETRIEVER" + genny = recorder.seen["genny"] + assert genny["openinference.span.kind"] == "LLM" + assert genny["gen_ai.operation.name"] == "chat" + assert genny["gen_ai.request.model"] == "gpt-4o" + assert genny["gen_ai.provider.name"] == "openai" From 368e67881b0d966b0257a84676c84d7f9cf68b08 Mon Sep 17 00:00:00 2001 From: Pablo <39369995+PabloPardoGarcia@users.noreply.github.com> Date: Wed, 29 Jul 2026 17:51:46 +0200 Subject: [PATCH 2/2] =?UTF-8?q?feat:=20debounced=20partial=20spans=20?= =?UTF-8?q?=E2=80=94=20delay=20emission,=20cancel=20on=20fast=20finish=20(?= =?UTF-8?q?#38)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/glassflow/client.py | 8 +- src/glassflow/config.py | 35 +++++++ src/glassflow/pending.py | 189 +++++++++++++++++++++++++++++++++--- tests/test_pending_delay.py | 141 +++++++++++++++++++++++++++ 4 files changed, 360 insertions(+), 13 deletions(-) create mode 100644 tests/test_pending_delay.py 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]