This is an automated email from the ASF dual-hosted git repository.
dstandish pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 76bea5421ac Move otel tracing helper stuff to _shared (#63566)
76bea5421ac is described below
commit 76bea5421ac99a50530f4cb12587a50a5c0c785b
Author: Daniel Standish <[email protected]>
AuthorDate: Mon Mar 16 07:13:35 2026 -0700
Move otel tracing helper stuff to _shared (#63566)
This stuff is not needed in airflow core. It is not public interface. And
it may at some point be needed by task sdk.
---
airflow-core/src/airflow/__init__.py | 2 +-
airflow-core/src/airflow/models/dagrun.py | 2 +-
airflow-core/src/airflow/settings.py | 4 ++--
airflow-core/tests/unit/models/test_dagrun.py | 4 ++--
.../src/airflow_shared}/observability/traces/__init__.py | 11 ++++++-----
5 files changed, 12 insertions(+), 11 deletions(-)
diff --git a/airflow-core/src/airflow/__init__.py
b/airflow-core/src/airflow/__init__.py
index 302c275f60b..177cdbd15ee 100644
--- a/airflow-core/src/airflow/__init__.py
+++ b/airflow-core/src/airflow/__init__.py
@@ -92,7 +92,7 @@ __lazy_imports: dict[str, tuple[str, str, bool]] = {
"Stats": (".observability.stats", "Stats", True),
"Trace": (".observability.trace", "Trace", True),
"metrics": (".observability.metrics", "", True),
- "traces": (".observability.traces", "", True),
+ "traces": ("._shared.observability.traces", "", True),
}
if TYPE_CHECKING:
# These objects are imported by PEP-562, however, static analyzers and
IDE's
diff --git a/airflow-core/src/airflow/models/dagrun.py
b/airflow-core/src/airflow/models/dagrun.py
index 61242e45390..9923781fbe5 100644
--- a/airflow-core/src/airflow/models/dagrun.py
+++ b/airflow-core/src/airflow/models/dagrun.py
@@ -62,6 +62,7 @@ from sqlalchemy.sql.functions import coalesce
from airflow._shared.observability.metrics.dual_stats_manager import
DualStatsManager
from airflow._shared.observability.metrics.stats import Stats
+from airflow._shared.observability.traces import new_dagrun_trace_carrier,
override_ids
from airflow._shared.timezones import timezone
from airflow.callbacks.callback_requests import DagCallbackRequest,
DagRunContext
from airflow.configuration import conf as airflow_conf
@@ -75,7 +76,6 @@ from airflow.models.taskinstance import TaskInstance as TI
from airflow.models.taskinstancehistory import TaskInstanceHistory as TIH
from airflow.models.tasklog import LogTemplate
from airflow.models.taskmap import TaskMap
-from airflow.observability.traces import new_dagrun_trace_carrier, override_ids
from airflow.serialization.definitions.deadline import
SerializedReferenceModels
from airflow.serialization.definitions.notset import NOTSET, ArgNotSet,
is_arg_set
from airflow.ti_deps.dep_context import DepContext
diff --git a/airflow-core/src/airflow/settings.py
b/airflow-core/src/airflow/settings.py
index 49d46f652c6..223ec291b4d 100644
--- a/airflow-core/src/airflow/settings.py
+++ b/airflow-core/src/airflow/settings.py
@@ -38,7 +38,7 @@ from sqlalchemy.ext.asyncio import (
)
from sqlalchemy.orm import scoped_session, sessionmaker
-from airflow.observability.traces import configure_otel
+from airflow._shared.observability.traces import configure_otel
try:
from sqlalchemy.ext.asyncio import async_sessionmaker
@@ -724,7 +724,7 @@ def initialize():
load_policy_plugins(policy_mgr)
import_local_settings()
configure_logging()
- configure_otel()
+ configure_otel(conf)
configure_adapters()
# The webservers import this file from models.py with the default settings.
diff --git a/airflow-core/tests/unit/models/test_dagrun.py
b/airflow-core/tests/unit/models/test_dagrun.py
index 14722f83b0c..2bed00cf75c 100644
--- a/airflow-core/tests/unit/models/test_dagrun.py
+++ b/airflow-core/tests/unit/models/test_dagrun.py
@@ -3238,7 +3238,7 @@ class TestDagRunTracing:
from opentelemetry.sdk.trace.export.in_memory_span_exporter import
InMemorySpanExporter
from opentelemetry.trace.propagation.tracecontext import
TraceContextTextMapPropagator
- from airflow.observability.traces import OverrideableRandomIdGenerator
+ from airflow._shared.observability.traces import
OverrideableRandomIdGenerator
in_mem_exporter = InMemorySpanExporter()
provider = TracerProvider(id_generator=OverrideableRandomIdGenerator())
@@ -3279,7 +3279,7 @@ class TestDagRunTracing:
from opentelemetry.sdk.trace.export.in_memory_span_exporter import
InMemorySpanExporter
from opentelemetry.trace import StatusCode
- from airflow.observability.traces import OverrideableRandomIdGenerator
+ from airflow._shared.observability.traces import
OverrideableRandomIdGenerator
in_mem_exporter = InMemorySpanExporter()
provider = TracerProvider(id_generator=OverrideableRandomIdGenerator())
diff --git a/airflow-core/src/airflow/observability/traces/__init__.py
b/shared/observability/src/airflow_shared/observability/traces/__init__.py
similarity index 95%
rename from airflow-core/src/airflow/observability/traces/__init__.py
rename to
shared/observability/src/airflow_shared/observability/traces/__init__.py
index 6bf0019f747..53163e45b97 100644
--- a/airflow-core/src/airflow/observability/traces/__init__.py
+++ b/shared/observability/src/airflow_shared/observability/traces/__init__.py
@@ -21,6 +21,7 @@ import logging
import os
from contextlib import contextmanager
from importlib.metadata import entry_points
+from typing import TYPE_CHECKING
from opentelemetry import context, trace
from opentelemetry.sdk.resources import Resource
@@ -30,8 +31,8 @@ from opentelemetry.sdk.trace.id_generator import
RandomIdGenerator
from opentelemetry.trace import NonRecordingSpan, SpanContext, TraceFlags
from opentelemetry.trace.propagation.tracecontext import
TraceContextTextMapPropagator
-from airflow.configuration import conf
-
+if TYPE_CHECKING:
+ from configparser import ConfigParser
log = logging.getLogger(__name__)
OVERRIDE_SPAN_ID_KEY = context.create_key("override_span_id")
@@ -80,7 +81,7 @@ def override_ids(trace_id, span_id, ctx=None):
context.detach(token)
-def _get_backcompat_config() -> tuple[str | None, Resource | None]:
+def _get_backcompat_config(conf: ConfigParser) -> tuple[str | None, Resource |
None]:
"""
Possibly get deprecated Airflow configs for otel.
@@ -128,7 +129,7 @@ def _load_exporter_from_env() -> SpanExporter:
return ep.load()()
-def configure_otel():
+def configure_otel(conf: ConfigParser):
otel_on = conf.getboolean("traces", "otel_on", fallback=False)
if not otel_on:
return
@@ -136,7 +137,7 @@ def configure_otel():
# ideally both endpoint and resource are None here
# they would only be something other than None if user is using deprecated
# Airflow-defined otel configs
- backcompat_endpoint, resource = _get_backcompat_config()
+ backcompat_endpoint, resource = _get_backcompat_config(conf)
# backcompat: if old-style host/port config provided an endpoint, set the
# env var so the exporter (loaded below) picks it up automatically
