xBis7 commented on code in PR #43941:
URL: https://github.com/apache/airflow/pull/43941#discussion_r2008698719
##########
airflow/traces/otel_tracer.py:
##########
@@ -130,106 +165,130 @@ def start_span(
)
return span
- def start_span_from_dagrun(
- self, dagrun, span_name: str | None = None, component: str = "dagrun",
links=None
+ def start_root_span(
+ self,
+ span_name: str,
+ component: str | None = None,
+ links=None,
+ start_time=None,
+ start_as_current: bool = True,
):
- """Produce a span from dag run."""
- # check if dagrun has configs
- conf = dagrun.conf
- trace_id = int(gen_trace_id(dag_run=dagrun, as_int=True))
- span_id = int(gen_dag_span_id(dag_run=dagrun, as_int=True))
-
- tracer = self.get_tracer(component=component, span_id=span_id,
trace_id=trace_id)
+ """Start a root span."""
+ # If no context is passed to the new span,
+ # then it will try to get the context of the current active span.
+ # Due to that, the context parameter can't be empty.
+ # It needs an invalid context in order to declare the new span as root.
+ invalid_span_ctx = SpanContext(
+ trace_id=INVALID_TRACE_ID, span_id=INVALID_SPAN_ID,
is_remote=True, trace_flags=TraceFlags(0x01)
+ )
+ invalid_ctx =
trace.set_span_in_context(NonRecordingSpan(invalid_span_ctx))
- tag_string = self.tag_string if self.tag_string else ""
- tag_string = tag_string + ("," + conf.get(TRACESTATE) if (conf and
conf.get(TRACESTATE)) else "")
+ if links is None:
+ _links = []
+ else:
+ _links = links
- if span_name is None:
- span_name = dagrun.dag_id
+ return self._new_span(
+ span_name=span_name,
+ parent_context=invalid_ctx,
+ component=component,
+ links=_links,
+ start_time=start_time,
+ start_as_current=start_as_current,
+ )
- _links = gen_links_from_kv_list(links) if links else []
+ def start_child_span(
+ self,
+ span_name: str,
+ parent_context: Context | None = None,
+ component: str | None = None,
+ links=None,
+ start_time=None,
+ start_as_current: bool = True,
+ ):
+ """Start a child span."""
+ if parent_context is None:
+ # If no context is passed, then use the current.
+ parent_span_context = trace.get_current_span().get_span_context()
+ parent_context =
trace.set_span_in_context(NonRecordingSpan(parent_span_context))
+ else:
+ context_val = next(iter(parent_context.values()))
+ parent_span_context = None
+ if isinstance(context_val, NonRecordingSpan):
+ parent_span_context = context_val.get_span_context()
- _links.append(
- Link(
- context=trace.get_current_span().get_span_context(),
- attributes={"meta.annotation_type": "link", "from":
"parenttrace"},
+ if links is None:
+ _links = []
+ else:
+ _links = links
+
+ if parent_span_context is not None:
+ _links.append(
+ Link(
+ context=parent_span_context,
+ attributes={"meta.annotation_type": "link", "from":
"parenttrace"},
+ )
)
- )
-
- if conf and conf.get(TRACEPARENT):
- # add the trace parent as the link
- _links.append(gen_link_from_traceparent(conf.get(TRACEPARENT)))
- span_ctx = SpanContext(
- trace_id=INVALID_TRACE_ID, span_id=INVALID_SPAN_ID,
is_remote=True, trace_flags=TraceFlags(0x01)
- )
- ctx = trace.set_span_in_context(NonRecordingSpan(span_ctx))
- span = tracer.start_as_current_span(
- name=span_name,
- context=ctx,
+ return self._new_span(
+ span_name=span_name,
+ parent_context=parent_context,
+ component=component,
links=_links,
- start_time=datetime_to_nano(dagrun.queued_at),
- attributes=parse_tracestate(tag_string),
+ start_time=start_time,
+ start_as_current=start_as_current,
)
- return span
- def start_span_from_taskinstance(
+ def _new_span(
self,
- ti,
- span_name: str | None = None,
- component: str = "taskinstance",
- child: bool = False,
+ span_name: str,
+ parent_context: Context | None = None,
+ component: str | None = None,
links=None,
+ start_time=None,
+ start_as_current: bool = True,
):
- """
- Create and start span from given task instance.
+ if component is None:
+ component = self.otel_service
- Essentially the span represents the ti itself if child == True, it
will create a 'child' span under the given span.
- """
- dagrun = ti.dag_run
- conf = dagrun.conf
- trace_id = int(gen_trace_id(dag_run=dagrun, as_int=True))
- span_id = int(gen_span_id(ti=ti, as_int=True))
- if span_name is None:
- span_name = ti.task_id
+ tracer = self.get_tracer(component=component)
- parent_id = span_id if child else int(gen_dag_span_id(dag_run=dagrun,
as_int=True))
+ if start_time is None:
+ start_time = pendulum.now("UTC")
Review Comment:
The result is the same
```
pendulum: 2025-03-22 06:58:29.245029+00:00
timezone.utcnow(): 2025-03-22 06:58:29.245175+00:00
```
There shouldn't be any issue. I made the change.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@airflow.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
