Repository: PrefectHQ/prefect
Pull request: log rendered custom `flow_run_name`
closes https://github.com/PrefectHQ/prefect/issues/16513

to log the correct name, we have to defer this info log until after we resolve the custom name

I changed `Created` to `Beginning` since we now "create" the flow run much earlier than we log this message, and I wanted to keep the log accurate - open to alternate wordings

---

### before
```python
In [1]: from prefect import flow

In [2]: @flow(flow_run_name='foo')
   ...: def f(): pass

In [3]:

In [3]: f()
13:10:12.057 | INFO    | prefect.engine - Created flow run 'poised-griffin' for flow 'f'
13:10:12.556 | INFO    | Flow run 'foo' - Finished in state Completed()
```
note `Created flow run 'poised-griffin' for flow 'f'`

### after
```python
In [1]: from prefect import flow

In [2]: @flow(flow_run_name='foo')
   ...: def f(): pass

In [3]: f()
13:09:35.053 | INFO    | Flow run 'foo' - Beginning flow run 'foo' for flow 'f'
13:09:35.194 | INFO    | Flow run 'foo' - Finished in state Completed()
```

Problem statement:
incorrect `flow_run_name` logged
### Bug summary

Even though the API is getting the right value, we are logging the auto-generated value

```python
In [1]: from prefect import flow

In [2]: @flow(flow_run_name='foo')
   ...: def f(): pass

In [3]: f()
12:29:41.538 | INFO    | prefect.engine - Created flow run 'fervent-deer' for flow 'f'
12:29:42.017 | INFO    | Flow run 'foo' - Finished in state Completed()
```




### Version info

```Text
main
```


### Additional context

_No response_

Hints from issue discussion:

Commit message: fix order of logging

Patch to review:
diff --git a/src/prefect/flow_engine.py b/src/prefect/flow_engine.py
index f9cd310e1a5f..57788d6c9ab0 100644
--- a/src/prefect/flow_engine.py
+++ b/src/prefect/flow_engine.py
@@ -490,24 +490,13 @@ def create_flow_run(self, client: SyncPrefectClient) -> FlowRun:
             ):
                 return subflow_run
 
-        flow_run = client.create_flow_run(
+        return client.create_flow_run(
             flow=self.flow,
             parameters=self.flow.serialize_parameters(parameters),
             state=Pending(),
             parent_task_run_id=getattr(parent_task_run, "id", None),
             tags=TagsContext.get().current_tags,
         )
-        if flow_run_ctx:
-            parent_logger = get_run_logger(flow_run_ctx)
-            parent_logger.info(
-                f"Created subflow run {flow_run.name!r} for flow {self.flow.name!r}"
-            )
-        else:
-            self.logger.info(
-                f"Created flow run {flow_run.name!r} for flow {self.flow.name!r}"
-            )
-
-        return flow_run
 
     def call_hooks(self, state: Optional[State] = None):
         if state is None:
@@ -606,6 +595,7 @@ def setup_run_context(self, client: Optional[SyncPrefectClient] = None):
             stack.enter_context(ConcurrencyContext())
 
             # set the logger to the flow run logger
+
             self.logger = flow_run_logger(flow_run=self.flow_run, flow=self.flow)
 
             # update the flow run name if necessary
@@ -622,6 +612,22 @@ def setup_run_context(self, client: Optional[SyncPrefectClient] = None):
                 )
                 self.flow_run.name = flow_run_name
                 self._flow_run_name_set = True
+
+            _logger, run_type = (
+                (get_run_logger(FlowRunContext.get()), "subflow")
+                if self.flow_run.parent_task_run_id
+                else (self.logger, "flow")
+            )
+
+            _logger.info(
+                f"Beginning {run_type} run {self.flow_run.name!r} for flow {self.flow.name!r}"
+            )
+
+            if flow_run_url := url_for(self.flow_run):
+                self.logger.info(
+                    f"View at {flow_run_url}", extra={"send_to_api": False}
+                )
+
             yield
 
     @contextmanager
@@ -635,12 +641,6 @@ def initialize_run(self):
 
             if not self.flow_run:
                 self.flow_run = self.create_flow_run(self.client)
-                flow_run_url = url_for(self.flow_run)
-
-                if flow_run_url:
-                    self.logger.info(
-                        f"View at {flow_run_url}", extra={"send_to_api": False}
-                    )
             else:
                 # Update the empirical policy to match the flow if it is not set
                 if self.flow_run.empirical_policy.retry_delay is None:
@@ -705,9 +705,11 @@ def initialize_run(self):
     @contextmanager
     def start(self) -> Generator[None, None, None]:
         with self.initialize_run():
