https://prefect.io logo
Title
a

Alex Papanicolaou

08/17/2020, 4:52 PM
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()
Would this work?
original_level = prefect.config.logging.level
prefect.config.logging.level = 'ERROR'
inner_flow.run()
prefect.config.logging.level = original_level
k

Kyle Moon-Wright

08/17/2020, 5:10 PM
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
.
a

Alex Papanicolaou

08/17/2020, 5:12 PM
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
not really surprised it doesn’t, of course.
k

Kyle Moon-Wright

08/17/2020, 5:16 PM
In which case filtering out the
inner_flow
to an agent that won't send logs to the UI may be your best bet.
a

Alex Papanicolaou

08/17/2020, 5:26 PM
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.">
Thanks for the help (or at least listening to me as I worked through it)!
k

Kyle Moon-Wright

08/17/2020, 5:27 PM
That was all you! Thanks for sharing!