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 2020/10/07 09:13:49 UTC
[GitHub] [airflow] michalmisiewicz opened a new pull request #11325: KubernetesPodOperator can retry log tailing in case of interruption
michalmisiewicz opened a new pull request #11325:
URL: https://github.com/apache/airflow/pull/11325
<!--
Thank you for contributing! Please make sure that your code changes
are covered with tests. And in case of new features or big changes
remember to adjust the documentation.
Feel free to ping committers for the review!
In case of existing issue, reference it using one of the following:
closes: #ISSUE
related: #ISSUE
How to write a good git commit message:
http://chris.beams.io/posts/git-commit/
-->
This PR adds ability for KubernetesPodOperator to retry log tailing in case of interruption, described in #11324.
I've also included fix for #10586. I've tested introducted changes to be stable when running modified version for several days.
Closes #11324 #10586
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] dimberman commented on a change in pull request #11325: KubernetesPodOperator should retry log tailing in case of interruption
Posted by GitBox <gi...@apache.org>.
dimberman commented on a change in pull request #11325:
URL: https://github.com/apache/airflow/pull/11325#discussion_r502011493
##########
File path: airflow/kubernetes/pod_launcher.py
##########
@@ -124,9 +125,21 @@ def monitor_pod(self, pod: V1Pod, get_logs: bool) -> Tuple[State, Optional[str]]
:return: Tuple[State, Optional[str]]
"""
if get_logs:
- logs = self.read_pod_logs(pod)
- for line in logs:
- self.log.info(line)
+ read_logs_since_sec = None
+ while True:
+ logs = self.read_pod_logs(pod, since_seconds=read_logs_since_sec)
+ for line in logs:
+ self.log.info(line)
+ curr_time = dt.now()
+ time.sleep(1)
+
+ if not self.base_container_is_running(pod):
+ break
+
+ self.log.warning('Pod %s log read interrupted', pod.metadata.name)
+ delta = dt.now() - curr_time
+ # Prefer logs duplication rather than loss
+ read_logs_since_sec = math.ceil(delta.total_seconds())
Review comment:
@michalmisiewicz that sounds perfect! +1 from me once that's 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] github-actions[bot] commented on pull request #11325: KubernetesPodOperator should retry log tailing in case of interruption
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #11325:
URL: https://github.com/apache/airflow/pull/11325#issuecomment-706189060
[The Workflow run](https://github.com/apache/airflow/actions/runs/297577902) is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks$,^Build docs$,^Spell check docs$,^Backport packages$,^Checks: Helm tests$,^Test OpenAPI*.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] dimberman merged pull request #11325: KubernetesPodOperator should retry log tailing in case of interruption
Posted by GitBox <gi...@apache.org>.
dimberman merged pull request #11325:
URL: https://github.com/apache/airflow/pull/11325
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] michalmisiewicz commented on a change in pull request #11325: KubernetesPodOperator should retry log tailing in case of interruption
Posted by GitBox <gi...@apache.org>.
michalmisiewicz commented on a change in pull request #11325:
URL: https://github.com/apache/airflow/pull/11325#discussion_r501820421
##########
File path: airflow/kubernetes/pod_launcher.py
##########
@@ -124,9 +125,21 @@ def monitor_pod(self, pod: V1Pod, get_logs: bool) -> Tuple[State, Optional[str]]
:return: Tuple[State, Optional[str]]
"""
if get_logs:
- logs = self.read_pod_logs(pod)
- for line in logs:
- self.log.info(line)
+ read_logs_since_sec = None
+ while True:
+ logs = self.read_pod_logs(pod, since_seconds=read_logs_since_sec)
+ for line in logs:
+ self.log.info(line)
+ curr_time = dt.now()
+ time.sleep(1)
+
+ if not self.base_container_is_running(pod):
+ break
+
+ self.log.warning('Pod %s log read interrupted', pod.metadata.name)
+ delta = dt.now() - curr_time
+ # Prefer logs duplication rather than loss
+ read_logs_since_sec = math.ceil(delta.total_seconds())
Review comment:
Good questions. `list_namespaced_pod` has option to prepend timestamp to log line. Example
```
2020-10-08T14:16:17.793417674Z 172.32.2.36 - - [08/Oct/2020:14:16:17 +0000] "GET /health HTTP/1.1" 200 187 "-" "kube-probe/1.17
```
We can enable this option, and split log on timestamp and message in `monitor_pod`. Timestamp is returned in RFC3339Nano format so it should be easily parsable using `pendulum`. `Pendulum` should handle possible timezone difference between Kubernetes API and pods.
If you are ok with new design I can refactor 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] dimberman commented on a change in pull request #11325: KubernetesPodOperator should retry log tailing in case of interruption
Posted by GitBox <gi...@apache.org>.
dimberman commented on a change in pull request #11325:
URL: https://github.com/apache/airflow/pull/11325#discussion_r502011493
##########
File path: airflow/kubernetes/pod_launcher.py
##########
@@ -124,9 +125,21 @@ def monitor_pod(self, pod: V1Pod, get_logs: bool) -> Tuple[State, Optional[str]]
:return: Tuple[State, Optional[str]]
"""
if get_logs:
- logs = self.read_pod_logs(pod)
- for line in logs:
- self.log.info(line)
+ read_logs_since_sec = None
+ while True:
+ logs = self.read_pod_logs(pod, since_seconds=read_logs_since_sec)
+ for line in logs:
+ self.log.info(line)
+ curr_time = dt.now()
+ time.sleep(1)
+
+ if not self.base_container_is_running(pod):
+ break
+
+ self.log.warning('Pod %s log read interrupted', pod.metadata.name)
+ delta = dt.now() - curr_time
+ # Prefer logs duplication rather than loss
+ read_logs_since_sec = math.ceil(delta.total_seconds())
Review comment:
@michalmisiewicz that sounds perfect! +1 from me once that's 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] dimberman commented on a change in pull request #11325: KubernetesPodOperator should retry log tailing in case of interruption
Posted by GitBox <gi...@apache.org>.
dimberman commented on a change in pull request #11325:
URL: https://github.com/apache/airflow/pull/11325#discussion_r501182226
##########
File path: airflow/kubernetes/pod_launcher.py
##########
@@ -124,9 +125,21 @@ def monitor_pod(self, pod: V1Pod, get_logs: bool) -> Tuple[State, Optional[str]]
:return: Tuple[State, Optional[str]]
"""
if get_logs:
- logs = self.read_pod_logs(pod)
- for line in logs:
- self.log.info(line)
+ read_logs_since_sec = None
+ while True:
+ logs = self.read_pod_logs(pod, since_seconds=read_logs_since_sec)
+ for line in logs:
+ self.log.info(line)
+ curr_time = dt.now()
+ time.sleep(1)
+
+ if not self.base_container_is_running(pod):
+ break
+
+ self.log.warning('Pod %s log read interrupted', pod.metadata.name)
+ delta = dt.now() - curr_time
+ # Prefer logs duplication rather than loss
+ read_logs_since_sec = math.ceil(delta.total_seconds())
Review comment:
Is there any chance of log lines being missed in the period between the logs being read and this time? Is there any way we can get the timestamp from the read_pod_log request?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] github-actions[bot] commented on pull request #11325: KubernetesPodOperator should retry log tailing in case of interruption
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #11325:
URL: https://github.com/apache/airflow/pull/11325#issuecomment-704900787
[The Build Workflow run](https://github.com/apache/airflow/actions/runs/293297900) is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks$,^Build docs$,^Spell check docs$,^Backport packages$,^Checks: Helm tests$,^Test OpenAPI*.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] boring-cyborg[bot] commented on pull request #11325: KubernetesPodOperator can retry log tailing in case of interruption
Posted by GitBox <gi...@apache.org>.
boring-cyborg[bot] commented on pull request #11325:
URL: https://github.com/apache/airflow/pull/11325#issuecomment-704804426
Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contribution Guide (https://github.com/apache/airflow/blob/master/CONTRIBUTING.rst)
Here are some useful points:
- Pay attention to the quality of your code (flake8, pylint and type annotations). Our [pre-commits]( https://github.com/apache/airflow/blob/master/STATIC_CODE_CHECKS.rst#prerequisites-for-pre-commit-hooks) will help you with that.
- In case of a new feature add useful documentation (in docstrings or in `docs/` directory). Adding a new operator? Check this short [guide](https://github.com/apache/airflow/blob/master/docs/howto/custom-operator.rst) Consider adding an example DAG that shows how users should use it.
- Consider using [Breeze environment](https://github.com/apache/airflow/blob/master/BREEZE.rst) for testing locally, itโs a heavy docker but it ships with a working Airflow and a lot of integrations.
- Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
- Please follow [ASF Code of Conduct](https://www.apache.org/foundation/policies/conduct) for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
- Be sure to read the [Airflow Coding style]( https://github.com/apache/airflow/blob/master/CONTRIBUTING.rst#coding-style-and-best-practices).
Apache Airflow is a community-driven project and together we are making it better ๐.
In case of doubts contact the developers at:
Mailing List: dev@airflow.apache.org
Slack: https://s.apache.org/airflow-slack
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] github-actions[bot] commented on pull request #11325: KubernetesPodOperator should retry log tailing in case of interruption
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #11325:
URL: https://github.com/apache/airflow/pull/11325#issuecomment-706231634
[The Workflow run](https://github.com/apache/airflow/actions/runs/297696278) is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks$,^Build docs$,^Spell check docs$,^Backport packages$,^Checks: Helm tests$,^Test OpenAPI*.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] michalmisiewicz commented on pull request #11325: KubernetesPodOperator should retry log tailing in case of interruption
Posted by GitBox <gi...@apache.org>.
michalmisiewicz commented on pull request #11325:
URL: https://github.com/apache/airflow/pull/11325#issuecomment-706226801
New version fixes encoding.
Before
```
[2020-09-23 13:04:03,723] {pod_launcher.py:156} INFO - b'Max workers: 100\n'
[2020-09-23 13:04:09,569] {pod_launcher.py:156} INFO - b'All cases collected, total: 30000\n'
```
After
```
[2020-10-09 14:40:28,752] {pod_launcher.py:136} INFO - Max workers: 100
[2020-10-09 14:40:29,445] {pod_launcher.py:136} INFO - All cases collected, total: 30000
```
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] dimberman commented on pull request #11325: KubernetesPodOperator should retry log tailing in case of interruption
Posted by GitBox <gi...@apache.org>.
dimberman commented on pull request #11325:
URL: https://github.com/apache/airflow/pull/11325#issuecomment-706436188
Thank you for this @michalmisiewicz!
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] michalmisiewicz commented on pull request #11325: KubernetesPodOperator can retry log tailing in case of interruption
Posted by GitBox <gi...@apache.org>.
michalmisiewicz commented on pull request #11325:
URL: https://github.com/apache/airflow/pull/11325#issuecomment-704807888
@kaxil @dimberman @mik-laj
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] michalmisiewicz commented on a change in pull request #11325: KubernetesPodOperator should retry log tailing in case of interruption
Posted by GitBox <gi...@apache.org>.
michalmisiewicz commented on a change in pull request #11325:
URL: https://github.com/apache/airflow/pull/11325#discussion_r501820421
##########
File path: airflow/kubernetes/pod_launcher.py
##########
@@ -124,9 +125,21 @@ def monitor_pod(self, pod: V1Pod, get_logs: bool) -> Tuple[State, Optional[str]]
:return: Tuple[State, Optional[str]]
"""
if get_logs:
- logs = self.read_pod_logs(pod)
- for line in logs:
- self.log.info(line)
+ read_logs_since_sec = None
+ while True:
+ logs = self.read_pod_logs(pod, since_seconds=read_logs_since_sec)
+ for line in logs:
+ self.log.info(line)
+ curr_time = dt.now()
+ time.sleep(1)
+
+ if not self.base_container_is_running(pod):
+ break
+
+ self.log.warning('Pod %s log read interrupted', pod.metadata.name)
+ delta = dt.now() - curr_time
+ # Prefer logs duplication rather than loss
+ read_logs_since_sec = math.ceil(delta.total_seconds())
Review comment:
Good questions. `list_namespaced_pod` has option to prepend timestamp to log line. Example
```
2020-10-08T14:16:17.793417674Z 172.32.2.36 - - [08/Oct/2020:14:16:17 +0000] "GET /health HTTP/1.1" 200 187 "-" "kube-probe/1.17
```
We can enable this option, and split log on timestamp and message in `monitor_pod`. Timestamp is returned in RFC3339Nano format so it should be easily parsable using `pendulum`. `Pendulum` should handle possible timezone difference between Kubernetes API and pods.
If you are ok with new design I can refactor 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] boring-cyborg[bot] commented on pull request #11325: KubernetesPodOperator should retry log tailing in case of interruption
Posted by GitBox <gi...@apache.org>.
boring-cyborg[bot] commented on pull request #11325:
URL: https://github.com/apache/airflow/pull/11325#issuecomment-706436134
Awesome work, congrats on your first merged pull request!
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org