Fixed hanged KubernetesPodOperator

[moiseenkov]

Fixes: #23497

• Inserted a container status check before pulling longs from a stream, because KubernetesPodOperator hangs when attempting to read logs of the terminated container. With the current fix we make sure that the container is alive or was terminated no longer than the specified timeout before reading logs.
• Updated unit tests

[Taragolis]

nit: datetime

[Taragolis]

Look nice for me. However better that also someone else look on this changes

[potiuk]

LGTM too. @dstandish @jedcunningham - maybe you can take a look too?

[moiseenkov]

@dimberman , @dstandish , @jedcunningham, hi.
Could you please review my fixes?

[potiuk]

Needs re-review @dimberman @dstandish I guess.

stale

[dstandish]

Hi, I dismissed my old review, so it's not blocking.

I do have a suggestion though I'm sorry if it's a bit late in the game. And maybe it doesn't have to be done in this PR.

But so the thing that stuck out to me when looking at this is, we do a kube api call (in logs_available) every chunk in the log stream. This seems like it could result in a lot of calls and depending on how many such processes on the cluster could cause problems. Just a hunch I guess. But so it would seem to me that to avoid this, perhaps you could run the logs_available check in a thread, just have it run periodically, like once every 30 seconds or something, and then when it returns false, just set a stop boolean on the consumer so that it knows to exit the loop. This decouples the checking from the log stream so that you that checking does not increase in response to log volume.

[dstandish]

OK I just experimented with our own "event scheduler" helper and it seems that we could use it to limit calls without managing threads. Here's a code sample:

import time

from airflow.utils.event_scheduler import EventScheduler

class Tracker:
    stop = False
    counter = 0

def hello(tracker: Tracker):
    tracker.counter += 1
    print("hi! %s" % tracker.counter)
    if tracker.counter > 10:
        tracker.stop = True

e = EventScheduler()
tracker = Tracker()
e.call_regular_interval(2, hello, (tracker,))

while True:
    e.run(blocking=False)
    time.sleep(0.5)
    if tracker.stop is True:
        break

I believe what this does is, keep track of how long it's been since hello has been called. and then in every loop, it checks whether it needs to run, and runs it if it does. So you could replace hello with the call to check if pod still running. And it won't make those network calls more than it has to.

[dimberman]

LGTM Let's merge when tests pass

[moiseenkov]

@dstandish , thank you for proposing a nicer approach.
You mentioned earlier that we do k8s API calls every chunk. It is true for earlier versions of this PR, but currently the API response is being cached for 120 seconds. Please, take a look at the method PodLogsConsumer.read_pod() - it invokes PodManager.read_pod() only if the cache is empty or last call was more than self.read_pod_cache_timeout seconds ago (by default 120s). Thus there's no such problem now, however the EventScheduler approach would look nicer, and the code can always be refactored if needed.

[dstandish]

Ah ok thanks 👍