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/05/11 22:28:08 UTC
[GitHub] [airflow] schattian opened a new pull request, #23658: 23497 tasks stuck
schattian opened a new pull request, #23658:
URL: https://github.com/apache/airflow/pull/23658
related: #23618
fixes leaking loop in https://github.com/apache/airflow/commit/67474cccb612fad39a2afbeee16014909447d242. That also involved changing `get_event_loop` to be `new_event_loop` so it is closable (otherwise, parallel usage of the function could try to reuse a closed loop).
That also fixes a previously broken (modified) test. Notice that one call to container_is_running was leaked to one test that only made 3 itself.
--
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] schattian commented on pull request #23658: Fix k8s pod.execute randomly stuck indefinitely by logs consumption (#23497)
Posted by GitBox <gi...@apache.org>.
schattian commented on PR #23658:
URL: https://github.com/apache/airflow/pull/23658#issuecomment-1160837468
@dstandish i've forgot, but what happens in such case (and same for whenever the pod is deleted for some other external issue that could cause, e.g: pod not found) is that the kubectl stream api returns immediately, so this is not an issue (see eg cncf kube pod operator).
I've just made the functions private
--
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] schattian commented on pull request #23658: Fix k8s pod.execute randomly stuck indefinitely by logs consumption (#23497)
Posted by GitBox <gi...@apache.org>.
schattian commented on PR #23658:
URL: https://github.com/apache/airflow/pull/23658#issuecomment-1157407782
>one alternative i'll just mention is we could handle this by doing our own "following" instead of relying on the stream. i.e. we do our own loop and consume the latest logs, then sleep a second and do the same.
you mean calling X amount of times the kubectl logs api? I think that's a worse than a follow..
Mostly because you'll have some lines printed twice (and could be expensive for some dags)
>one is I think you need to kill the logging process when on_kill is called.
Sorry, i dont think i understand what's the point here
>the other is, if you get an error in await_container_completion, your subprocess won't be killed. so, you probably should modify so that if you get an unexpected error after process has started, that the process will be killed.
Hmm but in that case, aren't we going to propagate a SIGTERM? (ie killing the subprocess implicitly)?
I was just relying on that, but if that's not how it works / could change, i will modify that.
>also, i'd make "private" the two methods you've added
👍🏻
>separate question for you @schattian. suppose the stream hangs, and the pod is still running. if we terminate the process and restart logs consumption, will it get logs from the new location? or will it still look at the original location and therefore any new logging is unreachable and it will immediately hang?
That is correct. A call to kubectl logs api after the stream hanged will return the newest logs.
But, there are a few problems with that:
1. The pod could be deleted, and in that case you'll receive no logs.
2. We determine that the stream hangs as "if the container is not running and stream consumption didnt finish", as there is not a better way to do it (please let me know if you have one). In that case
I have mentioned that in the first pr https://github.com/apache/airflow/pull/23618#discussion_r869096030
I think it adds some complexity that's not worth as this bugs doesn't occur so often
--
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] potiuk commented on pull request #23658: Fix k8s pod.execute randomly stuck indefinitely by logs consumption (#23497)
Posted by GitBox <gi...@apache.org>.
potiuk commented on PR #23658:
URL: https://github.com/apache/airflow/pull/23658#issuecomment-1161622080
> @potiuk uhm i dont see how it helps. the ci code was the same as before except the new additions, and tests fail for those.
>
> I've updated it btw, lets see
I think this is pretty obvious - if there was an issue with some tests at the moment you branched off no matter how hard you try it will stay there if you don't rebase - if it was fixed in one of the 102 commits you will get it fixed.
Why do you think it would work otherwise?
--
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] schattian commented on pull request #23658: Fix k8s pod.execute randomly stuck indefinitely by logs consumption (#23497)
Posted by GitBox <gi...@apache.org>.
schattian commented on PR #23658:
URL: https://github.com/apache/airflow/pull/23658#issuecomment-1161455305
hm, are calls to method wrapped by a private function mocked?
I dont see any other way the tests can timeout with the new changes. They are not flaky, only fail consistently after the addition
--
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 #23658: Fix k8s pod.execute randomly stuck indefinitely by logs consumption (#23497)
Posted by GitBox <gi...@apache.org>.
dstandish commented on code in PR #23658:
URL: https://github.com/apache/airflow/pull/23658#discussion_r896002411
##########
airflow/providers/cncf/kubernetes/utils/pod_manager.py:
##########
@@ -193,6 +195,40 @@ def follow_container_logs(self, pod: V1Pod, container_name: str) -> PodLoggingSt
)
return self.fetch_container_logs(pod=pod, container_name=container_name, follow=True)
+ def log_iterable(self, logs: Iterable[bytes]) -> Optional[DateTime]:
+ timestamp = None
+ for line in logs:
+ timestamp, message = self.parse_log_line(line.decode('utf-8', errors="backslashreplace"))
+ self.log.info(message)
+ return timestamp
+
+ def consume_container_logs_stream(
+ self, pod: V1Pod, container_name: str, stream: Iterable[bytes]
+ ) -> Optional[DateTime]:
+ def log_iterable_and_set_value(timestamp):
+ dt = self.log_iterable(stream)
+ if dt is not None:
+ timestamp.value = dt.timestamp() # type: ignore[attr-defined]
Review Comment:
nevermind i see 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
[GitHub] [airflow] potiuk commented on pull request #23658: Fix k8s pod.execute randomly stuck indefinitely by logs consumption (#23497)
Posted by GitBox <gi...@apache.org>.
potiuk commented on PR #23658:
URL: https://github.com/apache/airflow/pull/23658#issuecomment-1161468036
@schattian ALWAYS rebase your changes when you attempt to see if the tests are passing. This way you avpod wasting your and committers time on trying to solve problems that are Likely already solved.
You are 102 commits behind. Things ARE not going to work.
--
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] potiuk commented on pull request #23658: Fix k8s pod.execute randomly stuck indefinitely by logs consumption (#23497)
Posted by GitBox <gi...@apache.org>.
potiuk commented on PR #23658:
URL: https://github.com/apache/airflow/pull/23658#issuecomment-1146688374
@dstandish @ashb @ephraimbuddy @jedcunningham - I think we need some of your's expert knowledge here
--
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] schattian commented on pull request #23658: Fix k8s pod.execute randomly stuck indefinitely by logs consumption (#23497)
Posted by GitBox <gi...@apache.org>.
schattian commented on PR #23658:
URL: https://github.com/apache/airflow/pull/23658#issuecomment-1161515273
@potiuk uhm i dont see how it helps.
the ci code was the same as before except the new additions, and tests fail for those.
I've updated it btw
--
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] potiuk commented on pull request #23658: Fix k8s pod.execute randomly stuck indefinitely by logs consumption (#23497)
Posted by GitBox <gi...@apache.org>.
potiuk commented on PR #23658:
URL: https://github.com/apache/airflow/pull/23658#issuecomment-1161626353
> I think this is pretty obvious - if there was an issue with some tests at the moment you branched off no matter how hard you try it will stay there if you don't rebase - if it was fixed in one of the 102 commits you will get it fixed.
BTW. It's not guaranteed to work. It just removes one of the potential failure reasons out of the possible issues.
--
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] schattian commented on pull request #23658: Fix k8s pod.execute randomly stuck indefinitely by logs consumption (#23497)
Posted by GitBox <gi...@apache.org>.
schattian commented on PR #23658:
URL: https://github.com/apache/airflow/pull/23658#issuecomment-1161628922
yeah thats what i was thinking, added by making the functions private.
What i think happens is that the methods that are looking to be mocked are inside a private function.
I dont see how to change that behaviour other than making the functions non-private again.
--
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] potiuk commented on pull request #23658: Fix k8s pod.execute randomly stuck indefinitely by logs consumption (#23497)
Posted by GitBox <gi...@apache.org>.
potiuk commented on PR #23658:
URL: https://github.com/apache/airflow/pull/23658#issuecomment-1141963136
(And Sorry for long delays in review @schattian - we had Airflow Summit last week which kept many of the reviewers busy).
--
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 #23658: Fix k8s pod.execute randomly stuck indefinitely by logs consumption (#23497)
Posted by GitBox <gi...@apache.org>.
dstandish commented on code in PR #23658:
URL: https://github.com/apache/airflow/pull/23658#discussion_r890509834
##########
airflow/providers/cncf/kubernetes/utils/pod_manager.py:
##########
@@ -256,7 +293,7 @@ def consume_logs(*, since_time: Optional[DateTime] = None, follow: bool = True)
time.sleep(1)
def await_container_completion(self, pod: V1Pod, container_name: str) -> None:
- while not self.container_is_running(pod=pod, container_name=container_name):
+ while self.container_is_running(pod=pod, container_name=container_name):
Review Comment:
ah, done here: https://github.com/apache/airflow/pull/23883/files now just need to rebase
--
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] potiuk commented on pull request #23658: Fix k8s pod.execute randomly stuck indefinitely by logs consumption (#23497)
Posted by GitBox <gi...@apache.org>.
potiuk commented on PR #23658:
URL: https://github.com/apache/airflow/pull/23658#issuecomment-1125394833
Yeah. Using Asyncio is kinda tricky for standalone operator. It sounds a bit tricky with this approach. Seems that we are trying to start asyncio.loop to run asyncio operations and then we kill the thread after closing the operator. This sounds rather complex as we do not really achieve the "async" promises.
Just a thought - maybe we should somehew plug it in the "defferable" operators and triggerer ? I am not sure if that would be efficient for getting the logs using triggerer and triggers ? Or maybe we should abandon the idea of async operations here as they do not really give us anything and only cause troublesome testing and difficult debugging in the future?
@andrewgodwin - maybe you can share your thoughts on that ?
--
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] schattian commented on pull request #23658: Fix k8s pod.execute randomly stuck indefinitely by logs consumption (#23497)
Posted by GitBox <gi...@apache.org>.
schattian commented on PR #23658:
URL: https://github.com/apache/airflow/pull/23658#issuecomment-1135138024
@potiuk could you please take a look at this one? would be nice to have, personally I experience this bug regularly (once a week) on gke.
--
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 pull request #23658: Fix k8s pod.execute randomly stuck indefinitely by logs consumption (#23497)
Posted by GitBox <gi...@apache.org>.
dstandish commented on PR #23658:
URL: https://github.com/apache/airflow/pull/23658#issuecomment-1157816945
>> one is I think you need to kill the logging process when on_kill is called.
>Sorry, i dont think i understand what's the point here
in operators, if we kill the task (e.g. from UI) shouldn't we also kill the logging subprocess? if we don't do that, will it continue running e.g. on celery worker?
--
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] schattian commented on pull request #23658: Fix k8s pod.execute randomly stuck indefinitely by logs consumption (#23497)
Posted by GitBox <gi...@apache.org>.
schattian commented on PR #23658:
URL: https://github.com/apache/airflow/pull/23658#issuecomment-1128131089
@potiuk i replaced the asyncio-based approach with an executor to use multiprocessing.Process instead.
I added one more test case also for the case api hangs + no logs.
Imho this is the best to do to not overcomplicate this, wdyt?
--
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] github-actions[bot] commented on pull request #23658: Fix k8s pod.execute randomly stuck indefinitely by logs consumption (#23497)
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on PR #23658:
URL: https://github.com/apache/airflow/pull/23658#issuecomment-1141963662
The PR is likely OK to be merged with just subset of tests for default Python and Database versions without running the full matrix of tests, because it does not modify the core of Airflow. If the committers decide that the full tests matrix is needed, they will add the label 'full tests needed'. Then you should rebase to the latest main or amend the last commit of the PR, and push it with --force-with-lease.
--
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 #23658: Fix k8s pod.execute randomly stuck indefinitely by logs consumption (#23497)
Posted by GitBox <gi...@apache.org>.
dstandish commented on code in PR #23658:
URL: https://github.com/apache/airflow/pull/23658#discussion_r890437916
##########
airflow/providers/cncf/kubernetes/utils/pod_manager.py:
##########
@@ -193,6 +195,40 @@ def follow_container_logs(self, pod: V1Pod, container_name: str) -> PodLoggingSt
)
return self.fetch_container_logs(pod=pod, container_name=container_name, follow=True)
+ def log_iterable(self, logs: Iterable[bytes]) -> Optional[DateTime]:
+ timestamp = None
+ for line in logs:
+ timestamp, message = self.parse_log_line(line.decode('utf-8', errors="backslashreplace"))
+ self.log.info(message)
+ return timestamp
+
+ def consume_container_logs_stream(
+ self, pod: V1Pod, container_name: str, stream: Iterable[bytes]
+ ) -> Optional[DateTime]:
+ def log_iterable_and_set_value(timestamp):
+ dt = self.log_iterable(stream)
+ if dt is not None:
+ timestamp.value = dt.timestamp() # type: ignore[attr-defined]
Review Comment:
shouldn't this be `timestamp.value = dt`? ultimately this gets returned through `consume_logs` which returns pendulum.DateTime
--
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 #23658: Fix k8s pod.execute randomly stuck indefinitely by logs consumption (#23497)
Posted by GitBox <gi...@apache.org>.
dstandish commented on code in PR #23658:
URL: https://github.com/apache/airflow/pull/23658#discussion_r895935762
##########
airflow/providers/cncf/kubernetes/utils/pod_manager.py:
##########
@@ -256,7 +293,7 @@ def consume_logs(*, since_time: Optional[DateTime] = None, follow: bool = True)
time.sleep(1)
def await_container_completion(self, pod: V1Pod, container_name: str) -> None:
- while not self.container_is_running(pod=pod, container_name=container_name):
+ while self.container_is_running(pod=pod, container_name=container_name):
Review Comment:
sure!
--
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] schattian commented on pull request #23658: Fix k8s pod.execute randomly stuck indefinitely by logs consumption (#23497)
Posted by GitBox <gi...@apache.org>.
schattian commented on PR #23658:
URL: https://github.com/apache/airflow/pull/23658#issuecomment-1124935866
Now added explicit shutdown of the executor. However.. it's strange, `loop.close` should be shutting down the default executor: https://github.dev/python/cpython/blob/f882d33778ee2625ab32d90e28edb6878fb8af93/Lib/asyncio/base_events.py#L678-L683
If that doesn't work it is probably a bug somewhere or a typo im not seeing...
--
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] schattian commented on pull request #23658: Fix k8s pod.execute randomly stuck indefinitely by logs consumption (#23497)
Posted by GitBox <gi...@apache.org>.
schattian commented on PR #23658:
URL: https://github.com/apache/airflow/pull/23658#issuecomment-1130172267
Ok, now i dont see any relation between the failing tests and the pr.
--
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 #23658: Fix k8s pod.execute randomly stuck indefinitely by logs consumption (#23497)
Posted by GitBox <gi...@apache.org>.
dstandish commented on code in PR #23658:
URL: https://github.com/apache/airflow/pull/23658#discussion_r895937987
##########
airflow/providers/cncf/kubernetes/utils/pod_manager.py:
##########
@@ -193,6 +195,40 @@ def follow_container_logs(self, pod: V1Pod, container_name: str) -> PodLoggingSt
)
return self.fetch_container_logs(pod=pod, container_name=container_name, follow=True)
+ def log_iterable(self, logs: Iterable[bytes]) -> Optional[DateTime]:
+ timestamp = None
+ for line in logs:
+ timestamp, message = self.parse_log_line(line.decode('utf-8', errors="backslashreplace"))
+ self.log.info(message)
+ return timestamp
+
+ def consume_container_logs_stream(
+ self, pod: V1Pod, container_name: str, stream: Iterable[bytes]
+ ) -> Optional[DateTime]:
+ def log_iterable_and_set_value(timestamp):
+ dt = self.log_iterable(stream)
+ if dt is not None:
+ timestamp.value = dt.timestamp() # type: ignore[attr-defined]
Review Comment:
ok but when this value is returned by consume_logs, you must ensure that it is converted back to DateTime right? perhaps you have done that somewhere?
--
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] potiuk commented on a diff in pull request #23658: Fix k8s pod.execute randomly stuck indefinitely by logs consumption (#23497)
Posted by GitBox <gi...@apache.org>.
potiuk commented on code in PR #23658:
URL: https://github.com/apache/airflow/pull/23658#discussion_r895049912
##########
airflow/providers/cncf/kubernetes/utils/pod_manager.py:
##########
@@ -256,7 +293,7 @@ def consume_logs(*, since_time: Optional[DateTime] = None, follow: bool = True)
time.sleep(1)
def await_container_completion(self, pod: V1Pod, container_name: str) -> None:
- while not self.container_is_running(pod=pod, container_name=container_name):
+ while self.container_is_running(pod=pod, container_name=container_name):
Review Comment:
It looks green. @dstandish - do you approve :)?
--
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 pull request #23658: Fix k8s pod.execute randomly stuck indefinitely by logs consumption (#23497)
Posted by GitBox <gi...@apache.org>.
dstandish commented on PR #23658:
URL: https://github.com/apache/airflow/pull/23658#issuecomment-1162063256
> yeah thats what i was thinking, added by making the functions private.
What i think happens is that the methods that are looking to be mocked are inside a private function.
I dont see how to change that behaviour other than making the functions non-private again
wellp sorry yeah if they can't be private they can't be private. i just want to minimize surface area of promises we are making. when we have "public" methods we can't change them without deprecation.
re
> @dstandish i've forgot, but what happens in such case (and same for whenever the pod is deleted for some other external issue that could cause, for example, pod not found) is that the kubectl stream api returns immediately, so this is not an issue (see eg cncf kube pod operator on_kill def).
can you clarify... when _what_ happens?
--
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] github-actions[bot] closed pull request #23658: Fix k8s pod.execute randomly stuck indefinitely by logs consumption (#23497)
Posted by GitBox <gi...@apache.org>.
github-actions[bot] closed pull request #23658: Fix k8s pod.execute randomly stuck indefinitely by logs consumption (#23497)
URL: https://github.com/apache/airflow/pull/23658
--
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] schattian commented on pull request #23658: Fix k8s pod.execute randomly stuck indefinitely by logs consumption (#23497)
Posted by GitBox <gi...@apache.org>.
schattian commented on PR #23658:
URL: https://github.com/apache/airflow/pull/23658#issuecomment-1142318827
> (And Sorry for long delays in review @schattian - we had Airflow Summit last week which kept many of the reviewers busy).
didnt know that. no worries! and thanks for the rev
--
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 pull request #23658: Fix k8s pod.execute randomly stuck indefinitely by logs consumption (#23497)
Posted by GitBox <gi...@apache.org>.
dstandish commented on PR #23658:
URL: https://github.com/apache/airflow/pull/23658#issuecomment-1154242663
also, i'd make "private" the two methods you've added
--
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] potiuk commented on pull request #23658: Fix k8s pod.execute randomly stuck indefinitely by logs consumption (#23497)
Posted by GitBox <gi...@apache.org>.
potiuk commented on PR #23658:
URL: https://github.com/apache/airflow/pull/23658#issuecomment-1161624767
Now - it seem you do have a real issue - lilely introduced by your changes:
You need to (as usual) start breeze in your version (build image) - reproduce it by running tests in the same environment as in CI and fix.
--
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 pull request #23658: Fix k8s pod.execute randomly stuck indefinitely by logs consumption (#23497)
Posted by GitBox <gi...@apache.org>.
dstandish commented on PR #23658:
URL: https://github.com/apache/airflow/pull/23658#issuecomment-1163585574
> > can you clarify... when _what_ happens?
>
> The function consuming the logs (the process) will return immediately so it'll terminate.
>
> That's because the stream api will return.
Oh i think i understand what you're saying. You are saying that, since on_kill terminates the pod, the log stream will close and the process will die of its own accord, so there's no need to manage the termination of that process directly.
--
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] github-actions[bot] commented on pull request #23658: Fix k8s pod.execute randomly stuck indefinitely by logs consumption (#23497)
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on PR #23658:
URL: https://github.com/apache/airflow/pull/23658#issuecomment-1207302337
This pull request has been automatically marked as stale because it has not had recent activity. It will be closed in 5 days if no further activity occurs. Thank you for your contributions.
--
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] schattian commented on pull request #23658: Fix k8s pod.execute randomly stuck indefinitely by logs consumption (#23497)
Posted by GitBox <gi...@apache.org>.
schattian commented on PR #23658:
URL: https://github.com/apache/airflow/pull/23658#issuecomment-1162150792
> can you clarify... when _what_ happens?
The function consuming the logs (the process) will return immediately so it'll terminate.
That's because the stream api will return.
--
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 #23658: Fix k8s pod.execute randomly stuck indefinitely by logs consumption (#23497)
Posted by GitBox <gi...@apache.org>.
dstandish commented on code in PR #23658:
URL: https://github.com/apache/airflow/pull/23658#discussion_r890423955
##########
airflow/providers/cncf/kubernetes/utils/pod_manager.py:
##########
@@ -256,7 +293,7 @@ def consume_logs(*, since_time: Optional[DateTime] = None, follow: bool = True)
time.sleep(1)
def await_container_completion(self, pod: V1Pod, container_name: str) -> None:
- while not self.container_is_running(pod=pod, container_name=container_name):
+ while self.container_is_running(pod=pod, container_name=container_name):
Review Comment:
(it's less complicated, more obvious fix than the other part so we can merge it faster and focus more cleanly on the other bit in this PR)
--
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] schattian commented on a diff in pull request #23658: Fix k8s pod.execute randomly stuck indefinitely by logs consumption (#23497)
Posted by GitBox <gi...@apache.org>.
schattian commented on code in PR #23658:
URL: https://github.com/apache/airflow/pull/23658#discussion_r890881497
##########
airflow/providers/cncf/kubernetes/utils/pod_manager.py:
##########
@@ -256,7 +293,7 @@ def consume_logs(*, since_time: Optional[DateTime] = None, follow: bool = True)
time.sleep(1)
def await_container_completion(self, pod: V1Pod, container_name: str) -> None:
- while not self.container_is_running(pod=pod, container_name=container_name):
+ while self.container_is_running(pod=pod, container_name=container_name):
Review Comment:
nice - yep, i though on doing so in a diff pr but im using that func.
Will rebase
--
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] schattian commented on pull request #23658: Fix k8s pod.execute randomly stuck indefinitely by logs consumption (#23497)
Posted by GitBox <gi...@apache.org>.
schattian commented on PR #23658:
URL: https://github.com/apache/airflow/pull/23658#issuecomment-1148325090
> @schattian any reason you are using multiprocessing instead of a thread?
Yep. The thread could hang indefinitely, as the kube logs api stream could do, so I need to terminate the process.
However, i am not very knowledgeable on python's api for this kind of thing, so open to suggestions.
--
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] potiuk commented on pull request #23658: Fix k8s pod.execute randomly stuck indefinitely by logs consumption (#23497)
Posted by GitBox <gi...@apache.org>.
potiuk commented on PR #23658:
URL: https://github.com/apache/airflow/pull/23658#issuecomment-1148519571
> > @schattian any reason you are using multiprocessing instead of a thread?
>
> @dstandish Yep. The thread could hang indefinitely, as the kube logs api stream could do, so I need to terminate the process.
>
> However, i am not very knowledgeable on python's api for this kind of thing, so open to suggestions.
OK. I will wait for merging before releasing the new wave of providers then - I think it might need some more deliberation (We will include it in the next wave then).
--
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 pull request #23658: Fix k8s pod.execute randomly stuck indefinitely by logs consumption (#23497)
Posted by GitBox <gi...@apache.org>.
dstandish commented on PR #23658:
URL: https://github.com/apache/airflow/pull/23658#issuecomment-1154232958
> Yep. The thread could hang indefinitely, as the kube logs api stream could do, so I need to terminate the process.
>
> However, i am not very knowledgeable on python's api for this kind of thing, so open to suggestions.
ok yeah, thing i learned: killing threads is not well-supported. interestingly, [something similar is done for ECS](https://github.com/apache/airflow/blob/main/airflow/providers/amazon/aws/operators/ecs.py#L98), and threads are used there, but they don't deal with indefinite hanging problem (and therefore don't have to forcibly terminate it).
one alternative i'll just mention is we could handle this by doing our own "following" instead of relying on the stream. i.e. we do our own loop and consume the latest logs, then sleep a second and do the same. this might allow for a cleaner solution.
but, multiprocessing seems ok. i'll just mention a few things.
one is I think you need to kill the logging process when `on_kill` is called.
the other is, if you get an error in `await_container_completion`, your subprocess won't be killed. so, you probably should modify so that if you get an unexpected error after process has started, that the process will be killed.
also, i would try to make note somewhere that this is due to a problem with kubernetes, and what we are doing here is a hack that should be removed once the issue has been resolved upstream (and propagated sufficiently, which of course could be a long time from now). the upstream issue is documented here https://github.com/kubernetes/kubernetes/issues/59902.
--
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] schattian commented on pull request #23658: Fix k8s pod.execute randomly stuck indefinitely by logs consumption (#23497)
Posted by GitBox <gi...@apache.org>.
schattian commented on PR #23658:
URL: https://github.com/apache/airflow/pull/23658#issuecomment-1160477301
> in operators, if we kill the task (e.g. from UI) shouldn't we also kill the logging subprocess? if we don't do that, will it continue running e.g. on celery worker?
Ah, okay, yeah that's what i was trying to answer with:
>Hmm but in that case, aren't we going to propagate a SIGTERM? (ie killing the subprocess implicitly)?
>I was just relying on that, but if that's not how it works / could change, i will modify that.
wdyt?
--
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] schattian commented on a diff in pull request #23658: Fix k8s pod.execute randomly stuck indefinitely by logs consumption (#23497)
Posted by GitBox <gi...@apache.org>.
schattian commented on code in PR #23658:
URL: https://github.com/apache/airflow/pull/23658#discussion_r890885849
##########
airflow/providers/cncf/kubernetes/utils/pod_manager.py:
##########
@@ -193,6 +195,40 @@ def follow_container_logs(self, pod: V1Pod, container_name: str) -> PodLoggingSt
)
return self.fetch_container_logs(pod=pod, container_name=container_name, follow=True)
+ def log_iterable(self, logs: Iterable[bytes]) -> Optional[DateTime]:
+ timestamp = None
+ for line in logs:
+ timestamp, message = self.parse_log_line(line.decode('utf-8', errors="backslashreplace"))
+ self.log.info(message)
+ return timestamp
+
+ def consume_container_logs_stream(
+ self, pod: V1Pod, container_name: str, stream: Iterable[bytes]
+ ) -> Optional[DateTime]:
+ def log_iterable_and_set_value(timestamp):
+ dt = self.log_iterable(stream)
+ if dt is not None:
+ timestamp.value = dt.timestamp() # type: ignore[attr-defined]
Review Comment:
i cannot do it with dt as i am using a rawvalue('f') -float-. Therefore, i need to use the unix ts instead to pass 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] dstandish commented on a diff in pull request #23658: Fix k8s pod.execute randomly stuck indefinitely by logs consumption (#23497)
Posted by GitBox <gi...@apache.org>.
dstandish commented on code in PR #23658:
URL: https://github.com/apache/airflow/pull/23658#discussion_r890420936
##########
airflow/providers/cncf/kubernetes/utils/pod_manager.py:
##########
@@ -256,7 +293,7 @@ def consume_logs(*, since_time: Optional[DateTime] = None, follow: bool = True)
time.sleep(1)
def await_container_completion(self, pod: V1Pod, container_name: str) -> None:
- while not self.container_is_running(pod=pod, container_name=container_name):
+ while self.container_is_running(pod=pod, container_name=container_name):
Review Comment:
this looks like it is unrelated to your main change. could you please raise a separatae PR for this change specifically?
--
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 pull request #23658: Fix k8s pod.execute randomly stuck indefinitely by logs consumption (#23497)
Posted by GitBox <gi...@apache.org>.
dstandish commented on PR #23658:
URL: https://github.com/apache/airflow/pull/23658#issuecomment-1147757565
@schattian any reason you are using multiprocessing instead of a thread?
--
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 pull request #23658: Fix k8s pod.execute randomly stuck indefinitely by logs consumption (#23497)
Posted by GitBox <gi...@apache.org>.
dstandish commented on PR #23658:
URL: https://github.com/apache/airflow/pull/23658#issuecomment-1154252279
separate question for you @schattian. suppose the stream hangs, and the pod is still running. if we terminate the process and restart logs consumption, will it get logs from the new location? or will it still look at the original location and therefore any new logging is unreachable and it will immediately hang?
--
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] schattian commented on pull request #23658: Fix k8s pod.execute randomly stuck indefinitely by logs consumption (#23497)
Posted by GitBox <gi...@apache.org>.
schattian commented on PR #23658:
URL: https://github.com/apache/airflow/pull/23658#issuecomment-1125009898
Ok, i think i figured out the problem. `shutdown_default_executor` is not available until py3.9.
Therefore i need to create an executor instead of rely in the default one (notice `loop.close` does not wait until thread of executor finishes)
--
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] schattian commented on pull request #23658: Fix k8s pod.execute randomly stuck indefinitely by logs consumption (#23497)
Posted by GitBox <gi...@apache.org>.
schattian commented on PR #23658:
URL: https://github.com/apache/airflow/pull/23658#issuecomment-1125475091
@potiuk I had one thing in mind to try to reduce the complexity of this, releasing the connection (response.release_conn). Based on https://github.dev/urllib3/urllib3/blob/37a67739ba9a7835886555744e9dabe3009e538a/src/urllib3/response.py#L708-L712 seems could work.
Much harder to test though as rn is just a simple generator.
--
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] schattian commented on a diff in pull request #23658: Fix k8s pod.execute randomly stuck indefinitely by logs consumption (#23497)
Posted by GitBox <gi...@apache.org>.
schattian commented on code in PR #23658:
URL: https://github.com/apache/airflow/pull/23658#discussion_r875764928
##########
kubernetes_tests/test_kubernetes_pod_operator.py:
##########
@@ -376,7 +377,9 @@ def test_port(self):
assert self.expected_pod == actual_pod
def test_volume_mount(self):
- with mock.patch.object(PodManager, 'log') as mock_logger:
+ with mock.patch.object(
Review Comment:
notice a non-process safe counter makes this test flaky or impossible to do as the call of log.info happens in a process.
--
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] schattian commented on a diff in pull request #23658: Fix k8s pod.execute randomly stuck indefinitely by logs consumption (#23497)
Posted by GitBox <gi...@apache.org>.
schattian commented on code in PR #23658:
URL: https://github.com/apache/airflow/pull/23658#discussion_r875764928
##########
kubernetes_tests/test_kubernetes_pod_operator.py:
##########
@@ -376,7 +377,9 @@ def test_port(self):
assert self.expected_pod == actual_pod
def test_volume_mount(self):
- with mock.patch.object(PodManager, 'log') as mock_logger:
+ with mock.patch.object(
Review Comment:
notice a non-thread-safe counter (default) makes this test flaky or impossible to do as the call of log.info happens in a process.
--
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