You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "AutomationDev85 (via GitHub)" <gi...@apache.org> on 2024/10/21 12:04:26 UTC
[PR] [Edge] Set state of Edge Worker after worker timeout and add statistic export [airflow]
AutomationDev85 opened a new pull request, #43215:
URL: https://github.com/apache/airflow/pull/43215
Idea behind this PR is that the Executor checks during heartbeat if worker heartbeat timed out and sets the state of the worker to UNKOWN, to make sure that the state of the worker is up to date.
This PR also adds statsd export of the worker state. With this the current status of the worker is visible in the prometheus.
--
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
Re: [PR] [Edge] Add metrics export and reset state of Edge Worker after worker timeout [airflow]
Posted by "AutomationDev85 (via GitHub)" <gi...@apache.org>.
AutomationDev85 commented on code in PR #43215:
URL: https://github.com/apache/airflow/pull/43215#discussion_r1811946924
##########
providers/src/airflow/providers/edge/CHANGELOG.rst:
##########
@@ -27,6 +27,14 @@
Changelog
---------
+0.3.0pre0
+.........
+
+Misc
+~~~~
+
+* ``Edge Worker state is set ot unkown if worker heartbeat timed out.``
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
Re: [PR] [Edge] Add metrics export and reset state of Edge Worker after worker timeout [airflow]
Posted by "AutomationDev85 (via GitHub)" <gi...@apache.org>.
AutomationDev85 commented on code in PR #43215:
URL: https://github.com/apache/airflow/pull/43215#discussion_r1814718174
##########
providers/src/airflow/providers/edge/executors/edge_executor.py:
##########
@@ -86,9 +88,30 @@ def execute_async(
)
)
- @provide_session
- def sync(self, session: Session = NEW_SESSION) -> None:
- """Sync will get called periodically by the heartbeat method."""
+ def _check_worker_liveness(self, session: Session) -> bool:
+ """Reset worker state if heartbeat timed out."""
+ changed = False
+ heartbeat_interval: int = conf.getint("edge", "heartbeat_interval")
+ lifeless_workers: list[EdgeWorkerModel] = (
+ session.query(EdgeWorkerModel)
+ .filter(
+ EdgeWorkerModel.state != EdgeWorkerState.UNKNOWN
+ and EdgeWorkerModel.state != EdgeWorkerState.OFFLINE
+ and EdgeWorkerModel.last_update
+ < (timezone.utcnow() - timedelta(seconds=heartbeat_interval * 5))
Review Comment:
Fixed it now
--
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
Re: [PR] [Edge] Add metrics export and reset state of Edge Worker after worker timeout [airflow]
Posted by "jscheffl (via GitHub)" <gi...@apache.org>.
jscheffl commented on code in PR #43215:
URL: https://github.com/apache/airflow/pull/43215#discussion_r1814547417
##########
providers/src/airflow/providers/edge/executors/edge_executor.py:
##########
@@ -86,9 +88,30 @@ def execute_async(
)
)
- @provide_session
- def sync(self, session: Session = NEW_SESSION) -> None:
- """Sync will get called periodically by the heartbeat method."""
+ def _check_worker_liveness(self, session: Session) -> bool:
+ """Reset worker state if heartbeat timed out."""
+ changed = False
+ heartbeat_interval: int = conf.getint("edge", "heartbeat_interval")
+ lifeless_workers: list[EdgeWorkerModel] = (
+ session.query(EdgeWorkerModel)
+ .filter(
+ EdgeWorkerModel.state != EdgeWorkerState.UNKNOWN
+ and EdgeWorkerModel.state != EdgeWorkerState.OFFLINE
+ and EdgeWorkerModel.last_update
+ < (timezone.utcnow() - timedelta(seconds=heartbeat_interval * 5))
Review Comment:
I believe the ORM mapper rather wants it as array of filters like the following?
```suggestion
EdgeWorkerModel.state.not_in(EdgeWorkerState.UNKNOWN, EdgeWorkerState.OFFLINE),
EdgeWorkerModel.last_update < (timezone.utcnow() - timedelta(seconds=heartbeat_interval * 5))
```
--
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
Re: [PR] [Edge] Add metrics export and reset state of Edge Worker after worker timeout [airflow]
Posted by "AutomationDev85 (via GitHub)" <gi...@apache.org>.
AutomationDev85 commented on code in PR #43215:
URL: https://github.com/apache/airflow/pull/43215#discussion_r1811946630
##########
providers/src/airflow/providers/edge/executors/edge_executor.py:
##########
@@ -86,9 +88,32 @@ def execute_async(
)
)
- @provide_session
- def sync(self, session: Session = NEW_SESSION) -> None:
- """Sync will get called periodically by the heartbeat method."""
+ def _check_alive_worker(self, session: Session) -> bool:
+ """Reset worker state if heartbeat timed out."""
+ changed = False
+ heartbeat_interval: int = conf.getint("edge", "heartbeat_interval")
+ workers: list[EdgeWorkerModel] = session.query(EdgeWorkerModel).all()
+
+ for worker in workers:
+ connected = 1
+ if not worker.last_update or worker.last_update < (
+ timezone.utcnow() - timedelta(seconds=heartbeat_interval * 5)
+ ):
Review Comment:
done
##########
providers/src/airflow/providers/edge/executors/edge_executor.py:
##########
@@ -86,9 +88,32 @@ def execute_async(
)
)
- @provide_session
- def sync(self, session: Session = NEW_SESSION) -> None:
- """Sync will get called periodically by the heartbeat method."""
+ def _check_alive_worker(self, session: Session) -> bool:
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
Re: [PR] [Edge] Add metrics export and reset state of Edge Worker after worker timeout [airflow]
Posted by "jscheffl (via GitHub)" <gi...@apache.org>.
jscheffl merged PR #43215:
URL: https://github.com/apache/airflow/pull/43215
--
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
Re: [PR] [Edge] Add metrics export and reset state of Edge Worker after worker timeout [airflow]
Posted by "jscheffl (via GitHub)" <gi...@apache.org>.
jscheffl commented on code in PR #43215:
URL: https://github.com/apache/airflow/pull/43215#discussion_r1810696885
##########
providers/src/airflow/providers/edge/executors/edge_executor.py:
##########
@@ -86,9 +88,30 @@ def execute_async(
)
)
- @provide_session
- def sync(self, session: Session = NEW_SESSION) -> None:
- """Sync will get called periodically by the heartbeat method."""
+ def _check_worker_liveness(self, session: Session) -> bool:
+ """Reset worker state if heartbeat timed out."""
+ changed = False
+ heartbeat_interval: int = conf.getint("edge", "heartbeat_interval")
+ lifeless_workers: list[EdgeWorkerModel] = (
+ session.query(EdgeWorkerModel)
+ .filter(
+ EdgeWorkerModel.state != EdgeWorkerState.UNKNOWN
+ and EdgeWorkerModel.last_update
+ < (timezone.utcnow() - timedelta(seconds=heartbeat_interval * 5))
+ )
+ .all()
+ )
+
+ for worker in lifeless_workers:
+ changed = True
+ worker.state = EdgeWorkerState.UNKNOWN
+ session.add(worker)
Review Comment:
I believe from ORM model this is not correct. You don't want to ADD a new worker record but just want to update. In this case you just need to `commit()` on the session to write changes back to DB.
--
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
Re: [PR] [Edge] Set state of Edge Worker after worker timeout and add statistic export [airflow]
Posted by "jscheffl (via GitHub)" <gi...@apache.org>.
jscheffl commented on code in PR #43215:
URL: https://github.com/apache/airflow/pull/43215#discussion_r1808866814
##########
providers/src/airflow/providers/edge/executors/edge_executor.py:
##########
@@ -86,9 +88,32 @@ def execute_async(
)
)
- @provide_session
- def sync(self, session: Session = NEW_SESSION) -> None:
- """Sync will get called periodically by the heartbeat method."""
+ def _check_alive_worker(self, session: Session) -> bool:
Review Comment:
Naming suggestion: `worker liveness`?
```suggestion
def _check_worker_liveness(self, session: Session) -> bool:
```
##########
providers/src/airflow/providers/edge/CHANGELOG.rst:
##########
@@ -27,6 +27,14 @@
Changelog
---------
+0.3.0pre0
+.........
+
+Misc
+~~~~
+
+* ``Edge Worker state is set ot unkown if worker heartbeat timed out.``
Review Comment:
Typo?
##########
providers/src/airflow/providers/edge/executors/edge_executor.py:
##########
@@ -86,9 +88,32 @@ def execute_async(
)
)
- @provide_session
- def sync(self, session: Session = NEW_SESSION) -> None:
- """Sync will get called periodically by the heartbeat method."""
+ def _check_alive_worker(self, session: Session) -> bool:
+ """Reset worker state if heartbeat timed out."""
+ changed = False
+ heartbeat_interval: int = conf.getint("edge", "heartbeat_interval")
+ workers: list[EdgeWorkerModel] = session.query(EdgeWorkerModel).all()
+
+ for worker in workers:
+ connected = 1
+ if not worker.last_update or worker.last_update < (
+ timezone.utcnow() - timedelta(seconds=heartbeat_interval * 5)
+ ):
Review Comment:
Can you please make this selection into the query to push down the filter to the DB level, not calling `all()` on the database?
--
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
Re: [PR] [Edge] Add metrics export and reset state of Edge Worker after worker timeout [airflow]
Posted by "jscheffl (via GitHub)" <gi...@apache.org>.
jscheffl commented on code in PR #43215:
URL: https://github.com/apache/airflow/pull/43215#discussion_r1814541220
##########
providers/src/airflow/providers/edge/CHANGELOG.rst:
##########
@@ -27,6 +27,14 @@
Changelog
---------
+0.3.0pre0
+.........
+
+Misc
+~~~~
+
+* ``Edge Worker exports metrics and state is set to unkown if worker heartbeat timed out.``
Review Comment:
Two things make sense to have two lines.
```suggestion
* ``Edge Worker exports metrics``
* ``State is set to unknown if worker heartbeat times out.``
```
--
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
Re: [PR] [Edge] Add metrics export and reset state of Edge Worker after worker timeout [airflow]
Posted by "AutomationDev85 (via GitHub)" <gi...@apache.org>.
AutomationDev85 commented on code in PR #43215:
URL: https://github.com/apache/airflow/pull/43215#discussion_r1814613164
##########
providers/src/airflow/providers/edge/executors/edge_executor.py:
##########
@@ -86,9 +88,30 @@ def execute_async(
)
)
- @provide_session
- def sync(self, session: Session = NEW_SESSION) -> None:
- """Sync will get called periodically by the heartbeat method."""
+ def _check_worker_liveness(self, session: Session) -> bool:
+ """Reset worker state if heartbeat timed out."""
+ changed = False
+ heartbeat_interval: int = conf.getint("edge", "heartbeat_interval")
+ lifeless_workers: list[EdgeWorkerModel] = (
+ session.query(EdgeWorkerModel)
+ .filter(
+ EdgeWorkerModel.state != EdgeWorkerState.UNKNOWN
+ and EdgeWorkerModel.state != EdgeWorkerState.OFFLINE
+ and EdgeWorkerModel.last_update
+ < (timezone.utcnow() - timedelta(seconds=heartbeat_interval * 5))
Review Comment:
EdgeWorkerModel.state.not_in is not working. I changed the code to a different check.
--
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
Re: [PR] [Edge] Add metrics export and reset state of Edge Worker after worker timeout [airflow]
Posted by "AutomationDev85 (via GitHub)" <gi...@apache.org>.
AutomationDev85 commented on code in PR #43215:
URL: https://github.com/apache/airflow/pull/43215#discussion_r1811947098
##########
providers/src/airflow/providers/edge/executors/edge_executor.py:
##########
@@ -86,9 +88,30 @@ def execute_async(
)
)
- @provide_session
- def sync(self, session: Session = NEW_SESSION) -> None:
- """Sync will get called periodically by the heartbeat method."""
+ def _check_worker_liveness(self, session: Session) -> bool:
+ """Reset worker state if heartbeat timed out."""
+ changed = False
+ heartbeat_interval: int = conf.getint("edge", "heartbeat_interval")
+ lifeless_workers: list[EdgeWorkerModel] = (
+ session.query(EdgeWorkerModel)
+ .filter(
+ EdgeWorkerModel.state != EdgeWorkerState.UNKNOWN
+ and EdgeWorkerModel.last_update
+ < (timezone.utcnow() - timedelta(seconds=heartbeat_interval * 5))
+ )
+ .all()
+ )
+
+ for worker in lifeless_workers:
+ changed = True
+ worker.state = EdgeWorkerState.UNKNOWN
+ session.add(worker)
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