Hi, we made a small adaptation to the task runner to make it run a little faster locally so we could use prefect as a slick local DAG executor. When running one of these flows within a regular flow, we get all the logs for both flows when we really don’t want the inner flow logs. How would we shut off logs for the inner flow? Was thinking something like wrapping the inner flow like this:

with prefect.context(loglevel="ERROR"):
    inner_flow.run()

Hey @Alex Papanicolaou, My initial idea was to filter out the

inner_flow

to an agent with

log_to_cloud=False

through the use of labels, however there is likely a better way of doing this. That snippet you posted may be better for differentiating the types of errors you want for each

inner_flow

.

Doesn’t seem to work:

>>> prefect.config.logging.level = 'ERROR'
>>> print(prefect.config.logging)
{'level': 'ERROR', 'format': '[%(asctime)s] %(levelname)s - %(name)s | %(message)s', 'log_attributes': [], 'datefmt': '%Y-%m-%d %H:%M:%S', 'log_to_cloud': False, 'extra_loggers': []}
>>> flow.run(x=1, y=2)
[2020-08-17 17:11:57] INFO - prefect.FlowRunner | Beginning Flow run for 'test'
[2020-08-17 17:11:57] INFO - prefect.FlowRunner | Starting flow run.
[2020-08-17 17:11:57] INFO - prefect.TaskRunner | Task 'y': Starting task run...
[2020-08-17 17:11:57] INFO - prefect.TaskRunner | Task 'y': finished task run for task with final state: 'Success'
[2020-08-17 17:11:57] INFO - prefect.TaskRunner | Task 'x': Starting task run...
[2020-08-17 17:11:57] INFO - prefect.TaskRunner | Task 'x': finished task run for task with final state: 'Success'
[2020-08-17 17:11:57] INFO - prefect.TaskRunner | Task 'add': Starting task run...
[2020-08-17 17:11:57] INFO - prefect.TaskRunner | Task 'add': finished task run for task with final state: 'Success'
[2020-08-17 17:11:57] INFO - prefect.FlowRunner | Flow run SUCCESS: all reference tasks succeeded

In which case filtering out the

inner_flow

to an agent that won't send logs to the UI may be your best bet.

ah, so we want to run this inner flow purely locally in the same process. No agents or any of the nice extra Prefect components. We had written our own DAG executor but found Prefect was basically a much better version so we changed a few minor things to make it run very cleanly locally. Here is what works.

>>> prefect.utilities.logging.prefect_logger.setLevel("ERROR")
>>> flow.run(x=1, y=2)
<Success: "All reference tasks succeeded.">

That was all you! Thanks for sharing!