From e19cb24cbe26d6a8760f63d99a9e0fbcee3d824d Mon Sep 17 00:00:00 2001 From: Andrey Belonogov Date: Wed, 28 Oct 2026 16:25:05 -0700 Subject: [PATCH 1/2] feat(python): add set_attributes and custom span/log processors Co-authored-by: Cursor --- .../observability-python/README.md | 65 ++ .../ldobserve/_otel/configuration.py | 41 +- .../ldobserve/_otel/global_attributes.py | 163 +++++ .../ldobserve/_otel/processors.py | 119 ++++ .../observability-python/ldobserve/config.py | 35 +- .../observability-python/ldobserve/observe.py | 58 +- .../ldobserve/test_global_attributes.py | 617 ++++++++++++++++++ .../ldobserve/test_observe_instance.py | 23 +- 8 files changed, 1092 insertions(+), 29 deletions(-) create mode 100644 sdk/@launchdarkly/observability-python/ldobserve/_otel/global_attributes.py create mode 100644 sdk/@launchdarkly/observability-python/ldobserve/_otel/processors.py create mode 100644 sdk/@launchdarkly/observability-python/ldobserve/test_global_attributes.py diff --git a/sdk/@launchdarkly/observability-python/README.md b/sdk/@launchdarkly/observability-python/README.md index cbc5605d81..c20ef096a8 100644 --- a/sdk/@launchdarkly/observability-python/README.md +++ b/sdk/@launchdarkly/observability-python/README.md @@ -37,6 +37,71 @@ ldclient.set_config(Config("your-sdk-key", plugins=[ ))])) ``` +## Global attributes + +Use `observe.set_attributes` to add attributes to every span, log record, and metric data point +recorded after the call, including spans and logs from automatic instrumentation and the standard +`logging` module. Calls merge by key; use `observe.remove_attributes` to remove keys. + +```python +from ldobserve import observe + +# Safe to call before the plugin is initialized. +observe.set_attributes({"tenant.id": "acme", "deployment.region": "us-east-1"}) + +# Attributes passed explicitly take precedence over global attributes. +observe.record_count("orders.placed", 1, {"tenant.id": "override"}) + +observe.remove_attributes(["deployment.region"]) +``` + +- Precedence: attributes set on the span, log, or metric > global attributes > resource attributes. + Global attributes only fill in keys that are not already present. +- A span gets the global attributes that are set when it starts. +- Global attributes are stored in memory, shared by all threads in the process, and are not persisted. + +## Custom span and log processors + +Pass OpenTelemetry processors with the `span_processors` and `log_processors` options. They run in +the order given, after global attributes are applied and before data is exported, so changes they +make are exported. An exception raised by a processor is logged and does not stop export. + +```python +import typing + +from opentelemetry.context import Context +from opentelemetry.sdk._logs import LogData, LogRecordProcessor +from opentelemetry.sdk.trace import Span, SpanProcessor + +from ldobserve import ObservabilityConfig, ObservabilityPlugin + + +class TeamSpanProcessor(SpanProcessor): + def on_start(self, span: Span, parent_context: typing.Optional[Context] = None) -> None: + span.set_attribute("team", "checkout") + + +class RedactingLogProcessor(LogRecordProcessor): + def emit(self, log_data: LogData) -> None: + if "user.email" in log_data.log_record.attributes: + log_data.log_record.attributes["user.email"] = "[redacted]" + + def shutdown(self) -> None: + pass + + def force_flush(self, timeout_millis: int = 30000) -> bool: + return True + + +ObservabilityPlugin( + ObservabilityConfig( + service_name="your-service-name", + span_processors=[TeamSpanProcessor()], + log_processors=[RedactingLogProcessor()], + ) +) +``` + LaunchDarkly overview ------------------------- [LaunchDarkly](https://www.launchdarkly.com) is a feature management platform that serves trillions of feature flags daily to help teams build better software, faster. [Get started](https://docs.launchdarkly.com/home/getting-started) using LaunchDarkly today! diff --git a/sdk/@launchdarkly/observability-python/ldobserve/_otel/configuration.py b/sdk/@launchdarkly/observability-python/ldobserve/_otel/configuration.py index 39602bb0c9..ef3c0fd678 100644 --- a/sdk/@launchdarkly/observability-python/ldobserve/_otel/configuration.py +++ b/sdk/@launchdarkly/observability-python/ldobserve/_otel/configuration.py @@ -3,6 +3,7 @@ import typing from grpc import Compression from ldobserve._otel._sampling.custom_sampler import CustomSampler +from ldobserve._otel.processors import add_log_processors, add_span_processors from ldobserve._otel.sampling_log_exporter import SamplingLogExporter from ldobserve._otel.sampling_trace_exporter import SamplingTraceExporter from ldobserve._graph.get_sampling_config import get_sampling_config @@ -107,30 +108,38 @@ def __init__(self, project_id: str, config: _ProcessedConfig): ) self._tracer_provider = TracerProvider(resource=resource) - self._tracer_provider.add_span_processor( - BatchSpanProcessor( - SamplingTraceExporter( - sampler=sampler, - endpoint=f"{self._config.otlp_endpoint}/v1/traces", - compression=Compression.Gzip, - timeout=_SCHEDULE_DELAY_MILLIS, + add_span_processors( + self._tracer_provider, + self._config.span_processors, + [ + BatchSpanProcessor( + SamplingTraceExporter( + sampler=sampler, + endpoint=f"{self._config.otlp_endpoint}/v1/traces", + compression=Compression.Gzip, + timeout=_SCHEDULE_DELAY_MILLIS, + ) ) - ) + ], ) otel_trace.set_tracer_provider(self._tracer_provider) self._tracer = self._tracer_provider.get_tracer(__name__) self._logger_provider = LoggerProvider(resource=resource) - self._logger_provider.add_log_record_processor( - BatchLogRecordProcessor( - SamplingLogExporter( - sampler=sampler, - endpoint=f"{self._config.otlp_endpoint}/v1/logs", - compression=Compression.Gzip, - timeout=_SCHEDULE_DELAY_MILLIS, + add_log_processors( + self._logger_provider, + self._config.log_processors, + [ + BatchLogRecordProcessor( + SamplingLogExporter( + sampler=sampler, + endpoint=f"{self._config.otlp_endpoint}/v1/logs", + compression=Compression.Gzip, + timeout=_SCHEDULE_DELAY_MILLIS, + ) ) - ) + ], ) _logs.set_logger_provider(self._logger_provider) diff --git a/sdk/@launchdarkly/observability-python/ldobserve/_otel/global_attributes.py b/sdk/@launchdarkly/observability-python/ldobserve/_otel/global_attributes.py new file mode 100644 index 0000000000..b684cd076a --- /dev/null +++ b/sdk/@launchdarkly/observability-python/ldobserve/_otel/global_attributes.py @@ -0,0 +1,163 @@ +"""Process-wide global attributes applied to spans, logs, and metrics. + +The store is a module-level, immutable snapshot that is rebuilt under a lock on +every change. Readers (span/log processors and metric recording) only read the +current snapshot reference, so they never contend with writers and never observe +a partially-applied update. The store is usable before the SDK is initialized. +""" + +import threading +import types +import typing + +from opentelemetry.attributes import BoundedAttributes, _clean_attribute +from opentelemetry.context import Context +from opentelemetry.sdk._logs import LogData, LogRecordProcessor +from opentelemetry.sdk.trace import Span, SpanProcessor +from opentelemetry.util.types import Attributes, AttributeValue + +_lock = threading.Lock() +_attributes: typing.Mapping[str, AttributeValue] = types.MappingProxyType({}) + + +def get_global_attributes() -> typing.Mapping[str, AttributeValue]: + """Return the current immutable snapshot of the global attributes.""" + return _attributes + + +def set_global_attributes(attributes: typing.Mapping[str, AttributeValue]) -> None: + global _attributes + if not attributes: + return + + cleaned: dict[str, AttributeValue] = {} + for key, value in attributes.items(): + # Normalizes sequences to tuples; logs a warning for invalid keys/values. + cleaned_value = _clean_attribute(key, value, None) + if cleaned_value is None: + continue + cleaned[key] = typing.cast(AttributeValue, cleaned_value) + + if not cleaned: + return + + with _lock: + merged = dict(_attributes) + merged.update(cleaned) + _attributes = types.MappingProxyType(merged) + + +def remove_global_attributes(keys: typing.Iterable[str]) -> None: + global _attributes + if isinstance(keys, str): + keys = (keys,) + keys = tuple(keys) + if not keys: + return + + with _lock: + if not any(key in _attributes for key in keys): + return + remaining = {k: v for k, v in _attributes.items() if k not in keys} + _attributes = types.MappingProxyType(remaining) + + +def clear_global_attributes() -> None: + global _attributes + with _lock: + _attributes = types.MappingProxyType({}) + + +def merge_global_attributes( + attributes: typing.Optional[Attributes], +) -> typing.Optional[Attributes]: + """Return ``attributes`` with any missing global attributes filled in. + + Explicitly provided attributes always win over global attributes. + """ + snapshot = _attributes + if not snapshot: + return attributes + if not attributes: + return dict(snapshot) + merged = dict(snapshot) + merged.update(attributes) + return merged + + +def _missing_attributes( + existing: typing.Optional[typing.Mapping[str, typing.Any]], + snapshot: typing.Mapping[str, AttributeValue], + maxlen: typing.Optional[int], +) -> dict[str, AttributeValue]: + existing = existing or {} + missing = {k: v for k, v in snapshot.items() if k not in existing} + # BoundedAttributes evicts its oldest entry once full; only fill the + # remaining capacity so global attributes never displace explicit ones. + if maxlen is not None: + capacity = max(0, maxlen - len(existing)) + if len(missing) > capacity: + missing = dict(list(missing.items())[:capacity]) + return missing + + +class GlobalAttributesSpanProcessor(SpanProcessor): + """Fills in global attributes that are not already set when a span starts.""" + + def on_start( + self, span: Span, parent_context: typing.Optional[Context] = None + ) -> None: + snapshot = _attributes + if not snapshot: + return + # `Span.attributes` is a read-only view; the limit lives on the + # underlying BoundedAttributes. + maxlen = getattr(getattr(span, "_attributes", None), "maxlen", None) + missing = _missing_attributes(span.attributes, snapshot, maxlen) + if missing: + span.set_attributes(missing) + + def force_flush(self, timeout_millis: int = 30000) -> bool: + # The base implementation returns None, which the provider treats as a + # failed flush and stops before reaching the exporting processors. + return True + + +class GlobalAttributesLogRecordProcessor(LogRecordProcessor): + """Fills in global attributes that are not already set on each log record.""" + + def emit(self, log_data: LogData) -> None: + snapshot = _attributes + if not snapshot: + return + record = log_data.log_record + existing = record.attributes + missing = _missing_attributes( + existing, snapshot, getattr(existing, "maxlen", None) + ) + if not missing: + return + + if isinstance(existing, BoundedAttributes) and not getattr( + existing, "_immutable", False + ): + for key, value in missing.items(): + existing[key] = value + return + + merged: dict[str, typing.Any] = dict(missing) + if existing: + merged.update(existing) + record.attributes = BoundedAttributes( + maxlen=getattr(existing, "maxlen", None), + attributes=merged, + immutable=False, + max_value_len=getattr(existing, "max_value_len", None), + extended_attributes=True, + ) + + def shutdown(self) -> None: + pass + + def force_flush(self, timeout_millis: int = 30000) -> bool: + return True diff --git a/sdk/@launchdarkly/observability-python/ldobserve/_otel/processors.py b/sdk/@launchdarkly/observability-python/ldobserve/_otel/processors.py new file mode 100644 index 0000000000..d18608420c --- /dev/null +++ b/sdk/@launchdarkly/observability-python/ldobserve/_otel/processors.py @@ -0,0 +1,119 @@ +"""Processor pipeline assembly and isolation of user-provided processors. + +Pipeline order for both spans and logs is: + +1. The global attributes processor. +2. User processors, in the order they were configured. +3. The SDK's exporting processors. + +User processors are wrapped so an exception raised by one of them is logged and +never prevents the remaining processors (including export) from running. +""" + +import logging +import typing + +from opentelemetry.context import Context +from opentelemetry.sdk._logs import LogData, LoggerProvider, LogRecordProcessor +from opentelemetry.sdk.trace import ReadableSpan, Span, SpanProcessor, TracerProvider + +from ldobserve._otel.global_attributes import ( + GlobalAttributesLogRecordProcessor, + GlobalAttributesSpanProcessor, +) + +_logger = logging.getLogger(__name__) + + +class SafeSpanProcessor(SpanProcessor): + """Delegates to a user span processor, logging instead of raising.""" + + def __init__(self, processor: SpanProcessor): + self._processor = processor + + def on_start( + self, span: Span, parent_context: typing.Optional[Context] = None + ) -> None: + try: + self._processor.on_start(span, parent_context=parent_context) + except Exception: + _logger.exception("Span processor %r raised in on_start", self._processor) + + def on_end(self, span: ReadableSpan) -> None: + try: + self._processor.on_end(span) + except Exception: + _logger.exception("Span processor %r raised in on_end", self._processor) + + def shutdown(self) -> None: + try: + self._processor.shutdown() + except Exception: + _logger.exception("Span processor %r raised in shutdown", self._processor) + + def force_flush(self, timeout_millis: int = 30000) -> bool: + try: + # Processors that don't implement force_flush return None; only an + # explicit False is a failed flush. + return self._processor.force_flush(timeout_millis) is not False + except Exception: + _logger.exception( + "Span processor %r raised in force_flush", self._processor + ) + # The provider stops flushing at the first False; keep going so the + # exporting processors still flush. + return True + + +class SafeLogRecordProcessor(LogRecordProcessor): + """Delegates to a user log record processor, logging instead of raising.""" + + def __init__(self, processor: LogRecordProcessor): + self._processor = processor + + def emit(self, log_data: LogData) -> None: + try: + self._processor.emit(log_data) + except Exception: + _logger.exception("Log processor %r raised in emit", self._processor) + + def shutdown(self) -> None: + try: + self._processor.shutdown() + except Exception: + _logger.exception("Log processor %r raised in shutdown", self._processor) + + def force_flush(self, timeout_millis: int = 30000) -> bool: + try: + # Processors that don't implement force_flush return None; only an + # explicit False is a failed flush. + return self._processor.force_flush(timeout_millis) is not False + except Exception: + _logger.exception("Log processor %r raised in force_flush", self._processor) + # The provider stops flushing at the first False; keep going so the + # exporting processors still flush. + return True + + +def add_span_processors( + tracer_provider: TracerProvider, + user_processors: typing.Sequence[SpanProcessor], + export_processors: typing.Sequence[SpanProcessor], +) -> None: + tracer_provider.add_span_processor(GlobalAttributesSpanProcessor()) + for processor in user_processors: + tracer_provider.add_span_processor(SafeSpanProcessor(processor)) + for processor in export_processors: + tracer_provider.add_span_processor(processor) + + +def add_log_processors( + logger_provider: LoggerProvider, + user_processors: typing.Sequence[LogRecordProcessor], + export_processors: typing.Sequence[LogRecordProcessor], +) -> None: + logger_provider.add_log_record_processor(GlobalAttributesLogRecordProcessor()) + for processor in user_processors: + logger_provider.add_log_record_processor(SafeLogRecordProcessor(processor)) + for processor in export_processors: + logger_provider.add_log_record_processor(processor) diff --git a/sdk/@launchdarkly/observability-python/ldobserve/config.py b/sdk/@launchdarkly/observability-python/ldobserve/config.py index 045f10c39b..c63157e529 100644 --- a/sdk/@launchdarkly/observability-python/ldobserve/config.py +++ b/sdk/@launchdarkly/observability-python/ldobserve/config.py @@ -1,9 +1,11 @@ from dataclasses import dataclass import logging import os -from typing import Optional +from typing import Optional, Sequence from opentelemetry.instrumentation.logging import OTEL_PYTHON_LOG_CORRELATION +from opentelemetry.sdk._logs import LogRecordProcessor +from opentelemetry.sdk.trace import SpanProcessor from opentelemetry.sdk.environment_variables import ( OTEL_EXPORTER_OTLP_ENDPOINT, _OTEL_PYTHON_LOGGING_AUTO_INSTRUMENTATION_ENABLED, @@ -136,6 +138,32 @@ class ObservabilityConfig: Defaults to True. """ + span_processors: Optional[Sequence[SpanProcessor]] = None + """ + Additional OpenTelemetry span processors, invoked in order for every span. + + They run after global attributes (see :func:`ldobserve.observe.set_attributes`) are applied + and before spans are exported, so attributes set in `on_start` are exported. An exception + raised by a processor is logged and does not prevent the span from being exported. + + For example: + >>> from opentelemetry.sdk.trace import SpanProcessor + >>> class TenantSpanProcessor(SpanProcessor): + ... def on_start(self, span, parent_context=None): + ... span.set_attribute("tenant.id", current_tenant_id()) + >>> ObservabilityConfig(span_processors=[TenantSpanProcessor()]) + """ + + log_processors: Optional[Sequence[LogRecordProcessor]] = None + """ + Additional OpenTelemetry log record processors, invoked in order for every log record. + + They run after global attributes (see :func:`ldobserve.observe.set_attributes`) are applied + and before log records are exported, so changes made to `log_data.log_record` in `emit` are + exported. An exception raised by a processor is logged and does not prevent the log record + from being exported. + """ + def __getitem__(self, key: str): return getattr(self, key) @@ -154,6 +182,8 @@ class _ProcessedConfig: backend_url: str process_resources: bool os_resources: bool + span_processors: list[SpanProcessor] + log_processors: list[LogRecordProcessor] def __init__(self, config: ObservabilityConfig): self.otlp_endpoint = config.otlp_endpoint or os.getenv( @@ -210,3 +240,6 @@ def __init__(self, config: ObservabilityConfig): if config.os_resources is not None else _DEFAULT_OS_RESOURCES ) + + self.span_processors = list(config.span_processors or []) + self.log_processors = list(config.log_processors or []) diff --git a/sdk/@launchdarkly/observability-python/ldobserve/observe.py b/sdk/@launchdarkly/observability-python/ldobserve/observe.py index 590d93ec4b..967407fde2 100644 --- a/sdk/@launchdarkly/observability-python/ldobserve/observe.py +++ b/sdk/@launchdarkly/observability-python/ldobserve/observe.py @@ -6,9 +6,14 @@ from opentelemetry.sdk._logs import LoggingHandler from opentelemetry.trace import Span, Tracer import opentelemetry.trace as trace -from opentelemetry.util.types import Attributes +from opentelemetry.util.types import Attributes, AttributeValue from opentelemetry._logs import get_logger_provider from ldobserve._otel.configuration import _OTELConfiguration +from ldobserve._otel.global_attributes import ( + merge_global_attributes, + remove_global_attributes, + set_global_attributes, +) from ldobserve._source_context import exception_attributes from ldobserve._util.dict import flatten_dict @@ -74,14 +79,14 @@ def record_metric( ): if name not in self._gauges: self._gauges[name] = self._otel_configuration.meter.create_gauge(name) - self._gauges[name].set(value, attributes=attributes) + self._gauges[name].set(value, attributes=merge_global_attributes(attributes)) def record_count( self, name: str, value: int, attributes: typing.Optional[Attributes] = None ): if name not in self._counters: self._counters[name] = self._otel_configuration.meter.create_counter(name) - self._counters[name].add(value, attributes=attributes) + self._counters[name].add(value, attributes=merge_global_attributes(attributes)) def record_incr(self, name: str, attributes: typing.Optional[Attributes] = None): return self.record_count(name, 1, attributes) @@ -93,7 +98,9 @@ def record_histogram( self._histograms[name] = self._otel_configuration.meter.create_histogram( name ) - self._histograms[name].record(value, attributes=attributes) + self._histograms[name].record( + value, attributes=merge_global_attributes(attributes) + ) def record_up_down_counter( self, name: str, value: int, attributes: typing.Optional[Attributes] = None @@ -102,7 +109,9 @@ def record_up_down_counter( self._up_down_counters[name] = ( self._otel_configuration.meter.create_up_down_counter(name) ) - self._up_down_counters[name].add(value, attributes=attributes) + self._up_down_counters[name].add( + value, attributes=merge_global_attributes(attributes) + ) def log( self, message: str, level: int, attributes: typing.Optional[Attributes] = None @@ -274,6 +283,45 @@ def record_log( _use_instance(lambda instance: instance.log(message, level, attributes)) +def set_attributes(attributes: typing.Mapping[str, AttributeValue]) -> None: + """ + Set global attributes that are added to every span, log record, and metric data point + recorded after this call. Attributes are merged by key into the existing global attributes. + + Attributes provided on an individual span, log, or metric take precedence over global + attributes, and global attributes take precedence over resource attributes. A span uses the + global attributes that are set when it starts. + + Global attributes are kept in memory for the lifetime of the process and are shared by all + threads. They may be set before the observability plugin is initialized. + + Example: + import ldobserve.observe as observe + + observe.set_attributes({"tenant.id": "acme", "deployment.region": "us-east-1"}) + + :param attributes: the attributes to set. Values must be valid OpenTelemetry attribute values; + invalid values are ignored. + :return: None + """ + set_global_attributes(attributes) + + +def remove_attributes(keys: typing.Iterable[str]) -> None: + """ + Remove global attributes previously set with :func:`set_attributes`. + + Example: + import ldobserve.observe as observe + + observe.remove_attributes(["tenant.id"]) + + :param keys: the keys of the global attributes to remove. Unknown keys are ignored. + :return: None + """ + remove_global_attributes(keys) + + def logging_handler() -> logging.Handler: """A logging handler implementing `logging.Handler` that allows plugging LaunchDarkly Observability into your existing logging setup. Standard logging will be automatically instrumented unless diff --git a/sdk/@launchdarkly/observability-python/ldobserve/test_global_attributes.py b/sdk/@launchdarkly/observability-python/ldobserve/test_global_attributes.py new file mode 100644 index 0000000000..a7f36decab --- /dev/null +++ b/sdk/@launchdarkly/observability-python/ldobserve/test_global_attributes.py @@ -0,0 +1,617 @@ +import logging +import threading +import typing + +import pytest +from opentelemetry.attributes import BoundedAttributes +from opentelemetry.context import Context +from opentelemetry.sdk.util.instrumentation import InstrumentationScope +from opentelemetry.sdk._logs import ( + LogData, + LoggerProvider, + LogRecord, + LogRecordProcessor, +) +from opentelemetry.sdk._logs.export import InMemoryLogExporter, SimpleLogRecordProcessor +from opentelemetry.sdk.trace import ReadableSpan, Span, SpanProcessor, TracerProvider +from opentelemetry.sdk.trace.export import SimpleSpanProcessor +from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter + +import ldobserve.observe as observe +from ldobserve._otel import configuration as otel_configuration +from ldobserve._otel.global_attributes import ( + GlobalAttributesLogRecordProcessor, + clear_global_attributes, + get_global_attributes, +) +from ldobserve._otel.logging_handler import LDLoggingHandler +from ldobserve._otel.processors import add_log_processors, add_span_processors +from ldobserve.config import ObservabilityConfig, _ProcessedConfig +from ldobserve.observe import _ObserveInstance +from ldobserve.test_observe_instance import OTELTestConfiguration + +_raising_processors: list["_Raising"] = [] + + +@pytest.fixture(autouse=True) +def reset_global_attributes(): + clear_global_attributes() + yield + clear_global_attributes() + # Providers also shut down at interpreter exit; don't raise from there. + for processor in _raising_processors: + processor.armed = False + _raising_processors.clear() + _ObserveInstance._gauges.clear() + _ObserveInstance._counters.clear() + _ObserveInstance._histograms.clear() + _ObserveInstance._up_down_counters.clear() + + +def _make_instance(**kwargs) -> tuple[_ObserveInstance, OTELTestConfiguration]: + config = OTELTestConfiguration(**kwargs) + return _ObserveInstance("test-project", config), config + + +def _finished_span(config: OTELTestConfiguration, name: str) -> ReadableSpan: + spans = [s for s in config.span_exporter.get_finished_spans() if s.name == name] + assert len(spans) == 1 + return spans[0] + + +def _finished_logs(config: OTELTestConfiguration, body: str) -> list[LogData]: + return [ + log + for log in config.log_exporter.get_finished_logs() + if log.log_record.body == body + ] + + +def _data_point(config: OTELTestConfiguration, name: str): + config.force_flush_metrics() + data = config.metric_reader.get_metrics_data() + for resource_metrics in data.resource_metrics: + for scope_metrics in resource_metrics.scope_metrics: + for metric in scope_metrics.metrics: + if metric.name == name: + return metric.data.data_points[0] + raise AssertionError(f"metric {name} not found") + + +class RecordingSpanProcessor(SpanProcessor): + def __init__(self, name: str, calls: list[str]): + self.name = name + self.calls = calls + self.seen_attributes: list[dict] = [] + self.flushed = False + self.shut_down = False + + def on_start( + self, span: Span, parent_context: typing.Optional[Context] = None + ) -> None: + self.calls.append(self.name) + self.seen_attributes.append(dict(span.attributes or {})) + span.set_attribute(f"processed.{self.name}", True) + span.set_attribute("processed.last", self.name) + + def on_end(self, span: ReadableSpan) -> None: + self.calls.append(f"{self.name}.end") + + def force_flush(self, timeout_millis: int = 30000) -> bool: + self.flushed = True + return True + + def shutdown(self) -> None: + self.shut_down = True + + +class _Raising: + def __init__(self): + self.armed = True + _raising_processors.append(self) + + def _raise(self, message: str): + if self.armed: + raise RuntimeError(message) + + +class RaisingSpanProcessor(_Raising, SpanProcessor): + def on_start( + self, span: Span, parent_context: typing.Optional[Context] = None + ) -> None: + self._raise("on_start failed") + + def on_end(self, span: ReadableSpan) -> None: + self._raise("on_end failed") + + def force_flush(self, timeout_millis: int = 30000) -> bool: + self._raise("force_flush failed") + return True + + def shutdown(self) -> None: + self._raise("shutdown failed") + + +class RecordingLogProcessor(LogRecordProcessor): + def __init__(self, name: str, calls: list[str]): + self.name = name + self.calls = calls + self.seen_attributes: list[dict] = [] + self.flushed = False + self.shut_down = False + + def emit(self, log_data: LogData) -> None: + self.calls.append(self.name) + attributes = log_data.log_record.attributes + self.seen_attributes.append(dict(attributes or {})) + attributes[f"processed.{self.name}"] = True + attributes["processed.last"] = self.name + + def force_flush(self, timeout_millis: int = 30000) -> bool: + self.flushed = True + return True + + def shutdown(self) -> None: + self.shut_down = True + + +class RaisingLogProcessor(_Raising, LogRecordProcessor): + def emit(self, log_data: LogData) -> None: + self._raise("emit failed") + + def force_flush(self, timeout_millis: int = 30000) -> bool: + self._raise("force_flush failed") + return True + + def shutdown(self) -> None: + self._raise("shutdown failed") + + +# Store + + +def test_set_attributes_merges_by_key(): + observe.set_attributes({"a": "1", "b": 2}) + observe.set_attributes({"b": 3, "c": True}) + + assert dict(get_global_attributes()) == {"a": "1", "b": 3, "c": True} + + +def test_remove_attributes(): + observe.set_attributes({"a": "1", "b": 2, "c": 3.5}) + observe.remove_attributes(["a", "c", "missing"]) + + assert dict(get_global_attributes()) == {"b": 2} + + observe.remove_attributes(["b"]) + assert dict(get_global_attributes()) == {} + + +def test_remove_attributes_accepts_a_single_key_string(): + observe.set_attributes({"tenant": "acme", "t": "x"}) + observe.remove_attributes("tenant") + + assert dict(get_global_attributes()) == {"t": "x"} + + +def test_snapshot_is_immutable_and_replaced_on_change(): + observe.set_attributes({"a": "1"}) + snapshot = get_global_attributes() + + with pytest.raises(TypeError): + snapshot["b"] = "2" # type: ignore[index] + + observe.set_attributes({"b": "2"}) + assert dict(snapshot) == {"a": "1"} + assert dict(get_global_attributes()) == {"a": "1", "b": "2"} + + +def test_set_attributes_ignores_invalid_values(): + observe.set_attributes( + { + "valid": "ok", + "list": ["x", "y"], + "object": object(), # type: ignore[dict-item] + "mixed": ["x", 1], # type: ignore[list-item] + "": "empty key", + } + ) + + assert dict(get_global_attributes()) == {"valid": "ok", "list": ("x", "y")} + + +def test_set_attributes_before_init_does_not_warn(caplog): + assert not observe.is_initialized() + with caplog.at_level(logging.WARNING): + observe.set_attributes({"before.init": "yes"}) + observe.remove_attributes(["unknown"]) + + assert "before it was initialized" not in caplog.text + assert dict(get_global_attributes()) == {"before.init": "yes"} + + +def test_concurrent_set_and_remove(): + def writer(index: int): + for i in range(200): + observe.set_attributes({f"key.{index}": i}) + observe.remove_attributes([f"key.{index}"]) + observe.set_attributes({f"key.{index}": "final"}) + + threads = [threading.Thread(target=writer, args=(i,)) for i in range(8)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + + assert dict(get_global_attributes()) == {f"key.{i}": "final" for i in range(8)} + + +# Spans + + +def test_globals_set_before_init_apply_to_spans(): + observe.set_attributes({"tenant.id": "acme"}) + instance, config = _make_instance() + + with instance.start_span("manual"): + pass + + assert _finished_span(config, "manual").attributes["tenant.id"] == "acme" + + +def test_span_attributes_win_over_globals(): + instance, config = _make_instance() + observe.set_attributes({"shared": "global", "only.global": "g"}) + + with instance.start_span("manual", attributes={"shared": "span"}): + pass + + attributes = _finished_span(config, "manual").attributes + assert attributes["shared"] == "span" + assert attributes["only.global"] == "g" + + +def test_span_set_attribute_after_start_wins_over_globals(): + instance, config = _make_instance() + observe.set_attributes({"shared": "global"}) + + with instance.start_span("manual") as span: + span.set_attribute("shared", "explicit") + + assert _finished_span(config, "manual").attributes["shared"] == "explicit" + + +def test_span_uses_globals_current_at_start(): + instance, config = _make_instance() + observe.set_attributes({"phase": "start"}) + + with instance.start_span("manual"): + observe.set_attributes({"phase": "later", "added.later": True}) + observe.remove_attributes(["phase"]) + + attributes = _finished_span(config, "manual").attributes + assert attributes["phase"] == "start" + assert "added.later" not in attributes + + +def test_instrumented_spans_get_globals(): + _, config = _make_instance() + observe.set_attributes({"tenant.id": "acme"}) + + # Instrumentation libraries obtain their own tracer from the provider. + tracer = config._tracer_provider.get_tracer("opentelemetry.instrumentation.test") + with tracer.start_as_current_span("instrumented"): + pass + + assert _finished_span(config, "instrumented").attributes["tenant.id"] == "acme" + + +def test_exception_spans_get_globals(): + instance, config = _make_instance() + observe.set_attributes({"tenant.id": "acme"}) + + instance.record_exception(ValueError("boom")) + + assert ( + _finished_span(config, "launchdarkly.error").attributes["tenant.id"] == "acme" + ) + + +def test_globals_do_not_displace_span_attributes_at_capacity(): + instance, config = _make_instance() + config._tracer_provider._span_limits.max_span_attributes = 2 + observe.set_attributes({"global.a": "a", "global.b": "b"}) + + with instance.start_span("full", attributes={"x": 1}): + pass + + attributes = _finished_span(config, "full").attributes + assert attributes["x"] == 1 + assert len(attributes) == 2 + + +# Logs + + +def test_record_log_gets_globals_and_call_attributes_win(): + instance, config = _make_instance() + observe.set_attributes({"shared": "global", "tenant.id": "acme"}) + + instance.log("hello", logging.WARNING, {"shared": "call"}) + + logs = _finished_logs(config, "hello") + assert len(logs) == 1 + attributes = logs[0].log_record.attributes + assert attributes["shared"] == "call" + assert attributes["tenant.id"] == "acme" + + +def test_stdlib_logging_records_get_globals(): + exporter = InMemoryLogExporter() + provider = LoggerProvider() + add_log_processors(provider, [], [SimpleLogRecordProcessor(exporter)]) + handler = LDLoggingHandler(level=logging.NOTSET, logger_provider=provider) + app_logger = logging.getLogger("ldobserve.test.stdlib") + app_logger.addHandler(handler) + app_logger.setLevel(logging.INFO) + app_logger.propagate = False + observe.set_attributes({"tenant.id": "acme", "shared": "global"}) + + try: + app_logger.info("bridged", extra={"shared": "extra"}) + finally: + app_logger.removeHandler(handler) + + logs = exporter.get_finished_logs() + assert len(logs) == 1 + attributes = logs[0].log_record.attributes + assert attributes["tenant.id"] == "acme" + assert attributes["shared"] == "extra" + + +def test_globals_win_over_resource_attributes_on_logs(): + instance, config = _make_instance(service_name="resource-service") + observe.set_attributes({"service.name": "global-service"}) + + instance.log("resource", logging.WARNING) + + log = _finished_logs(config, "resource")[0].log_record + assert log.attributes["service.name"] == "global-service" + assert log.resource.attributes["service.name"] == "resource-service" + + +@pytest.mark.parametrize( + "attributes", + [None, BoundedAttributes(maxlen=10, attributes={"shared": "record"})], + ids=["none", "immutable"], +) +def test_log_processor_replaces_missing_or_immutable_attributes(attributes): + observe.set_attributes({"shared": "global", "tenant.id": "acme"}) + record = LogRecord(body="replaced") + record.attributes = attributes + log_data = LogData(record, InstrumentationScope("test")) + + GlobalAttributesLogRecordProcessor().emit(log_data) + + expected = {"shared": "global" if attributes is None else "record"} + expected["tenant.id"] = "acme" + assert dict(log_data.log_record.attributes) == expected + assert log_data.log_record.dropped_attributes == 0 + + +# Metrics + + +@pytest.mark.parametrize( + "record", + [ + lambda i, name, attrs: i.record_metric(name, 1.0, attrs), + lambda i, name, attrs: i.record_count(name, 1, attrs), + lambda i, name, attrs: i.record_incr(name, attrs), + lambda i, name, attrs: i.record_histogram(name, 1.0, attrs), + lambda i, name, attrs: i.record_up_down_counter(name, 1, attrs), + ], + ids=["gauge", "count", "incr", "histogram", "up_down_counter"], +) +def test_metrics_get_globals_and_call_attributes_win(record): + instance, config = _make_instance() + observe.set_attributes({"shared": "global", "tenant.id": "acme"}) + + record(instance, "metric.with.attrs", {"shared": "call"}) + point = _data_point(config, "metric.with.attrs") + assert dict(point.attributes) == {"shared": "call", "tenant.id": "acme"} + + +def test_metrics_without_call_attributes_get_globals(): + instance, config = _make_instance() + observe.set_attributes({"tenant.id": "acme"}) + + instance.record_count("metric.no.attrs", 1) + + point = _data_point(config, "metric.no.attrs") + assert dict(point.attributes) == {"tenant.id": "acme"} + + +def test_metrics_without_globals_are_unchanged(): + instance, config = _make_instance() + + instance.record_count("metric.plain", 1) + + assert not _data_point(config, "metric.plain").attributes + + +# Custom processors + + +def test_span_processors_run_in_order_and_mutations_are_exported(): + calls: list[str] = [] + first = RecordingSpanProcessor("first", calls) + second = RecordingSpanProcessor("second", calls) + instance, config = _make_instance(span_processors=[first, second]) + observe.set_attributes({"tenant.id": "acme"}) + + with instance.start_span("processed"): + pass + + assert calls == ["first", "second", "first.end", "second.end"] + # Global attributes are applied before user processors run. + assert first.seen_attributes[0]["tenant.id"] == "acme" + assert second.seen_attributes[0]["processed.first"] is True + + attributes = _finished_span(config, "processed").attributes + assert attributes["processed.first"] is True + assert attributes["processed.second"] is True + assert attributes["processed.last"] == "second" + assert attributes["tenant.id"] == "acme" + + +def test_raising_span_processor_does_not_break_export(caplog): + calls: list[str] = [] + after = RecordingSpanProcessor("after", calls) + instance, config = _make_instance(span_processors=[RaisingSpanProcessor(), after]) + + with caplog.at_level(logging.ERROR): + with instance.start_span("survives"): + pass + + assert "raised in on_start" in caplog.text + assert "raised in on_end" in caplog.text + + assert _finished_span(config, "survives").attributes["processed.after"] is True + assert calls == ["after", "after.end"] + + +def test_span_processor_flush_and_shutdown_are_forwarded(): + calls: list[str] = [] + recording = RecordingSpanProcessor("recording", calls) + _, config = _make_instance(span_processors=[RaisingSpanProcessor(), recording]) + + assert config._tracer_provider.force_flush() is True + assert recording.flushed + + config._tracer_provider.shutdown() + assert recording.shut_down + + +class NoFlushSpanProcessor(SpanProcessor): + """Relies on the base class force_flush, which returns None.""" + + +def test_force_flush_reaches_export_processors(): + calls: list[str] = [] + export = RecordingSpanProcessor("export", calls) + provider = TracerProvider() + add_span_processors(provider, [NoFlushSpanProcessor()], [export]) + + assert provider.force_flush() is True + assert export.flushed + + +def test_log_processors_run_in_order_and_mutations_are_exported(): + calls: list[str] = [] + first = RecordingLogProcessor("first", calls) + second = RecordingLogProcessor("second", calls) + instance, config = _make_instance(log_processors=[first, second]) + observe.set_attributes({"tenant.id": "acme"}) + + instance.log("processed", logging.WARNING) + + assert calls == ["first", "second"] + assert first.seen_attributes[0]["tenant.id"] == "acme" + assert second.seen_attributes[0]["processed.first"] is True + + attributes = _finished_logs(config, "processed")[0].log_record.attributes + assert attributes["processed.first"] is True + assert attributes["processed.second"] is True + assert attributes["processed.last"] == "second" + assert attributes["tenant.id"] == "acme" + + +def test_raising_log_processor_does_not_break_export(caplog): + calls: list[str] = [] + after = RecordingLogProcessor("after", calls) + instance, config = _make_instance(log_processors=[RaisingLogProcessor(), after]) + + with caplog.at_level(logging.ERROR, logger="ldobserve._otel.processors"): + instance.log("survives", logging.WARNING) + + assert "raised in emit" in caplog.text + + logs = _finished_logs(config, "survives") + assert len(logs) == 1 + assert logs[0].log_record.attributes["processed.after"] is True + assert calls == ["after"] + + +def test_log_processor_flush_and_shutdown_are_forwarded(): + calls: list[str] = [] + recording = RecordingLogProcessor("recording", calls) + _, config = _make_instance(log_processors=[RaisingLogProcessor(), recording]) + + assert config._logger_provider.force_flush() is True + assert recording.flushed + + config._logger_provider.shutdown() + assert recording.shut_down + + +# Configuration + + +def test_processed_config_defaults_to_no_processors(): + config = _ProcessedConfig(ObservabilityConfig()) + + assert config.span_processors == [] + assert config.log_processors == [] + + +def test_otel_configuration_wires_user_processors_before_export(mocker): + span_exporter = InMemorySpanExporter() + log_exporter = InMemoryLogExporter() + mocker.patch.object(otel_configuration, "get_sampling_config") + mocker.patch.object(otel_configuration, "OTLPMetricExporter") + mocker.patch.object(otel_configuration, "PeriodicExportingMetricReader") + mocker.patch.object(otel_configuration.otel_trace, "set_tracer_provider") + mocker.patch.object(otel_configuration._logs, "set_logger_provider") + mocker.patch.object(otel_configuration.metrics, "set_meter_provider") + mocker.patch.object( + otel_configuration, + "BatchSpanProcessor", + lambda _exporter: SimpleSpanProcessor(span_exporter), + ) + mocker.patch.object( + otel_configuration, + "BatchLogRecordProcessor", + lambda _exporter: SimpleLogRecordProcessor(log_exporter), + ) + + calls: list[str] = [] + span_processor = RecordingSpanProcessor("span", calls) + log_processor = RecordingLogProcessor("log", calls) + otel_config = otel_configuration._OTELConfiguration( + "test-project", + _ProcessedConfig( + ObservabilityConfig( + instrument_logging=False, + span_processors=[span_processor], + log_processors=[log_processor], + ) + ), + ) + observe.set_attributes({"tenant.id": "acme"}) + + with otel_config.tracer.start_as_current_span("wired"): + pass + otel_config._logger_provider.get_logger("test").emit(LogRecord(body="wired")) + + span = span_exporter.get_finished_spans()[0] + assert span.attributes["processed.span"] is True + assert span.attributes["tenant.id"] == "acme" + log = log_exporter.get_finished_logs()[0].log_record + assert log.attributes["processed.log"] is True + assert log.attributes["tenant.id"] == "acme" + assert span_processor.seen_attributes[0]["tenant.id"] == "acme" + assert log_processor.seen_attributes[0]["tenant.id"] == "acme" + + otel_config._tracer_provider.shutdown() + otel_config._logger_provider.shutdown() diff --git a/sdk/@launchdarkly/observability-python/ldobserve/test_observe_instance.py b/sdk/@launchdarkly/observability-python/ldobserve/test_observe_instance.py index 1d6ae9d82b..2b1a2b71b5 100644 --- a/sdk/@launchdarkly/observability-python/ldobserve/test_observe_instance.py +++ b/sdk/@launchdarkly/observability-python/ldobserve/test_observe_instance.py @@ -5,10 +5,10 @@ from dataclasses import dataclass import opentelemetry.trace as trace from opentelemetry import metrics as otel_metrics -from opentelemetry.sdk._logs import LoggerProvider, LoggingHandler +from opentelemetry.sdk._logs import LoggerProvider, LoggingHandler, LogRecordProcessor from opentelemetry.sdk.metrics import MeterProvider from opentelemetry.sdk.resources import Resource -from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.sdk.trace import SpanProcessor, TracerProvider from opentelemetry.sdk.trace.export import SimpleSpanProcessor from opentelemetry.util.types import Attributes from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter @@ -17,6 +17,7 @@ from ldobserve._otel.configuration import _OTELConfiguration +from ldobserve._otel.processors import add_log_processors, add_span_processors from ldobserve.observe import _ObserveInstance @@ -86,7 +87,11 @@ class OTELTestConfiguration(_OTELConfiguration): """Test implementation of OTELConfiguration that uses in-memory exporters""" def __init__( - self, project_id: str = "test-project", service_name: str = "test-service" + self, + project_id: str = "test-project", + service_name: str = "test-service", + span_processors: typing.Sequence[SpanProcessor] = (), + log_processors: typing.Sequence[LogRecordProcessor] = (), ): # Create test resource resource = Resource.create( @@ -100,8 +105,10 @@ def __init__( # Set up in-memory trace provider self._tracer_provider = TracerProvider(resource=resource) self._span_exporter = InMemorySpanExporter() - self._tracer_provider.add_span_processor( - SimpleSpanProcessor(self._span_exporter) + add_span_processors( + self._tracer_provider, + span_processors, + [SimpleSpanProcessor(self._span_exporter)], ) # Don't set global tracer provider to avoid conflicts @@ -118,8 +125,10 @@ def __init__( # Set up in-memory logs provider self._logger_provider = LoggerProvider(resource=resource) self._log_exporter = InMemoryLogExporter() - self._logger_provider.add_log_record_processor( - SimpleLogRecordProcessor(self._log_exporter) + add_log_processors( + self._logger_provider, + log_processors, + [SimpleLogRecordProcessor(self._log_exporter)], ) self._log_handler = LoggingHandler( logger_provider=self._logger_provider, level=logging.NOTSET From 9201b5059344730c19ffc9f81099ac7cfb3c5dbe Mon Sep 17 00:00:00 2001 From: Andrey Belonogov Date: Wed, 28 Oct 2026 17:17:56 -0700 Subject: [PATCH 2/2] chore(python): add global attributes repro to flask e2e app Co-authored-by: Cursor --- e2e/python-plugin-flask/README.md | 16 +++++++++++++- e2e/python-plugin-flask/flask_app/__init__.py | 19 +++++++++++++---- e2e/python-plugin-flask/flask_app/repro.py | 21 +++++++++++++++++++ 3 files changed, 51 insertions(+), 5 deletions(-) create mode 100644 e2e/python-plugin-flask/flask_app/repro.py diff --git a/e2e/python-plugin-flask/README.md b/e2e/python-plugin-flask/README.md index 63bd231a25..4d83bf65da 100644 --- a/e2e/python-plugin-flask/README.md +++ b/e2e/python-plugin-flask/README.md @@ -43,4 +43,18 @@ environment the data should land in. | `/manual-record-chained-exception` | A `raise ... from` chain, whose root cause frames and source must survive | | `/manual-span` | `observe.start_span` and a flag evaluation | | `/manual-record-log` | `observe.record_log` | -| `/record-metrics` | Counters, gauges, and histograms | \ No newline at end of file +| `/record-metrics` | Counters, gauges, and histograms | + +# Repro #1: global attributes + +`flask_app/repro.py` calls `observe.set_attributes` with +`utm_campaign=spring_promo` and `utm_source=email` before the LaunchDarkly client +and observability plugin are initialized. `/manual-span`, `/manual-record-log`, +and `/record-metrics` print and return what to look for in LaunchDarkly: + +- `[#1] expect utm_campaign=spring_promo on this row` on the span, log, and metrics. +- `[#1] expect utm_source=per_call (per-call wins over global)` on the + `manual-record-log` log, which passes `utm_source` itself. + +`make install` copies the local `@launchdarkly/observability-python` package into +the virtualenv, so rerun it after changing the SDK. \ No newline at end of file diff --git a/e2e/python-plugin-flask/flask_app/__init__.py b/e2e/python-plugin-flask/flask_app/__init__.py index 41848b0c0d..7415933488 100644 --- a/e2e/python-plugin-flask/flask_app/__init__.py +++ b/e2e/python-plugin-flask/flask_app/__init__.py @@ -6,12 +6,19 @@ from ldclient.config import Config from loguru import logger from opentelemetry import trace +from flask_app.repro import ( + EXPECT_GLOBAL, + EXPECT_PER_CALL, + apply_global_attributes, + expect, +) sdk_key = os.getenv("LAUNCHDARKLY_SDK_KEY") if not sdk_key: raise ValueError("LAUNCHDARKLY_SDK_KEY is not set") +apply_global_attributes() ldclient.set_config(Config(sdk_key, plugins=[ @@ -43,7 +50,7 @@ def manual_span(): client.variation("test", Context.create("bob"), default=False) span.set_attribute("my-attribute", "my-value") # Any user defined code I want to capture. - return 'Manual span' + return expect('Manual span', EXPECT_GLOBAL) @app.route('/manual-record-exception') def manual_record_exception(): @@ -75,8 +82,12 @@ def manual_record_chained_exception(): @app.route('/manual-record-log') def manual_record_log(): - observe.record_log("manual-record-log", logging.INFO, {"custom": "value"}) - return "Manually recorded log" + observe.record_log( + "manual-record-log", + logging.INFO, + {"custom": "value", "utm_source": "per_call"}, + ) + return expect("Manually recorded log", EXPECT_GLOBAL, EXPECT_PER_CALL) @app.route('/record-metrics') def record_metrics(): @@ -84,4 +95,4 @@ def record_metrics(): observe.record_count("manual-record-counter", 1) observe.record_histogram("manual-record-histogram", 1) observe.record_up_down_counter("manual-record-uptime", 1) - return "Manually recorded metric" + return expect("Manually recorded metric", EXPECT_GLOBAL) diff --git a/e2e/python-plugin-flask/flask_app/repro.py b/e2e/python-plugin-flask/flask_app/repro.py new file mode 100644 index 0000000000..2c16b9f793 --- /dev/null +++ b/e2e/python-plugin-flask/flask_app/repro.py @@ -0,0 +1,21 @@ +"""Repro #1: global attributes (e.g. UTM params) on all telemetry.""" + +from ldobserve import observe + +GLOBAL_ATTRIBUTES = {"utm_campaign": "spring_promo", "utm_source": "email"} + +EXPECT_GLOBAL = "[#1] expect utm_campaign=spring_promo on this row" +EXPECT_PER_CALL = "[#1] expect utm_source=per_call (per-call wins over global)" + + +def apply_global_attributes(): + # Called before the LaunchDarkly client and observability plugin are + # initialized, to show that global attributes can be set before init. + observe.set_attributes(GLOBAL_ATTRIBUTES) + + +def expect(response, *expectations): + """Print the expectations and append them to the route's response.""" + for expectation in expectations: + print(expectation) + return "\n".join((response, *expectations))