Fix labels used to find queued KubeExecutor pods

[jedcunningham]

We need to use the job_id used to queue the TI, not the current
schedulers job_id. These can differ naturally with HA schedulers and
with scheduler restarts (clearing "queued but not launched TIs" happens
before adoption).

[dstandish]

what if the scenario is this

pod is created, scheduler dies

scheduler back online

scheduler starts up the task again... i assume maybe it gets a new queued_by_job_id -- but the pod has the old id?

does the queued by id even matter? should we not just be looking at things that uniquely identify the TI?

[jedcunningham]

Note: It's using the queued_by_job_id from the TI, which will match the label. Eventually the pod will be adopted by another (or the new) scheduler, however, this can run before that happens (it's on an interval after all) AND this does run before adoption when a scheduler starts.

Now, does it even matter? For better or worse, we use the job_id to help determine which Airflow the task is part of. For example, we watch for events based on the job id as well:

airflow/airflow/executors/kubernetes_executor.py

Line 129 in 33a4502

kwargs = {'label_selector': f'airflow-worker={scheduler_job_id}'}

This becomes important if we consider a shared namespace with multiple Airflow worker pods in it. It becomes even more important if we have the same dags/tasks/scheduled runs. There are certainly still issues here, but this is at least status quo for now until we can properly fix everything.

[jedcunningham]

Here is another example in the adoption process:

airflow/airflow/executors/kubernetes_executor.py

Line 682 in 33a4502

kwargs = {'label_selector': f'airflow-worker={scheduler_job_id}'}

[dstandish]

Thanks

Note: It's using the queued_by_job_id from the TI, which will match the label. Eventually the pod will be adopted by another (or the new) scheduler, however, this can run before that happens (it's on an interval after all) AND this does run before adoption when a scheduler starts.

OK so you are saying that this process (i.e. read TI from db, and use it to build the labels, then search for the pod), generally speaking, will happen before the task would e.g. be failed and retried (at which time presumably it would get a new queued by job id, after which if the pod was still out there it would no longer be found in this way). If I have you right, then that makes sense and looks good.

This becomes important if we consider a shared namespace with multiple Airflow worker pods in it. It becomes even more important if we have the same dags/tasks/scheduled runs

Makes sense. But that would not sound like a good idea!

Separate note... airflow-worker is a misnomer right? That makes it sound like celery worker... though i have also seen it used to refer to k8s exec task pods.... and it's not that either....

But yeah i mean now that you mention it... scheduler job ids are probably just incrementing integers no? In that case it would not be the most rock solid of selectors in the shared namespace scenario. Though i'm sure would generally be very rare, maybe there should be a random cluster identifier created somewhere in the db with airflow db init that could be used as a more direct way of keeping things separate

[jedcunningham]

Yeah, that is exactly my plan eventually, generate a random identifier for the instance. Unfortunately I don't think we have anything we can use right now for that purpose.

[dstandish]

i tried assert_has_calls where you can specify the exact call list (which would be ever so marginally more direct than the approach here but apparently it does not work with MagicMock only Mock

[github-actions]

The PR most likely needs to run full matrix of tests because it modifies parts of the core of Airflow. However, committers might decide to merge it quickly and take the risk. If they don't merge it quickly - please rebase it to the latest main at your convenience, or amend the last commit of the PR, and push it with --force-with-lease.