You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@airflow.apache.org by "Vandon, Raphael" <va...@amazon.com.INVALID> on 2023/06/02 23:00:05 UTC
Deferrable operators in system tests, how?
Hello,
As we implement deferrable capabilities for operators, we're going to want to update the sample dags/system tests.
The problem we encounter at the moment is that system tests are run with pytest using the DebugExecutor, and it only runs the "worker part" of the operators, without handling the "deferred part" that is taking place in the Triggers. In system tests, a task ran in deferrable mode just hangs forever in the “deferred” state, never completing.
I see 3 approaches to that problem:
1. Patch the DebugExecutor (or an other executor that'd become the "official" executor for system tests) to handle deferred tasks. This bears some complexity, but is probably doable with some ad-hoc code that won't be the same as the "production" code but could execute the right things in order (Trigger.run after the TaskDeferred exception, and then the "execute_complete" method with the event returned)
2. Change the way we run system tests, stop using pytest, and start airflow in breeze, scheduling the example dags as actual dags. This has the upside of being the closest to prod, but it's also a more resource-intensive solution, and gathering the success/failure status and the logs will require some piping.
3. Do nothing, and forbid the use of deferrable in system tests. This is not great for obvious reasons.
It's a bit hard to say which option is "the best" before digging into it, but my preference would go to 1. from what I tell so far.
What do you community think about those options ? Did I forget a potential solution ?
Other providers who also have system tests and deferrable operators, what is your approach ?
Re: Deferrable operators in system tests, how?
Posted by "Ferruzzi, Dennis" <fe...@amazon.com.INVALID>.
FWIW, I'm liking the Breeze idea. I hadn't thought of it at first, but it seems to make the most sense.
- ferruzzi
________________________________
From: Beck, Vincent <vi...@amazon.com.INVALID>
Sent: Wednesday, June 7, 2023 7:21 AM
To: dev@airflow.apache.org
Subject: RE: [EXTERNAL]Deferrable operators in system tests, how?
CAUTION: This email originated from outside of the organization. Do not click links or open attachments unless you can confirm the sender and know the content is safe.
Same here, I like the breeze option
On 2023-06-07, 3:32 AM, "Pankaj Koti" <pankaj.koti@astronomer.io.INVA <ma...@astronomer.io.INVA>LID> wrote:
CAUTION: This email originated from outside of the organization. Do not click links or open attachments unless you can confirm the sender and know the content is safe.
+1 to both ideas.
I have a stronger +1 to include the flag with breeze since triggerer
is an Airflow component and breeze commands having support for
it would be nice.
Regards,
Pankaj Koti
*Senior Software Engineer, *OSS Engineering Team.
Location: Pune, India
Timezone: Indian Standard Time (IST)
Email: pankaj.koti@astronomer.io <ma...@astronomer.io>
Mobile: +91 9730079985
On Wed, Jun 7, 2023 at 5:07 AM Daniel Standish
<daniel.standish@astronomer.io.inva <ma...@astronomer.io.inva>lid> wrote:
> Don't think fixture would break that. It would just be test code not in the
> dag. It would just ensure that the triggerer is running before the tests
> that use the triggerer need it. But doing it in breeze makes more sense for
> sure. Although I suppose a combination approach could be considered EG, if
> not in the breeze environment, and no triggerer running in any case, then
> spin one up. But that might be overkill and over complicating it.
>
> On Tue, Jun 6, 2023, 4:24 PM Vandon, Raphael <vandonr@amazon.com.inva <ma...@amazon.com.inva>lid>
> wrote:
>
> > Thanks for pointing out solutions I hadn't considered.
> >
> > > we could simply have a pytest fixture that will do the job (based on
> > what Daniel proposed).
> > > It will not support running tests via Airflow UI or regular python
> > execution of the test files,
> > > but maybe it's a good idea that does not rely on docker/breeze.
> >
> > I like the idea that example dags are actual dags, that you can drop in
> > your dags folder and run (provided you have the right config).
> > As you mention, adding a pytest fixture to run the triggerer will
> probably
> > break that ?
> >
> > It'd also break the setup Phani described __, if I understood correctly ?
> >
> > >> Our approach is to spin up Airflow , install the providers that
> > currently have deferrable functionality,
> > >> and run system tests on those example DAGs
> >
> > So I'd be in favor of adding a flag to breeze to start a triggerer. It's
> > the most flexible option, lets people run tests in full airflow, in
> breeze,
> > or outside of breeze by running the triggerer themselves.
> >
> > It also sounds like it wouldn't be too hard to implement, even though I'm
> > not familiar with breeze code. I can try to give it a shot later this
> week
> > or the next.
> >
> >
> > ---------------------------------------------------------------------
> > To unsubscribe, e-mail: dev-unsubscribe@airflow.apache.org <ma...@airflow.apache.org>
> > For additional commands, e-mail: dev-help@airflow.apache.org <ma...@airflow.apache.org>
> >
>
Т���������������������������������������������������������������������ХF�V�7V'67&�&R�R���âFWb�V�7V'67&�&T�&f��r�6�R��&pФf�"FF�F����6����G2�R���âFWbֆV��&f��r�6�R��&p
Re: Deferrable operators in system tests, how?
Posted by "Beck, Vincent" <vi...@amazon.com.INVALID>.
Same here, I like the breeze option
On 2023-06-07, 3:32 AM, "Pankaj Koti" <pankaj.koti@astronomer.io.INVA <ma...@astronomer.io.INVA>LID> wrote:
CAUTION: This email originated from outside of the organization. Do not click links or open attachments unless you can confirm the sender and know the content is safe.
+1 to both ideas.
I have a stronger +1 to include the flag with breeze since triggerer
is an Airflow component and breeze commands having support for
it would be nice.
Regards,
Pankaj Koti
*Senior Software Engineer, *OSS Engineering Team.
Location: Pune, India
Timezone: Indian Standard Time (IST)
Email: pankaj.koti@astronomer.io <ma...@astronomer.io>
Mobile: +91 9730079985
On Wed, Jun 7, 2023 at 5:07 AM Daniel Standish
<daniel.standish@astronomer.io.inva <ma...@astronomer.io.inva>lid> wrote:
> Don't think fixture would break that. It would just be test code not in the
> dag. It would just ensure that the triggerer is running before the tests
> that use the triggerer need it. But doing it in breeze makes more sense for
> sure. Although I suppose a combination approach could be considered EG, if
> not in the breeze environment, and no triggerer running in any case, then
> spin one up. But that might be overkill and over complicating it.
>
> On Tue, Jun 6, 2023, 4:24 PM Vandon, Raphael <vandonr@amazon.com.inva <ma...@amazon.com.inva>lid>
> wrote:
>
> > Thanks for pointing out solutions I hadn't considered.
> >
> > > we could simply have a pytest fixture that will do the job (based on
> > what Daniel proposed).
> > > It will not support running tests via Airflow UI or regular python
> > execution of the test files,
> > > but maybe it's a good idea that does not rely on docker/breeze.
> >
> > I like the idea that example dags are actual dags, that you can drop in
> > your dags folder and run (provided you have the right config).
> > As you mention, adding a pytest fixture to run the triggerer will
> probably
> > break that ?
> >
> > It'd also break the setup Phani described __, if I understood correctly ?
> >
> > >> Our approach is to spin up Airflow , install the providers that
> > currently have deferrable functionality,
> > >> and run system tests on those example DAGs
> >
> > So I'd be in favor of adding a flag to breeze to start a triggerer. It's
> > the most flexible option, lets people run tests in full airflow, in
> breeze,
> > or outside of breeze by running the triggerer themselves.
> >
> > It also sounds like it wouldn't be too hard to implement, even though I'm
> > not familiar with breeze code. I can try to give it a shot later this
> week
> > or the next.
> >
> >
> > ---------------------------------------------------------------------
> > To unsubscribe, e-mail: dev-unsubscribe@airflow.apache.org <ma...@airflow.apache.org>
> > For additional commands, e-mail: dev-help@airflow.apache.org <ma...@airflow.apache.org>
> >
>
Re: Deferrable operators in system tests, how?
Posted by Pankaj Koti <pa...@astronomer.io.INVALID>.
+1 to both ideas.
I have a stronger +1 to include the flag with breeze since triggerer
is an Airflow component and breeze commands having support for
it would be nice.
Regards,
Pankaj Koti
*Senior Software Engineer, *OSS Engineering Team.
Location: Pune, India
Timezone: Indian Standard Time (IST)
Email: pankaj.koti@astronomer.io
Mobile: +91 9730079985
On Wed, Jun 7, 2023 at 5:07 AM Daniel Standish
<da...@astronomer.io.invalid> wrote:
> Don't think fixture would break that. It would just be test code not in the
> dag. It would just ensure that the triggerer is running before the tests
> that use the triggerer need it. But doing it in breeze makes more sense for
> sure. Although I suppose a combination approach could be considered EG, if
> not in the breeze environment, and no triggerer running in any case, then
> spin one up. But that might be overkill and over complicating it.
>
> On Tue, Jun 6, 2023, 4:24 PM Vandon, Raphael <va...@amazon.com.invalid>
> wrote:
>
> > Thanks for pointing out solutions I hadn't considered.
> >
> > > we could simply have a pytest fixture that will do the job (based on
> > what Daniel proposed).
> > > It will not support running tests via Airflow UI or regular python
> > execution of the test files,
> > > but maybe it's a good idea that does not rely on docker/breeze.
> >
> > I like the idea that example dags are actual dags, that you can drop in
> > your dags folder and run (provided you have the right config).
> > As you mention, adding a pytest fixture to run the triggerer will
> probably
> > break that ?
> >
> > It'd also break the setup Phani described __, if I understood correctly ?
> >
> > >> Our approach is to spin up Airflow , install the providers that
> > currently have deferrable functionality,
> > >> and run system tests on those example DAGs
> >
> > So I'd be in favor of adding a flag to breeze to start a triggerer. It's
> > the most flexible option, lets people run tests in full airflow, in
> breeze,
> > or outside of breeze by running the triggerer themselves.
> >
> > It also sounds like it wouldn't be too hard to implement, even though I'm
> > not familiar with breeze code. I can try to give it a shot later this
> week
> > or the next.
> >
> >
> > ---------------------------------------------------------------------
> > To unsubscribe, e-mail: dev-unsubscribe@airflow.apache.org
> > For additional commands, e-mail: dev-help@airflow.apache.org
> >
>
Re: Deferrable operators in system tests, how?
Posted by Daniel Standish <da...@astronomer.io.INVALID>.
Don't think fixture would break that. It would just be test code not in the
dag. It would just ensure that the triggerer is running before the tests
that use the triggerer need it. But doing it in breeze makes more sense for
sure. Although I suppose a combination approach could be considered EG, if
not in the breeze environment, and no triggerer running in any case, then
spin one up. But that might be overkill and over complicating it.
On Tue, Jun 6, 2023, 4:24 PM Vandon, Raphael <va...@amazon.com.invalid>
wrote:
> Thanks for pointing out solutions I hadn't considered.
>
> > we could simply have a pytest fixture that will do the job (based on
> what Daniel proposed).
> > It will not support running tests via Airflow UI or regular python
> execution of the test files,
> > but maybe it's a good idea that does not rely on docker/breeze.
>
> I like the idea that example dags are actual dags, that you can drop in
> your dags folder and run (provided you have the right config).
> As you mention, adding a pytest fixture to run the triggerer will probably
> break that ?
>
> It'd also break the setup Phani described __, if I understood correctly ?
>
> >> Our approach is to spin up Airflow , install the providers that
> currently have deferrable functionality,
> >> and run system tests on those example DAGs
>
> So I'd be in favor of adding a flag to breeze to start a triggerer. It's
> the most flexible option, lets people run tests in full airflow, in breeze,
> or outside of breeze by running the triggerer themselves.
>
> It also sounds like it wouldn't be too hard to implement, even though I'm
> not familiar with breeze code. I can try to give it a shot later this week
> or the next.
>
>
> ---------------------------------------------------------------------
> To unsubscribe, e-mail: dev-unsubscribe@airflow.apache.org
> For additional commands, e-mail: dev-help@airflow.apache.org
>
Re: Deferrable operators in system tests, how?
Posted by "Vandon, Raphael" <va...@amazon.com.INVALID>.
Thanks for pointing out solutions I hadn't considered.
> we could simply have a pytest fixture that will do the job (based on what Daniel proposed).
> It will not support running tests via Airflow UI or regular python execution of the test files,
> but maybe it's a good idea that does not rely on docker/breeze.
I like the idea that example dags are actual dags, that you can drop in your dags folder and run (provided you have the right config).
As you mention, adding a pytest fixture to run the triggerer will probably break that ?
It'd also break the setup Phani described __, if I understood correctly ?
>> Our approach is to spin up Airflow , install the providers that currently have deferrable functionality,
>> and run system tests on those example DAGs
So I'd be in favor of adding a flag to breeze to start a triggerer. It's the most flexible option, lets people run tests in full airflow, in breeze, or outside of breeze by running the triggerer themselves.
It also sounds like it wouldn't be too hard to implement, even though I'm not familiar with breeze code. I can try to give it a shot later this week or the next.
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@airflow.apache.org
For additional commands, e-mail: dev-help@airflow.apache.org
Re: Deferrable operators in system tests, how?
Posted by Jarek Potiuk <ja...@potiuk.com>.
There are two options I think:
1) rely on Breeze as execution environment for System Tests and add a flag
to start triggerer with system tests
2) add a pytest fixture that will do it automatically when system tests are
run via pytests (that could be done via tests/system/conftest.py)
I believe most of the system tests out there are run with Breeze. Breeze
gives a stable, reproducible execution environment for various tests, it's
been designed from the ground up to provide a "reproducible test
environment" - that was the main "function" of Breeze from the start. So we
could add a change to run triggerer when starting tests with breeze:
`--run-triggerer=True`. Plus we could describe, that you can also run the
deferrable system tests in regular "local" venv as long as you run
triggerer manually to support it.
But it does not have to be like it. It's also possible that we
could simply have a pytest fixture that will do the job (based on what
Daniel proposed).This way, whenever we run the system tests as "pytest"
test, we should get triggerrer up-and-running and torn down after. It will
not support running tests via Airflow UI or regular python execution of the
test files, but maybe it's a good idea that does not rely on docker/breeze.
J.
On Sat, Jun 3, 2023 at 8:45 AM Daniel Standish
<da...@astronomer.io.invalid> wrote:
> I just took a look and it turns out that DebugExecutor works fine with
> triggerer you just need to have one running.
>
> You could run one in a subprocess. I experimented with refactoring the
> subprocess hook for this purpose (so you can start the subprocess
> asynchronously) and then ran this dag with debug executor and it worked.
>
> from __future__ import annotations
>
> import os
>
> os.environ["AIRFLOW__CORE__EXECUTOR"] = "DebugExecutor"
>
> from datetime import timedelta
>
> import pendulum
>
> from airflow.models import DAG
> from airflow.sensors.time_delta import TimeDeltaSensorAsync
>
> with DAG(
> dag_id="example_sensors",
> schedule="@once",
> start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
> ) as dag:
> t0a = TimeDeltaSensorAsync(task_id="wait_some_seconds_async",
> delta=timedelta(seconds=2))
>
> from airflow.hooks.subprocess import SubprocessHook
>
> # let's get triggerer running in a subprocess
> hook = SubprocessHook()
> triggerer_process = hook.start_process(command=["airflow",
> "triggerer"], cwd="/tmp")
>
> # now let's run the dag
> dag.clear()
> dag.run()
>
> # now the dag has completed
>
> # kill triggerer
> hook.send_sigterm(triggerer_process)
> result = hook.process_output(subprocess=triggerer_process)
> assert result.exit_code == 0
>
> # now triggerer has exited and our system test is done
>
>
> So, something like this could be done.
>
Re: Deferrable operators in system tests, how?
Posted by Daniel Standish <da...@astronomer.io.INVALID>.
I just took a look and it turns out that DebugExecutor works fine with
triggerer you just need to have one running.
You could run one in a subprocess. I experimented with refactoring the
subprocess hook for this purpose (so you can start the subprocess
asynchronously) and then ran this dag with debug executor and it worked.
from __future__ import annotations
import os
os.environ["AIRFLOW__CORE__EXECUTOR"] = "DebugExecutor"
from datetime import timedelta
import pendulum
from airflow.models import DAG
from airflow.sensors.time_delta import TimeDeltaSensorAsync
with DAG(
dag_id="example_sensors",
schedule="@once",
start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
) as dag:
t0a = TimeDeltaSensorAsync(task_id="wait_some_seconds_async",
delta=timedelta(seconds=2))
from airflow.hooks.subprocess import SubprocessHook
# let's get triggerer running in a subprocess
hook = SubprocessHook()
triggerer_process = hook.start_process(command=["airflow",
"triggerer"], cwd="/tmp")
# now let's run the dag
dag.clear()
dag.run()
# now the dag has completed
# kill triggerer
hook.send_sigterm(triggerer_process)
result = hook.process_output(subprocess=triggerer_process)
assert result.exit_code == 0
# now triggerer has exited and our system test is done
So, something like this could be done.
Re: Deferrable operators in system tests, how?
Posted by Phani Kumar <ph...@astronomer.io.INVALID>.
My preference would be Option 2 as that is closest to prod. Our approach is
to spin up Airflow , install the providers that currently have deferrable
functionality, and run system tests on those example DAGs on a daily basis
. We are also running them on multiple executors (Celery, Kubernetes
deployments)
On Sat, Jun 3, 2023 at 4:30 AM Vandon, Raphael <va...@amazon.com.invalid>
wrote:
> Hello,
> As we implement deferrable capabilities for operators, we're going to want
> to update the sample dags/system tests.
> The problem we encounter at the moment is that system tests are run with
> pytest using the DebugExecutor, and it only runs the "worker part" of the
> operators, without handling the "deferred part" that is taking place in the
> Triggers. In system tests, a task ran in deferrable mode just hangs forever
> in the “deferred” state, never completing.
>
> I see 3 approaches to that problem:
> 1. Patch the DebugExecutor (or an other executor that'd become the
> "official" executor for system tests) to handle deferred tasks. This bears
> some complexity, but is probably doable with some ad-hoc code that won't be
> the same as the "production" code but could execute the right things in
> order (Trigger.run after the TaskDeferred exception, and then the
> "execute_complete" method with the event returned)
> 2. Change the way we run system tests, stop using pytest, and start
> airflow in breeze, scheduling the example dags as actual dags. This has the
> upside of being the closest to prod, but it's also a more
> resource-intensive solution, and gathering the success/failure status and
> the logs will require some piping.
> 3. Do nothing, and forbid the use of deferrable in system tests. This is
> not great for obvious reasons.
>
> It's a bit hard to say which option is "the best" before digging into it,
> but my preference would go to 1. from what I tell so far.
>
> What do you community think about those options ? Did I forget a potential
> solution ?
> Other providers who also have system tests and deferrable operators, what
> is your approach ?
>