feat(python): add set_attributes and custom span/log processors

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 |
+| `/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.

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,13 +82,17 @@ 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():
     observe.record_metric("manual-record-metric", 1)
     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)

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

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!

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)
 

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