Propagate trace context to triggered deployments from within parent run

docs/v3/api-ref/rest-api/server/schema.json

@@ -15840,6 +15840,40 @@
                         "title": "Idempotency Key",
                         "description": "An optional idempotency key. If a flow run with the same idempotency key has already been created, the existing flow run will be returned."
                     },
+                    "labels": {
+                        "anyOf": [
+                            {
+                                "additionalProperties": {
+                                    "anyOf": [
+                                        {
+                                            "type": "boolean"
+                                        },
+                                        {
+                                            "type": "integer"
+                                        },
+                                        {
+                                            "type": "number"
+                                        },
+                                        {
+                                            "type": "string"
+                                        }
+                                    ]
+                                },
+                                "type": "object"
+                            },
+                            {
+                                "type": "null"
+                            }
+                        ],
+                        "title": "Labels",
+                        "description": "A dictionary of key-value labels. Values can be strings, numbers, or booleans.",
+                        "examples": [
+                            {
+                                "key": "value1",
+                                "key2": 42
+                            }
+                        ]
+                    },
                     "parent_task_run_id": {
                         "anyOf": [
                             {

src/prefect/client/orchestration.py

@@ -606,6 +606,7 @@ async def create_flow_run_from_deployment(
         parent_task_run_id: Optional[UUID] = None,
         work_queue_name: Optional[str] = None,
         job_variables: Optional[dict[str, Any]] = None,
+        labels: Optional[KeyValueLabelsField] = None,
     ) -> FlowRun:
         """
         Create a flow run for a deployment.
@@ -651,6 +652,7 @@ async def create_flow_run_from_deployment(
             idempotency_key=idempotency_key,
             parent_task_run_id=parent_task_run_id,
             job_variables=job_variables,
+            labels=labels,
         )
 
         # done separately to avoid including this field in payloads sent to older API versions

src/prefect/client/schemas/actions.py

@@ -397,7 +397,7 @@ class DeploymentFlowRunCreate(ActionBaseModel):
     parent_task_run_id: Optional[UUID] = Field(None)
     work_queue_name: Optional[str] = Field(None)
     job_variables: Optional[dict] = Field(None)
-    labels: KeyValueLabelsField = Field(default_factory=dict)
+    labels: Optional[KeyValueLabelsField] = Field(default=None)
 
 
 class SavedSearchCreate(ActionBaseModel):

src/prefect/deployments/flow_runs.py

@@ -13,6 +13,11 @@
 from prefect.results import BaseResult, ResultRecordMetadata
 from prefect.states import Pending, Scheduled
 from prefect.tasks import Task
+from prefect.telemetry.run_telemetry import (
+    LABELS_TRACEPARENT_KEY,
+    TRACEPARENT_KEY,
+)
+from prefect.types import KeyValueLabels
 from prefect.utilities.asyncutils import sync_compatible
 from prefect.utilities.slugify import slugify
 
@@ -156,6 +161,17 @@ async def run_deployment(
     else:
         parent_task_run_id = None
 
+    if flow_run_ctx and flow_run_ctx.flow_run:
+        traceparent = flow_run_ctx.flow_run.labels.get(LABELS_TRACEPARENT_KEY)
+    else:
+        traceparent = None
+
+    trace_labels = {}
+    if traceparent:
+        carrier: KeyValueLabels = {TRACEPARENT_KEY: traceparent}
+
+        trace_labels = {LABELS_TRACEPARENT_KEY: carrier[TRACEPARENT_KEY]}
+
     flow_run = await client.create_flow_run_from_deployment(
         deployment.id,
         parameters=parameters,
@@ -166,6 +182,7 @@ async def run_deployment(
         parent_task_run_id=parent_task_run_id,
         work_queue_name=work_queue_name,
         job_variables=job_variables,
+        labels=trace_labels,
     )
 
     flow_run_id = flow_run.id

src/prefect/server/schemas/actions.py

@@ -585,6 +585,11 @@ class DeploymentFlowRunCreate(ActionBaseModel):
             " has already been created, the existing flow run will be returned."
         ),
     )
+    labels: Union[schemas.core.KeyValueLabels, None] = Field(
+        None,
+        description="A dictionary of key-value labels. Values can be strings, numbers, or booleans.",
+        examples=[{"key": "value1", "key2": 42}],
+    )
     parent_task_run_id: Optional[UUID] = Field(None)
     work_queue_name: Optional[str] = Field(None)
     job_variables: Optional[Dict[str, Any]] = Field(None)

tests/deployment/test_flow_runs.py

@@ -9,14 +9,17 @@
 from httpx import Response
 
 from prefect import flow
+from prefect.client.schemas.responses import DeploymentResponse
 from prefect.context import FlowRunContext
 from prefect.deployments import run_deployment
+from prefect.flow_engine import run_flow_async
 from prefect.server.schemas.core import TaskRunResult
 from prefect.settings import (
     PREFECT_API_URL,
 )
 from prefect.tasks import task
 from prefect.utilities.slugify import slugify
+from tests.telemetry.instrumentation_tester import InstrumentationTester
 
 if TYPE_CHECKING:
     from prefect.client.orchestration import PrefectClient
@@ -433,3 +436,60 @@ async def foo():
                 )
             ]
         }
+
+    async def test_propagates_otel_trace_to_deployment_flow_run(
+        self,
+        test_deployment: DeploymentResponse,
+        instrumentation: InstrumentationTester,
+        prefect_client: "PrefectClient",
+    ):
+        """Test that OTEL trace context gets propagated from parent flow to deployment flow run"""
+        deployment = test_deployment
+        mock_traceparent = "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01"
+
+        @flow(name="child-flow")
+        async def child_flow() -> None:
+            pass
+
+        flow_id = await prefect_client.create_flow(child_flow)
+
+        deployment_id = await prefect_client.create_deployment(
+            name="foo-deployment", flow_id=flow_id, parameter_openapi_schema={}
+        )
+        deployment = await prefect_client.read_deployment(deployment_id)
+
+        @flow(name="parent-flow")
+        async def parent_flow():
+            # Set OTEL context in the parent flow's labels
+            flow_run = FlowRunContext.get().flow_run
+
+            flow_run.labels["__OTEL_TRACEPARENT"] = mock_traceparent
+
+            return await run_deployment(
+                f"foo/{deployment.name}",
+                timeout=0,
+                poll_interval=0,
+            )
+
+        parent_state = await parent_flow(return_state=True)
+        child_flow_run = await parent_state.result()
+
+        await run_flow_async(child_flow, child_flow_run)
+        # Get all spans
+        spans = instrumentation.get_finished_spans()
+
+        # Find parent flow span
+        parent_span = next(
+            span
+            for span in spans
+            if span.attributes.get("prefect.flow.name") == "parent-flow"
+        )
+        child_span = next(
+            span
+            for span in spans
+            if span.attributes.get("prefect.flow.name") == "child-flow"
+        )
+        assert child_span
+        assert parent_span
+
+        assert mock_traceparent == child_flow_run.labels["__OTEL_TRACEPARENT"]

ui-v2/src/api/prefect.ts

@@ -5434,6 +5434,13 @@ export interface components {
              * @description An optional idempotency key. If a flow run with the same idempotency key has already been created, the existing flow run will be returned.
              */
             idempotency_key?: string | null;
+            /**
+             * Labels
+             * @description A dictionary of key-value labels. Values can be strings, numbers, or booleans.
+             */
+            labels?: {
+                [key: string]: boolean | number | string;
+            } | null;
             /** Parent Task Run Id */
             parent_task_run_id?: string | null;
             /** Work Queue Name */