rename add_span to add_debug_span

apache · xBis7 · Apr 6, 2025 · Apr 6, 2025 · Apr 6, 2025 · Apr 7, 2025
commit 4769ac980c0fdd9e4d0e29da6322c0dedf010f3f
@@ -33,7 +33,7 @@
 from airflow.models import Log
 from airflow.stats import Stats
 from airflow.traces import NO_TRACE_ID
-from airflow.traces.tracer import DebugTrace, Trace, add_span, gen_context
+from airflow.traces.tracer import DebugTrace, Trace, add_debug_span, gen_context
 from airflow.traces.utils import gen_span_id_from_ti_key, gen_trace_id
 from airflow.utils.log.logging_mixin import LoggingMixin
 from airflow.utils.state import TaskInstanceState
@@ -270,7 +270,7 @@ def sync(self) -> None:
         Executors should override this to perform gather statuses.
         """
 
-    @add_span
+    @add_debug_span
     def heartbeat(self) -> None:
         """Heartbeat sent to trigger new jobs."""
         open_slots = self.parallelism - len(self.running)
@@ -373,7 +373,7 @@ def order_queued_tasks_by_priority(self) -> list[tuple[TaskInstanceKey, QueuedTa
             reverse=True,
         )
 
-    @add_span
+    @add_debug_span
     def trigger_tasks(self, open_slots: int) -> None:
         """
         Initiate async execution of the queued tasks, up to the number of available slots.
@@ -465,7 +465,7 @@ def trigger_tasks(self, open_slots: int) -> None:
         elif workload_list:
             self._process_workloads(workload_list)  # type: ignore[attr-defined]
 
-    @add_span
+    @add_debug_span
     def _process_tasks(self, task_tuples: list[TaskTuple]) -> None:
         for key, command, queue, executor_config in task_tuples:
             task_instance = self.queued_tasks[key][3]  # TaskInstance in fourth element

@@ -32,7 +32,7 @@
 from airflow.listeners.listener import get_listener_manager
 from airflow.models.base import ID_LEN, Base
 from airflow.stats import Stats
-from airflow.traces.tracer import DebugTrace, add_span
+from airflow.traces.tracer import DebugTrace, add_debug_span
 from airflow.utils import timezone
 from airflow.utils.helpers import convert_camel_to_snake
 from airflow.utils.log.logging_mixin import LoggingMixin
@@ -385,7 +385,7 @@ def execute_job(job: Job, execute_callable: Callable[[], int | None]) -> int | N
     return ret
 
 
-@add_span
+@add_debug_span
 def perform_heartbeat(
     job: Job, heartbeat_callback: Callable[[Session], None], only_if_necessary: bool
 ) -> None:

@@ -67,7 +67,7 @@
 from airflow.ti_deps.dependencies_states import EXECUTION_STATES
 from airflow.timetables.simple import AssetTriggeredTimetable
 from airflow.traces import utils as trace_utils
-from airflow.traces.tracer import DebugTrace, Trace, add_span
+from airflow.traces.tracer import DebugTrace, Trace, add_debug_span
 from airflow.utils import timezone
 from airflow.utils.dates import datetime_to_nano
 from airflow.utils.event_scheduler import EventScheduler
@@ -1489,7 +1489,7 @@ def _mark_backfills_complete(self, session: Session = NEW_SESSION) -> None:
         for b in backfills:
             b.completed_at = now
 
-    @add_span
+    @add_debug_span
     def _create_dag_runs(self, dag_models: Collection[DagModel], session: Session) -> None:
         """Create a DAG run and update the dag_model to control if/when the next DAGRun should be created."""
         # Bulk Fetch DagRuns with dag_id and logical_date same
@@ -1674,7 +1674,7 @@ def _should_update_dag_next_dagruns(
             return False
         return True
 
-    @add_span
+    @add_debug_span
     def _start_queued_dagruns(self, session: Session) -> None:
         """Find DagRuns in queued state and decide moving them to running state."""
         # added all() to save runtime, otherwise query is executed more than once
@@ -1691,7 +1691,7 @@ def _start_queued_dagruns(self, session: Session) -> None:
         )
         active_runs_of_dags = Counter({(dag_id, br_id): num for dag_id, br_id, num in session.execute(query)})
 
-        @add_span
+        @add_debug_span
         def _update_state(dag: DAG, dag_run: DagRun):
             span = Trace.get_current_span()
             span.set_attributes(

@@ -61,7 +61,7 @@
 )
 from airflow.sdk.execution_time.supervisor import WatchedSubprocess, make_buffered_socket_reader
 from airflow.stats import Stats
-from airflow.traces.tracer import DebugTrace, Trace, add_span
+from airflow.traces.tracer import DebugTrace, Trace, add_debug_span
 from airflow.triggers import base as events
 from airflow.utils import timezone
 from airflow.utils.helpers import log_filename_template_renderer
@@ -484,14 +484,14 @@ def heartbeat(self):
     def heartbeat_callback(self, session: Session | None = None) -> None:
         Stats.incr("triggerer_heartbeat", 1, 1)
 
-    @add_span
+    @add_debug_span
     def load_triggers(self):
         """Query the database for the triggers we're supposed to be running and update the runner."""
         Trigger.assign_unassigned(self.job.id, self.capacity, self.health_check_threshold)
         ids = Trigger.ids_for_triggerer(self.job.id)
         self.update_triggers(set(ids))
 
-    @add_span
+    @add_debug_span
     def handle_events(self):
         """Dispatch outbound events to the Trigger model which pushes them to the relevant task instances."""
         while self.events:
@@ -502,12 +502,12 @@ def handle_events(self):
             # Emit stat event
             Stats.incr("triggers.succeeded")
 
-    @add_span
+    @add_debug_span
     def clean_unused(self):
         """Clean out unused or finished triggers."""
         Trigger.clean_unused()
 
-    @add_span
+    @add_debug_span
     def handle_failed_triggers(self):
         """
         Handle "failed" triggers. - ones that errored or exited before they sent an event.

diff --git a/airflow-core/src/airflow/traces/tracer.py b/airflow-core/src/airflow/traces/tracer.py
@@ -44,7 +44,7 @@ def gen_links_from_kv_list(list):
     return gen_links_from_kv_list(list)
 
 
-def add_span(func):
+def add_debug_span(func):
     """Decorate a function with span."""
     func_name = func.__name__
     qual_name = func.__qualname__