You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by GitBox <gi...@apache.org> on 2022/10/18 17:50:24 UTC
[GitHub] [airflow] mobuchowski opened a new pull request, #27113: notification: add dag run state notification system
mobuchowski opened a new pull request, #27113:
URL: https://github.com/apache/airflow/pull/27113
Draft:
Add system that enables listeners to be notified about DagRun state changes.
TODO:
- start events
- improve tests
- backfill job should work as well
- performance test?
Signed-off-by: Maciej Obuchowski <ob...@gmail.com>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a diff in pull request #27113: notification: add dag run state notification system
Posted by GitBox <gi...@apache.org>.
ashb commented on code in PR #27113:
URL: https://github.com/apache/airflow/pull/27113#discussion_r1016360230
##########
airflow/jobs/base_job.py:
##########
@@ -110,6 +111,7 @@ def __init__(self, executor=None, heartrate=None, *args, **kwargs):
self.heartrate = heartrate
self.unixname = getuser()
self.max_tis_per_query: int = conf.getint('scheduler', 'max_tis_per_query')
+ get_listener_manager().hook.on_starting()
Review Comment:
> can't implement the same hookspec in one file - especially with same-name methods
Plugins can be classes, not just modules.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] dstandish commented on a diff in pull request #27113: notification: add dag run state notification system
Posted by GitBox <gi...@apache.org>.
dstandish commented on code in PR #27113:
URL: https://github.com/apache/airflow/pull/27113#discussion_r1004758527
##########
airflow/jobs/scheduler_job.py:
##########
@@ -745,6 +757,9 @@ def _execute(self) -> None:
self._run_scheduler_loop()
+ if self.enabled_dagrun_listener:
Review Comment:
yeah like ... what do you think about that @mobuchowski ? if it is feasible to add a null notifier, i.e. some kind of base class that doesn't do anything, then maybe we could get rid of all of these if checks, and maybe that would be better. what do you think of the thought?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] mobuchowski commented on a diff in pull request #27113: notification: add dag run state notification system
Posted by GitBox <gi...@apache.org>.
mobuchowski commented on code in PR #27113:
URL: https://github.com/apache/airflow/pull/27113#discussion_r1015781659
##########
airflow/jobs/base_job.py:
##########
@@ -110,6 +111,7 @@ def __init__(self, executor=None, heartrate=None, *args, **kwargs):
self.heartrate = heartrate
self.unixname = getuser()
self.max_tis_per_query: int = conf.getint('scheduler', 'max_tis_per_query')
+ get_listener_manager().hook.on_starting()
Review Comment:
The question here is really if we need this - having conflicting names for different scopes would be more painful for the plugin author.
I can imagine some pain points:
- need to remember to mark method with particular hookspec
- can't implement the same hookspec in one file - especially with same-name methods, so you're forced to have multi-file plugin.
- Airflow plugin loading API gets more confusing, since you need to know a priori which hookspec the plugin implements
Seems like the easier solution would be to "namespace" them just by name. Like; `airflow_scheduler_on_starting` and `airflow_backfill_on_starting` if those were to be two different methods.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] mobuchowski commented on a diff in pull request #27113: notification: add dag run state notification system
Posted by GitBox <gi...@apache.org>.
mobuchowski commented on code in PR #27113:
URL: https://github.com/apache/airflow/pull/27113#discussion_r1015305002
##########
airflow/listeners/listener.py:
##########
@@ -47,6 +50,15 @@ def __init__(self):
def has_listeners(self) -> bool:
return len(self.pm.get_plugins()) > 0
+ @property
+ def has_scheduler_listeners(self) -> bool:
+ for plugin in self.pm.get_plugins():
+ if inspect.ismodule(plugin):
Review Comment:
Removed that in favor of `lifecycle methods`.
##########
airflow/listeners/listener.py:
##########
@@ -33,6 +34,8 @@
_listener_manager = None
+_scheduler_hooks = ["on_dag_run_success", "on_dag_run_failure"]
Review Comment:
Removed those
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] mobuchowski commented on a diff in pull request #27113: notification: add dag run state notification system
Posted by GitBox <gi...@apache.org>.
mobuchowski commented on code in PR #27113:
URL: https://github.com/apache/airflow/pull/27113#discussion_r1029553731
##########
airflow/jobs/scheduler_job.py:
##########
@@ -798,7 +798,7 @@ def _update_dag_run_state_for_paused_dags(self, session: Session = NEW_SESSION)
)
for dag_run in dag_runs:
dag_run.dag = dag
- _, callback_to_run = dag_run.update_state(execute_callbacks=False)
+ dag_run, callback_to_run = dag_run.update_state(execute_callbacks=False)
Review Comment:
I've used it before, removed in the latest PR version and this leftover stayed. Fixed.
BTW, I think it's not detected by static checks because it's used in the same block before anyway.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] mobuchowski commented on pull request #27113: notification: add dag run state notification system
Posted by GitBox <gi...@apache.org>.
mobuchowski commented on PR #27113:
URL: https://github.com/apache/airflow/pull/27113#issuecomment-1306006380
@ashb implementations are supposed to import `hookimpl` marker - not `hookspec`, which is used to define interface.
`hookimpl` did not move: https://github.com/apache/airflow/blob/main/airflow/listeners/__init__.py#L22
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] uranusjr commented on pull request #27113: notification: add dag run state notification system
Posted by GitBox <gi...@apache.org>.
uranusjr commented on PR #27113:
URL: https://github.com/apache/airflow/pull/27113#issuecomment-1305470615
This is implemented significantly differently from task instance state listeners. Is there a reason for this? Should we refactor how the task instance listeners are implemented to match this?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a diff in pull request #27113: notification: add dag run state notification system
Posted by GitBox <gi...@apache.org>.
ashb commented on code in PR #27113:
URL: https://github.com/apache/airflow/pull/27113#discussion_r1015570248
##########
airflow/jobs/base_job.py:
##########
@@ -110,6 +111,7 @@ def __init__(self, executor=None, heartrate=None, *args, **kwargs):
self.heartrate = heartrate
self.unixname = getuser()
self.max_tis_per_query: int = conf.getint('scheduler', 'max_tis_per_query')
+ get_listener_manager().hook.on_starting()
Review Comment:
I think this is my only question: should we combine all the hookspecs in to a single hook, or would this be "clearer"/more maintainable in future as
```suggestion
get_listener_manager().lifecycle_hook.on_starting()
```
Possible reason for wanting this: it allows us to have conflicting names for different scopes. I suspect this could be "we'll do it when we need it" since it is "internal"
##########
airflow/listeners/spec/lifecycle.py:
##########
@@ -0,0 +1,38 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from pluggy import HookspecMarker
+
+hookspec = HookspecMarker("airflow")
Review Comment:
I think each one marker should have a different name, otherwise they are effectively the same spec aren't they? (I'm not 100% up to speed on Pluggy, so this may _not_ be the case)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] mobuchowski commented on a diff in pull request #27113: notification: add dag run state notification system
Posted by GitBox <gi...@apache.org>.
mobuchowski commented on code in PR #27113:
URL: https://github.com/apache/airflow/pull/27113#discussion_r1004635203
##########
airflow/models/dagrun.py:
##########
@@ -580,11 +591,11 @@ def recalculate(self) -> _UnfinishedStates:
self.set_state(DagRunState.FAILED)
if execute_callbacks:
dag.handle_callback(self, success=False, reason='task_failure', session=session)
- elif dag.has_on_failure_callback:
+ elif dag.has_on_failure_callback or notification:
from airflow.models.dag import DagModel
dag_model = DagModel.get_dagmodel(dag.dag_id, session)
- callback = DagCallbackRequest(
+ cb = DagCallbackRequest(
Review Comment:
Changed that, also the notification now does not rely on `DagCallbackRequest` - since we don't always have it, like in `BackfillJob`.
##########
airflow/models/dagrun.py:
##########
@@ -593,17 +604,22 @@ def recalculate(self) -> _UnfinishedStates:
msg='task_failure',
)
+ if notification:
+ notification(dag_run=self, callback_request=cb)
+ if dag.has_on_success_callback:
Review Comment:
Changed that, also the notification now does not rely on DagCallbackRequest - since we don't always have it, like in BackfillJob.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a diff in pull request #27113: notification: add dag run state notification system
Posted by GitBox <gi...@apache.org>.
ashb commented on code in PR #27113:
URL: https://github.com/apache/airflow/pull/27113#discussion_r1029543297
##########
airflow/jobs/scheduler_job.py:
##########
@@ -798,7 +798,7 @@ def _update_dag_run_state_for_paused_dags(self, session: Session = NEW_SESSION)
)
for dag_run in dag_runs:
dag_run.dag = dag
- _, callback_to_run = dag_run.update_state(execute_callbacks=False)
+ dag_run, callback_to_run = dag_run.update_state(execute_callbacks=False)
Review Comment:
Oh I guess dag_run isn't used after this call
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] dstandish commented on a diff in pull request #27113: notification: add dag run state notification system
Posted by GitBox <gi...@apache.org>.
dstandish commented on code in PR #27113:
URL: https://github.com/apache/airflow/pull/27113#discussion_r998550969
##########
airflow/models/dagrun.py:
##########
@@ -593,17 +604,22 @@ def recalculate(self) -> _UnfinishedStates:
msg='task_failure',
)
+ if notification:
+ notification(dag_run=self, callback_request=cb)
+ if dag.has_on_success_callback:
Review Comment:
maybe better would be to not combine the notification into this if / else block in the way you've done ... the nested branching could be more complicated than necessary? e.g. case in point here this is supposed to be `has_on_failure_callback` i think. i'm not 100% sure there's a better way, but there _could_ be and it might be worth giving it a think
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] mobuchowski commented on pull request #27113: notification: add dag run state notification system
Posted by GitBox <gi...@apache.org>.
mobuchowski commented on PR #27113:
URL: https://github.com/apache/airflow/pull/27113#issuecomment-1307503678
@ashb hopefully we'll reach consensus about that till next release 🙂
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] mobuchowski commented on a diff in pull request #27113: notification: add dag run state notification system
Posted by GitBox <gi...@apache.org>.
mobuchowski commented on code in PR #27113:
URL: https://github.com/apache/airflow/pull/27113#discussion_r1015305683
##########
airflow/jobs/scheduler_job.py:
##########
@@ -1568,3 +1590,21 @@ def _cleanup_stale_dags(self, session: Session = NEW_SESSION) -> None:
dag.is_active = False
SerializedDagModel.remove_dag(dag_id=dag.dag_id, session=session)
session.flush()
+
+ def notify_dagrun_state_changed(self, dag_run: DagRun, msg: str = ""):
+ if not self.enabled_dagrun_listener or not get_listener_manager().has_scheduler_listeners:
+ return
+
+ if dag_run.state == DagRunState.RUNNING:
+ self._notification_threadpool.submit( # type: ignore[union-attr]
+ get_listener_manager().hook.on_dag_run_start, dag_run=dag_run, msg=msg
+ )
+ elif dag_run.state == DagRunState.SUCCESS:
+ self._notification_threadpool.submit( # type: ignore[union-attr]
+ get_listener_manager().hook.on_dag_run_success, dag_run=dag_run, msg=msg
+ )
+ elif dag_run.state == DagRunState.FAILED:
+ self._notification_threadpool.submit( # type: ignore[union-attr]
+ get_listener_manager().hook.on_dag_run_failure, dag_run=dag_run, msg=msg
+ )
Review Comment:
Removed threadpool.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] mobuchowski commented on a diff in pull request #27113: notification: add dag run state notification system
Posted by GitBox <gi...@apache.org>.
mobuchowski commented on code in PR #27113:
URL: https://github.com/apache/airflow/pull/27113#discussion_r1029408755
##########
airflow/jobs/backfill_job.py:
##########
@@ -946,3 +952,12 @@ def query(result, items):
self.log.info("Reset the following %s TaskInstances:\n\t%s", len(reset_tis), task_instance_str)
return len(reset_tis)
+
+ def notify_dagrun_state_changed(self, dag_run: DagRun, msg: str = ""):
+ if dag_run.state == DagRunState.RUNNING:
+ get_listener_manager().hook.on_dag_run_running(dag_run=dag_run, msg=msg)
+ elif dag_run.state == DagRunState.SUCCESS:
+ get_listener_manager().hook.on_dag_run_success(dag_run=dag_run, msg=msg)
+ elif dag_run.state == DagRunState.FAILED:
+ get_listener_manager().hook.on_dag_run_failed(dag_run=dag_run, msg=msg)
+ # deliberately not notifying on QUEUED
Review Comment:
`QUEUED` requires running in more processes, like webserver, or `TriggerDagRunOperator`. We can add coverage later, for now I think promising that we'll notify about queued will be untrue.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] mobuchowski commented on a diff in pull request #27113: notification: add dag run state notification system
Posted by GitBox <gi...@apache.org>.
mobuchowski commented on code in PR #27113:
URL: https://github.com/apache/airflow/pull/27113#discussion_r1029274401
##########
airflow/jobs/base_job.py:
##########
@@ -252,6 +255,7 @@ def run(self):
self.state = State.FAILED
raise
finally:
+ get_listener_manager().hook.before_stopping()
Review Comment:
Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb merged pull request #27113: notification: add dag run state notification system
Posted by GitBox <gi...@apache.org>.
ashb merged PR #27113:
URL: https://github.com/apache/airflow/pull/27113
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] mobuchowski commented on a diff in pull request #27113: notification: add dag run state notification system
Posted by GitBox <gi...@apache.org>.
mobuchowski commented on code in PR #27113:
URL: https://github.com/apache/airflow/pull/27113#discussion_r1016433904
##########
airflow/jobs/base_job.py:
##########
@@ -110,6 +111,7 @@ def __init__(self, executor=None, heartrate=None, *args, **kwargs):
self.heartrate = heartrate
self.unixname = getuser()
self.max_tis_per_query: int = conf.getint('scheduler', 'max_tis_per_query')
+ get_listener_manager().hook.on_starting()
Review Comment:
Yes, but still, you can't easily share state - your `lifecycle` plugin class is not the same as `dag_run` plugin class
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a diff in pull request #27113: notification: add dag run state notification system
Posted by GitBox <gi...@apache.org>.
ashb commented on code in PR #27113:
URL: https://github.com/apache/airflow/pull/27113#discussion_r1007040063
##########
airflow/config_templates/config.yml:
##########
@@ -2169,6 +2169,13 @@
type: string
example: ~
default: "15"
+ - name: enable_dagrun_listener_notifications
+ description: |
+ Enable emitting dagrun listener notifications in scheduler.
+ version_added: 2.5.0
+ type: boolean
+ example: ~
+ default: "False"
Review Comment:
Not sure we need a config setting for this really (Thinking: we already have so many config options, adding another one should be avoided unless we really need it)
##########
airflow/listeners/listener.py:
##########
@@ -47,6 +50,15 @@ def __init__(self):
def has_listeners(self) -> bool:
return len(self.pm.get_plugins()) > 0
+ @property
+ def has_scheduler_listeners(self) -> bool:
+ for plugin in self.pm.get_plugins():
+ if inspect.ismodule(plugin):
Review Comment:
Can you explain what's going on here? Why do we need to check if its a module? Pluggy supports adding classes to I thought? (But mostly: why do we care?
##########
airflow/jobs/backfill_job.py:
##########
@@ -18,6 +18,7 @@
from __future__ import annotations
import time
+from concurrent.futures import Executor, ThreadPoolExecutor
Review Comment:
```suggestion
from concurrent.futures import Executor as FutureExecutor, ThreadPoolExecutor
```
(and matching else where: to avoid confusing with Airflow's own Executor class
##########
airflow/listeners/listener.py:
##########
@@ -47,6 +50,15 @@ def __init__(self):
def has_listeners(self) -> bool:
return len(self.pm.get_plugins()) > 0
+ @property
Review Comment:
```suggestion
@cached_property
```
Once we've computed this once per process it can't change again.
##########
airflow/listeners/listener.py:
##########
@@ -33,6 +34,8 @@
_listener_manager = None
+_scheduler_hooks = ["on_dag_run_success", "on_dag_run_failure"]
Review Comment:
Given these hooks are also called from backfill:
```suggestion
_dagrun_hooks = ["on_dag_run_success", "on_dag_run_failure"]
```
##########
airflow/jobs/scheduler_job.py:
##########
@@ -1568,3 +1590,21 @@ def _cleanup_stale_dags(self, session: Session = NEW_SESSION) -> None:
dag.is_active = False
SerializedDagModel.remove_dag(dag_id=dag.dag_id, session=session)
session.flush()
+
+ def notify_dagrun_state_changed(self, dag_run: DagRun, msg: str = ""):
+ if not self.enabled_dagrun_listener or not get_listener_manager().has_scheduler_listeners:
+ return
+
+ if dag_run.state == DagRunState.RUNNING:
+ self._notification_threadpool.submit( # type: ignore[union-attr]
+ get_listener_manager().hook.on_dag_run_start, dag_run=dag_run, msg=msg
+ )
+ elif dag_run.state == DagRunState.SUCCESS:
+ self._notification_threadpool.submit( # type: ignore[union-attr]
+ get_listener_manager().hook.on_dag_run_success, dag_run=dag_run, msg=msg
+ )
+ elif dag_run.state == DagRunState.FAILED:
+ self._notification_threadpool.submit( # type: ignore[union-attr]
+ get_listener_manager().hook.on_dag_run_failure, dag_run=dag_run, msg=msg
+ )
Review Comment:
I'm not sure the threadpool belongs in here.
If we make that the _plugin's_ responsibility then
a) I don't think forcing a threadpool (of a fixed size, but that could be config driven) on users of this hook is required;
b) We probably don't need `has_scheduler_listeners` anymore;
b) This PR becomes a lot smaller.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] mobuchowski commented on a diff in pull request #27113: notification: add dag run state notification system
Posted by GitBox <gi...@apache.org>.
mobuchowski commented on code in PR #27113:
URL: https://github.com/apache/airflow/pull/27113#discussion_r1007072665
##########
airflow/config_templates/config.yml:
##########
@@ -2169,6 +2169,13 @@
type: string
example: ~
default: "15"
+ - name: enable_dagrun_listener_notifications
+ description: |
+ Enable emitting dagrun listener notifications in scheduler.
+ version_added: 2.5.0
+ type: boolean
+ example: ~
+ default: "False"
Review Comment:
We probably can work with `if there's scheduler plugin registered, just do it`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] mobuchowski commented on pull request #27113: notification: add dag run state notification system
Posted by GitBox <gi...@apache.org>.
mobuchowski commented on PR #27113:
URL: https://github.com/apache/airflow/pull/27113#issuecomment-1307350276
@ashb okay. Let's determine first if we need `namespaced` or `scoped` plugin methods. I'm not sure about it - but you and other committers know much better what Airflow needs. Plugins don't have to implement every method - in fact, they can implement whatever they want, one, all or many, so this isn't a problem. Also, I don't see need to physically separate `dag run listener` from `task instance listener`. In my plugin, those would be in the same python module that just provides plumbing to pass the data further, but maybe there is some need for that.
If we determine the need we might decide on an implementation, where I would prefer to have logical separation - maybe similar to as we have now - common methods are close together, and can be possibly prefixed with some `namespace` to separate them logically.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] mobuchowski commented on a diff in pull request #27113: notification: add dag run state notification system
Posted by GitBox <gi...@apache.org>.
mobuchowski commented on code in PR #27113:
URL: https://github.com/apache/airflow/pull/27113#discussion_r1015304699
##########
airflow/jobs/backfill_job.py:
##########
@@ -18,6 +18,7 @@
from __future__ import annotations
import time
+from concurrent.futures import Executor, ThreadPoolExecutor
Review Comment:
Removed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] dstandish commented on a diff in pull request #27113: notification: add dag run state notification system
Posted by GitBox <gi...@apache.org>.
dstandish commented on code in PR #27113:
URL: https://github.com/apache/airflow/pull/27113#discussion_r1004671681
##########
airflow/jobs/scheduler_job.py:
##########
@@ -1298,7 +1324,11 @@ def _schedule_dag_run(
self._verify_integrity_if_dag_changed(dag_run=dag_run, session=session)
# TODO[HA]: Rename update_state -> schedule_dag_run, ?? something else?
- schedulable_tis, callback_to_run = dag_run.update_state(session=session, execute_callbacks=False)
+ schedulable_tis, callback_to_run = dag_run.update_state(
+ session=session,
+ execute_callbacks=False,
+ notification=self.notify_dagrun_state_changed if self.enabled_dagrun_listener else None,
Review Comment:
it seems cleaner to have a "null notification" that is the default than to do this if / else
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a diff in pull request #27113: notification: add dag run state notification system
Posted by GitBox <gi...@apache.org>.
ashb commented on code in PR #27113:
URL: https://github.com/apache/airflow/pull/27113#discussion_r1028105968
##########
airflow/jobs/base_job.py:
##########
@@ -149,6 +151,7 @@ def is_alive(self, grace_multiplier=2.1):
@provide_session
def kill(self, session=None):
"""Handles on_kill callback and updates state in database."""
+ get_listener_manager().hook.before_stopping()
Review Comment:
We shouldn't need this here and in the `finally` block below do we?
##########
airflow/listeners/spec/lifecycle.py:
##########
@@ -0,0 +1,38 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from pluggy import HookspecMarker
+
+hookspec = HookspecMarker("airflow")
+
+
+@hookspec
+def on_starting():
+ """
+ Called when Airflow component (scheduler, worker) starts.
+ It's guaranteed this will be called before any other plugin method.
+ """
+
+
+@hookspec
+def before_stopping():
+ """
+ Called before Airflow component (scheduler, worker) stops.
Review Comment:
```suggestion
Called before Airflow component (scheduler, worker) stops.
```
##########
airflow/jobs/base_job.py:
##########
@@ -252,6 +255,7 @@ def run(self):
self.state = State.FAILED
raise
finally:
+ get_listener_manager().hook.before_stopping()
Review Comment:
```suggestion
get_listener_manager().hook.before_stopping(self)
```
##########
airflow/jobs/base_job.py:
##########
@@ -110,6 +111,7 @@ def __init__(self, executor=None, heartrate=None, *args, **kwargs):
self.heartrate = heartrate
self.unixname = getuser()
self.max_tis_per_query: int = conf.getint("scheduler", "max_tis_per_query")
+ get_listener_manager().hook.on_starting()
Review Comment:
```suggestion
get_listener_manager().hook.on_starting(self)
```
(Because this is also called for LocalTaskJob, so being able to tell them apart is useful)
##########
airflow/jobs/backfill_job.py:
##########
@@ -946,3 +952,12 @@ def query(result, items):
self.log.info("Reset the following %s TaskInstances:\n\t%s", len(reset_tis), task_instance_str)
return len(reset_tis)
+
+ def notify_dagrun_state_changed(self, dag_run: DagRun, msg: str = ""):
+ if dag_run.state == DagRunState.RUNNING:
+ get_listener_manager().hook.on_dag_run_running(dag_run=dag_run, msg=msg)
+ elif dag_run.state == DagRunState.SUCCESS:
+ get_listener_manager().hook.on_dag_run_success(dag_run=dag_run, msg=msg)
+ elif dag_run.state == DagRunState.FAILED:
+ get_listener_manager().hook.on_dag_run_failed(dag_run=dag_run, msg=msg)
+ # deliberately not notifying on QUEUED
Review Comment:
This shouldn't be job specific right? Can't this live directly in DagRun itself?
##########
airflow/jobs/backfill_job.py:
##########
@@ -946,3 +952,12 @@ def query(result, items):
self.log.info("Reset the following %s TaskInstances:\n\t%s", len(reset_tis), task_instance_str)
return len(reset_tis)
+
+ def notify_dagrun_state_changed(self, dag_run: DagRun, msg: str = ""):
+ if dag_run.state == DagRunState.RUNNING:
+ get_listener_manager().hook.on_dag_run_running(dag_run=dag_run, msg=msg)
+ elif dag_run.state == DagRunState.SUCCESS:
+ get_listener_manager().hook.on_dag_run_success(dag_run=dag_run, msg=msg)
+ elif dag_run.state == DagRunState.FAILED:
+ get_listener_manager().hook.on_dag_run_failed(dag_run=dag_run, msg=msg)
+ # deliberately not notifying on QUEUED
Review Comment:
We should have queued on the spec, and it's up to the listener to ignore it if it wants.
Additionally I wonder if we should just have a single `dag_run_state_changed` type event instead of one for each and every state we might have in a dag run?
##########
airflow/listeners/spec/lifecycle.py:
##########
@@ -0,0 +1,38 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from pluggy import HookspecMarker
+
+hookspec = HookspecMarker("airflow")
+
+
+@hookspec
+def on_starting():
+ """
+ Called when Airflow component (scheduler, worker) starts.
Review Comment:
```suggestion
Called when Airflow component (scheduler, worker) starts.
```
##########
airflow/listeners/spec/lifecycle.py:
##########
@@ -0,0 +1,38 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from pluggy import HookspecMarker
+
+hookspec = HookspecMarker("airflow")
+
+
+@hookspec
+def on_starting():
+ """
+ Called when Airflow component (scheduler, worker) starts.
Review Comment:
This should probably recieve the `Job` object so the plugin/listener can tell what kind of thing it's being run inside of
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] mobuchowski commented on a diff in pull request #27113: notification: add dag run state notification system
Posted by GitBox <gi...@apache.org>.
mobuchowski commented on code in PR #27113:
URL: https://github.com/apache/airflow/pull/27113#discussion_r1029273880
##########
airflow/jobs/backfill_job.py:
##########
@@ -946,3 +952,12 @@ def query(result, items):
self.log.info("Reset the following %s TaskInstances:\n\t%s", len(reset_tis), task_instance_str)
return len(reset_tis)
+
+ def notify_dagrun_state_changed(self, dag_run: DagRun, msg: str = ""):
+ if dag_run.state == DagRunState.RUNNING:
+ get_listener_manager().hook.on_dag_run_running(dag_run=dag_run, msg=msg)
+ elif dag_run.state == DagRunState.SUCCESS:
+ get_listener_manager().hook.on_dag_run_success(dag_run=dag_run, msg=msg)
+ elif dag_run.state == DagRunState.FAILED:
+ get_listener_manager().hook.on_dag_run_failed(dag_run=dag_run, msg=msg)
+ # deliberately not notifying on QUEUED
Review Comment:
Fixed.
##########
airflow/listeners/spec/lifecycle.py:
##########
@@ -0,0 +1,38 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from pluggy import HookspecMarker
+
+hookspec = HookspecMarker("airflow")
+
+
+@hookspec
+def on_starting():
+ """
+ Called when Airflow component (scheduler, worker) starts.
+ It's guaranteed this will be called before any other plugin method.
+ """
+
+
+@hookspec
+def before_stopping():
+ """
+ Called before Airflow component (scheduler, worker) stops.
Review Comment:
Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] dstandish commented on a diff in pull request #27113: notification: add dag run state notification system
Posted by GitBox <gi...@apache.org>.
dstandish commented on code in PR #27113:
URL: https://github.com/apache/airflow/pull/27113#discussion_r998546502
##########
airflow/models/dagrun.py:
##########
@@ -580,11 +591,11 @@ def recalculate(self) -> _UnfinishedStates:
self.set_state(DagRunState.FAILED)
if execute_callbacks:
dag.handle_callback(self, success=False, reason='task_failure', session=session)
- elif dag.has_on_failure_callback:
+ elif dag.has_on_failure_callback or notification:
from airflow.models.dag import DagModel
dag_model = DagModel.get_dagmodel(dag.dag_id, session)
- callback = DagCallbackRequest(
+ cb = DagCallbackRequest(
Review Comment:
`cb` vs `callback` doesn't really signal what's the difference here. perhaps you can find a way to signal the intention. e.g. `returned_callback` vs `callback`
something like that
##########
airflow/models/dagrun.py:
##########
@@ -593,17 +604,22 @@ def recalculate(self) -> _UnfinishedStates:
msg='task_failure',
)
+ if notification:
+ notification(dag_run=self, callback_request=cb)
+ if dag.has_on_success_callback:
Review Comment:
maybe better would be to not combine the notification into this if / else block in the way you've done ... the nested branching could be more complicated than necessary? e.g. case in point here this is supposed to be `has_on_success_callback` i think. i'm not 100% sure there's a better way, but there _could_ be and it might be worth giving it a think
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] mobuchowski commented on a diff in pull request #27113: notification: add dag run state notification system
Posted by GitBox <gi...@apache.org>.
mobuchowski commented on code in PR #27113:
URL: https://github.com/apache/airflow/pull/27113#discussion_r1029274102
##########
airflow/listeners/spec/lifecycle.py:
##########
@@ -0,0 +1,38 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from pluggy import HookspecMarker
+
+hookspec = HookspecMarker("airflow")
+
+
+@hookspec
+def on_starting():
+ """
+ Called when Airflow component (scheduler, worker) starts.
Review Comment:
Fixed.
##########
airflow/jobs/base_job.py:
##########
@@ -110,6 +111,7 @@ def __init__(self, executor=None, heartrate=None, *args, **kwargs):
self.heartrate = heartrate
self.unixname = getuser()
self.max_tis_per_query: int = conf.getint("scheduler", "max_tis_per_query")
+ get_listener_manager().hook.on_starting()
Review Comment:
Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a diff in pull request #27113: notification: add dag run state notification system
Posted by GitBox <gi...@apache.org>.
ashb commented on code in PR #27113:
URL: https://github.com/apache/airflow/pull/27113#discussion_r1029542941
##########
airflow/jobs/scheduler_job.py:
##########
@@ -798,7 +798,7 @@ def _update_dag_run_state_for_paused_dags(self, session: Session = NEW_SESSION)
)
for dag_run in dag_runs:
dag_run.dag = dag
- _, callback_to_run = dag_run.update_state(execute_callbacks=False)
+ dag_run, callback_to_run = dag_run.update_state(execute_callbacks=False)
Review Comment:
I think
```suggestion
_, callback_to_run = dag_run.update_state(execute_callbacks=False)
```
given this function is defined as `tuple[list[TI], DagCallbackRequest | None]`
(I'm surprised this passes tests _at all_!!)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] mobuchowski commented on a diff in pull request #27113: notification: add dag run state notification system
Posted by GitBox <gi...@apache.org>.
mobuchowski commented on code in PR #27113:
URL: https://github.com/apache/airflow/pull/27113#discussion_r1004635203
##########
airflow/models/dagrun.py:
##########
@@ -580,11 +591,11 @@ def recalculate(self) -> _UnfinishedStates:
self.set_state(DagRunState.FAILED)
if execute_callbacks:
dag.handle_callback(self, success=False, reason='task_failure', session=session)
- elif dag.has_on_failure_callback:
+ elif dag.has_on_failure_callback or notification:
from airflow.models.dag import DagModel
dag_model = DagModel.get_dagmodel(dag.dag_id, session)
- callback = DagCallbackRequest(
+ cb = DagCallbackRequest(
Review Comment:
Reverted that part of the code.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on pull request #27113: notification: add dag run state notification system
Posted by GitBox <gi...@apache.org>.
ashb commented on PR #27113:
URL: https://github.com/apache/airflow/pull/27113#issuecomment-1306913653
If we have a single hookimpl we should probably also have a single hookspec; there doesn't seem any point in splitting one but not the other.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] mobuchowski commented on a diff in pull request #27113: notification: add dag run state notification system
Posted by GitBox <gi...@apache.org>.
mobuchowski commented on code in PR #27113:
URL: https://github.com/apache/airflow/pull/27113#discussion_r1005817540
##########
airflow/jobs/scheduler_job.py:
##########
@@ -745,6 +757,9 @@ def _execute(self) -> None:
self._run_scheduler_loop()
+ if self.enabled_dagrun_listener:
Review Comment:
We can just check for that in `notify_dagrun_state_changed`, and I changed it to just do that
##########
airflow/jobs/scheduler_job.py:
##########
@@ -1298,7 +1324,11 @@ def _schedule_dag_run(
self._verify_integrity_if_dag_changed(dag_run=dag_run, session=session)
# TODO[HA]: Rename update_state -> schedule_dag_run, ?? something else?
- schedulable_tis, callback_to_run = dag_run.update_state(session=session, execute_callbacks=False)
+ schedulable_tis, callback_to_run = dag_run.update_state(
+ session=session,
+ execute_callbacks=False,
+ notification=self.notify_dagrun_state_changed if self.enabled_dagrun_listener else None,
Review Comment:
Done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] mobuchowski commented on pull request #27113: notification: add dag run state notification system
Posted by GitBox <gi...@apache.org>.
mobuchowski commented on PR #27113:
URL: https://github.com/apache/airflow/pull/27113#issuecomment-1305596298
@uranusjr we have some problems with how `task instance state listener` works. One is that the Airflow object we get from `SqlAlchemy` event mechanism can be different from the one where it's executed.
For example https://github.com/apache/airflow/pull/27167 https://github.com/apache/airflow/pull/21157
Another problem is that in `task instance state listener` we're getting notified about state change in `after_flush` - which is the only `SqlAlchemy` event that has the database changes we need and occurs after actual execution. The problem is that `database transaction` haven't yet finished, so we need to run our code in a separate thread. The worker then can exit by `os._exit()` so we can't introduce any more sophisticated mechanism to manage that. One of the problems here is this Snowflake issue: https://github.com/OpenLineage/OpenLineage/issues/1226
One solution for threading issues in `task instance state listener` on Airflow side would be to split the listener to listen to two events: first would get actual changes and cache them locally, second would notify plugin about those changes after database transaction ended. I think it's more risky solution for scheduler, since in contrast to worker process, where there is only one task instance run at one time, scheduler could cache a lot more data. And still, this would not solve the first issue about differing objects.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org