-            with trace.use_span(
-                self._telemetry.span
-            ) if self._telemetry.span else nullcontext():
+            with (
+                trace.use_span(self._telemetry.span)
+                if self._telemetry.span
+                else nullcontext()
+            ):
                 self.begin_run()
 
                 if self.state.is_running():
@@ -1052,24 +1054,13 @@ async def create_flow_run(self, client: PrefectClient) -> FlowRun:
             ):
                 return subflow_run
 
-        flow_run = await client.create_flow_run(
+        return await client.create_flow_run(
             flow=self.flow,
             parameters=self.flow.serialize_parameters(parameters),
             state=Pending(),
             parent_task_run_id=getattr(parent_task_run, "id", None),
             tags=TagsContext.get().current_tags,
         )
-        if flow_run_ctx:
-            parent_logger = get_run_logger(flow_run_ctx)
-            parent_logger.info(
-                f"Created subflow run {flow_run.name!r} for flow {self.flow.name!r}"
-            )
-        else:
-            self.logger.info(
-                f"Created flow run {flow_run.name!r} for flow {self.flow.name!r}"
-            )
-
-        return flow_run
 
     async def call_hooks(self, state: Optional[State] = None):
         if state is None:
@@ -1184,6 +1175,22 @@ async def setup_run_context(self, client: Optional[PrefectClient] = None):
                 )
                 self.flow_run.name = flow_run_name
                 self._flow_run_name_set = True
+
+            _logger, run_type = (
+                (get_run_logger(FlowRunContext.get()), "subflow")
+                if self.flow_run.parent_task_run_id
+                else (self.logger, "flow")
+            )
+
+            _logger.info(
+                f"Beginning {run_type} run {self.flow_run.name!r} for flow {self.flow.name!r}"
+            )
+
+            if flow_run_url := url_for(self.flow_run):
+                self.logger.info(
+                    f"View at {flow_run_url}", extra={"send_to_api": False}
+                )
+
             yield
 
     @asynccontextmanager
@@ -1267,9 +1274,11 @@ async def initialize_run(self):
     @asynccontextmanager
     async def start(self) -> AsyncGenerator[None, None]:
         async with self.initialize_run():
-            with trace.use_span(
-                self._telemetry.span
-            ) if self._telemetry.span else nullcontext():
+            with (
+                trace.use_span(self._telemetry.span)
+                if self._telemetry.span
+                else nullcontext()
+            ):
                 await self.begin_run()
 
                 if self.state.is_running():
diff --git a/tests/test_flows.py b/tests/test_flows.py
index 733529e54677..596ebc947cff 100644
--- a/tests/test_flows.py
+++ b/tests/test_flows.py
@@ -2829,6 +2829,25 @@ def flow_method():
         assert mocked_flow_method.call_count == 2
         assert generate_flow_run_name.call_count == 2
 
+    async def test_both_engines_logs_custom_flow_run_name(
+        self, caplog: pytest.LogCaptureFixture
+    ):
+        @flow(flow_run_name="very-bespoke-name")
+        def test():
+            pass
+
+        test()
+
+        assert "Beginning flow run 'very-bespoke-name'" in caplog.text
+
+        @flow(flow_run_name="very-bespoke-async-name")
+        async def test_async():
+            pass
+
+        await test_async()
+
+        assert "Beginning flow run 'very-bespoke-async-name'" in caplog.text
+
 
 def create_hook(mock_obj):
     def my_hook(flow, flow_run, state):


Code review comments:
diff_hunk: @@ -1184,6 +1175,22 @@ async def setup_run_context(self, client: Optional[PrefectClient] = None):
                 )
                 self.flow_run.name = flow_run_name
                 self._flow_run_name_set = True
+
+            _logger, run_type = (
+                (get_run_logger(FlowRunContext.get()), "subflow")
+                if self.flow_run.parent_task_run_id
+                else (self.logger, "flow")
+            )
path: src/prefect/flow_engine.py
text: @user1:
For readability:
```suggestion
            if self.flow_run.parent_task_run_id:
                _logger = get_run_logger(FlowRunContext.get())
                run_type = "subflow"
            else:
                _logger = self.logger
                run_type = "flow"
```

diff_hunk: @@ -622,6 +612,22 @@ def setup_run_context(self, client: Optional[SyncPrefectClient] = None):
                 )
                 self.flow_run.name = flow_run_name
                 self._flow_run_name_set = True
+
+            _logger, run_type = (
+                (get_run_logger(FlowRunContext.get()), "subflow")
+                if self.flow_run.parent_task_run_id
+                else (self.logger, "flow")
+            )
path: src/prefect/flow_engine.py
text: @user1:
For readability:
```suggestion
            if self.flow_run.parent_task_run_id:
                _logger = get_run_logger(FlowRunContext.get())
                run_type = "subflow"
            else:
                _logger = self.logger
                run_type = "flow"
```