feat: propagate trace context from user app

We may want to propagate the traceparent from an instrumented app
running Prefect deployments so that we get end-to-end tracing from
the app to the Prefect workers.

Signed-off-by: Fatih Acar <fatih@opsmill.com>

src/prefect/deployments/flow_runs.py

@@ -4,6 +4,8 @@
 
 import anyio
 import pendulum
+from opentelemetry import trace
+from opentelemetry.instrumentation.utils import is_instrumentation_enabled
 
 import prefect
 from prefect.client.schemas import FlowRun
@@ -13,9 +15,7 @@
 from prefect.results import ResultRecordMetadata
 from prefect.states import Pending, Scheduled
 from prefect.tasks import Task
-from prefect.telemetry.run_telemetry import (
-    LABELS_TRACEPARENT_KEY,
-)
+from prefect.telemetry.run_telemetry import LABELS_TRACEPARENT_KEY, RunTelemetry
 from prefect.utilities.asyncutils import sync_compatible
 from prefect.utilities.slugify import slugify
 
@@ -164,6 +164,8 @@ async def run_deployment(
 
     if flow_run_ctx and flow_run_ctx.flow_run:
         traceparent = flow_run_ctx.flow_run.labels.get(LABELS_TRACEPARENT_KEY)
+    elif is_instrumentation_enabled():
+        traceparent = RunTelemetry.traceparent_from_span(span=trace.get_current_span())
     else:
         traceparent = None
 

src/prefect/telemetry/run_telemetry.py

@@ -132,7 +132,7 @@ def _start_span(
             },
         )
 
-        if traceparent := self._traceparent_from_span(self.span):
+        if traceparent := RunTelemetry.traceparent_from_span(self.span):
             run.labels[LABELS_TRACEPARENT_KEY] = traceparent
 
         return traceparent, self.span
@@ -150,7 +150,8 @@ def _trace_context_from_labels(
         carrier = {TRACEPARENT_KEY: traceparent}
         return propagate.extract(carrier)
 
-    def _traceparent_from_span(self, span: Span) -> str | None:
+    @staticmethod
+    def traceparent_from_span(span: Span) -> str | None:
         carrier: dict[str, Any] = {}
         propagate.inject(carrier, context=trace.set_span_in_context(span))
         return carrier.get(TRACEPARENT_KEY)