Incorrect try number subtraction producing invalid span id for OTEL a…

…irflow (issue #41501) (#41502) (#41535) * Fix for issue #39336 * removed unnecessary import (cherry picked from commit dd3c3a7) Co-authored-by: Howard Yoo <32691630+howardyoo@users.noreply.github.com>
apache · Aug 16, 2024 · 30cfb73 · 30cfb73
1 parent 5f8230b
commit 30cfb73
Show file tree

Hide file tree

Showing 5 changed files with 5 additions and 10 deletions.
diff --git a/airflow/executors/base_executor.py b/airflow/executors/base_executor.py
@@ -467,7 +467,7 @@ def success(self, key: TaskInstanceKey, info=None) -> None:
                 span.set_attribute("dag_id", key.dag_id)
                 span.set_attribute("run_id", key.run_id)
                 span.set_attribute("task_id", key.task_id)
-                span.set_attribute("try_number", key.try_number - 1)
+                span.set_attribute("try_number", key.try_number)
 
         self.change_state(key, TaskInstanceState.SUCCESS, info)
 

diff --git a/airflow/executors/local_executor.py b/airflow/executors/local_executor.py
@@ -277,7 +277,7 @@ def execute_async(
                 span.set_attribute("dag_id", key.dag_id)
                 span.set_attribute("run_id", key.run_id)
                 span.set_attribute("task_id", key.task_id)
-                span.set_attribute("try_number", key.try_number - 1)
+                span.set_attribute("try_number", key.try_number)
                 span.set_attribute("commands_to_run", str(command))
 
             local_worker = LocalWorker(self.executor.result_queue, key=key, command=command)

diff --git a/airflow/executors/sequential_executor.py b/airflow/executors/sequential_executor.py
@@ -76,7 +76,7 @@ def execute_async(
             span.set_attribute("dag_id", key.dag_id)
             span.set_attribute("run_id", key.run_id)
             span.set_attribute("task_id", key.task_id)
-            span.set_attribute("try_number", key.try_number - 1)
+            span.set_attribute("try_number", key.try_number)
             span.set_attribute("commands_to_run", str(self.commands_to_run))
 
     def sync(self) -> None:

diff --git a/airflow/jobs/scheduler_job_runner.py b/airflow/jobs/scheduler_job_runner.py
@@ -837,7 +837,7 @@ def _process_executor_events(self, executor: BaseExecutor, session: Session) ->
                 span.set_attribute("hostname", ti.hostname)
                 span.set_attribute("log_url", ti.log_url)
                 span.set_attribute("operator", str(ti.operator))
-                span.set_attribute("try_number", ti.try_number - 1)
+                span.set_attribute("try_number", ti.try_number)
                 span.set_attribute("executor_state", state)
                 span.set_attribute("job_id", ti.job_id)
                 span.set_attribute("pool", ti.pool)

diff --git a/airflow/traces/utils.py b/airflow/traces/utils.py
@@ -22,7 +22,6 @@
 
 from airflow.traces import NO_TRACE_ID
 from airflow.utils.hashlib_wrapper import md5
-from airflow.utils.state import TaskInstanceState
 
 if TYPE_CHECKING:
     from airflow.models import DagRun, TaskInstance
@@ -75,12 +74,8 @@ def gen_dag_span_id(dag_run: DagRun, as_int: bool = False) -> str | int:
 def gen_span_id(ti: TaskInstance, as_int: bool = False) -> str | int:
     """Generate span id from the task instance."""
     dag_run = ti.dag_run
-    if ti.state == TaskInstanceState.SUCCESS or ti.state == TaskInstanceState.FAILED:
-        try_number = ti.try_number - 1
-    else:
-        try_number = ti.try_number
     return _gen_id(
-        [dag_run.dag_id, dag_run.run_id, ti.task_id, str(try_number)],
+        [dag_run.dag_id, dag_run.run_id, ti.task_id, str(ti.try_number)],
         as_int,
         SPAN_ID,
     )