[AIRFLOW-3516] Support to create k8 worker pods in batches

[ramandumcs]

Make sure you have checked all steps below.

Jira

• My PR addresses the following Airflow Jira issues and references them in the PR title. For example, "[AIRFLOW-XXX] My Airflow PR"
  • https://issues.apache.org/jira/browse/AIRFLOW-XXX
  • In case you are fixing a typo in the documentation you can prepend your commit with [AIRFLOW-XXX], code changes always need a Jira issue.

Description

• Here are some details about my PR, including screenshots of any UI changes:

Tests

• My PR adds the following unit tests OR does not need testing for this extremely good reason:

Commits

• My commits all reference Jira issues in their subject lines, and I have squashed multiple commits if they address the same issue. In addition, my commits follow the guidelines from "How to write a good git commit message":
  1. Subject is separated from body by a blank line
  2. Subject is limited to 50 characters (not including Jira issue reference)
  3. Subject does not end with a period
  4. Subject uses the imperative mood ("add", not "adding")
  5. Body wraps at 72 characters
  6. Body explains "what" and "why", not "how"

Documentation

• In case of new functionality, my PR adds documentation that describes how to use it.
  • When adding new operators/hooks/sensors, the autoclass documentation generation needs to be added.
  • All the public functions and the classes in the PR contain docstrings that explain what it does

Code Quality

• Passes flake8

[codecov-io]

Codecov Report

Merging #4434 into master will increase coverage by 3.82%.
The diff coverage is n/a.

@@            Coverage Diff             @@
##           master    #4434      +/-   ##
==========================================
+ Coverage   74.76%   78.58%   +3.82%     
==========================================
  Files         429      204     -225     
  Lines       29649    16482   -13167     
==========================================
- Hits        22167    12953    -9214     
+ Misses       7482     3529    -3953

Impacted Files	Coverage Δ
airflow/operators/hive_stats_operator.py	0% <0%> (-100%)	⬇️
airflow/operators/presto_to_mysql.py	0% <0%> (-100%)	⬇️
airflow/www/api/experimental/endpoints.py	71.12% <0%> (-19.02%)	⬇️
airflow/executors/__init__.py	63.46% <0%> (-17.31%)	⬇️
airflow/api/common/experimental/delete_dag.py	84% <0%> (-4%)	⬇️
airflow/utils/sqlalchemy.py	78.57% <0%> (-3.25%)	⬇️
airflow/utils/db.py	32.28% <0%> (-2.2%)	⬇️
airflow/www_rbac/views.py	73.5% <0%> (-1.86%)	⬇️
airflow/plugins_manager.py	92.13% <0%> (-1.2%)	⬇️
airflow/www/views.py	69.82% <0%> (-0.36%)	⬇️
... and 245 more

---

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update dba13d9...167e035. Read the comment docs.

[OmerJog]

This isn't related to this PR.
Please open a new PR for this.

Also, once removing it please squash your commits

[ramandumcs]

Thanks @OmerJog. Updated the PR

[feng-tao]

PTAL @dimberman

[dimberman]

@feng-tao apologies for the delay, getting over a cold. Will check out now.

[dimberman]

@ramandumcs What is the use-case where a user would want to launch in batches rather than greedily launch tasks as they're received?

[ramandumcs]

Thanks @dimberman for looking in to this PR.

As per current implementation K8 Executor submits/creates one k8 worker pod per scheduler loop.
Scheduler creates a single k8 worker pod inside self.executor.heartbeat() function of jobs.py file. In self.executor.heartbeat() function it calls self.sync() which submits/creates single k8 worker pod.
task_queue might have 100s of tasks to be run but only one task gets submitted per scheduler loop which increases the task scheduling latency/delay.

Each scheduling loop takes a minimum of 1 second so scheduling latency of last task in the task queue with 1000 tasks will be atleast 1000 seconds.
(We have also observed scheuling loop sometimes taking ~ 2 to 3 seconds which further increases the task latency)

We have a use case to run 1000s of concurrent tasks and we started using Airflow with K8 executor where we observed and investigated this behaviour.
So we are proposing a fix to have some control over the number of worker pods that get submitted per loop. Ideal scenarion might haven been to submit all the tasks per loop but it can impact the scheduler's processing of DAG files as these are synchronous call.
So we made this configurable.
Please let us know if it makes sense and please also share your thoughts/comments

[dimberman]

@ramandumcs ohh ok that makes total sense. This LGTM. @kaxil is it too late to put this in 1.10.2?

[dimberman]

@feng-tao good to go on my end!

[feng-tao]

thanks @dimberman , @ramandumcs , couples of qqs for my understanding:

1. do we need to add the config in default_test.cfg?
2. is there a way to test the behavior?
3. My understanding for the change is that previously we launch one pod per single heartbeat call, now we could launch multiple pods(not with multithreading) inside single heartbeat call?

[ramandumcs]

Yup, Now we are launching multiple pods per heartbeat call. I will see if we can add test cases for this particular behavior